From 5f6deee7962e84c197307ff2197f37dec9afdcec Mon Sep 17 00:00:00 2001 From: souplearn Date: Sat, 25 Jul 2026 02:06:21 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E5=A2=9E=E5=8A=A0=E6=8A=A5=E5=B7=A5?= =?UTF-8?q?=E5=BD=92=E5=B1=9E=E6=8B=86=E5=88=86=E6=89=A7=E8=A1=8C=E8=AE=A1?= =?UTF-8?q?=E5=88=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../2026-07-25-report-time-allocation.md | 1836 +++++++++++++++++ 1 file changed, 1836 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-25-report-time-allocation.md diff --git a/docs/superpowers/plans/2026-07-25-report-time-allocation.md b/docs/superpowers/plans/2026-07-25-report-time-allocation.md new file mode 100644 index 0000000..3550f1d --- /dev/null +++ b/docs/superpowers/plans/2026-07-25-report-time-allocation.md @@ -0,0 +1,1836 @@ +# 报工跨日期工时归属拆分 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 将跨夜班、跨自然日的报工从“整条记录只有一个报工日期”改为“原始报工一条,统计归属按有效工时拆到多个日期”。 + +**Architecture:** 后端新增 `ProductionReportAllocation` 归属明细表和 `report_allocations` 服务,报工提交、自动提交、管理员审核更改后统一刷新归属明细。统计类接口按 `allocation_date` 查询和聚合;原始记录、审核、作废、更改水印继续基于 `ProductionReport`。 + +**Tech Stack:** FastAPI, SQLAlchemy 2, MySQL, Pytest, openpyxl, 微信小程序原生 WXML/WXSS/JS。 + +--- + +## Scope + +本计划实现以下内容: + +- 空白作息区间算加班。 +- 普通报工按工序明细有效工时拆分归属日期。 +- 数量、换料次数、参考工资按有效工时比例拆分。 +- 清洗等零工时报工使用原始 `report_date` 兜底,不做工时比例拆分。 +- 经理看板、导出、对账小账本、设备/模具使用统计切换到归属明细口径。 +- 管理员审核页和经理侧展示归属摘要。 +- 提供 MySQL 表结构迁移和历史归属明细回填脚本。 + +本计划不做: + +- 不拆分原始 `ProductionReport`。 +- 不让管理员手工编辑 `report_date`。 +- 不批量修改历史 `ProductionReport.report_date`。 + +## File Map + +Backend repo root: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint` + +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/models.py` + - 新增 `ProductionReportAllocation` 模型和关系。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/schemas.py` + - 新增归属输出 schema,并在 `ReportItemOut`、`ReportOut`、`DashboardRow` 中追加兼容字段。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/metrics.py` + - 空白作息区间计入加班。 + - 暴露可复用的 item 时间片/班种拆分辅助函数。 +- Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_allocations.py` + - 归属明细计算、摘要格式化、数据库刷新。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/serializers.py` + - 输出归属摘要和 item 级归属明细。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py` + - 员工提交和清洗提交后刷新归属明细。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/auto_submit.py` + - 系统自动提交后刷新归属明细。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reviews.py` + - 管理员审核/更改保存后刷新归属明细。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_lifecycle.py` + - 彻底删除作废记录时同步删除归属明细。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/dashboard.py` + - 经理看板按归属明细聚合。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/excel_export.py` + - 经理导出增加“原始报工日期”和“统计归属日期”口径。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reconciliation.py` + - 对账小账本按归属明细月份汇总最后工序成品数量。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/usage_stats.py` + - 设备/模具统计按归属明细日期过滤。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/usage_stats.py` + - 支持从归属明细构造使用统计行,报工次数按原始报工 ID 去重。 +- Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_allocations.py` + - 创建表并回填历史归属明细。 +- Test `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_metrics.py` + - 增加空白区间算加班测试。 +- Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_allocations.py` + - 覆盖跨日期拆分、比例拆分、零工时兜底。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_usage_stats.py` + - 覆盖归属明细使用统计和报工次数去重。 +- Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_dashboard_allocations.py` + - 覆盖经理看板按 `allocation_date` 聚合。 +- Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_reconciliation_allocations.py` + - 覆盖对账小账本按归属月份取最后工序成品。 + +Frontend repo root: `/Users/souplearn/Gitlab/app/JhHardwareWRS` + +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js` + - 映射 `allocation_summary_text` 和 `allocations`。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxml` + - 管理员审核页展示归属摘要。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxss` + - 增加归属摘要样式。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml` + - 经理看板展示统计归属日期说明。 +- Modify `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxss` + - 增加归属说明样式。 + +--- + +### Task 1: Lock Current Metrics Behavior And Add Blank-Interval Overtime Tests + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_metrics.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/metrics.py` + +- [ ] **Step 1: Add failing tests for blank intervals** + +Append these tests to `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_metrics.py`: + +```python +def test_morning_blank_interval_counts_as_overtime(): + start_at = datetime(2026, 7, 25, 6, 0, 0) + end_at = datetime(2026, 7, 25, 8, 0, 0) + item = Item(good_qty=10) + + metrics = calculate_report_metrics(start_at, end_at, [item]) + + assert metrics["duration_minutes"] == 120 + assert metrics["effective_minutes"] == 120 + assert metrics["shift_day_minutes"] == 0 + assert metrics["shift_overtime_minutes"] == 120 + assert metrics["shift_night_minutes"] == 0 + assert item.allocated_minutes == 120 + + +def test_dinner_blank_interval_counts_as_overtime(): + start_at = datetime(2026, 7, 25, 17, 20, 0) + end_at = datetime(2026, 7, 25, 18, 0, 0) + item = Item(good_qty=10) + + metrics = calculate_report_metrics(start_at, end_at, [item]) + + assert metrics["duration_minutes"] == 40 + assert metrics["effective_minutes"] == 40 + assert metrics["shift_day_minutes"] == 0 + assert metrics["shift_overtime_minutes"] == 40 + assert metrics["shift_night_minutes"] == 0 + assert item.allocated_minutes == 40 +``` + +- [ ] **Step 2: Run the failing tests** + +Run: + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_metrics.py::test_morning_blank_interval_counts_as_overtime tests/test_metrics.py::test_dinner_blank_interval_counts_as_overtime -v +``` + +Expected before implementation: both tests fail because blank intervals currently do not contribute to `shift_overtime_minutes`. + +- [ ] **Step 3: Implement blank interval overtime in metrics** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/metrics.py`, add helpers near `_overlap_minutes`: + +```python +def _clip_interval( + interval_start: datetime, + interval_end: datetime, + boundary_start: datetime, + boundary_end: datetime, +) -> tuple[datetime, datetime] | None: + start = max(interval_start, boundary_start) + end = min(interval_end, boundary_end) + if end <= start: + return None + return start, end + + +def _merge_intervals(intervals: list[tuple[datetime, datetime]]) -> list[tuple[datetime, datetime]]: + ordered = sorted(intervals, key=lambda item: item[0]) + merged: list[tuple[datetime, datetime]] = [] + for start, end in ordered: + if not merged or start > merged[-1][1]: + merged.append((start, end)) + elif end > merged[-1][1]: + merged[-1] = (merged[-1][0], end) + return merged + + +def _subtract_intervals( + base_start: datetime, + base_end: datetime, + occupied: list[tuple[datetime, datetime]], +) -> list[tuple[datetime, datetime]]: + gaps: list[tuple[datetime, datetime]] = [] + cursor = base_start + for start, end in _merge_intervals(occupied): + if start > cursor: + gaps.append((cursor, start)) + if end > cursor: + cursor = end + if cursor < base_end: + gaps.append((cursor, base_end)) + return gaps + + +def _calendar_dates_between(start_at: datetime, end_at: datetime): + current = start_at.date() + last = end_at.date() + while current <= last: + yield current + current += timedelta(days=1) + + +def _blank_overtime_intervals( + current_date: date, + schedule: WorkScheduleConfig, +) -> list[tuple[datetime, datetime]]: + day_start = datetime.combine(current_date, time.min) + day_end = day_start + timedelta(days=1) + occupied: list[tuple[datetime, datetime]] = [] + raw_intervals = [ + _schedule_interval(current_date, schedule.day_start, schedule.day_end), + _schedule_interval(current_date, schedule.overtime_start, schedule.overtime_end), + _schedule_interval(current_date, schedule.night_start, schedule.night_end), + _schedule_interval(current_date - timedelta(days=1), schedule.night_start, schedule.night_end), + ] + for interval_start, interval_end in raw_intervals: + clipped = _clip_interval(interval_start, interval_end, day_start, day_end) + if clipped is not None: + occupied.append(clipped) + return _subtract_intervals(day_start, day_end, occupied) +``` + +Then inside `_raw_shift_minutes`, after the existing `for bucket, intervals in shift_intervals.items()` block, add: + +```python + for blank_start, blank_end in _blank_overtime_intervals(current_date, active_schedule): + minutes["overtime"] += _overlap_minutes(start_at, end_at, blank_start, blank_end) +``` + +- [ ] **Step 4: Run targeted metrics tests** + +Run: + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_metrics.py -v +``` + +Expected: all `test_metrics.py` tests pass. + +- [ ] **Step 5: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/services/metrics.py tests/test_metrics.py +git commit -m "feat: 空白作息区间计入加班工时" +``` + +--- + +### Task 2: Add Allocation Model And MySQL Migration + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/models.py` +- Create: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_allocations.py` + +- [ ] **Step 1: Add SQLAlchemy model** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/models.py`, add this model after `ProductionReportItem`: + +```python +class ProductionReportAllocation(Base): + __tablename__ = "production_report_allocations" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + report_id: Mapped[int] = mapped_column(ForeignKey("production_reports.id"), nullable=False) + report_item_id: Mapped[int | None] = mapped_column(ForeignKey("production_report_items.id", ondelete="CASCADE")) + attendance_point_name: Mapped[str] = mapped_column(String(128), nullable=False, default="") + employee_phone: Mapped[str] = mapped_column(String(20), nullable=False, default="") + allocation_date: Mapped[date] = mapped_column(Date, nullable=False) + day_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0) + overtime_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0) + night_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0) + effective_minutes: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False, default=0) + good_qty: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0) + defect_qty: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0) + scrap_qty: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0) + changeover_count: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0) + reference_wage: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, default=0) + created_at: Mapped[datetime] = mapped_column(DateTime, default=now, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=now, onupdate=now, nullable=False) + + report: Mapped[ProductionReport] = relationship(back_populates="allocations") + item: Mapped[ProductionReportItem | None] = relationship(back_populates="allocations") + + __table_args__ = ( + Index("idx_report_allocations_report", "report_id"), + Index("idx_report_allocations_item", "report_item_id"), + Index("idx_report_allocations_date", "allocation_date"), + Index("idx_report_allocations_point_date", "attendance_point_name", "allocation_date"), + Index("idx_report_allocations_employee_date", "employee_phone", "allocation_date"), + ) +``` + +Add relationships: + +```python +# inside ProductionReport + allocations: Mapped[list["ProductionReportAllocation"]] = relationship( + back_populates="report", + cascade="all, delete-orphan", + ) + +# inside ProductionReportItem + allocations: Mapped[list["ProductionReportAllocation"]] = relationship( + back_populates="item", + cascade="all, delete-orphan", + ) +``` + +- [ ] **Step 2: Add migration script** + +Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_allocations.py`: + +```python +from pathlib import Path +import sys + +from sqlalchemy import text + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from app.database import engine # noqa: E402 + + +def _table_exists(conn, table_name: str) -> bool: + return bool(conn.execute( + text(""" + SELECT COUNT(1) + FROM information_schema.tables + WHERE table_schema = DATABASE() + AND table_name = :table_name + """), + {"table_name": table_name}, + ).scalar()) + + +def _index_exists(conn, table_name: str, index_name: str) -> bool: + return bool(conn.execute( + text(""" + SELECT COUNT(1) + FROM information_schema.statistics + WHERE table_schema = DATABASE() + AND table_name = :table_name + AND index_name = :index_name + """), + {"table_name": table_name, "index_name": index_name}, + ).scalar()) + + +def _ensure_schema() -> None: + with engine.begin() as conn: + if not _table_exists(conn, "production_report_allocations"): + conn.execute(text(""" + CREATE TABLE production_report_allocations ( + id BIGINT NOT NULL AUTO_INCREMENT, + report_id BIGINT NOT NULL, + report_item_id BIGINT NULL, + attendance_point_name VARCHAR(128) NOT NULL DEFAULT '', + employee_phone VARCHAR(20) NOT NULL DEFAULT '', + allocation_date DATE NOT NULL, + day_minutes DECIMAL(10,2) NOT NULL DEFAULT 0, + overtime_minutes DECIMAL(10,2) NOT NULL DEFAULT 0, + night_minutes DECIMAL(10,2) NOT NULL DEFAULT 0, + effective_minutes DECIMAL(10,2) NOT NULL DEFAULT 0, + good_qty DECIMAL(12,2) NOT NULL DEFAULT 0, + defect_qty DECIMAL(12,2) NOT NULL DEFAULT 0, + scrap_qty DECIMAL(12,2) NOT NULL DEFAULT 0, + changeover_count DECIMAL(12,2) NOT NULL DEFAULT 0, + reference_wage DECIMAL(12,2) NOT NULL DEFAULT 0, + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL, + PRIMARY KEY (id), + CONSTRAINT fk_report_allocations_report + FOREIGN KEY (report_id) REFERENCES production_reports(id) + ON DELETE CASCADE, + CONSTRAINT fk_report_allocations_item + FOREIGN KEY (report_item_id) REFERENCES production_report_items(id) + ON DELETE CASCADE + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + """)) + indexes = { + "idx_report_allocations_report": "CREATE INDEX idx_report_allocations_report ON production_report_allocations (report_id)", + "idx_report_allocations_item": "CREATE INDEX idx_report_allocations_item ON production_report_allocations (report_item_id)", + "idx_report_allocations_date": "CREATE INDEX idx_report_allocations_date ON production_report_allocations (allocation_date)", + "idx_report_allocations_point_date": "CREATE INDEX idx_report_allocations_point_date ON production_report_allocations (attendance_point_name, allocation_date)", + "idx_report_allocations_employee_date": "CREATE INDEX idx_report_allocations_employee_date ON production_report_allocations (employee_phone, allocation_date)", + } + for index_name, sql in indexes.items(): + if not _index_exists(conn, "production_report_allocations", index_name): + conn.execute(text(sql)) + + +def main() -> None: + _ensure_schema() + print("report allocations schema migrated") + + +if __name__ == "__main__": + main() +``` + +- [ ] **Step 3: Verify model import** + +Run: + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +python - <<'PY' +from app.models import ProductionReportAllocation +print(ProductionReportAllocation.__tablename__) +PY +``` + +Expected: + +```text +production_report_allocations +``` + +- [ ] **Step 4: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/models.py scripts/migrate_report_allocations.py +git commit -m "feat: 增加报工归属明细表" +``` + +--- + +### Task 3: Build Allocation Calculation Service + +**Files:** +- Create: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_allocations.py` +- Create: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_allocations.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/metrics.py` + +- [ ] **Step 1: Add failing allocation tests** + +Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_allocations.py`: + +```python +from datetime import date, datetime +from types import SimpleNamespace + +from app.services.report_allocations import build_report_allocation_drafts, allocation_summary_text +from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG + + +def _item(**overrides): + values = { + "id": 10, + "attendance_point_name": "嘉恒", + "device_no": "28#", + "project_no": "P1", + "product_name": "产品A", + "process_name": "1", + "stamping_method": "普通", + "process_unit_price_yuan": 1, + "good_qty": 650, + "defect_qty": 65, + "scrap_qty": 6.5, + "changeover_count": 13, + "allocated_minutes": 650, + "started_at": None, + } + values.update(overrides) + return SimpleNamespace(**values) + + +def _report(start_at, end_at, *items, report_date=date(2026, 7, 25)): + return SimpleNamespace( + id=1, + attendance_point_name="嘉恒", + employee_phone="13800000000", + report_date=report_date, + start_at=start_at, + end_at=end_at, + is_voided=False, + items=list(items), + session=SimpleNamespace(devices=[]), + ) + + +def test_split_midnight_report_to_previous_night_and_current_day(): + item = _item() + report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) + + drafts = build_report_allocation_drafts(report, DEFAULT_WORK_SCHEDULE_CONFIG) + + assert [(row.report_item_id, row.allocation_date) for row in drafts] == [ + (10, date(2026, 7, 24)), + (10, date(2026, 7, 25)), + ] + previous, current = drafts + assert previous.night_minutes == 350 + assert previous.effective_minutes == 350 + assert previous.good_qty == 350 + assert previous.defect_qty == 35 + assert previous.scrap_qty == 3.5 + assert previous.changeover_count == 7 + assert previous.reference_wage == 350 + assert current.day_minutes == 180 + assert current.overtime_minutes == 120 + assert current.effective_minutes == 300 + assert current.good_qty == 300 + assert current.defect_qty == 30 + assert current.scrap_qty == 3 + assert current.changeover_count == 6 + assert current.reference_wage == 300 + + +def test_one_minute_night_tail_only_allocates_one_minute_to_previous_day(): + item = _item(good_qty=301, defect_qty=0, scrap_qty=0, changeover_count=0, process_unit_price_yuan=2) + report = _report(datetime(2026, 7, 25, 5, 59), datetime(2026, 7, 25, 11, 0), item) + + drafts = build_report_allocation_drafts(report, DEFAULT_WORK_SCHEDULE_CONFIG) + + previous, current = drafts + assert previous.allocation_date == date(2026, 7, 24) + assert previous.night_minutes == 1 + assert previous.good_qty == 1 + assert previous.reference_wage == 2 + assert current.allocation_date == date(2026, 7, 25) + assert current.overtime_minutes == 120 + assert current.day_minutes == 180 + assert current.good_qty == 300 + assert current.reference_wage == 600 + + +def test_zero_effective_minutes_falls_back_to_original_report_date(): + item = _item( + id=20, + stamping_method="清洗", + allocated_minutes=0, + good_qty=100, + defect_qty=0, + scrap_qty=0, + process_unit_price_yuan=0, + ) + report = _report( + datetime(2026, 7, 25, 9, 0), + datetime(2026, 7, 25, 9, 0), + item, + report_date=date(2026, 7, 25), + ) + + drafts = build_report_allocation_drafts(report, DEFAULT_WORK_SCHEDULE_CONFIG) + + assert len(drafts) == 1 + assert drafts[0].allocation_date == date(2026, 7, 25) + assert drafts[0].effective_minutes == 0 + assert drafts[0].good_qty == 100 + + +def test_allocation_summary_text_groups_dates_and_shift_kinds(): + item = _item() + report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) + drafts = build_report_allocation_drafts(report, DEFAULT_WORK_SCHEDULE_CONFIG) + + assert allocation_summary_text(drafts) == "2026-07-24 夜班5.83小时;2026-07-25 白班3小时、加班2小时" +``` + +- [ ] **Step 2: Run failing allocation tests** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_report_allocations.py -v +``` + +Expected before implementation: import failure for `app.services.report_allocations`. + +- [ ] **Step 3: Expose item time range helpers from metrics** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/metrics.py`, add a small dataclass near the Protocol classes: + +```python +@dataclass(frozen=True) +class ItemTimeRange: + item: ReportItemLike + start_at: datetime + end_at: datetime +``` + +Add `from dataclasses import dataclass` at the top. + +Then add this public helper below `_allocate_item_minutes`: + +```python +def item_time_ranges( + start_at: datetime, + end_at: datetime, + items: list[ReportItemLike], + devices: list[ReportDeviceLike] | None = None, +) -> list[ItemTimeRange]: + segment_ranges = _device_segment_ranges(start_at, end_at, devices) + if not items: + return [] + if not segment_ranges: + return [ItemTimeRange(item=item, start_at=start_at, end_at=end_at) for item in items if end_at > start_at] + + items_by_device: dict[str, list[ReportItemLike]] = defaultdict(list) + for item in items: + mold_key = _mold_key( + getattr(item, "product_name", "") or item.device_no, + getattr(item, "process_name", ""), + ) + items_by_device[mold_key].append(item) + + ranges: list[ItemTimeRange] = [] + allocated_item_ids: set[int] = set() + for device_no, device_ranges in segment_ranges.items(): + device_items = items_by_device.get(device_no, []) + if not device_items: + continue + for segment_start, segment_end in device_ranges: + grouped_items: dict[datetime, list[ReportItemLike]] = defaultdict(list) + for item in device_items: + item_start = getattr(item, "started_at", None) or segment_start + grouped_items[_clamp_datetime(item_start, segment_start, segment_end)].append(item) + first_item_start = min(grouped_items) + if first_item_start > segment_start: + grouped_items[segment_start].extend(grouped_items.pop(first_item_start)) + ordered_starts = sorted(grouped_items) + for index, item_start in enumerate(ordered_starts): + item_end = segment_end + if index + 1 < len(ordered_starts): + item_end = ordered_starts[index + 1] + if item_end <= item_start: + continue + for item in grouped_items[item_start]: + ranges.append(ItemTimeRange(item=item, start_at=item_start, end_at=item_end)) + allocated_item_ids.add(id(item)) + + unmatched_items = [item for item in items if id(item) not in allocated_item_ids] + for item in unmatched_items: + ranges.append(ItemTimeRange(item=item, start_at=start_at, end_at=end_at)) + return ranges +``` + +- [ ] **Step 4: Implement report allocation service** + +Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_allocations.py` with these public APIs: + +```python +from collections import defaultdict +from dataclasses import dataclass +from datetime import date, datetime, timedelta + +from sqlalchemy import delete +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, + format_hours, + item_time_ranges, + shift_minutes_between, +) +from app.services.work_schedule import DEFAULT_WORK_SCHEDULE_CONFIG, WorkScheduleConfig +from app.timezone import now + + +@dataclass(frozen=True) +class ReportAllocationDraft: + report_id: int + 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 _night_allocation_date(segment_start: datetime, schedule: WorkScheduleConfig) -> date: + night_end = _parse_time(schedule.night_end) + if segment_start.time() < night_end: + return segment_start.date() - timedelta(days=1) + return segment_start.date() + + +def _parse_time(value: str): + hour, minute = [int(part) for part in str(value).split(":", 1)] + return datetime.min.time().replace(hour=hour, minute=minute) + + +def _date_for_bucket(bucket: str, segment_start: datetime, schedule: WorkScheduleConfig) -> date: + if bucket == "night": + return _night_allocation_date(segment_start, schedule) + return segment_start.date() + + +def _split_range_by_hour_boundaries(start_at: datetime, end_at: datetime) -> list[tuple[datetime, datetime]]: + boundaries = {start_at, end_at} + cursor = start_at.replace(minute=0, second=0, microsecond=0) + while cursor <= end_at: + boundaries.add(cursor) + cursor += timedelta(hours=1) + return [ + (start, end) + for start, end in zip(sorted(boundaries), sorted(boundaries)[1:]) + if end > start + ] + + +def _empty_bucket_minutes() -> dict[str, float]: + return {key: 0.0 for key in SHIFT_BUCKETS} + + +def _item_reference_wage(item) -> float: + return as_float(getattr(item, "good_qty", 0)) * as_float(getattr(item, "process_unit_price_yuan", 0)) + + +def _zero_effective_fallback(report: ProductionReport, item) -> ReportAllocationDraft: + return ReportAllocationDraft( + report_id=report.id, + 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=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 build_report_allocation_drafts( + report: ProductionReport, + schedule: WorkScheduleConfig | None = None, +) -> list[ReportAllocationDraft]: + active_schedule = schedule or DEFAULT_WORK_SCHEDULE_CONFIG + items = list(getattr(report, "items", []) or []) + if not items: + return [] + devices = getattr(getattr(report, "session", None), "devices", None) + ranges = item_time_ranges(report.start_at, report.end_at, items, devices) + grouped_minutes: dict[tuple[int | None, date], dict[str, float]] = defaultdict(_empty_bucket_minutes) + + for item_range in ranges: + for slice_start, slice_end in _split_range_by_hour_boundaries(item_range.start_at, item_range.end_at): + shift_minutes = shift_minutes_between(slice_start, slice_end, schedule=active_schedule) + for bucket in SHIFT_BUCKETS: + minutes = as_float(shift_minutes.get(bucket, 0)) + if minutes <= 0: + continue + allocation_date = _date_for_bucket(bucket, slice_start, active_schedule) + item_id = getattr(item_range.item, "id", None) + grouped_minutes[(item_id, allocation_date)][bucket] += minutes + + drafts: list[ReportAllocationDraft] = [] + for item in items: + item_id = getattr(item, "id", None) + item_rows = [ + (allocation_date, minutes) + for (row_item_id, allocation_date), minutes in grouped_minutes.items() + if row_item_id == item_id + ] + total_effective = sum(sum(as_float(minutes.get(key, 0)) for key in SHIFT_BUCKETS) for _date, minutes in item_rows) + if total_effective <= 0: + drafts.append(_zero_effective_fallback(report, item)) + continue + for allocation_date, minutes in sorted(item_rows, key=lambda row: row[0]): + effective = sum(as_float(minutes.get(key, 0)) for key in SHIFT_BUCKETS) + if effective <= 0: + continue + ratio = effective / total_effective + drafts.append( + ReportAllocationDraft( + report_id=report.id, + report_item_id=item_id, + attendance_point_name=str(getattr(report, "attendance_point_name", "") or ""), + employee_phone=str(getattr(report, "employee_phone", "") or ""), + allocation_date=allocation_date, + day_minutes=round2(minutes.get("day", 0)), + overtime_minutes=round2(minutes.get("overtime", 0)), + night_minutes=round2(minutes.get("night", 0)), + effective_minutes=round2(effective), + 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(_item_reference_wage(item) * ratio), + ) + ) + return sorted(drafts, key=lambda row: (row.report_item_id or 0, row.allocation_date)) + + +def allocation_summary_text(rows: list[ReportAllocationDraft | ProductionReportAllocation]) -> str: + grouped: dict[date, dict[str, float]] = defaultdict(_empty_bucket_minutes) + for row in rows: + grouped[row.allocation_date]["day"] += as_float(row.day_minutes) + grouped[row.allocation_date]["overtime"] += as_float(row.overtime_minutes) + grouped[row.allocation_date]["night"] += as_float(row.night_minutes) + parts: list[str] = [] + labels = {"day": "白班", "overtime": "加班", "night": "夜班"} + for allocation_date, minutes in sorted(grouped.items()): + shift_parts = [ + f"{labels[key]}{format_hours(minutes.get(key, 0))}小时" + for key in ("day", "overtime", "night") + if round2(minutes.get(key, 0)) > 0 + ] + if shift_parts: + parts.append(f"{allocation_date.isoformat()} {'、'.join(shift_parts)}") + return ";".join(parts) + + +def refresh_report_allocations( + db: Session, + report: ProductionReport, + schedule: WorkScheduleConfig | None = None, + *, + commit: bool = False, +) -> list[ProductionReportAllocation]: + db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report.id)) + timestamp = now() + 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, + created_at=timestamp, + updated_at=timestamp, + ) + for draft in build_report_allocation_drafts(report, schedule) + ] + db.add_all(rows) + if commit: + db.commit() + return rows +``` + +During implementation, prefer extracting helpers from this block rather than copying logic into routers. + +- [ ] **Step 5: Run allocation tests** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_report_allocations.py tests/test_metrics.py -v +``` + +Expected: all tests pass. + +- [ ] **Step 6: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/services/metrics.py app/services/report_allocations.py tests/test_report_allocations.py tests/test_metrics.py +git commit -m "feat: 增加报工工时归属拆分计算" +``` + +--- + +### Task 4: Refresh Allocations In Report Lifecycle + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/auto_submit.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reviews.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_lifecycle.py` +- Test: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_allocations.py` + +- [ ] **Step 1: Add lifecycle test for refresh** + +Append this test to `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_allocations.py`: + +```python +def test_refresh_report_allocations_replaces_existing_rows(): + from sqlalchemy import create_engine, select + from sqlalchemy.orm import sessionmaker + + from app.models import Base, ProductionReportAllocation + from app.services.report_allocations import refresh_report_allocations + + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + Session = sessionmaker(bind=engine) + db = Session() + item = _item(id=10) + report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item) + + refresh_report_allocations(db, report, DEFAULT_WORK_SCHEDULE_CONFIG, commit=True) + refresh_report_allocations(db, report, DEFAULT_WORK_SCHEDULE_CONFIG, commit=True) + + rows = db.scalars(select(ProductionReportAllocation).order_by(ProductionReportAllocation.allocation_date)).all() + assert len(rows) == 2 + assert [row.allocation_date for row in rows] == [date(2026, 7, 24), date(2026, 7, 25)] +``` + +- [ ] **Step 2: Wire normal report submission** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py`, import: + +```python +from app.services.report_allocations import refresh_report_allocations +``` + +In `submit_report`, after `db.add(report)` and before `db.commit()`, add: + +```python + db.flush() + refresh_report_allocations(db, report, schedule=schedule, commit=False) +``` + +- [ ] **Step 3: Wire cleaning submission** + +In `submit_cleaning_report`, after `db.add(report)` and before `db.commit()`, add: + +```python + db.flush() + refresh_report_allocations(db, report, schedule=get_work_schedule_config(db, point_name), commit=False) +``` + +Expected behavior: cleaning has zero effective minutes, so allocation rows use original `report_date` and carry quantity. + +- [ ] **Step 4: Wire auto submit** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/auto_submit.py`, import: + +```python +from app.services.report_allocations import refresh_report_allocations +``` + +In `auto_submit_session`, after `db.add(report)`, add: + +```python + db.flush() + refresh_report_allocations(db, report, schedule=schedule, commit=False) +``` + +- [ ] **Step 5: Wire admin approve/edit** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reviews.py`, import: + +```python +from app.services.report_allocations import refresh_report_allocations +``` + +In `approve_report`, after `_recalculate_report_metrics(report, schedule)` and before final `db.commit()`, add: + +```python + refresh_report_allocations(db, report, schedule=schedule, commit=False) +``` + +- [ ] **Step 6: Delete allocations when purging expired voided reports** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_lifecycle.py`, import `ProductionReportAllocation` and add this before deleting `ProductionReportItem`: + +```python + db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report.id)) +``` + +- [ ] **Step 7: Run lifecycle-adjacent tests** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_report_allocations.py tests/test_review_corrections.py -v +``` + +Expected: all tests pass. + +- [ ] **Step 8: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/routers/reports.py app/services/auto_submit.py app/routers/reviews.py app/services/report_lifecycle.py tests/test_report_allocations.py +git commit -m "feat: 报工保存后刷新归属明细" +``` + +--- + +### Task 5: Expose Allocation Summary In API Responses + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/schemas.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/serializers.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reviews.py` + +- [ ] **Step 1: Add schema fields** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/schemas.py`, add: + +```python +class ReportAllocationOut(BaseModel): + 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 +``` + +Add to `ReportItemOut`: + +```python + allocations: list[ReportAllocationOut] = Field(default_factory=list) +``` + +Add to `ReportOut`: + +```python + allocation_summary_text: str = "" +``` + +Add to `DashboardRow`: + +```python + source_report_date: date | None = None + source_report_dates: list[date] = Field(default_factory=list) +``` + +- [ ] **Step 2: Serialize allocations** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/serializers.py`, import: + +```python +from app.schemas import ReportAllocationOut +from app.services.report_allocations import allocation_summary_text +``` + +Add helper: + +```python +def report_allocation_out(row) -> ReportAllocationOut: + return ReportAllocationOut( + allocation_date=row.allocation_date, + day_minutes=as_float(row.day_minutes), + overtime_minutes=as_float(row.overtime_minutes), + night_minutes=as_float(row.night_minutes), + effective_minutes=as_float(row.effective_minutes), + good_qty=as_float(row.good_qty), + defect_qty=as_float(row.defect_qty), + scrap_qty=as_float(row.scrap_qty), + changeover_count=as_float(row.changeover_count), + reference_wage=as_float(row.reference_wage), + ) +``` + +Update `report_item_out` to set: + +```python + allocations=[report_allocation_out(row) for row in getattr(item, "allocations", [])], +``` + +Update `report_out` to set: + +```python + allocation_summary_text=allocation_summary_text(list(getattr(report, "allocations", []) or [])), +``` + +- [ ] **Step 3: Eager load allocations in report queries** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py` and `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reviews.py`, update report queries that already use `selectinload(ProductionReport.items)` to also load: + +```python +selectinload(ProductionReport.allocations) +selectinload(ProductionReport.items).selectinload(ProductionReportItem.allocations) +``` + +Add `ProductionReportItem` import where needed. + +- [ ] **Step 4: Smoke test response serialization** + +Run: + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_report_allocations.py tests/test_review_corrections.py -v +``` + +Expected: tests pass and import errors do not occur. + +- [ ] **Step 5: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/schemas.py app/services/serializers.py app/routers/reports.py app/routers/reviews.py +git commit -m "feat: 输出报工归属摘要" +``` + +--- + +### Task 6: Switch Manager Dashboard And Excel Export To Allocations + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/dashboard.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/excel_export.py` +- Create: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_dashboard_allocations.py` + +- [ ] **Step 1: Add failing dashboard aggregation test** + +Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_dashboard_allocations.py` with a focused helper-level test around a new function `_dashboard_rows_from_allocations`. + +```python +from datetime import date +from types import SimpleNamespace + +from app.routers.dashboard import _dashboard_rows_from_allocations + + +def test_dashboard_groups_by_allocation_date_not_original_report_date(): + employee = SimpleNamespace(name="张三", is_temporary=False) + report = SimpleNamespace( + id=1, + attendance_point_name="嘉恒", + report_date=date(2026, 7, 25), + employee_phone="13800000000", + employee=employee, + is_system_auto_submitted=False, + is_voided=False, + is_multi_person_assistant=False, + review_remark="", + audit_logs=[], + ) + item = SimpleNamespace( + id=10, + report=report, + attendance_point_name="嘉恒", + project_no="P1", + product_name="产品A", + process_name="1", + stamping_method="普通", + material_code="M1", + material_name="料A", + raw_material_batch_no="B1", + device_no="28#", + operator_count=1, + process_unit_price_yuan=2, + standard_beat=60, + good_qty=650, + defect_qty=0, + scrap_qty=0, + ) + allocations = [ + SimpleNamespace( + report=report, + item=item, + report_id=1, + report_item_id=10, + allocation_date=date(2026, 7, 24), + day_minutes=0, + overtime_minutes=0, + night_minutes=350, + effective_minutes=350, + good_qty=350, + defect_qty=0, + scrap_qty=0, + changeover_count=0, + reference_wage=700, + ), + SimpleNamespace( + report=report, + item=item, + report_id=1, + report_item_id=10, + allocation_date=date(2026, 7, 25), + day_minutes=180, + overtime_minutes=120, + night_minutes=0, + effective_minutes=300, + good_qty=300, + defect_qty=0, + scrap_qty=0, + changeover_count=0, + reference_wage=600, + ), + ] + + rows = _dashboard_rows_from_allocations(allocations) + + assert [row.report_date for row in rows] == [date(2026, 7, 25), date(2026, 7, 24)] + assert [row.total_good_qty for row in rows] == [300, 350] + assert [row.reference_wage for row in rows] == [600, 700] + assert all(row.report_count == 1 for row in rows) +``` + +- [ ] **Step 2: Implement allocation query in dashboard router** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/dashboard.py`, import `ProductionReportAllocation` and replace date filtering in `_dashboard_rows`: + +```python +query = ( + select(ProductionReportAllocation) + .join(ProductionReport, ProductionReport.id == ProductionReportAllocation.report_id) + .join(ProductionReportItem, ProductionReportItem.id == ProductionReportAllocation.report_item_id) + .options( + selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.employee), + selectinload(ProductionReportAllocation.report).selectinload(ProductionReport.audit_logs), + selectinload(ProductionReportAllocation.item), + ) + .where(ProductionReport.status == ReportStatus.approved) +) +if not include_voided: + query = query.where(ProductionReport.is_voided.is_(False)) +if start_date: + query = query.where(ProductionReportAllocation.allocation_date >= start_date) +if end_date: + query = query.where(ProductionReportAllocation.allocation_date <= end_date) +allocations = db.scalars(query).all() +return _dashboard_rows_from_allocations(allocations) +``` + +Implement `_dashboard_rows_from_allocations(allocations: list) -> list[DashboardRow]` by moving the existing grouping logic from `_dashboard_rows`, with these substitutions: + +- group key uses `allocation.allocation_date` instead of `report.report_date` +- `effective_minutes` uses `allocation.effective_minutes` +- shift minutes use `allocation.day_minutes`, `allocation.overtime_minutes`, `allocation.night_minutes` +- quantities use `allocation.good_qty`, `allocation.defect_qty + allocation.scrap_qty` +- `changeover_count` uses `allocation.changeover_count` +- `reference_wage` uses `allocation.reference_wage` +- `report_count` uses `len(group["report_ids"])` +- `source_report_dates` collects `report.report_date` + +- [ ] **Step 3: Update Excel headers** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/excel_export.py`, change dashboard headers from: + +```python +"报工日期", +``` + +to: + +```python +"统计归属日期", +"原始报工日期", +``` + +When appending rows, write: + +```python +row.report_date.isoformat(), +"、".join(sorted(item.isoformat() for item in getattr(row, "source_report_dates", []) or [row.source_report_date or row.report_date])), +``` + +Ensure all later column indexes used for red fills shift by one column. + +- [ ] **Step 4: Run dashboard/export tests** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_dashboard_allocations.py tests/test_usage_stats.py -v +``` + +Expected: dashboard allocation test passes; existing usage stats tests still pass. + +- [ ] **Step 5: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/routers/dashboard.py app/services/excel_export.py tests/test_dashboard_allocations.py +git commit -m "feat: 经理看板按归属日期汇总" +``` + +--- + +### Task 7: Switch Reconciliation Ledger To Allocations + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reconciliation.py` +- Create: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_reconciliation_allocations.py` + +- [ ] **Step 1: Add query helper for reported quantities** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reconciliation.py`, change `_reported_good_quantities` to query `ProductionReportAllocation` joined to `ProductionReportItem` and `ProductionReport`. + +The selected month must be: + +```python +extract("month", ProductionReportAllocation.allocation_date).label("month") +``` + +The selected quantity must be: + +```python +func.sum(ProductionReportAllocation.good_qty).label("good_qty") +``` + +The year filter must be: + +```python +extract("year", ProductionReportAllocation.allocation_date) == year +``` + +- [ ] **Step 2: Update reconciliation years** + +In `list_reconciliation_years`, read years from `ProductionReportAllocation.allocation_date` joined to approved, unvoided reports: + +```python +select(extract("year", ProductionReportAllocation.allocation_date)) +``` + +- [ ] **Step 3: Add focused reconciliation test** + +Create `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_reconciliation_allocations.py` with a pure helper test. Extract the final row folding into a helper named `_fold_reported_good_quantity_rows` so the test does not depend on a database fixture. + +Expected scenario: + +```text +产品A最后工序为2序; +同一原始报工拆到2026-07和2026-08; +7月对账小账本只显示7月 allocation.good_qty; +8月只显示8月 allocation.good_qty。 +``` + +Use exact expected values: + +```python +assert totals[("嘉恒", "产品A", 7)] == 350 +assert totals[("嘉恒", "产品A", 8)] == 300 +``` + +- [ ] **Step 4: Run reconciliation tests** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_reconciliation_allocations.py -v +``` + +Expected: tests pass. + +- [ ] **Step 5: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/routers/reconciliation.py tests/test_reconciliation_allocations.py +git commit -m "feat: 对账小账本按归属日期统计" +``` + +--- + +### Task 8: Switch Usage Stats To Allocations + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/usage_stats.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/usage_stats.py` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_usage_stats.py` + +- [ ] **Step 1: Update accumulator to dedupe report count** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/usage_stats.py`, add `report_ids` to `UsageStatRow`: + +```python + report_ids: set[int] = field(default_factory=set, repr=False) +``` + +Change `UsageStatAccumulator.add` signature: + +```python + report_id: int | None = None, +``` + +Replace: + +```python + row.report_count += 1 +``` + +with: + +```python + if report_id is None: + row.report_count += 1 + elif report_id not in row.report_ids: + row.report_ids.add(report_id) + row.report_count += 1 +``` + +Pass `report_id` from both report-item and allocation paths. + +- [ ] **Step 2: Add allocation-aware builder** + +Add a function: + +```python +def build_usage_stats_from_allocations( + *, + allocations: list, + category: str, + equipment_type_by_key: dict[tuple[str, str], str], +) -> list[UsageStatRow]: + reports = [] + for allocation in allocations: + item = allocation.item + proxy = SimpleNamespace( + attendance_point_name=item.attendance_point_name, + product_name=item.product_name, + process_name=item.process_name, + stamping_method=item.stamping_method, + operator_count=item.operator_count, + device_no=item.device_no, + allocated_minutes=allocation.effective_minutes, + good_qty=allocation.good_qty, + report_id=allocation.report_id, + ) + reports.append(SimpleNamespace(items=[proxy])) + return build_usage_stats( + reports=reports, + category=category, + equipment_type_by_key=equipment_type_by_key, + ) +``` + +Add `from types import SimpleNamespace` at the top. + +When implementing, also modify `_iter_report_items` or the `acc.add` calls so `report_id` reaches `UsageStatAccumulator.add`. + +- [ ] **Step 3: Update usage stats router query** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/usage_stats.py`, import `ProductionReportAllocation` and query allocations joined to approved, unvoided reports. + +Date filters must use: + +```python +ProductionReportAllocation.allocation_date >= start_date +ProductionReportAllocation.allocation_date <= end_date +``` + +Summary/export/detail should call `build_usage_stats_from_allocations`. + +Daily detail rows should use `allocation.allocation_date`. + +Report detail rows should show `allocation.allocation_date` as `report_date` so the detail page follows the selected statistics date. + +- [ ] **Step 4: Add usage stats tests** + +Append to `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_usage_stats.py`: + +```python +def test_usage_stats_dedupes_report_count_for_split_allocations(): + allocation_a = SimpleNamespace( + report_id=1, + effective_minutes=350, + good_qty=350, + item=_item(device_no="28#", allocated_minutes=0, good_qty=0), + ) + allocation_b = SimpleNamespace( + report_id=1, + effective_minutes=300, + good_qty=300, + item=_item(device_no="28#", allocated_minutes=0, good_qty=0), + ) + + rows = build_usage_stats_from_allocations( + allocations=[allocation_a, allocation_b], + category="device", + equipment_type_by_key={("嘉恒", "28#"): "冲压设备"}, + ) + + assert len(rows) == 1 + assert rows[0].value == 650 + assert rows[0].report_count == 1 +``` + +- [ ] **Step 5: Run usage stats tests** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest tests/test_usage_stats.py -v +``` + +Expected: all tests pass. + +- [ ] **Step 6: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add app/routers/usage_stats.py app/services/usage_stats.py tests/test_usage_stats.py +git commit -m "feat: 设备模具统计按归属日期汇总" +``` + +--- + +### Task 9: Frontend Display For Allocation Summary + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxml` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxss` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml` +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxss` + +- [ ] **Step 1: Map new fields in API** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js`, add: + +```javascript +const toCamelAllocation = row => ({ + allocationDate: row.allocation_date || '', + dayMinutes: Number(row.day_minutes || 0), + overtimeMinutes: Number(row.overtime_minutes || 0), + nightMinutes: Number(row.night_minutes || 0), + effectiveMinutes: Number(row.effective_minutes || 0), + goodQty: Number(row.good_qty || 0), + defectQty: Number(row.defect_qty || 0), + scrapQty: Number(row.scrap_qty || 0), + changeoverCount: Number(row.changeover_count || 0), + referenceWage: Number(row.reference_wage || 0), +}) +``` + +In `toCamelReportItem`, add: + +```javascript + allocations: (row.allocations || []).map(toCamelAllocation), +``` + +In `toCamelReport`, add: + +```javascript + allocationSummaryText: row.allocation_summary_text || '', +``` + +In dashboard row mapper, map: + +```javascript + sourceReportDate: row.source_report_date || '', + sourceReportDates: row.source_report_dates || [], +``` + +- [ ] **Step 2: Show summary in review page** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxml`, below `{{item.metrics.shiftDistributionText || ''}}`, add: + +```xml +归属拆分:{{item.allocationSummaryText}} +``` + +- [ ] **Step 3: Show source date in dashboard** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml`, below the row title date, add: + +```xml +原始报工日期 {{item.sourceReportDates.join('、')}} +``` + +- [ ] **Step 4: Add compact styles** + +In both WXSS files, add: + +```css +.allocation-summary { + display: block; + margin-top: 6rpx; + color: #4b5563; + font-size: 24rpx; + line-height: 1.45; +} +``` + +- [ ] **Step 5: Frontend smoke check** + +Open WeChat DevTools and verify: + +```text +管理员审核页:跨日期记录显示“归属拆分:...” +经理看板:统计日期仍在主标题;跨日期聚合行额外显示原始报工日期 +旧记录没有 allocation_summary_text 时页面不显示空行 +``` + +- [ ] **Step 6: Commit** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS +git add utils/api.js pages/review/review.wxml pages/review/review.wxss pages/dashboard/dashboard.wxml pages/dashboard/dashboard.wxss +git commit -m "feat: 展示报工归属拆分摘要" +``` + +--- + +### Task 10: Run Migration And Backfill Verification + +**Files:** +- Modify: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_allocations.py` +- Run: `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_allocations.py` + +- [ ] **Step 1: Add backfill logic to migration script** + +In `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_allocations.py`, add imports: + +```python +from sqlalchemy.orm import selectinload + +from app.database import SessionLocal, engine # noqa: E402 +from app.models import ProductionReport, WorkSession # noqa: E402 +from app.services.report_allocations import refresh_report_allocations # noqa: E402 +from app.services.work_schedule import get_work_schedule_config # noqa: E402 +``` + +Replace the existing `from app.database import engine` import with the import above. + +Add: + +```python +def _backfill() -> tuple[int, int]: + processed = 0 + skipped = 0 + with SessionLocal() as db: + reports = db.scalars( + select(ProductionReport) + .options( + selectinload(ProductionReport.items), + selectinload(ProductionReport.session).selectinload(WorkSession.devices), + ) + .order_by(ProductionReport.id.asc()) + ).all() + for report in reports: + if report is None: + skipped += 1 + continue + schedule = get_work_schedule_config(db, report.attendance_point_name) + refresh_report_allocations(db, report, schedule=schedule, commit=False) + processed += 1 + if processed % 200 == 0: + db.commit() + db.commit() + return processed, skipped +``` + +Add `select` to the SQLAlchemy import: + +```python +from sqlalchemy import select, text +``` + +Update `main`: + +```python +def main() -> None: + _ensure_schema() + processed, skipped = _backfill() + print(f"report allocations migrated, processed={processed}, skipped={skipped}") +``` + +- [ ] **Step 2: Run migration on local dev DB** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +python scripts/migrate_report_allocations.py +``` + +Expected: + +```text +report allocations migrated, processed=, skipped=0 +``` + +- [ ] **Step 3: Verify table and indexes** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +python - <<'PY' +from sqlalchemy import text +from app.database import engine + +with engine.connect() as conn: + table_count = conn.execute(text(""" + SELECT COUNT(1) + FROM information_schema.tables + WHERE table_schema = DATABASE() + AND table_name = 'production_report_allocations' + """)).scalar() + row_count = conn.execute(text("SELECT COUNT(1) FROM production_report_allocations")).scalar() + indexes = conn.execute(text(""" + SELECT index_name + FROM information_schema.statistics + WHERE table_schema = DATABASE() + AND table_name = 'production_report_allocations' + ORDER BY index_name + """)).all() +print({"table_count": table_count, "row_count": row_count, "indexes": [row[0] for row in indexes]}) +PY +``` + +Expected: + +```text +table_count is 1 +row_count is greater than or equal to the number of production report items that can be allocated +indexes include idx_report_allocations_point_date +``` + +- [ ] **Step 4: Verify sample cross-date data** + +If local DB has no cross-date sample, create one through the app flow or by controlled test data. Then run: + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +python - <<'PY' +from sqlalchemy import text +from app.database import engine + +with engine.connect() as conn: + rows = conn.execute(text(""" + SELECT report_id, report_item_id, allocation_date, + day_minutes, overtime_minutes, night_minutes, + effective_minutes, good_qty, reference_wage + FROM production_report_allocations + WHERE report_id = :report_id + ORDER BY allocation_date, report_item_id + """), {"report_id": 1}).mappings().all() +for row in rows: + print(dict(row)) +PY +``` + +Expected for a `00:10-11:00` style report: + +```text +one row for previous date with night_minutes > 0 +one row for current date with day_minutes and overtime_minutes > 0 +sum(good_qty) equals original item good_qty within rounding tolerance +sum(reference_wage) equals original item reference wage within rounding tolerance +``` + +- [ ] **Step 5: Commit migration backfill update** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git add scripts/migrate_report_allocations.py +git commit -m "feat: 回填历史报工归属明细" +``` + +--- + +### Task 11: Full Verification + +**Files:** +- Verify backend repo `/Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint` +- Verify frontend repo `/Users/souplearn/Gitlab/app/JhHardwareWRS` + +- [ ] **Step 1: Run backend tests** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +pytest -v +``` + +Expected: all backend tests pass. + +- [ ] **Step 2: Run targeted import smoke** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +python - <<'PY' +from app.models import ProductionReportAllocation +from app.services.report_allocations import build_report_allocation_drafts +from app.routers.dashboard import _dashboard_rows +print("ok") +PY +``` + +Expected: + +```text +ok +``` + +- [ ] **Step 3: Manual app verification** + +Use local backend and WeChat DevTools: + +```text +1. 创建或找一条 00:10-11:00 的测试报工。 +2. 管理员审核通过。 +3. 管理员审核页显示归属拆分摘要。 +4. 经理看板按前一天和当天拆成统计行。 +5. 经理导出 Excel 有“统计归属日期”和“原始报工日期”。 +6. 对账小账本按统计归属月份显示最后工序成品数量。 +7. 设备/模具统计按统计归属日期过滤,报工次数不重复放大。 +8. 作废该报工后,看板、对账、使用统计不再计入。 +9. 撤销作废后,看板、对账、使用统计恢复。 +``` + +- [ ] **Step 4: Final git status** + +```bash +cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint +git status --short +cd /Users/souplearn/Gitlab/app/JhHardwareWRS +git status --short +``` + +Expected: + +```text +only intentional commits remain; no unrelated user files are staged +``` + +--- + +## Self-Review Checklist + +- Spec coverage: + - 原始报工不拆:Tasks 2-6 keep `ProductionReport` unchanged as source record. + - 归属明细表:Task 2. + - 空白区间算加班:Task 1. + - 数量和工资按工时比例拆:Task 3. + - 清洗零工时兜底:Task 3. + - 提交、自动提交、管理员审核刷新归属:Task 4. + - 经理看板和导出:Task 6. + - 对账小账本:Task 7. + - 设备/模具统计:Task 8. + - 前端展示:Task 9. + - 历史回填:Task 10. + +- Placeholder scan: + - No unresolved placeholder markers. + - No unresolved business decision. + - Report count is fixed as original report ID dedupe. + +- Type consistency: + - Model name: `ProductionReportAllocation`. + - Service functions: `build_report_allocation_drafts`, `allocation_summary_text`, `refresh_report_allocations`. + - API fields: `allocation_summary_text`, `allocations`, `source_report_date`, `source_report_dates`.