souplearn 25ed4e2da8 docs: 增加报工超额快照执行计划

2026-07-25 19:45:08 +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: 在报工提交和管理员修正时，对产品工序数量是否超过材料/BOM/前序可用量进行校验，并把校验结果按当时状态固化为快照，供审核、记录、看板和导出展示。

Architecture: 后端新增“超额快照”字段和 report_over_limit 服务，所有判断都落在报工明细层，报工主表只保存汇总标识用于列表快速展示。员工提交、清洗外普通报工、系统自动提交后的管理员补填、管理员审核修改后统一刷新超额快照；后续 ERP 追加生产出库不会重算历史快照。

Tech Stack: FastAPI, SQLAlchemy 2, MySQL, Pytest, openpyxl, 微信小程序原生 WXML/WXSS/JS。

Scope

实现内容：

第一道数字工序按提交时 ERP 批次累计出库重量和产品毛重校验最大可报工数量。
后续数字工序按提交时前一道数字工序累计成品数量校验当前工序累计消耗数量。
当前工序消耗数量 = 成品数量 + 不良数量 + 报废数量。
前序可用数量 = 前序成品数量。
超额状态按提交/管理员修正时固化，不随 ERP 后续补料动态改变。
无法校验与超额分开：无法校验只提醒，不按超额标红。
报工卡片、审核页、记录页、经理看板、经理看板导出展示超额/无法校验信息。
数据库迁移脚本直接执行并用 information_schema 或目标查询验证。

不实现内容：

不阻止员工提交。
不把 ERP 后续出库反向分配到历史报工。
不校验清洗、杂活、非数字工序。
不新建 ERP 投料页面，第一版直接读取现有 pp_production_batch_ledger.total_issued_weight_kg。

Business Rules

工序范围

仅普通产品数字工序参与校验。
process_name 需能解析为正整数，例如 1、1序 均按 1 处理。
杂活、清洗冲压方式、非数字工序直接跳过。
多人工序继续参与，因为它是普通产品工序。

第一道工序

最大可报工数量 = 提交时批次累计生产出库重量 / 产品毛重
本次消耗数量 = 成品 + 不良 + 报废
当前累计消耗 = 历史未作废且已提交的同考勤点/同项目/同产品/同批次/同工序消耗 + 本次消耗

若 当前累计消耗 > 最大可报工数量，则本明细超额。

后续工序

本次消耗数量 = 成品 + 不良 + 报废
当前累计消耗 = 历史未作废且已提交的同考勤点/同项目/同产品/同批次/当前工序消耗 + 本次消耗
前序可用数量 = 历史未作废且已提交的同考勤点/同项目/同产品/同批次/前一道数字工序成品数量

若 当前累计消耗 > 前序可用数量，则本明细超额。

前一道工序按产品清单中同考勤点、同项目、同产品、同批次无关的数字工序序列查找；如果只有 1序、3序，则 3序 的前序是 1序。

统计范围

参与累计的报工：

未作废 + 待审核 + 审核通过

不参与累计的报工：

已作废
已拒绝
未提交的上班班次

当前正在保存的报工需要纳入累计，但不能重复计算自身。实现时以 exclude_report_id 排除本报工历史行，再把本次明细加到累计中。

快照固化

员工提交时报工明细保存 over_limit_check_source = "submit"。
管理员审核页修改关键字段后，保存 over_limit_check_source = "review"。
ERP 后续补料不触发历史快照重算。
作废/撤销作废不修改该报工自身快照，但未来新报工累计时会排除作废记录。

File Map

Backend repo root: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint

Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/models.py
- ProductionReport 增加 has_over_limit、over_limit_summary。
- ProductionReportItem 增加明细级快照字段。
Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/schemas.py
- ReportItemOut、ReportOut、DashboardRow 增加超额输出字段。
Create /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_over_limit.py
- 解析工序、读取 ERP 批次重量、计算累计、刷新快照。
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
- 管理员审核/修正后刷新超额快照。
Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/auto_submit.py
- 系统自动提交生成空报工后不强判超额；管理员补填审核时刷新。
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/scripts/migrate_report_over_limit_snapshot.py
- MySQL 补字段、索引并验证。
Create /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py
- 服务级、提交、审核刷新测试。
Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_dashboard_allocations.py
- 经理看板超额字段测试。

Frontend repo root: /Users/souplearn/Gitlab/app/JhHardwareWRS

Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js
- 映射超额字段。
Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.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/records/records.wxml
- 员工记录页水印和明细提示。
Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/records/records.wxss
- 记录页样式。
Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml
- 经理看板水印和原因。
Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxss
- 看板样式。

Task 1: Add Database Fields And Migration

Files:

Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/models.py
Create: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_over_limit_snapshot.py
Test: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py
Step 1: Write failing schema test

Create /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py with this initial content:

from datetime import datetime

from sqlalchemy import BigInteger, create_engine, inspect
from sqlalchemy.ext.compiler import compiles

from app.database import Base


@compiles(BigInteger, "sqlite")
def _compile_big_integer_for_sqlite(type_, compiler, **kw):
    return "INTEGER"


def test_report_over_limit_columns_exist_on_sqlite_metadata():
    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
    Base.metadata.create_all(engine)
    inspector = inspect(engine)

    report_columns = {column["name"] for column in inspector.get_columns("production_reports")}
    item_columns = {column["name"] for column in inspector.get_columns("production_report_items")}

    assert {"has_over_limit", "over_limit_summary"}.issubset(report_columns)
    assert {
        "over_limit_status",
        "over_limit_type",
        "over_limit_reason",
        "over_limit_checked_at",
        "over_limit_check_source",
        "over_limit_batch_no",
        "over_limit_product_gross_weight_kg",
        "over_limit_issued_weight_kg",
        "over_limit_max_reportable_qty",
        "over_limit_previous_good_qty",
        "over_limit_current_cumulative_qty",
        "over_limit_current_report_qty",
    }.issubset(item_columns)

Step 2: Run schema test and verify failure

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_report_over_limit_columns_exist_on_sqlite_metadata -v

Expected: FAIL because the new columns do not exist.

Step 3: Add model fields

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/models.py, add these fields to ProductionReport after auto_submit_reason:

    has_over_limit: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    over_limit_summary: Mapped[str | None] = mapped_column(String(500))

