fix: 修正归属工时舍入残差

app/services/report_allocations.py

@@ -240,6 +240,58 @@ def _balance_minutes_to_targets(
             _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:
@@ -367,6 +419,11 @@ def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | 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:

tests/test_report_allocations.py

@@ -154,6 +154,27 @@ def test_no_device_shared_full_report_splits_effective_minutes_between_items():
     ]
 
 
+def test_short_shared_report_rounding_residual_keeps_aggregate_night_total():
+    start_at = datetime(2026, 5, 3, 1, 4, 50)
+    end_at = datetime(2026, 5, 3, 1, 4, 53)
+    first_item = _item(id=1, good_qty=1)
+    second_item = _item(id=2, good_qty=1)
+    report = _report(start_at, end_at, [first_item, second_item])
+
+    metrics = calculate_report_metrics(start_at, end_at, [first_item, second_item])
+    rows = build_report_allocation_drafts(report)
+
+    assert metrics["effective_minutes"] == 0.05
+    assert metrics["shift_night_minutes"] == 0.05
+    assert [row.allocation_date for row in rows] == [date(2026, 5, 2), date(2026, 5, 2)]
+    assert [row.effective_minutes for row in rows] == [0.03, 0.02]
+    assert round(sum(row.effective_minutes for row in rows), 2) == 0.05
+    assert round(sum(row.night_minutes for row in rows), 2) == 0.05
+    assert round(sum(row.day_minutes for row in rows), 2) == 0
+    assert round(sum(row.overtime_minutes for row in rows), 2) == 0
+    assert all(row.effective_minutes == round(row.day_minutes + row.overtime_minutes + row.night_minutes, 2) for row in rows)
+
+
 def test_items_sharing_same_started_at_split_the_segment_minutes():
     started_at = datetime(2026, 7, 25, 8, 0)
     first_item = _item(id=1, good_qty=10, started_at=started_at)