niuniu/JhHardwareWRS
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
25ed4e2da8
JhHardwareWRS/docs/superpowers/plans/2026-07-25-report-time-allocation.md
souplearn 5f6deee796 docs: 增加报工归属拆分执行计划
2026-07-25 02:06:21 +08:00

63 KiB
Raw Blame History

报工跨日期工时归属拆分 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:

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:

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:

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:

        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:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
pytest tests/test_metrics.py -v

Expected: all test_metrics.py tests pass.

  • Step 5: Commit
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:

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:

# 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:

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:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
python - <<'PY'
from app.models import ProductionReportAllocation
print(ProductionReportAllocation.__tablename__)
PY

Expected:

production_report_allocations
  • Step 4: Commit
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:

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
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:

@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:

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:

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
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
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:

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:

from app.services.report_allocations import refresh_report_allocations

In submit_report, after db.add(report) and before db.commit(), add:

    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:

    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:

from app.services.report_allocations import refresh_report_allocations

In auto_submit_session, after db.add(report), add:

    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:

from app.services.report_allocations import refresh_report_allocations

In approve_report, after _recalculate_report_metrics(report, schedule) and before final db.commit(), add:

    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:

        db.execute(delete(ProductionReportAllocation).where(ProductionReportAllocation.report_id == report.id))
  • Step 7: Run lifecycle-adjacent tests
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
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:

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:

    allocations: list[ReportAllocationOut] = Field(default_factory=list)

Add to ReportOut:

    allocation_summary_text: str = ""

Add to DashboardRow:

    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:

from app.schemas import ReportAllocationOut
from app.services.report_allocations import allocation_summary_text

Add helper:

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:

        allocations=[report_allocation_out(row) for row in getattr(item, "allocations", [])],

Update report_out to set:

        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:

selectinload(ProductionReport.allocations)
selectinload(ProductionReport.items).selectinload(ProductionReportItem.allocations)

Add ProductionReportItem import where needed.

  • Step 4: Smoke test response serialization

Run:

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
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.

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:

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:

"报工日期",

to:

"统计归属日期",
"原始报工日期",

When appending rows, write:

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
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
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:

extract("month", ProductionReportAllocation.allocation_date).label("month")

The selected quantity must be:

func.sum(ProductionReportAllocation.good_qty).label("good_qty")

The year filter must be:

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:

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:

产品A最后工序为2序；
同一原始报工拆到2026-07和2026-08；
7月对账小账本只显示7月 allocation.good_qty；
8月只显示8月 allocation.good_qty。

Use exact expected values:

assert totals[("嘉恒", "产品A", 7)] == 350
assert totals[("嘉恒", "产品A", 8)] == 300
  • Step 4: Run reconciliation tests
cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
pytest tests/test_reconciliation_allocations.py -v

Expected: tests pass.

  • Step 5: Commit
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:

    report_ids: set[int] = field(default_factory=set, repr=False)

Change UsageStatAccumulator.add signature:

        report_id: int | None = None,

Replace:

        row.report_count += 1

with:

        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:

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:

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:

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
cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
pytest tests/test_usage_stats.py -v

Expected: all tests pass.

  • Step 6: Commit
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:

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:

    allocations: (row.allocations || []).map(toCamelAllocation),

In toCamelReport, add:

    allocationSummaryText: row.allocation_summary_text || '',

In dashboard row mapper, map:

    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:

<text wx:if="{{item.allocationSummaryText}}" class="allocation-summary">归属拆分：{{item.allocationSummaryText}}</text>
  • Step 3: Show source date in dashboard

In /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml, below the row title date, add:

<text wx:if="{{item.sourceReportDates && item.sourceReportDates.length}}" class="allocation-summary">原始报工日期 {{item.sourceReportDates.join('、')}}</text>
  • Step 4: Add compact styles

In both WXSS files, add:

.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:

管理员审核页：跨日期记录显示“归属拆分：...”
经理看板：统计日期仍在主标题；跨日期聚合行额外显示原始报工日期
旧记录没有 allocation_summary_text 时页面不显示空行
  • Step 6: Commit
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:

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:

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:

from sqlalchemy import select, text

Update main:

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
cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
python scripts/migrate_report_allocations.py

Expected:

report allocations migrated, processed=<number>, skipped=0
  • Step 3: Verify table and indexes
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:

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:

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:

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
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

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
pytest -v

Expected: all backend tests pass.

  • Step 2: Run targeted import smoke
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:

ok
  • Step 3: Manual app verification

Use local backend and WeChat DevTools:

1. 创建或找一条 00:10-11:00 的测试报工。
2. 管理员审核通过。
3. 管理员审核页显示归属拆分摘要。
4. 经理看板按前一天和当天拆成统计行。
5. 经理导出 Excel 有“统计归属日期”和“原始报工日期”。
6. 对账小账本按统计归属月份显示最后工序成品数量。
7. 设备/模具统计按统计归属日期过滤，报工次数不重复放大。
8. 作废该报工后，看板、对账、使用统计不再计入。
9. 撤销作废后，看板、对账、使用统计恢复。
  • Step 4: Final git status
cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git status --short
cd /Users/souplearn/Gitlab/app/JhHardwareWRS
git status --short

Expected:

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.