Add these fields to ProductionReportItem after remark:

    over_limit_status: Mapped[str] = mapped_column(String(32), nullable=False, default="not_checked")
    over_limit_type: Mapped[str | None] = mapped_column(String(64))
    over_limit_reason: Mapped[str | None] = mapped_column(String(500))
    over_limit_checked_at: Mapped[datetime | None] = mapped_column(DateTime)
    over_limit_check_source: Mapped[str | None] = mapped_column(String(32))
    over_limit_batch_no: Mapped[str | None] = mapped_column(String(128))
    over_limit_product_gross_weight_kg: Mapped[float | None] = mapped_column(Numeric(12, 4))
    over_limit_issued_weight_kg: Mapped[float | None] = mapped_column(Numeric(18, 6))
    over_limit_max_reportable_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))
    over_limit_previous_good_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))
    over_limit_current_cumulative_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))
    over_limit_current_report_qty: Mapped[float | None] = mapped_column(Numeric(12, 2))

Add indexes to ProductionReport.__table_args__:

        Index("idx_reports_over_limit", "has_over_limit", "submitted_at"),

Add indexes to ProductionReportItem.__table_args__:

        Index("idx_report_items_over_limit_status", "over_limit_status"),
        Index(
            "idx_report_items_process_batch",
            "attendance_point_name",
            "project_no",
            "product_name",
            "raw_material_batch_no",
            "process_name",
        ),

Step 4: Run schema test and verify pass

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_report_over_limit_columns_exist_on_sqlite_metadata -v

Expected: PASS.

Step 5: Create MySQL migration script

Create /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_over_limit_snapshot.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


REPORT_COLUMNS = {
    "has_over_limit": "ADD COLUMN has_over_limit TINYINT(1) NOT NULL DEFAULT 0 AFTER auto_submit_reason",
    "over_limit_summary": "ADD COLUMN over_limit_summary VARCHAR(500) NULL AFTER has_over_limit",
}

ITEM_COLUMNS = {
    "over_limit_status": "ADD COLUMN over_limit_status VARCHAR(32) NOT NULL DEFAULT 'not_checked' AFTER remark",
    "over_limit_type": "ADD COLUMN over_limit_type VARCHAR(64) NULL AFTER over_limit_status",
    "over_limit_reason": "ADD COLUMN over_limit_reason VARCHAR(500) NULL AFTER over_limit_type",
    "over_limit_checked_at": "ADD COLUMN over_limit_checked_at DATETIME NULL AFTER over_limit_reason",
    "over_limit_check_source": "ADD COLUMN over_limit_check_source VARCHAR(32) NULL AFTER over_limit_checked_at",
    "over_limit_batch_no": "ADD COLUMN over_limit_batch_no VARCHAR(128) NULL AFTER over_limit_check_source",
    "over_limit_product_gross_weight_kg": "ADD COLUMN over_limit_product_gross_weight_kg DECIMAL(12,4) NULL AFTER over_limit_batch_no",
    "over_limit_issued_weight_kg": "ADD COLUMN over_limit_issued_weight_kg DECIMAL(18,6) NULL AFTER over_limit_product_gross_weight_kg",
    "over_limit_max_reportable_qty": "ADD COLUMN over_limit_max_reportable_qty DECIMAL(12,2) NULL AFTER over_limit_issued_weight_kg",
    "over_limit_previous_good_qty": "ADD COLUMN over_limit_previous_good_qty DECIMAL(12,2) NULL AFTER over_limit_max_reportable_qty",
    "over_limit_current_cumulative_qty": "ADD COLUMN over_limit_current_cumulative_qty DECIMAL(12,2) NULL AFTER over_limit_previous_good_qty",
    "over_limit_current_report_qty": "ADD COLUMN over_limit_current_report_qty DECIMAL(12,2) NULL AFTER over_limit_current_cumulative_qty",
}

INDEXES = {
    "production_reports": {
        "idx_reports_over_limit": "CREATE INDEX idx_reports_over_limit ON production_reports (has_over_limit, submitted_at)",
    },
    "production_report_items": {
        "idx_report_items_over_limit_status": "CREATE INDEX idx_report_items_over_limit_status ON production_report_items (over_limit_status)",
        "idx_report_items_process_batch": (
            "CREATE INDEX idx_report_items_process_batch "
            "ON production_report_items (attendance_point_name, project_no, product_name, raw_material_batch_no, process_name)"
        ),
    },
}


def column_exists(conn, table_name: str, column_name: str) -> bool:
    return bool(
        conn.execute(
            text(
                """
                SELECT COUNT(*)
                FROM information_schema.columns
                WHERE table_schema = DATABASE()
                  AND table_name = :table_name
                  AND column_name = :column_name
                """
            ),
            {"table_name": table_name, "column_name": column_name},
        ).scalar_one()
    )


