from collections import defaultdict from dataclasses import dataclass, replace from datetime import date, datetime, timedelta from sqlalchemy import delete, select from sqlalchemy.orm import Session from app.models import ProductionReport, ProductionReportAllocation from app.services.common import as_float, round2 from app.services.metrics import ( SHIFT_BUCKETS, SHIFT_LABELS, _blank_overtime_intervals, _device_item_time_ranges, _device_segment_ranges, _iter_shift_base_dates, _mold_key, _overlap_minutes, _schedule_interval, calculate_report_metrics, format_hours, item_time_ranges, minutes_between, ) from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG, WorkScheduleConfig @dataclass(frozen=True) class ReportAllocationDraft: report_id: int | None report_item_id: int | None attendance_point_name: str employee_phone: str allocation_date: date day_minutes: float = 0 overtime_minutes: float = 0 night_minutes: float = 0 effective_minutes: float = 0 good_qty: float = 0 defect_qty: float = 0 scrap_qty: float = 0 changeover_count: float = 0 reference_wage: float = 0 def _add_minutes( rows: dict[date, dict[str, float]], allocation_date: date, bucket: str, minutes: float, ) -> None: if minutes <= 0: return rows[allocation_date][bucket] += minutes def _range_allocation_minutes( start_at: datetime, end_at: datetime, schedule: WorkScheduleConfig | None = None, ) -> dict[date, dict[str, float]]: active_schedule = schedule or DEFAULT_WORK_SCHEDULE_CONFIG rows: dict[date, dict[str, float]] = defaultdict(lambda: {key: 0.0 for key in SHIFT_BUCKETS}) if end_at <= start_at: return rows for current_date in _iter_shift_base_dates(start_at, end_at): meal_intervals = [ _schedule_interval(current_date, active_schedule.lunch_start, active_schedule.lunch_end), _schedule_interval(current_date, active_schedule.dinner_start, active_schedule.dinner_end), ] shift_intervals = { "day": [_schedule_interval(current_date, active_schedule.day_start, active_schedule.day_end)], "overtime": [ _schedule_interval(current_date, active_schedule.overtime_start, active_schedule.overtime_end) ], "night": [_schedule_interval(current_date, active_schedule.night_start, active_schedule.night_end)], } for bucket, intervals in shift_intervals.items(): for interval_start, interval_end in intervals: overlap_start = max(start_at, interval_start) overlap_end = min(end_at, interval_end) if overlap_end <= overlap_start: continue meal_minutes = sum( _overlap_minutes(overlap_start, overlap_end, meal_start, meal_end) for meal_start, meal_end in meal_intervals if ( start_at <= meal_start and end_at >= meal_end and end_at > meal_end + timedelta(minutes=5) ) ) minutes = max(0.0, minutes_between(overlap_start, overlap_end) - meal_minutes) if bucket == "night": _add_minutes(rows, current_date, "night", minutes) else: _add_minutes(rows, overlap_start.date(), bucket, minutes) for blank_start, blank_end in _blank_overtime_intervals(current_date, active_schedule): overlap_start = max(start_at, blank_start) overlap_end = min(end_at, blank_end) if overlap_end <= overlap_start: continue meal_minutes = sum( _overlap_minutes(overlap_start, overlap_end, meal_start, meal_end) for meal_start, meal_end in meal_intervals if ( start_at <= meal_start and end_at >= meal_end and end_at > meal_end + timedelta(minutes=5) ) ) minutes = max(0.0, minutes_between(overlap_start, overlap_end) - meal_minutes) _add_minutes(rows, overlap_start.date(), "overtime", minutes) return rows def _report_devices(report) -> list | None: session = getattr(report, "session", None) devices = getattr(session, "devices", None) if devices is not None: return list(devices) devices = getattr(report, "devices", None) return list(devices) if devices is not None else None def _item_reference_wage(item) -> float: return as_float(getattr(item, "good_qty", 0)) * as_float(getattr(item, "process_unit_price_yuan", 0)) def _allocation_targets(report, items: list, schedule: WorkScheduleConfig | None = None): metrics = calculate_report_metrics( getattr(report, "start_at"), getattr(report, "end_at"), items, break_minutes=as_float(getattr(report, "break_minutes", 0)), devices=_report_devices(report), schedule=schedule, ) return ( {id(item): as_float(getattr(item, "allocated_minutes", 0)) for item in items}, { "day": as_float(metrics.get("shift_day_minutes", 0)), "overtime": as_float(metrics.get("shift_overtime_minutes", 0)), "night": as_float(metrics.get("shift_night_minutes", 0)), }, ) def _minutes_total(rows: dict[date, dict[str, float]]) -> float: return sum(sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS) for minutes in rows.values()) def _empty_shift_minutes() -> dict[str, float]: return {key: 0.0 for key in SHIFT_BUCKETS} def _add_date_bucket_minutes( target: dict[date, dict[str, float]], source: dict[date, dict[str, float]], *, scale: float = 1.0, ) -> None: for allocation_date, minutes in source.items(): row = target.setdefault(allocation_date, _empty_shift_minutes()) for key in SHIFT_BUCKETS: row[key] += minutes.get(key, 0.0) * scale def _bucket_totals(rows_by_date: dict[date, dict[str, float]]) -> dict[str, float]: return { key: sum(minutes.get(key, 0.0) for minutes in rows_by_date.values()) for key in SHIFT_BUCKETS } def _item_bucket_totals(minutes_by_item: dict[int, dict[date, dict[str, float]]]) -> dict[str, float]: totals = _empty_shift_minutes() for rows_by_date in minutes_by_item.values(): for key, value in _bucket_totals(rows_by_date).items(): totals[key] += value return totals def _allocate_unassigned_minutes_to_existing_items( minutes_by_item: dict[int, dict[date, dict[str, float]]], unassigned_minutes: dict[date, dict[str, float]], ) -> None: unassigned_total = _minutes_total(unassigned_minutes) if unassigned_total <= 0 or not minutes_by_item: return raw_item_totals = { item_id: _minutes_total(rows_by_date) for item_id, rows_by_date in minutes_by_item.items() } raw_total = sum(raw_item_totals.values()) item_count = len(minutes_by_item) for item_id, rows_by_date in minutes_by_item.items(): weight = raw_item_totals[item_id] / raw_total if raw_total > 0 else 1 / item_count _add_date_bucket_minutes(rows_by_date, unassigned_minutes, scale=weight) def _scale_item_minutes_to_bucket_targets( minutes_by_item: dict[int, dict[date, dict[str, float]]], target_buckets: dict[str, float], ) -> None: raw_buckets = _item_bucket_totals(minutes_by_item) for key in SHIFT_BUCKETS: raw_value = raw_buckets.get(key, 0.0) target_value = target_buckets.get(key, 0.0) if raw_value <= 0 or abs(raw_value - target_value) <= 0.000001: continue scale = target_value / raw_value for rows_by_date in minutes_by_item.values(): for minutes in rows_by_date.values(): minutes[key] = minutes.get(key, 0.0) * scale def _scale_bucket_minutes( minutes_by_item: dict[int, dict[date, dict[str, float]]], bucket: str, scale: float, ) -> None: for rows_by_date in minutes_by_item.values(): for minutes in rows_by_date.values(): minutes[bucket] = minutes.get(bucket, 0.0) * scale def _balance_minutes_to_targets( minutes_by_item: dict[int, dict[date, dict[str, float]]], target_minutes_by_item: dict[int, float], target_buckets: dict[str, float], ) -> None: for _ in range(100): for item_id, rows_by_date in minutes_by_item.items(): raw_total = _minutes_total(rows_by_date) target_total = target_minutes_by_item.get(item_id, raw_total) if raw_total <= 0: continue scale = target_total / raw_total if abs(scale - 1) <= 0.000000001: continue for minutes in rows_by_date.values(): for key in SHIFT_BUCKETS: minutes[key] = minutes.get(key, 0.0) * scale raw_buckets = _item_bucket_totals(minutes_by_item) for key in SHIFT_BUCKETS: raw_value = raw_buckets.get(key, 0.0) target_value = target_buckets.get(key, 0.0) if raw_value <= 0: continue scale = target_value / raw_value if abs(scale - 1) <= 0.000000001: continue _scale_bucket_minutes(minutes_by_item, key, scale) def _round_bucket_cells_to_targets( minutes_by_item: dict[int, dict[date, dict[str, float]]], target_buckets: dict[str, float], item_order: dict[int, int], ) -> None: for bucket in SHIFT_BUCKETS: cells: list[tuple[float, int, int, date]] = [] for item_id, rows_by_date in minutes_by_item.items(): for allocation_date, minutes in rows_by_date.items(): value = minutes.get(bucket, 0.0) if value > 0: cents_value = max(0.0, value * 100) floor_cents = int(cents_value + 0.000000001) cells.append((cents_value - floor_cents, floor_cents, item_order.get(item_id, 0), allocation_date)) minutes[bucket] = floor_cents / 100 else: minutes[bucket] = 0.0 target_cents = int(round(round2(target_buckets.get(bucket, 0.0)) * 100)) floor_total = sum(cell[1] for cell in cells) residual = target_cents - floor_total if not cells or residual == 0: continue ordered = sorted( cells, key=lambda cell: (-cell[0], cell[2], cell[3].isoformat()), ) if residual < 0: ordered = sorted( cells, key=lambda cell: (cell[0], cell[2], cell[3].isoformat()), ) cents_by_cell: dict[tuple[int, date], int] = { (cell[2], cell[3]): cell[1] for cell in cells } steps = abs(residual) for index in range(steps): _, _, item_index, allocation_date = ordered[index % len(ordered)] key = (item_index, allocation_date) if residual > 0: cents_by_cell[key] += 1 elif cents_by_cell[key] > 0: cents_by_cell[key] -= 1 item_id_by_order = {order: item_id for item_id, order in item_order.items()} for (item_index, allocation_date), cents in cents_by_cell.items(): item_id = item_id_by_order[item_index] minutes_by_item[item_id][allocation_date][bucket] = cents / 100 def _items_by_device(items: list) -> dict[str, list]: grouped: dict[str, list] = defaultdict(list) for item in items: mold_key = _mold_key( getattr(item, "product_name", "") or item.device_no, getattr(item, "process_name", ""), ) grouped[mold_key].append(item) return grouped def _device_minutes_by_item(report, items: list, schedule: WorkScheduleConfig | None = None): start_at = getattr(report, "start_at") end_at = getattr(report, "end_at") segment_ranges = _device_segment_ranges(start_at, end_at, _report_devices(report)) if not segment_ranges: return None items_by_device = _items_by_device(items) minutes_by_item: dict[int, dict[date, dict[str, float]]] = {} target_minutes: dict[date, dict[str, float]] = defaultdict(_empty_shift_minutes) unassigned_minutes: dict[date, dict[str, float]] = defaultdict(_empty_shift_minutes) allocated_item_ids: set[int] = set() for device_no, ranges in segment_ranges.items(): device_items = items_by_device.get(device_no, []) for segment_start, segment_end in ranges: segment_minutes = _range_allocation_minutes(segment_start, segment_end, schedule) _add_date_bucket_minutes(target_minutes, segment_minutes) if not device_items: _add_date_bucket_minutes(unassigned_minutes, segment_minutes) continue for item_range in _device_item_time_ranges(device_items, segment_start, segment_end): item_minutes = minutes_by_item.setdefault(id(item_range.item), defaultdict(_empty_shift_minutes)) range_minutes = _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule) _add_date_bucket_minutes( item_minutes, range_minutes, scale=as_float(getattr(item_range, "allocation_weight", 1.0)), ) for item in device_items: allocated_item_ids.add(id(item)) unmatched_items = [item for item in items if id(item) not in allocated_item_ids] if unmatched_items and _minutes_total(unassigned_minutes) > 0: target_buckets = _bucket_totals(target_minutes) matched_buckets = _item_bucket_totals(minutes_by_item) unassigned_buckets = _bucket_totals(unassigned_minutes) for unmatched_item in unmatched_items: item_minutes = minutes_by_item.setdefault(id(unmatched_item), defaultdict(_empty_shift_minutes)) for allocation_date, minutes in unassigned_minutes.items(): row = item_minutes.setdefault(allocation_date, _empty_shift_minutes()) for key in SHIFT_BUCKETS: remaining = max(0.0, target_buckets.get(key, 0.0) - matched_buckets.get(key, 0.0)) raw_unassigned = unassigned_buckets.get(key, 0.0) if remaining <= 0 or raw_unassigned <= 0: continue row[key] += minutes.get(key, 0.0) * (remaining / raw_unassigned) / len(unmatched_items) return minutes_by_item _allocate_unassigned_minutes_to_existing_items(minutes_by_item, unassigned_minutes) _scale_item_minutes_to_bucket_targets(minutes_by_item, _bucket_totals(target_minutes)) return minutes_by_item def _range_minutes_by_item(report, items: list, schedule: WorkScheduleConfig | None = None): ranges = item_time_ranges(getattr(report, "start_at"), getattr(report, "end_at"), items, _report_devices(report)) minutes_by_item: dict[int, dict[date, dict[str, float]]] = {} for item_range in ranges: item_key = id(item_range.item) target = minutes_by_item.setdefault(item_key, defaultdict(_empty_shift_minutes)) for allocation_date, minutes in _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule).items(): row = target.setdefault(allocation_date, _empty_shift_minutes()) for key in SHIFT_BUCKETS: row[key] += minutes.get(key, 0.0) * as_float( getattr(item_range, "allocation_weight", 1.0) ) return minutes_by_item def _fallback_row(report, item) -> ReportAllocationDraft: return ReportAllocationDraft( report_id=getattr(report, "id", None), report_item_id=getattr(item, "id", None), attendance_point_name=str(getattr(report, "attendance_point_name", "") or ""), employee_phone=str(getattr(report, "employee_phone", "") or ""), allocation_date=getattr(report, "report_date"), good_qty=round2(getattr(item, "good_qty", 0)), defect_qty=round2(getattr(item, "defect_qty", 0)), scrap_qty=round2(getattr(item, "scrap_qty", 0)), changeover_count=round2(getattr(item, "changeover_count", 0)), reference_wage=round2(_item_reference_wage(item)), ) def _draft_from_minutes(report, item, allocation_date: date, minutes: dict[str, float], ratio: float) -> ReportAllocationDraft: reference_wage = _item_reference_wage(item) day_minutes = round2(minutes.get("day", 0)) overtime_minutes = round2(minutes.get("overtime", 0)) night_minutes = round2(minutes.get("night", 0)) return ReportAllocationDraft( report_id=getattr(report, "id", None), report_item_id=getattr(item, "id", None), attendance_point_name=str(getattr(report, "attendance_point_name", "") or ""), employee_phone=str(getattr(report, "employee_phone", "") or ""), allocation_date=allocation_date, day_minutes=day_minutes, overtime_minutes=overtime_minutes, night_minutes=night_minutes, effective_minutes=round2(day_minutes + overtime_minutes + night_minutes), good_qty=round2(as_float(getattr(item, "good_qty", 0)) * ratio), defect_qty=round2(as_float(getattr(item, "defect_qty", 0)) * ratio), scrap_qty=round2(as_float(getattr(item, "scrap_qty", 0)) * ratio), changeover_count=round2(as_float(getattr(item, "changeover_count", 0)) * ratio), reference_wage=round2(reference_wage * ratio), ) SPLIT_RESIDUAL_FIELDS = ( "good_qty", "defect_qty", "scrap_qty", "changeover_count", "reference_wage", ) def _item_split_targets(item) -> dict[str, float]: return { "good_qty": round2(getattr(item, "good_qty", 0)), "defect_qty": round2(getattr(item, "defect_qty", 0)), "scrap_qty": round2(getattr(item, "scrap_qty", 0)), "changeover_count": round2(getattr(item, "changeover_count", 0)), "reference_wage": round2(_item_reference_wage(item)), } def _balance_item_split_residuals( drafts: list[ReportAllocationDraft], targets: dict[str, float], ) -> list[ReportAllocationDraft]: if not drafts: return drafts mutable_values = [ {field: int(round(as_float(getattr(draft, field)) * 100)) for field in SPLIT_RESIDUAL_FIELDS} for draft in drafts ] for field in SPLIT_RESIDUAL_FIELDS: target_cents = int(round(round2(targets[field]) * 100)) current_cents = sum(values[field] for values in mutable_values) residual = target_cents - current_cents if residual == 0: continue if residual > 0: ordered_indexes = sorted( range(len(drafts)), key=lambda index: ( -as_float(drafts[index].effective_minutes), drafts[index].allocation_date.isoformat(), index, ), ) else: ordered_indexes = sorted( range(len(drafts)), key=lambda index: ( -mutable_values[index][field], -as_float(drafts[index].effective_minutes), drafts[index].allocation_date.isoformat(), index, ), ) step = 1 if residual > 0 else -1 remaining = abs(residual) while remaining > 0: changed = False for index in ordered_indexes: if step < 0 and mutable_values[index][field] <= 0: continue mutable_values[index][field] += step remaining -= 1 changed = True if remaining == 0: break if not changed: break return [ replace( draft, **{ field: mutable_values[index][field] / 100 for field in SPLIT_RESIDUAL_FIELDS }, ) for index, draft in enumerate(drafts) ] def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None = None) -> list[ReportAllocationDraft]: items = list(getattr(report, "items", []) or []) if not items: return [] minutes_by_item = _device_minutes_by_item(report, items, schedule) if minutes_by_item is None: minutes_by_item = _range_minutes_by_item(report, items, schedule) item_targets, bucket_targets = _allocation_targets(report, items, schedule) _balance_minutes_to_targets(minutes_by_item, item_targets, bucket_targets) _round_bucket_cells_to_targets( minutes_by_item, bucket_targets, {id(item): index for index, item in enumerate(items)}, ) rows: list[ReportAllocationDraft] = [] for item in items: item_minutes = minutes_by_item.get(id(item), {}) total_effective = sum(sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS) for minutes in item_minutes.values()) if total_effective <= 0: rows.append(_fallback_row(report, item)) continue item_rows: list[ReportAllocationDraft] = [] for allocation_date in sorted(item_minutes): minutes = item_minutes[allocation_date] effective = sum(minutes.get(key, 0.0) for key in SHIFT_BUCKETS) if effective <= 0: continue item_rows.append(_draft_from_minutes(report, item, allocation_date, minutes, effective / total_effective)) rows.extend(_balance_item_split_residuals(item_rows, _item_split_targets(item))) return rows def allocation_summary_text(rows) -> str: grouped: dict[date, dict[str, float]] = defaultdict(lambda: {key: 0.0 for key in SHIFT_BUCKETS}) for row in rows: allocation_date = getattr(row, "allocation_date") grouped[allocation_date]["day"] += as_float(getattr(row, "day_minutes", 0)) grouped[allocation_date]["overtime"] += as_float(getattr(row, "overtime_minutes", 0)) grouped[allocation_date]["night"] += as_float(getattr(row, "night_minutes", 0)) parts: list[str] = [] for allocation_date in sorted(grouped): shift_text = "、".join( f"{SHIFT_LABELS[key]}{format_hours(grouped[allocation_date].get(key, 0))}小时" for key in SHIFT_BUCKETS if round2(grouped[allocation_date].get(key, 0)) > 0 ) if shift_text: parts.append(f"{allocation_date.isoformat()} {shift_text}") return "；".join(parts) def refresh_report_allocations( db: Session, report, schedule: WorkScheduleConfig | None = None, *, commit: bool = False, ) -> list[ProductionReportAllocation]: """Rebuild allocation rows with metrics semantics, updating item allocation-derived fields.""" report_id = getattr(report, "id") if isinstance(report_id, int): db.execute( select(ProductionReport.id) .where(ProductionReport.id == report_id) .with_for_update() ).scalar_one_or_none() db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report_id)) rows = [ ProductionReportAllocation( report_id=draft.report_id, report_item_id=draft.report_item_id, attendance_point_name=draft.attendance_point_name, employee_phone=draft.employee_phone, allocation_date=draft.allocation_date, day_minutes=draft.day_minutes, overtime_minutes=draft.overtime_minutes, night_minutes=draft.night_minutes, effective_minutes=draft.effective_minutes, good_qty=draft.good_qty, defect_qty=draft.defect_qty, scrap_qty=draft.scrap_qty, changeover_count=draft.changeover_count, reference_wage=draft.reference_wage, ) for draft in build_report_allocation_drafts(report, schedule) ] db.add_all(rows) db.flush() if commit: db.commit() return rows