fix: 修正共享报工段工时归属拆分

app/services/metrics.py

@@ -35,6 +35,7 @@ class ItemTimeRange:
     item: ReportItemLike
     start_at: datetime
     end_at: datetime
+    allocation_weight: float = 1.0
 
 
 SHIFT_BUCKETS = ("day", "overtime", "night")
@@ -419,8 +420,17 @@ def _device_item_time_ranges(
         item_end = segment_end
         if index + 1 < len(ordered_starts):
             item_end = ordered_starts[index + 1]
-        for item in grouped_items[item_start]:
-            ranges.append(ItemTimeRange(item=item, start_at=item_start, end_at=item_end))
+        items_at_start = grouped_items[item_start]
+        allocation_weight = 1 / len(items_at_start)
+        for item in items_at_start:
+            ranges.append(
+                ItemTimeRange(
+                    item=item,
+                    start_at=item_start,
+                    end_at=item_end,
+                    allocation_weight=allocation_weight,
+                )
+            )
     return ranges
 
 
@@ -435,7 +445,11 @@ def item_time_ranges(
 
     segment_ranges = _device_segment_ranges(start_at, end_at, devices)
     if not segment_ranges:
-        return [ItemTimeRange(item=item, start_at=start_at, end_at=end_at) for item in items]
+        allocation_weight = 1 / len(items)
+        return [
+            ItemTimeRange(item=item, start_at=start_at, end_at=end_at, allocation_weight=allocation_weight)
+            for item in items
+        ]
 
     items_by_device: dict[str, list[ReportItemLike]] = defaultdict(list)
     for item in items:

app/services/report_allocations.py

@@ -188,7 +188,9 @@ def build_report_allocation_drafts(report, schedule: WorkScheduleConfig | None =
         target = minutes_by_item.setdefault(item_key, defaultdict(lambda: {key: 0.0 for key in SHIFT_BUCKETS}))
         for allocation_date, minutes in _range_allocation_minutes(item_range.start_at, item_range.end_at, schedule).items():
             for key in SHIFT_BUCKETS:
-                target[allocation_date][key] += minutes.get(key, 0.0)
+                target[allocation_date][key] += minutes.get(key, 0.0) * as_float(
+                    getattr(item_range, "allocation_weight", 1.0)
+                )
 
     rows: list[ReportAllocationDraft] = []
     for item in items:

tests/test_report_allocations.py

@@ -47,7 +47,26 @@ def _item(
     )
 
 
-def _report(start_at: datetime, end_at: datetime, item, *, id: int = 1, report_date: date | None = None):
+def _device(scanned_at: datetime):
+    return SimpleNamespace(
+        device_no="23#",
+        process_name="冲压",
+        scanned_at=scanned_at,
+        released_at=None,
+        release_reason=None,
+        sort_order=0,
+    )
+
+
+def _report(
+    start_at: datetime,
+    end_at: datetime,
+    item,
+    *,
+    id: int = 1,
+    report_date: date | None = None,
+    devices: list | None = None,
+):
     return SimpleNamespace(
         id=id,
         attendance_point_name="总厂",
@@ -55,8 +74,8 @@ def _report(start_at: datetime, end_at: datetime, item, *, id: int = 1, report_d
         report_date=report_date or start_at.date(),
         start_at=start_at,
         end_at=end_at,
-        items=[item],
-        session=SimpleNamespace(devices=[]),
+        items=item if isinstance(item, list) else [item],
+        session=SimpleNamespace(devices=devices or []),
     )
 
 
@@ -113,6 +132,42 @@ def test_one_minute_before_night_end_allocates_to_previous_date():
     assert rows[1].reference_wage == 600
 
 
+def test_no_device_shared_full_report_splits_effective_minutes_between_items():
+    first_item = _item(id=1, good_qty=10)
+    second_item = _item(id=2, good_qty=20)
+    report = _report(
+        datetime(2026, 7, 25, 8, 0),
+        datetime(2026, 7, 25, 11, 0),
+        [first_item, second_item],
+    )
+
+    rows = build_report_allocation_drafts(report)
+
+    assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [
+        (1, 90, 90),
+        (2, 90, 90),
+    ]
+
+
+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)
+    second_item = _item(id=2, good_qty=20, started_at=started_at)
+    report = _report(
+        started_at,
+        datetime(2026, 7, 25, 11, 0),
+        [first_item, second_item],
+        devices=[_device(started_at)],
+    )
+
+    rows = build_report_allocation_drafts(report)
+
+    assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [
+        (1, 90, 90),
+        (2, 90, 90),
+    ]
+
+
 def test_zero_effective_minutes_falls_back_to_original_report_date():
     item = _item(good_qty=5, defect_qty=1, scrap_qty=0.5, changeover_count=2, process_unit_price_yuan=3)
     report = _report(