fix: 平衡归属工时班次汇总

app/services/report_allocations.py

@@ -130,8 +130,8 @@ def _item_reference_wage(item) -> float:
     return as_float(getattr(item, "good_qty", 0)) * as_float(getattr(item, "process_unit_price_yuan", 0))
 
 
-def _item_target_minutes(report, items: list, schedule: WorkScheduleConfig | None = None) -> dict[int, float]:
+def _allocation_targets(report, items: list, schedule: WorkScheduleConfig | None = None):
-    calculate_report_metrics(
+    metrics = calculate_report_metrics(
         getattr(report, "start_at"),
         getattr(report, "end_at"),
         items,
@@ -139,7 +139,14 @@ def _item_target_minutes(report, items: list, schedule: WorkScheduleConfig | Non
         devices=_report_devices(report),
         schedule=schedule,
     )
-    return {id(item): as_float(getattr(item, "allocated_minutes", 0)) for item in items}
+    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:
@@ -193,20 +200,45 @@ def _scale_item_minutes_to_bucket_targets(
                 minutes[key] = minutes.get(key, 0.0) * scale
 
 
-def _scale_item_minutes_to_item_targets(
+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 or abs(raw_total - target_total) <= 0.000001:
+            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 _items_by_device(items: list) -> dict[str, list]:
     grouped: dict[str, list] = defaultdict(list)
@@ -333,7 +365,8 @@ def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None =
     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)
-    _scale_item_minutes_to_item_targets(minutes_by_item, _item_target_minutes(report, items, schedule))
+    item_targets, bucket_targets = _allocation_targets(report, items, schedule)
+    _balance_minutes_to_targets(minutes_by_item, item_targets, bucket_targets)
 
     rows: list[ReportAllocationDraft] = []
     for item in items:

tests/test_report_allocations.py

@@ -329,6 +329,15 @@ def test_item_totals_anchor_to_metrics_when_unmatched_remaining_buckets_exceed_i
         2: 360,
         3: 60,
     }
+    assert {
+        "day": round(sum(row.day_minutes for row in rows), 2),
+        "overtime": round(sum(row.overtime_minutes for row in rows), 2),
+        "night": round(sum(row.night_minutes for row in rows), 2),
+    } == {
+        "day": metrics["shift_day_minutes"],
+        "overtime": metrics["shift_overtime_minutes"],
+        "night": metrics["shift_night_minutes"],
+    }
     assert round(sum(row.effective_minutes for row in rows), 2) == 660
     assert all(row.effective_minutes == row.day_minutes + row.overtime_minutes + row.night_minutes for row in rows)