def index_exists(conn, table_name: str, index_name: str) -> bool:
    return bool(
        conn.execute(
            text(
                """
                SELECT COUNT(*)
                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_one()
    )


def ensure_columns(conn) -> None:
    for column_name, ddl in REPORT_COLUMNS.items():
        if not column_exists(conn, "production_reports", column_name):
            conn.execute(text(f"ALTER TABLE production_reports {ddl}"))
    for column_name, ddl in ITEM_COLUMNS.items():
        if not column_exists(conn, "production_report_items", column_name):
            conn.execute(text(f"ALTER TABLE production_report_items {ddl}"))


def ensure_indexes(conn) -> None:
    for table_name, indexes in INDEXES.items():
        for index_name, ddl in indexes.items():
            if not index_exists(conn, table_name, index_name):
                conn.execute(text(ddl))


def verify(conn) -> None:
    missing = []
    for column_name in REPORT_COLUMNS:
        if not column_exists(conn, "production_reports", column_name):
            missing.append(f"production_reports.{column_name}")
    for column_name in ITEM_COLUMNS:
        if not column_exists(conn, "production_report_items", column_name):
            missing.append(f"production_report_items.{column_name}")
    for table_name, indexes in INDEXES.items():
        for index_name in indexes:
            if not index_exists(conn, table_name, index_name):
                missing.append(f"{table_name}.{index_name}")
    if missing:
        raise RuntimeError("report over limit migration missing objects: " + ", ".join(missing))


def main() -> None:
    with engine.begin() as conn:
        ensure_columns(conn)
        ensure_indexes(conn)
        verify(conn)
    print("report over limit snapshot migrated")


if __name__ == "__main__":
    main()

Step 6: Compile migration script

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m py_compile scripts/migrate_report_over_limit_snapshot.py

Expected: exit code 0.

Step 7: Commit

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git add app/models.py scripts/migrate_report_over_limit_snapshot.py tests/test_report_over_limit.py
git commit -m "feat: 增加报工超额快照字段"

Expected: commit succeeds.

Task 2: Implement Over-Limit Calculation Service

Files:

Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py
Create: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_over_limit.py
Step 1: Add service tests for process parsing and current quantity

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py:

from decimal import Decimal
from types import SimpleNamespace

from app.services.report_over_limit import (
    OVER_LIMIT_STATUS_EXCEEDED,
    OVER_LIMIT_STATUS_NOT_CHECKED,
    OVER_LIMIT_STATUS_OK,
    OVER_LIMIT_STATUS_UNCHECKABLE,
    current_report_qty,
    parse_process_order,
)


def test_parse_process_order_accepts_numeric_process_names():
    assert parse_process_order("1") == 1
    assert parse_process_order("1序") == 1
    assert parse_process_order(" 02 序 ") == 2


def test_parse_process_order_rejects_special_and_text_process_names():
    assert parse_process_order("杂活") is None
    assert parse_process_order("清洗") is None
    assert parse_process_order("第一序") is None


def test_current_report_qty_uses_good_defect_and_scrap():
    item = SimpleNamespace(good_qty=Decimal("90"), defect_qty=Decimal("5"), scrap_qty=Decimal("5"))

    assert current_report_qty(item) == 100

Step 2: Run tests and verify failure

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_parse_process_order_accepts_numeric_process_names tests/test_report_over_limit.py::test_parse_process_order_rejects_special_and_text_process_names tests/test_report_over_limit.py::test_current_report_qty_uses_good_defect_and_scrap -v

Expected: FAIL because app.services.report_over_limit does not exist.

Step 3: Create basic service helpers

Create /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_over_limit.py:

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
import re
from typing import Any

from sqlalchemy import func, select, text
from sqlalchemy.orm import Session

from app.models import Product, ProductionReport, ProductionReportItem, ReportStatus
from app.services.common import as_float
from app.services.product_rules import is_cleaning_item, is_misc_item


OVER_LIMIT_STATUS_NOT_CHECKED = "not_checked"
OVER_LIMIT_STATUS_OK = "ok"
OVER_LIMIT_STATUS_EXCEEDED = "exceeded"
OVER_LIMIT_STATUS_UNCHECKABLE = "uncheckable"

OVER_LIMIT_TYPE_FIRST_PROCESS_MATERIAL = "first_process_material"
OVER_LIMIT_TYPE_PREVIOUS_PROCESS = "previous_process"
OVER_LIMIT_TYPE_UNCHECKABLE = "uncheckable"

CHECK_SOURCE_SUBMIT = "submit"
CHECK_SOURCE_REVIEW = "review"

PROCESS_RE = re.compile(r"^\\s*(\\d+)\\s*(?:序)?\\s*$")
QUANT_2 = Decimal("0.01")
QUANT_4 = Decimal("0.0001")
QUANT_6 = Decimal("0.000001")


@dataclass(frozen=True)
class OverLimitResult:
    status: str
    over_limit_type: str | None
    reason: str | None
    batch_no: str | None
    product_gross_weight_kg: Decimal | None
    issued_weight_kg: Decimal | None
    max_reportable_qty: Decimal | None
    previous_good_qty: Decimal | None
    current_cumulative_qty: Decimal | None
    current_report_qty: Decimal


def decimal_value(value: Any) -> Decimal:
    if value is None:
        return Decimal("0")
    if isinstance(value, Decimal):
        return value
    try:
        return Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        return Decimal("0")


def round2(value: Any) -> Decimal:
    return decimal_value(value).quantize(QUANT_2, rounding=ROUND_HALF_UP)


def round4(value: Any) -> Decimal:
    return decimal_value(value).quantize(QUANT_4, rounding=ROUND_HALF_UP)


def round6(value: Any) -> Decimal:
    return decimal_value(value).quantize(QUANT_6, rounding=ROUND_HALF_UP)


def parse_process_order(process_name: str | None) -> int | None:
    match = PROCESS_RE.match(str(process_name or ""))
    if not match:
        return None
    value = int(match.group(1))
    return value if value > 0 else None


def current_report_qty(item: ProductionReportItem) -> Decimal:
    return round2(
        decimal_value(getattr(item, "good_qty", 0))
        + decimal_value(getattr(item, "defect_qty", 0))
        + decimal_value(getattr(item, "scrap_qty", 0))
    )

Step 4: Run tests and verify pass

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_parse_process_order_accepts_numeric_process_names tests/test_report_over_limit.py::test_parse_process_order_rejects_special_and_text_process_names tests/test_report_over_limit.py::test_current_report_qty_uses_good_defect_and_scrap -v

Expected: PASS.

Step 5: Add tests for first-process material limit

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py:

from sqlalchemy.orm import sessionmaker

from app.models import Product, ProductionReport, ProductionReportItem, ReportStatus
from app.services.report_over_limit import evaluate_report_item_over_limit


def _sqlite_db():
    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine, future=True)
    return SessionLocal()


def test_first_process_exceeds_material_limit_with_submit_snapshot(monkeypatch):
    db = _sqlite_db()
    try:
        product = Product(
            attendance_point_name="总厂",
            project_no="P1",
            product_name="产品A",
            device_no="",
            process_name="1",
            product_gross_weight_kg=Decimal("2"),
            standard_beat=1,
        )
        report = ProductionReport(
            id=1,
            session_id=1,
            attendance_point_name="总厂",
            employee_phone="13800000000",
            report_date=datetime(2026, 7, 25).date(),
            start_at=datetime(2026, 7, 25, 8),
            end_at=datetime(2026, 7, 25, 10),
            status=ReportStatus.pending,
        )
        item = ProductionReportItem(
            id=1,
            report=report,
            attendance_point_name="总厂",
            device_no="A",
            project_no="P1",
            product_name="产品A",
            raw_material_batch_no="LOT-1",
            process_name="1",
            good_qty=60,
            defect_qty=0,
            scrap_qty=0,
            standard_beat=1,
        )
        db.add_all([product, report, item])
        db.flush()
        monkeypatch.setattr(
            "app.services.report_over_limit.read_batch_issued_weight_kg",
            lambda db, batch_no, project_no, product_name: Decimal("100"),
        )

        result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)

        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
        assert result.max_reportable_qty == Decimal("50.00")
        assert result.current_report_qty == Decimal("60.00")
        assert result.current_cumulative_qty == Decimal("60.00")
        assert "1序累计报工数量超过本批次材料可支撑数量" in result.reason
    finally:
        db.close()

Step 6: Run first-process test and verify failure

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_first_process_exceeds_material_limit_with_submit_snapshot -v

Expected: FAIL because evaluate_report_item_over_limit is not implemented.

Step 7: Implement ERP weight lookup and first-process logic

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_over_limit.py:

def read_batch_issued_weight_kg(
    db: Session,
    batch_no: str | None,
    project_no: str | None,
    product_name: str | None,
) -> Decimal | None:
    normalized_batch_no = str(batch_no or "").strip()
    if not normalized_batch_no:
        return None
    row = db.execute(
        text(
            """
            SELECT total_issued_weight_kg
            FROM pp_production_batch_ledger
            WHERE material_lot_no = :batch_no
              AND COALESCE(miniapp_selectable, 1) = 1
            ORDER BY updated_at DESC, id DESC
            LIMIT 1
            """
        ),
        {"batch_no": normalized_batch_no},
    ).first()
    if row is None or row[0] is None:
        return None
    return round6(row[0])


def _product_for_item(db: Session, report: ProductionReport, item: ProductionReportItem) -> Product | None:
    return db.scalar(
        select(Product).where(
            Product.attendance_point_name == report.attendance_point_name,
            Product.project_no == item.project_no,
            Product.product_name == item.product_name,
            Product.process_name == (item.process_name or ""),
            Product.device_no == "",
        )
    )


def _all_product_process_orders(db: Session, report: ProductionReport, item: ProductionReportItem) -> list[int]:
    rows = db.execute(
        select(Product.process_name).where(
            Product.attendance_point_name == report.attendance_point_name,
            Product.project_no == item.project_no,
            Product.product_name == item.product_name,
            Product.device_no == "",
        )
    ).all()
    orders = sorted({order for row in rows if (order := parse_process_order(row[0])) is not None})
    return orders


def _previous_process_order(db: Session, report: ProductionReport, item: ProductionReportItem, current_order: int) -> int | None:
    previous_orders = [order for order in _all_product_process_orders(db, report, item) if order < current_order]
    return previous_orders[-1] if previous_orders else None


def _item_scope_filter(report: ProductionReport, item: ProductionReportItem, process_order: int):
    process_name_texts = {str(process_order), f"{process_order}序"}
    return (
        ProductionReportItem.attendance_point_name == report.attendance_point_name,
        ProductionReportItem.project_no == item.project_no,
        ProductionReportItem.product_name == item.product_name,
        ProductionReportItem.raw_material_batch_no == item.raw_material_batch_no,
        ProductionReportItem.process_name.in_(process_name_texts),
    )


def _submitted_unvoided_filter(exclude_report_id: int | None):
    conditions = (
        ProductionReportItem.report_id == ProductionReport.id,
        ProductionReport.is_voided.is_(False),
        ProductionReport.status.in_([ReportStatus.pending, ReportStatus.approved]),
    )
    if exclude_report_id is not None:
        conditions = (*conditions, ProductionReport.id != exclude_report_id)
    return conditions


def _historical_consumed_qty(
    db: Session,
    report: ProductionReport,
    item: ProductionReportItem,
    process_order: int,
    exclude_report_id: int | None,
) -> Decimal:
    value = db.scalar(
        select(
            func.coalesce(
                func.sum(
                    ProductionReportItem.good_qty
                    + ProductionReportItem.defect_qty
                    + ProductionReportItem.scrap_qty
                ),
                0,
            )
        ).where(
            *_item_scope_filter(report, item, process_order),
            *_submitted_unvoided_filter(exclude_report_id),
        )
    )
    return round2(value)


def _historical_good_qty(
    db: Session,
    report: ProductionReport,
    item: ProductionReportItem,
    process_order: int,
    exclude_report_id: int | None,
) -> Decimal:
    value = db.scalar(
        select(func.coalesce(func.sum(ProductionReportItem.good_qty), 0)).where(
            *_item_scope_filter(report, item, process_order),
            *_submitted_unvoided_filter(exclude_report_id),
        )
    )
    return round2(value)


def _unchecked(reason: str, item: ProductionReportItem) -> OverLimitResult:
    return OverLimitResult(
        status=OVER_LIMIT_STATUS_UNCHECKABLE,
        over_limit_type=OVER_LIMIT_TYPE_UNCHECKABLE,
        reason=reason,
        batch_no=getattr(item, "raw_material_batch_no", None),
        product_gross_weight_kg=None,
        issued_weight_kg=None,
        max_reportable_qty=None,
        previous_good_qty=None,
        current_cumulative_qty=None,
        current_report_qty=current_report_qty(item),
    )


def evaluate_report_item_over_limit(
    db: Session,
    report: ProductionReport,
    item: ProductionReportItem,
    *,
    exclude_report_id: int | None,
) -> OverLimitResult:
    if is_cleaning_item(item) or is_misc_item(item):
        return OverLimitResult(
            status=OVER_LIMIT_STATUS_NOT_CHECKED,
            over_limit_type=None,
            reason=None,
            batch_no=item.raw_material_batch_no,
            product_gross_weight_kg=None,
            issued_weight_kg=None,
            max_reportable_qty=None,
            previous_good_qty=None,
            current_cumulative_qty=None,
            current_report_qty=current_report_qty(item),
        )

    current_order = parse_process_order(item.process_name)
    if current_order is None:
        return _unchecked("非数字工序，无法校验超额", item)
    if not str(item.raw_material_batch_no or "").strip():
        return _unchecked("缺少材料库存批次号，无法校验超额", item)

    current_qty = current_report_qty(item)
    historical_qty = _historical_consumed_qty(db, report, item, current_order, exclude_report_id)
    current_cumulative_qty = round2(historical_qty + current_qty)
    previous_order = _previous_process_order(db, report, item, current_order)

    if previous_order is None:
        product = _product_for_item(db, report, item)
        gross_weight = round4(getattr(product, "product_gross_weight_kg", None)) if product else Decimal("0")
        if gross_weight <= 0:
            return _unchecked("缺少产品毛重，无法校验第一道工序超额", item)
        issued_weight = read_batch_issued_weight_kg(db, item.raw_material_batch_no, item.project_no, item.product_name)
        if issued_weight is None:
            return _unchecked("缺少批次投入重量，无法校验第一道工序超额", item)
        max_qty = round2(issued_weight / gross_weight)
        exceeded = current_cumulative_qty > max_qty
        return OverLimitResult(
            status=OVER_LIMIT_STATUS_EXCEEDED if exceeded else OVER_LIMIT_STATUS_OK,
            over_limit_type=OVER_LIMIT_TYPE_FIRST_PROCESS_MATERIAL if exceeded else None,
            reason=(
                f"{current_order}序累计报工数量超过本批次材料可支撑数量"
                if exceeded
                else None
            ),
            batch_no=item.raw_material_batch_no,
            product_gross_weight_kg=gross_weight,
            issued_weight_kg=issued_weight,
            max_reportable_qty=max_qty,
            previous_good_qty=None,
            current_cumulative_qty=current_cumulative_qty,
            current_report_qty=current_qty,
        )

    previous_good_qty = _historical_good_qty(db, report, item, previous_order, exclude_report_id)
    exceeded = current_cumulative_qty > previous_good_qty
    return OverLimitResult(
        status=OVER_LIMIT_STATUS_EXCEEDED if exceeded else OVER_LIMIT_STATUS_OK,
        over_limit_type=OVER_LIMIT_TYPE_PREVIOUS_PROCESS if exceeded else None,
        reason=(
            f"{current_order}序累计报工数量超过{previous_order}序累计成品数量"
            if exceeded
            else None
        ),
        batch_no=item.raw_material_batch_no,
        product_gross_weight_kg=None,
        issued_weight_kg=None,
        max_reportable_qty=None,
        previous_good_qty=previous_good_qty,
        current_cumulative_qty=current_cumulative_qty,
        current_report_qty=current_qty,
    )

Step 8: Run first-process test and verify pass

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_first_process_exceeds_material_limit_with_submit_snapshot -v

Expected: PASS.

Step 9: Add tests for previous-process rule

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py:

def test_later_process_exceeds_previous_good_quantity():
    db = _sqlite_db()
    try:
        db.add_all(
            [
                Product(attendance_point_name="总厂", project_no="P1", product_name="产品A", device_no="", process_name="1", standard_beat=1),
                Product(attendance_point_name="总厂", project_no="P1", product_name="产品A", device_no="", process_name="3", standard_beat=1),
            ]
        )
        previous_report = ProductionReport(
            id=1,
            session_id=1,
            attendance_point_name="总厂",
            employee_phone="13800000000",
            report_date=datetime(2026, 7, 25).date(),
            start_at=datetime(2026, 7, 25, 8),
            end_at=datetime(2026, 7, 25, 10),
            status=ReportStatus.approved,
        )
        previous_item = ProductionReportItem(
            id=1,
            report=previous_report,
            attendance_point_name="总厂",
            device_no="A",
            project_no="P1",
            product_name="产品A",
            raw_material_batch_no="LOT-1",
            process_name="1",
            good_qty=100,
            defect_qty=10,
            scrap_qty=5,
            standard_beat=1,
        )
        report = ProductionReport(
            id=2,
            session_id=2,
            attendance_point_name="总厂",
            employee_phone="13800000001",
            report_date=datetime(2026, 7, 25).date(),
            start_at=datetime(2026, 7, 25, 11),
            end_at=datetime(2026, 7, 25, 12),
            status=ReportStatus.pending,
        )
        item = ProductionReportItem(
            id=2,
            report=report,
            attendance_point_name="总厂",
            device_no="B",
            project_no="P1",
            product_name="产品A",
            raw_material_batch_no="LOT-1",
            process_name="3",
            good_qty=98,
            defect_qty=3,
            scrap_qty=2,
            standard_beat=1,
        )
        db.add_all([previous_report, previous_item, report, item])
        db.flush()

        result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=report.id)

        assert result.status == OVER_LIMIT_STATUS_EXCEEDED
        assert result.previous_good_qty == Decimal("100.00")
        assert result.current_report_qty == Decimal("103.00")
        assert result.current_cumulative_qty == Decimal("103.00")
        assert "3序累计报工数量超过1序累计成品数量" in result.reason
    finally:
        db.close()

Step 10: Run previous-process test

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_later_process_exceeds_previous_good_quantity -v

Expected: PASS.

Step 11: Commit

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git add app/services/report_over_limit.py tests/test_report_over_limit.py
git commit -m "feat: 增加报工超额校验服务"

Expected: commit succeeds.

Task 3: Refresh Snapshots On Submit And Review

Files:

Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_over_limit.py
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reviews.py
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py
Step 1: Add refresh snapshot test

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py:

from app.services.report_over_limit import refresh_report_over_limit_snapshot


def test_refresh_report_over_limit_snapshot_writes_item_and_report_summary(monkeypatch):
    db = _sqlite_db()
    try:
        product = Product(
            attendance_point_name="总厂",
            project_no="P1",
            product_name="产品A",
            device_no="",
            process_name="1",
            product_gross_weight_kg=Decimal("2"),
            standard_beat=1,
        )
        report = ProductionReport(
            id=1,
            session_id=1,
            attendance_point_name="总厂",
            employee_phone="13800000000",
            report_date=datetime(2026, 7, 25).date(),
            start_at=datetime(2026, 7, 25, 8),
            end_at=datetime(2026, 7, 25, 10),
            status=ReportStatus.pending,
        )
        item = ProductionReportItem(
            id=1,
            report=report,
            attendance_point_name="总厂",
            device_no="A",
            project_no="P1",
            product_name="产品A",
            raw_material_batch_no="LOT-1",
            process_name="1",
            good_qty=60,
            defect_qty=0,
            scrap_qty=0,
            standard_beat=1,
        )
        db.add_all([product, report, item])
        db.flush()
        monkeypatch.setattr(
            "app.services.report_over_limit.read_batch_issued_weight_kg",
            lambda db, batch_no, project_no, product_name: Decimal("100"),
        )

        refresh_report_over_limit_snapshot(db, report, source="submit")

        assert report.has_over_limit is True
        assert report.over_limit_summary == "1项超额"
        assert item.over_limit_status == OVER_LIMIT_STATUS_EXCEEDED
        assert item.over_limit_check_source == "submit"
        assert item.over_limit_batch_no == "LOT-1"
        assert item.over_limit_max_reportable_qty == Decimal("50.00")
    finally:
        db.close()

Step 2: Run refresh test and verify failure

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_refresh_report_over_limit_snapshot_writes_item_and_report_summary -v

Expected: FAIL because refresh_report_over_limit_snapshot is not implemented.

Step 3: Implement refresh function

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/report_over_limit.py:

def _apply_result_to_item(item: ProductionReportItem, result: OverLimitResult, *, source: str, checked_at: datetime) -> None:
    item.over_limit_status = result.status
    item.over_limit_type = result.over_limit_type
    item.over_limit_reason = result.reason
    item.over_limit_checked_at = checked_at
    item.over_limit_check_source = source
    item.over_limit_batch_no = result.batch_no
    item.over_limit_product_gross_weight_kg = result.product_gross_weight_kg
    item.over_limit_issued_weight_kg = result.issued_weight_kg
    item.over_limit_max_reportable_qty = result.max_reportable_qty
    item.over_limit_previous_good_qty = result.previous_good_qty
    item.over_limit_current_cumulative_qty = result.current_cumulative_qty
    item.over_limit_current_report_qty = result.current_report_qty


def refresh_report_over_limit_snapshot(
    db: Session,
    report: ProductionReport,
    *,
    source: str,
    checked_at: datetime | None = None,
) -> None:
    timestamp = checked_at or datetime.now()
    exceeded_count = 0
    uncheckable_count = 0
    for item in list(getattr(report, "items", []) or []):
        result = evaluate_report_item_over_limit(db, report, item, exclude_report_id=getattr(report, "id", None))
        _apply_result_to_item(item, result, source=source, checked_at=timestamp)
        if result.status == OVER_LIMIT_STATUS_EXCEEDED:
            exceeded_count += 1
        elif result.status == OVER_LIMIT_STATUS_UNCHECKABLE:
            uncheckable_count += 1

    report.has_over_limit = exceeded_count > 0
    if exceeded_count > 0 and uncheckable_count > 0:
        report.over_limit_summary = f"{exceeded_count}项超额，{uncheckable_count}项无法校验"
    elif exceeded_count > 0:
        report.over_limit_summary = f"{exceeded_count}项超额"
    elif uncheckable_count > 0:
        report.over_limit_summary = f"{uncheckable_count}项无法校验"
    else:
        report.over_limit_summary = None
    db.flush()

Step 4: Run refresh test and verify pass

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_refresh_report_over_limit_snapshot_writes_item_and_report_summary -v

Expected: PASS.

Step 5: Wire submit path

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py, add import:

from app.services.report_over_limit import refresh_report_over_limit_snapshot

After existing refresh_report_allocations(db, report, schedule=schedule, commit=False) in normal submit, add:

    refresh_report_over_limit_snapshot(db, report, source="submit", checked_at=submitted_at)

Do not add this to /cleaning submit because cleaning is excluded.

Step 6: Wire review path

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reviews.py, add import:

from app.services.report_over_limit import refresh_report_over_limit_snapshot

After existing refresh_report_allocations(db, report, schedule=schedule, commit=False) in approve endpoint, add:

    refresh_report_over_limit_snapshot(db, report, source="review", checked_at=now())

Step 7: Run report over-limit tests

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py -v

Expected: all tests pass.

Step 8: Commit

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git add app/services/report_over_limit.py app/routers/reports.py app/routers/reviews.py tests/test_report_over_limit.py
git commit -m "feat: 报工提交和审核刷新超额快照"

Expected: commit succeeds.

Task 4: Expose Snapshot Fields Through API

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/tests/test_report_over_limit.py
Step 1: Add serializer test

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_report_over_limit.py:

from app.services.serializers import report_out


def test_report_out_exposes_over_limit_snapshot_fields():
    report = ProductionReport(
        id=1,
        session_id=1,
        attendance_point_name="总厂",
        employee_phone="13800000000",
        report_date=datetime(2026, 7, 25).date(),
        start_at=datetime(2026, 7, 25, 8),
        end_at=datetime(2026, 7, 25, 10),
        status=ReportStatus.pending,
        has_over_limit=True,
        over_limit_summary="1项超额",
    )
    item = ProductionReportItem(
        id=1,
        report=report,
        attendance_point_name="总厂",
        device_no="A",
        project_no="P1",
        product_name="产品A",
        raw_material_batch_no="LOT-1",
        process_name="1",
        good_qty=60,
        defect_qty=0,
        scrap_qty=0,
        standard_beat=1,
        over_limit_status=OVER_LIMIT_STATUS_EXCEEDED,
        over_limit_type="first_process_material",
        over_limit_reason="1序累计报工数量超过本批次材料可支撑数量",
        over_limit_checked_at=datetime(2026, 7, 25, 10),
        over_limit_check_source="submit",
        over_limit_batch_no="LOT-1",
        over_limit_product_gross_weight_kg=Decimal("2.0000"),
        over_limit_issued_weight_kg=Decimal("100.000000"),
        over_limit_max_reportable_qty=Decimal("50.00"),
        over_limit_current_cumulative_qty=Decimal("60.00"),
        over_limit_current_report_qty=Decimal("60.00"),
    )
    report.items = [item]
    report.allocations = []
    report.audit_logs = []

    output = report_out(report)

    assert output.has_over_limit is True
    assert output.over_limit_summary == "1项超额"
    assert output.items[0].over_limit_status == OVER_LIMIT_STATUS_EXCEEDED
    assert output.items[0].over_limit_reason == "1序累计报工数量超过本批次材料可支撑数量"
    assert output.items[0].over_limit_max_reportable_qty == 50

Step 2: Run serializer test and verify failure

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_report_out_exposes_over_limit_snapshot_fields -v

Expected: FAIL because schema fields are missing.

Step 3: Add schema fields

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/schemas.py, add to ReportItemOut:

    over_limit_status: str = "not_checked"
    over_limit_type: str | None = None
    over_limit_reason: str | None = None
    over_limit_checked_at: datetime | None = None
    over_limit_check_source: str | None = None
    over_limit_batch_no: str | None = None
    over_limit_product_gross_weight_kg: float | None = None
    over_limit_issued_weight_kg: float | None = None
    over_limit_max_reportable_qty: float | None = None
    over_limit_previous_good_qty: float | None = None
    over_limit_current_cumulative_qty: float | None = None
    over_limit_current_report_qty: float | None = None

Add to ReportOut:

    has_over_limit: bool = False
    over_limit_summary: str | None = None

Step 4: Map serializer fields

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/services/serializers.py, add these keyword arguments in report_item_out:

        over_limit_status=item.over_limit_status or "not_checked",
        over_limit_type=item.over_limit_type,
        over_limit_reason=item.over_limit_reason,
        over_limit_checked_at=item.over_limit_checked_at,
        over_limit_check_source=item.over_limit_check_source,
        over_limit_batch_no=item.over_limit_batch_no,
        over_limit_product_gross_weight_kg=optional_float(item.over_limit_product_gross_weight_kg),
        over_limit_issued_weight_kg=optional_float(item.over_limit_issued_weight_kg),
        over_limit_max_reportable_qty=optional_float(item.over_limit_max_reportable_qty),
        over_limit_previous_good_qty=optional_float(item.over_limit_previous_good_qty),
        over_limit_current_cumulative_qty=optional_float(item.over_limit_current_cumulative_qty),
        over_limit_current_report_qty=optional_float(item.over_limit_current_report_qty),

Add these keyword arguments in report_out:

        has_over_limit=bool(report.has_over_limit),
        over_limit_summary=report.over_limit_summary,

Step 5: Run serializer test

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py::test_report_out_exposes_over_limit_snapshot_fields -v

Expected: PASS.

Step 6: Commit

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git add app/schemas.py app/services/serializers.py tests/test_report_over_limit.py
git commit -m "feat: 输出报工超额快照字段"

Expected: commit succeeds.

Task 5: Add Dashboard And Export Fields

Files:

Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/schemas.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/tests/test_dashboard_allocations.py
Step 1: Add dashboard test

Append to /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_dashboard_allocations.py:

def test_dashboard_rows_include_over_limit_summary_from_reports():
    report = SimpleNamespace(
        id=1,
        attendance_point_name="总厂",
        employee_phone="13800000000",
        report_date=date(2026, 7, 25),
        has_over_limit=True,
        over_limit_summary="1项超额",
        review_remark=None,
        employee=SimpleNamespace(name="张三", is_temporary=False),
        items=[],
    )
    item = SimpleNamespace(
        id=1,
        report=report,
        report_id=1,
        attendance_point_name="总厂",
        device_no="A",
        project_no="P1",
        product_name="产品A",
        material_code="M1",
        material_name="材料A",
        raw_material_batch_no="LOT-1",
        process_name="1",
        stamping_method="冲压",
        operator_count=1,
        process_unit_price_yuan=1,
        standard_beat=1,
        good_qty=60,
        defect_qty=0,
        scrap_qty=0,
        over_limit_status="exceeded",
        over_limit_reason="1序累计报工数量超过本批次材料可支撑数量",
    )
    report.items = [item]
    allocation = SimpleNamespace(
        report=report,
        item=item,
        report_id=1,
        allocation_date=date(2026, 7, 25),
        attendance_point_name="总厂",
        employee_phone="13800000000",
        good_qty=60,
        defect_qty=0,
        scrap_qty=0,
        effective_minutes=60,
        day_minutes=60,
        overtime_minutes=0,
        night_minutes=0,
        reference_wage=60,
        changeover_count=0,
    )

    rows = _dashboard_rows_from_allocations([allocation])

    assert rows[0].has_over_limit is True
    assert rows[0].over_limit_summary == "1项超额"
    assert rows[0].over_limit_reasons == "1序累计报工数量超过本批次材料可支撑数量"

Step 2: Run dashboard test and verify failure

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_dashboard_allocations.py::test_dashboard_rows_include_over_limit_summary_from_reports -v

Expected: FAIL because dashboard schema does not expose fields.

Step 3: Add dashboard schema fields

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/schemas.py, add to DashboardRow:

    has_over_limit: bool = False
    over_limit_summary: str | None = None
    over_limit_reasons: str | None = None

Step 4: Aggregate dashboard fields

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/dashboard.py, initialize group fields:

                "has_over_limit": False,
                "over_limit_summaries": [],
                "over_limit_summary_keys": set(),
                "over_limit_reasons": [],
                "over_limit_reason_keys": set(),

Inside allocation loop, after report = getattr(allocation, "report", None) and item = getattr(allocation, "item", None), add:

        if getattr(report, "has_over_limit", False):
            group["has_over_limit"] = True
        summary = str(getattr(report, "over_limit_summary", "") or "").strip()
        if summary and summary not in group["over_limit_summary_keys"]:
            group["over_limit_summary_keys"].add(summary)
            group["over_limit_summaries"].append(summary)
        reason = str(getattr(item, "over_limit_reason", "") or "").strip()
        if reason and reason not in group["over_limit_reason_keys"]:
            group["over_limit_reason_keys"].add(reason)
            group["over_limit_reasons"].append(reason)

When constructing DashboardRow, add:

                has_over_limit=group["has_over_limit"],
                over_limit_summary="；".join(group["over_limit_summaries"]) or None,
                over_limit_reasons="；".join(group["over_limit_reasons"]) or None,

Step 5: Add export columns

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/dashboard.py, extend DASHBOARD_EXPORT_HEADERS or equivalent export header mapping with:

    "has_over_limit": "是否超额",
    "over_limit_summary": "超额摘要",
    "over_limit_reasons": "超额原因",

Map boolean as:

"是" if row.has_over_limit else "否"

Step 6: Run dashboard/export tests

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_dashboard_allocations.py -v

Expected: PASS.

Step 7: Commit

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git add app/schemas.py app/routers/dashboard.py app/services/excel_export.py tests/test_dashboard_allocations.py
git commit -m "feat: 经理看板展示报工超额信息"

Expected: commit succeeds.

Task 6: Frontend Mapping And Display

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/records/records.wxml
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/records/records.wxss
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxss
Step 1: Map API fields

In /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js, in report item mapping, add:

    overLimitStatus: row.over_limit_status || 'not_checked',
    overLimitType: row.over_limit_type || '',
    overLimitReason: row.over_limit_reason || '',
    overLimitCheckedAt: row.over_limit_checked_at || '',
    overLimitCheckSource: row.over_limit_check_source || '',
    overLimitBatchNo: row.over_limit_batch_no || '',
    overLimitProductGrossWeightKg: row.over_limit_product_gross_weight_kg ?? '',
    overLimitIssuedWeightKg: row.over_limit_issued_weight_kg ?? '',
    overLimitMaxReportableQty: row.over_limit_max_reportable_qty ?? '',
    overLimitPreviousGoodQty: row.over_limit_previous_good_qty ?? '',
    overLimitCurrentCumulativeQty: row.over_limit_current_cumulative_qty ?? '',
    overLimitCurrentReportQty: row.over_limit_current_report_qty ?? '',

In report mapping, add:

    hasOverLimit: Boolean(row.has_over_limit),
    overLimitSummary: row.over_limit_summary || '',

In dashboard row mapping, add:

      hasOverLimit: Boolean(row.has_over_limit),
      overLimitSummary: row.over_limit_summary || '',
      overLimitReasons: row.over_limit_reasons || '',

Step 2: Add review page display

In /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxml, near existing card watermarks, add:

    <text wx:if="{{item.hasOverLimit}}" class="over-limit-watermark">超额</text>

Inside each detail card metadata area, add:

          <view wx:if="{{detail.overLimitStatus === 'exceeded' || detail.overLimitStatus === 'uncheckable'}}" class="over-limit-panel {{detail.overLimitStatus === 'exceeded' ? 'danger' : 'warn'}}">
            <text class="over-limit-title">{{detail.overLimitStatus === 'exceeded' ? '超额' : '无法校验'}}</text>
            <text class="over-limit-text">{{detail.overLimitReason || '-'}}</text>
            <text class="over-limit-text">本次消耗 {{detail.overLimitCurrentReportQty || '-'}}</text>
            <text class="over-limit-text">累计消耗 {{detail.overLimitCurrentCumulativeQty || '-'}}</text>
            <text wx:if="{{detail.overLimitMaxReportableQty}}" class="over-limit-text">最大可报 {{detail.overLimitMaxReportableQty}}</text>
            <text wx:if="{{detail.overLimitPreviousGoodQty}}" class="over-limit-text">前序可用 {{detail.overLimitPreviousGoodQty}}</text>
          </view>

Step 3: Add shared page styles

In /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxss, add:

.over-limit-watermark {
  position: absolute;
  right: 24rpx;
  top: 24rpx;
  transform: rotate(12deg);
  color: rgba(220, 38, 38, 0.85);
  border: 2rpx solid rgba(220, 38, 38, 0.55);
  border-radius: 8rpx;
  padding: 6rpx 14rpx;
  font-size: 24rpx;
  font-weight: 700;
}

.over-limit-panel {
  margin-top: 16rpx;
  padding: 14rpx 16rpx;
  border-radius: 8rpx;
  display: flex;
  flex-direction: column;
  gap: 6rpx;
}

.over-limit-panel.danger {
  background: #fef2f2;
  border: 1rpx solid #fecaca;
}

.over-limit-panel.warn {
  background: #fffbeb;
  border: 1rpx solid #fde68a;
}

.over-limit-title {
  font-size: 24rpx;
  font-weight: 700;
  color: #b91c1c;
}

.over-limit-text {
  font-size: 22rpx;
  color: #7f1d1d;
}

Add equivalent styles to records.wxss and dashboard.wxss, or reuse existing page-local class names with same CSS.

Step 4: Add records page display

In /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/records/records.wxml, add:

    <text wx:if="{{item.hasOverLimit}}" class="over-limit-watermark">超额</text>

Inside detail card:

            <view wx:if="{{detail.overLimitStatus === 'exceeded' || detail.overLimitStatus === 'uncheckable'}}" class="over-limit-panel {{detail.overLimitStatus === 'exceeded' ? 'danger' : 'warn'}}">
              <text class="over-limit-title">{{detail.overLimitStatus === 'exceeded' ? '超额' : '无法校验'}}</text>
              <text class="over-limit-text">{{detail.overLimitReason || '-'}}</text>
            </view>

Step 5: Add dashboard page display

In /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml, add:

    <text wx:if="{{item.hasOverLimit}}" class="over-limit-watermark">超额</text>

Near metrics or remark area:

      <view wx:if="{{item.overLimitSummary || item.overLimitReasons}}" class="over-limit-panel danger">
        <text class="over-limit-title">{{item.overLimitSummary || '超额'}}</text>
        <text wx:if="{{item.overLimitReasons}}" class="over-limit-text">{{item.overLimitReasons}}</text>
      </view>

Step 6: Run frontend syntax check

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS
node --check utils/api.js

Expected: exit code 0.

Step 7: Commit

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS
git add utils/api.js pages/review/review.wxml pages/review/review.wxss pages/records/records.wxml pages/records/records.wxss pages/dashboard/dashboard.wxml pages/dashboard/dashboard.wxss
git commit -m "feat: 小程序展示报工超额快照"

Expected: commit succeeds. Do not stage /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/config.js if it has unrelated local edits.

Task 7: Execute Database Migration And Verify

Files:

Execute: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/scripts/migrate_report_over_limit_snapshot.py
Step 1: Run migration

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python scripts/migrate_report_over_limit_snapshot.py

Expected:

report over limit snapshot migrated

Step 2: Verify MySQL columns and indexes

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python - <<'PY'
from sqlalchemy import text
from app.database import engine

checks = {
    "production_reports.has_over_limit": """
        SELECT COUNT(*) FROM information_schema.columns
        WHERE table_schema = DATABASE()
          AND table_name = 'production_reports'
          AND column_name = 'has_over_limit'
    """,
    "production_reports.over_limit_summary": """
        SELECT COUNT(*) FROM information_schema.columns
        WHERE table_schema = DATABASE()
          AND table_name = 'production_reports'
          AND column_name = 'over_limit_summary'
    """,
    "production_report_items.over_limit_status": """
        SELECT COUNT(*) FROM information_schema.columns
        WHERE table_schema = DATABASE()
          AND table_name = 'production_report_items'
          AND column_name = 'over_limit_status'
    """,
    "idx_reports_over_limit": """
        SELECT COUNT(*) FROM information_schema.statistics
        WHERE table_schema = DATABASE()
          AND table_name = 'production_reports'
          AND index_name = 'idx_reports_over_limit'
    """,
    "idx_report_items_process_batch": """
        SELECT COUNT(*) FROM information_schema.statistics
        WHERE table_schema = DATABASE()
          AND table_name = 'production_report_items'
          AND index_name = 'idx_report_items_process_batch'
    """,
}

with engine.connect() as conn:
    for name, sql in checks.items():
        value = conn.execute(text(sql)).scalar_one()
        print(f"{name}: {value}")
        if value < 1:
            raise SystemExit(f"missing {name}")
PY

Expected: every printed value is at least 1, command exits 0.

Step 3: Commit migration execution note only if a file changed

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git status --short

Expected: no file changes from executing the script. Do not create a commit for pure database execution.

Task 8: Full Verification

Files:

Verify backend and frontend repos.
Step 1: Run backend targeted tests

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest tests/test_report_over_limit.py tests/test_dashboard_allocations.py -v

Expected: all pass.

Step 2: Run backend full suite

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m pytest -v

Expected: all pass.

Step 3: Compile changed backend modules

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
.venv/bin/python -m py_compile \
  app/services/report_over_limit.py \
  scripts/migrate_report_over_limit_snapshot.py

Expected: exit code 0.

Step 4: Run backend whitespace check

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git diff --check

Expected: no output, exit code 0.

Step 5: Run frontend syntax check

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS
node --check utils/api.js

Expected: exit code 0.

Step 6: Check git status in both repos

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git status --short --branch

cd /Users/souplearn/Gitlab/app/JhHardwareWRS
git status --short --branch

Expected:

Backend contains only intended commits or is clean.
Frontend contains intended commit; unrelated utils/config.js must not be staged or committed.
Step 7: Final report

Report to the user:

已完成：
- 提交时固化超额快照
- 第一道工序按批次投入重量/产品毛重校验
- 后续工序按前序成品数量校验
- 审核、记录、经理看板、导出展示超额信息
- 已执行数据库脚本并验证字段/索引

验证：
- 后端 targeted tests: 通过
- 后端 full pytest: 通过
- 前端 node --check utils/api.js: 通过

Self-Review

Spec coverage: 覆盖提交时快照固化、第一道工序材料上限、后续工序前序上限、数量口径、页面展示、导出、数据库迁移和验证。
Placeholder scan: 已扫描占位词和空泛步骤，未发现需要补全的内容。
Type consistency: 后端字段统一使用 snake_case；前端映射统一使用 camelCase；状态值统一为 not_checked、ok、exceeded、uncheckable。
Scope check: 该计划是单一闭环功能，可独立测试和发布；不包含 ERP 投料页面和动态历史重算。

63 KiB Raw Blame History Unescape Escape

报工超额提交快照 Implementation Plan

Scope

Business Rules

工序范围

第一道工序

后续工序

统计范围

快照固化

File Map

Task 1: Add Database Fields And Migration

Task 2: Implement Over-Limit Calculation Service

Task 3: Refresh Snapshots On Submit And Review

Task 4: Expose Snapshot Fields Through API

Task 5: Add Dashboard And Export Fields

Task 6: Frontend Mapping And Display

Task 7: Execute Database Migration And Verify

Task 8: Full Verification

Self-Review

63 KiB

Raw Blame History