diff --git a/.impeccable/design.json b/.impeccable/design.json new file mode 100644 index 0000000..a7b7c17 --- /dev/null +++ b/.impeccable/design.json @@ -0,0 +1,162 @@ +{ + "schemaVersion": 2, + "generatedAt": "2026-06-12T00:00:00+08:00", + "title": "Design System: ForgeFlow ERP", + "extensions": { + "colorMeta": { + "canvas": { + "role": "neutral", + "displayName": "ERP Canvas", + "canonical": "#f0f2f5", + "tonalRamp": ["#ffffff", "#f5f7fb", "#f0f2f5", "#e9edf3"] + }, + "surface": { + "role": "neutral", + "displayName": "Work Surface", + "canonical": "#ffffff", + "tonalRamp": ["#ffffff", "#fbfdff", "#f8fafc"] + }, + "primary": { + "role": "primary", + "displayName": "Operational Blue", + "canonical": "#409eff", + "tonalRamp": ["#ecf5ff", "#d9ecff", "#409eff", "#2563eb"] + }, + "success": { + "role": "semantic", + "displayName": "Pass Green", + "canonical": "#67c23a", + "tonalRamp": ["#f0f9eb", "#e1f3d8", "#67c23a"] + }, + "warning": { + "role": "semantic", + "displayName": "Review Amber", + "canonical": "#e6a23c", + "tonalRamp": ["#fdf6ec", "#faecd8", "#e6a23c"] + }, + "danger": { + "role": "semantic", + "displayName": "Stop Red", + "canonical": "#f56c6c", + "tonalRamp": ["#fef0f0", "#fde2e2", "#f56c6c"] + } + }, + "typographyMeta": { + "title": { + "displayName": "Title", + "purpose": "Panel and drawer titles inside authenticated product screens." + }, + "body": { + "displayName": "Body", + "purpose": "Chinese-first operational UI text, descriptions, and table cells." + }, + "label": { + "displayName": "Label", + "purpose": "Field labels, chips, table headers, and compact metadata." + } + }, + "shadows": [ + { + "name": "low-surface", + "value": "0 6px 16px rgba(31, 45, 61, 0.08), 0 1px 2px rgba(31, 45, 61, 0.06)", + "purpose": "Tables, compact repeated cards, and low emphasis containers." + }, + { + "name": "panel-surface", + "value": "0 12px 28px rgba(31, 45, 61, 0.12), 0 3px 8px rgba(31, 45, 61, 0.08)", + "purpose": "Main panels and large workflow containers." + }, + { + "name": "floating-surface", + "value": "0 20px 46px rgba(31, 45, 61, 0.16), 0 8px 18px rgba(31, 45, 61, 0.1)", + "purpose": "Drawers, dropdowns, popovers, and overlays." + } + ], + "motion": [ + { + "name": "fast-state", + "value": "140ms ease", + "purpose": "Small button, link, and row hover feedback." + }, + { + "name": "normal-state", + "value": "220ms cubic-bezier(0.2, 0.8, 0.2, 1)", + "purpose": "Drawer and panel state transitions." + } + ], + "breakpoints": [ + { + "name": "tablet", + "value": "860px" + }, + { + "name": "desktop-collapse", + "value": "1180px" + } + ] + }, + "components": [ + { + "name": "Primary Button", + "kind": "button", + "refersTo": "button-primary", + "description": "Main action button for saving, creating, and confirming high-value ERP actions.", + "html": "", + "css": ".ds-btn-primary { display: inline-flex; align-items: center; justify-content: center; min-height: 38px; padding: 0 14px; border: 1px solid #409eff; border-radius: 12px; background: #409eff; color: #ffffff; font: 700 13px/1 'Helvetica Neue', Helvetica, 'PingFang SC', 'Microsoft YaHei', Arial, sans-serif; cursor: pointer; transition: background 140ms ease, border-color 140ms ease, box-shadow 140ms ease; } .ds-btn-primary:hover { background: #2563eb; border-color: #2563eb; } .ds-btn-primary:focus-visible { outline: none; box-shadow: 0 0 0 3px rgba(37, 99, 235, 0.18); }" + }, + { + "name": "Status Badge", + "kind": "chip", + "refersTo": "status-primary", + "description": "Compact semantic status label used across tables and detail drawers.", + "html": "Approved", + "css": ".ds-status { display: inline-flex; align-items: center; gap: 6px; min-height: 24px; padding: 2px 8px; border: 1px solid #d9ecff; border-radius: 4px; background: #ecf5ff; color: #409eff; font: 600 12px/1.45 'Helvetica Neue', Helvetica, 'PingFang SC', 'Microsoft YaHei', Arial, sans-serif; white-space: nowrap; } .ds-status span { width: 6px; height: 6px; border-radius: 999px; background: currentColor; box-shadow: 0 0 0 3px color-mix(in srgb, currentColor 16%, transparent); }" + }, + { + "name": "Data Table Container", + "kind": "custom", + "refersTo": "table-container", + "description": "Scrollable table frame with sticky headers for dense ERP records.", + "html": "
DocumentStatus
PO-001Pending
", + "css": ".ds-table { overflow: auto; border: 1px solid rgba(209, 218, 232, 0.96); border-radius: 16px; background: #ffffff; box-shadow: 0 6px 16px rgba(31, 45, 61, 0.08), 0 1px 2px rgba(31, 45, 61, 0.06); } .ds-table table { width: 100%; border-collapse: separate; border-spacing: 0; font: 14px/1.45 'Helvetica Neue', Helvetica, 'PingFang SC', 'Microsoft YaHei', Arial, sans-serif; } .ds-table th { position: sticky; top: 0; background: #f8fbff; color: #334155; font-size: 12px; font-weight: 750; text-align: left; } .ds-table th, .ds-table td { padding: 12px; border-bottom: 1px solid #ebeef5; } .ds-table td { color: #42526b; }" + } + ], + "narrative": { + "northStar": "The Clear Production Desk", + "overview": "ForgeFlow ERP is a clear, light, high-density operational interface for a hardware manufacturing company. It should help users trace orders, materials, batches, operations, documents, and costs without visual noise.", + "keyCharacteristics": [ + "Light routine ERP surfaces", + "Restrained blue action and selection language", + "Chinese-first practical product typography", + "Tables, drawers, forms, and status badges as core vocabulary", + "Formal document visuals only for business documents" + ], + "rules": [ + { + "name": "The Blue Rarity Rule", + "body": "Blue identifies current position and next action; it is not broad decoration.", + "section": "colors" + }, + { + "name": "The Fixed Scale Rule", + "body": "Product screens use fixed UI type sizes, not fluid hero headings.", + "section": "typography" + }, + { + "name": "The One Depth Signal Rule", + "body": "Do not stack thick borders, large radius, glossy gradients, and heavy shadow on one element.", + "section": "elevation" + } + ], + "dos": [ + "Do keep routine ERP screens light, restrained, and table-first.", + "Do standardize table, drawer, form, status badge, pagination, and action button behavior before styling individual pages.", + "Do provide loading, empty, error, no-permission, disabled, saving, and archive-failed states for critical flows." + ], + "donts": [ + "Don't make screens look obviously AI-generated through generic card grids, decorative gradients, glass effects, or ornamental animation.", + "Don't use side accent borders greater than 1px on cards or callouts.", + "Don't let old dark industrial styling bleed into everyday ERP work screens." + ] + } +} diff --git a/.impeccable/live/config.json b/.impeccable/live/config.json new file mode 100644 index 0000000..0358edf --- /dev/null +++ b/.impeccable/live/config.json @@ -0,0 +1,6 @@ +{ + "files": ["frontend/index.html"], + "insertBefore": "", + "commentSyntax": "html", + "cspChecked": true +} diff --git a/.impeccable/live/forgeflow-smoke-login.png b/.impeccable/live/forgeflow-smoke-login.png new file mode 100644 index 0000000..fdc5def Binary files /dev/null and b/.impeccable/live/forgeflow-smoke-login.png differ diff --git a/backend/app/api/routes/dashboard.py b/backend/app/api/routes/dashboard.py index 067288d..84cff0b 100644 --- a/backend/app/api/routes/dashboard.py +++ b/backend/app/api/routes/dashboard.py @@ -43,6 +43,7 @@ from app.schemas.dashboard import ( SalesOrderLifecycleRead, ) from app.services.auth import require_authenticated_user +from app.services.material_lifecycle_dag import build_material_lot_lifecycle_graph from app.services.mock_data import get_dashboard_payload router = APIRouter(dependencies=[Depends(require_authenticated_user)]) @@ -358,7 +359,7 @@ def _material_lot_rows(limit: int = 100): .outerjoin(issue_subquery, issue_subquery.c.source_lot_id == StockLot.id) .outerjoin(finished_subquery, finished_subquery.c.source_material_lot_id == StockLot.id) .outerjoin(delivery_subquery, delivery_subquery.c.source_material_lot_id == StockLot.id) - .where(Item.item_type == "RAW_MATERIAL") + .where(Item.item_type == "RAW_MATERIAL", Warehouse.warehouse_type == "RAW") .order_by(StockLot.created_at.desc(), StockLot.id.desc()) .limit(limit) ) @@ -387,13 +388,10 @@ def list_dashboard_material_lots( return [DashboardMaterialLotRead.model_validate(dict(row)) for row in rows] -@router.get("/material-lots/{lot_id}/lifecycle", response_model=MaterialLotLifecycleRead) -def get_material_lot_lifecycle(lot_id: int, db: Session = Depends(get_db)) -> MaterialLotLifecycleRead: - lot_row = db.execute(_material_lot_rows(limit=1).where(StockLot.id == lot_id)).mappings().first() - if not lot_row: - raise HTTPException(status_code=404, detail="原材料批次不存在") - - lot = DashboardMaterialLotRead.model_validate(dict(lot_row)) +def _build_legacy_material_lot_lifecycle_graph( + db: Session, + lot: DashboardMaterialLotRead, +) -> tuple[list[LifecycleNode], list[LifecycleEdge]]: nodes: list[LifecycleNode] = [] edges: list[LifecycleEdge] = [] node_ids: set[str] = set() @@ -465,7 +463,7 @@ def get_material_lot_lifecycle(lot_id: int, db: Session = Depends(get_db)) -> Ma .join(WorkOrder, WorkOrder.id == WorkOrderMaterialIssue.work_order_id) .join(Item, Item.id == WorkOrder.product_item_id) .outerjoin(StockLot, StockLot.id == WorkOrderMaterialIssue.source_lot_id) - .where(WorkOrderMaterialIssue.source_lot_id == lot_id) + .where(WorkOrderMaterialIssue.source_lot_id == lot.lot_id) .order_by(WorkOrderMaterialIssue.issue_time, WorkOrderMaterialIssue.id) ).mappings().all() work_order_ids = sorted({row["work_order_id"] for row in issues}) @@ -741,6 +739,19 @@ def get_material_lot_lifecycle(lot_id: int, db: Session = Depends(get_db)) -> Ma ) add_edge(_edge(f"fg-{row['completion_item_id']}", node_id, "库存发货")) + return nodes, edges + + +@router.get("/material-lots/{lot_id}/lifecycle", response_model=MaterialLotLifecycleRead) +def get_material_lot_lifecycle(lot_id: int, db: Session = Depends(get_db)) -> MaterialLotLifecycleRead: + lot_row = db.execute(_material_lot_rows(limit=1).where(StockLot.id == lot_id)).mappings().first() + if not lot_row: + raise HTTPException(status_code=404, detail="原材料批次不存在") + + lot = DashboardMaterialLotRead.model_validate(dict(lot_row)) + nodes, edges = build_material_lot_lifecycle_graph(db, lot) + if not any(node.type == "production_ledger" for node in nodes): + nodes, edges = _build_legacy_material_lot_lifecycle_graph(db, lot) return MaterialLotLifecycleRead(lot=lot, nodes=nodes, edges=edges) diff --git a/backend/app/services/material_lifecycle_dag.py b/backend/app/services/material_lifecycle_dag.py new file mode 100644 index 0000000..2f282b7 --- /dev/null +++ b/backend/app/services/material_lifecycle_dag.py @@ -0,0 +1,456 @@ +from __future__ import annotations + +from decimal import Decimal +from typing import Any + +from sqlalchemy import false, select +from sqlalchemy.orm import Session, aliased + +from app.models.master_data import Item +from app.models.operations import Delivery, DeliveryItem, OperationReport, ProductionBatchLedger, ProductionBatchLedgerTxn, StockLot +from app.models.org import Employee +from app.models.sales import Customer, SalesOrder +from app.schemas.dashboard import DashboardMaterialLotRead, LifecycleDetailItem, LifecycleEdge, LifecycleNode +from app.services.document_archives import production_document_batch_no_from_text +from app.services.system_config import get_smart_operation_report_enabled + + +def _num(value: Any) -> float: + if value is None: + return 0.0 + if isinstance(value, Decimal): + return float(value) + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + + +def _fmt_qty(value: Any, digits: int = 2) -> str: + numeric = _num(value) + if abs(numeric - round(numeric)) < 0.000001: + return f"{int(round(numeric))}" + return f"{numeric:.{digits}f}".rstrip("0").rstrip(".") + + +def _fmt_weight(value: Any) -> str: + return f"{_fmt_qty(value, 3)} kg" + + +def _fmt_datetime(value: Any) -> str: + if not value: + return "-" + return str(value).replace("T", " ")[:19] + + +def _detail(label: str, value: Any) -> LifecycleDetailItem: + text = "-" if value is None or value == "" else str(value) + return LifecycleDetailItem(label=label, value=text) + + +def _node( + *, + node_id: str, + node_type: str, + title: str, + stage: str, + level: int, + subtitle: str | None = None, + status: str | None = None, + tone: str = "normal", + summary: str | None = None, + metrics: list[tuple[str, Any]] | None = None, + details: list[tuple[str, Any]] | None = None, + raw: dict[str, Any] | None = None, +) -> LifecycleNode: + return LifecycleNode( + id=node_id, + type=node_type, + title=title, + subtitle=subtitle, + status=status, + stage=stage, + level=level, + tone=tone, + summary=summary, + metrics=[_detail(label, value) for label, value in (metrics or [])], + details=[_detail(label, value) for label, value in (details or [])], + raw=raw or {}, + ) + + +def _edge(source: str, target: str, label: str | None = None) -> LifecycleEdge: + return LifecycleEdge(id=f"{source}->{target}", source=source, target=target, label=label) + + +def _append_node(nodes: list[LifecycleNode], node_ids: set[str], node: LifecycleNode) -> None: + if node.id not in node_ids: + node_ids.add(node.id) + nodes.append(node) + + +def _append_edge(edges: list[LifecycleEdge], edge_ids: set[str], node_ids: set[str], edge: LifecycleEdge) -> None: + if edge.source in node_ids and edge.target in node_ids and edge.id not in edge_ids: + edge_ids.add(edge.id) + edges.append(edge) + + +def _settlement_group_key(txn: ProductionBatchLedgerTxn) -> tuple[str, str]: + batch_no = production_document_batch_no_from_text(txn.remark) + if batch_no: + return f"batch:{batch_no}", batch_no + return f"txn:{txn.id}", f"{txn.txn_type}-{txn.id}" + + +def _txn_node_type(txn_type: str) -> str: + return { + "成品入库": "finished_inbound", + "生产余料入库": "surplus_inbound", + "生产废料入库": "scrap_inbound", + "重新开工": "ledger_reopen", + }.get(txn_type, "ledger_writeoff") + + +def build_material_lot_lifecycle_graph( + db: Session, + lot: DashboardMaterialLotRead, +) -> tuple[list[LifecycleNode], list[LifecycleEdge]]: + nodes: list[LifecycleNode] = [] + edges: list[LifecycleEdge] = [] + node_ids: set[str] = set() + edge_ids: set[str] = set() + + raw_node_id = f"raw-lot-{lot.lot_id}" + _append_node( + nodes, + node_ids, + _node( + node_id=raw_node_id, + node_type="raw_material_lot", + title=lot.lot_no, + subtitle=f"{lot.material_code} · {lot.material_name}", + status=lot.status, + stage="原料入库", + level=0, + summary="原材料批次入库后，按生产台账追踪生产出库、库外余额、入库结算和销售去向。", + metrics=[ + ("入库重量", _fmt_weight(lot.inbound_weight_kg)), + ("剩余重量", _fmt_weight(lot.remaining_weight_kg)), + ], + details=[ + ("原材料编码", lot.material_code), + ("原材料名称", lot.material_name), + ("仓库", lot.warehouse_name), + ], + raw=lot.model_dump(), + ), + ) + + ledger_rows = db.execute( + select( + ProductionBatchLedger, + Item.item_code.label("product_code"), + Item.item_name.label("product_name"), + ) + .join(Item, Item.id == ProductionBatchLedger.product_item_id) + .where(ProductionBatchLedger.material_lot_id == lot.lot_id) + .order_by(ProductionBatchLedger.id) + ).all() + ledgers = [row.ProductionBatchLedger for row in ledger_rows] + ledger_ids = [int(ledger.id) for ledger in ledgers] + + txns_by_ledger_id: dict[int, list[ProductionBatchLedgerTxn]] = {ledger_id: [] for ledger_id in ledger_ids} + if ledger_ids: + txns = db.scalars( + select(ProductionBatchLedgerTxn) + .where(ProductionBatchLedgerTxn.production_ledger_id.in_(ledger_ids)) + .order_by(ProductionBatchLedgerTxn.biz_time, ProductionBatchLedgerTxn.id) + ).all() + for txn in txns: + txns_by_ledger_id.setdefault(int(txn.production_ledger_id), []).append(txn) + + finished_node_id_by_lot_id: dict[int, str] = {} + settlement_txn_types = {"成品入库", "生产余料入库", "生产废料入库", "结单核销", "该批材料结单", "重新开工"} + for row in ledger_rows: + ledger = row.ProductionBatchLedger + ledger_id = int(ledger.id) + ledger_node_id = f"ledger-{ledger_id}" + _append_node( + nodes, + node_ids, + _node( + node_id=ledger_node_id, + node_type="production_ledger", + title=f"{ledger.material_lot_no} · {row.product_name}", + subtitle=row.product_code, + status=ledger.status, + stage="生产台账", + level=1, + summary="一行生产台账代表同一材料库存批次号与产品组合，是材料批次闭环的主链路。", + metrics=[ + ("累计领料", _fmt_weight(ledger.total_issued_weight_kg)), + ("库外余额", _fmt_weight(ledger.outside_weight_kg)), + ("成品入库", _fmt_qty(ledger.finished_inbound_qty)), + ("余料入库", _fmt_weight(ledger.surplus_inbound_weight_kg)), + ("废料入库", _fmt_weight(ledger.scrap_inbound_weight_kg)), + ], + details=[ + ("产品编码", row.product_code), + ("产品名称", row.product_name), + ("首次领料", _fmt_datetime(ledger.first_issue_time)), + ("最近领料", _fmt_datetime(ledger.last_issue_time)), + ("结单次数", ledger.lock_count), + ("备注", ledger.remark), + ], + raw={ + "production_ledger_id": ledger.id, + "material_lot_id": ledger.material_lot_id, + "product_item_id": ledger.product_item_id, + }, + ), + ) + _append_edge(edges, edge_ids, node_ids, _edge(raw_node_id, ledger_node_id, "进入生产台账")) + + issue_txns = [txn for txn in txns_by_ledger_id.get(ledger_id, []) if txn.txn_type == "生产出库"] + issue_node_ids: list[str] = [] + for txn in issue_txns: + issue_node_id = f"issue-txn-{txn.id}" + issue_node_ids.append(issue_node_id) + _append_node( + nodes, + node_ids, + _node( + node_id=issue_node_id, + node_type="production_issue_txn", + title=f"生产出库 {txn.id}", + subtitle=ledger.material_lot_no, + status="POSTED", + tone="success", + stage="生产出库", + level=2, + summary="生产出库事务累加到生产台账，并增加该批材料的库外余额。", + metrics=[ + ("出库重量", _fmt_weight(txn.weight_delta_kg)), + ("出库后库外余额", _fmt_weight(txn.outside_weight_after_kg)), + ], + details=[ + ("业务时间", _fmt_datetime(txn.biz_time)), + ("来源单据", txn.source_doc_type), + ("来源ID", txn.source_doc_id), + ("备注", txn.remark), + ], + raw={"txn_id": txn.id, "production_ledger_id": txn.production_ledger_id}, + ), + ) + _append_edge(edges, edge_ids, node_ids, _edge(ledger_node_id, issue_node_id, "生产出库")) + + outside_node_id = f"outside-balance-{ledger_id}" + _append_node( + nodes, + node_ids, + _node( + node_id=outside_node_id, + node_type="ledger_outside_balance", + title="库外余额", + subtitle=ledger.material_lot_no, + status=ledger.status, + stage="库外闭环", + level=3, + summary="库外余额用于衡量该材料批次尚未通过成品、余料、废料入库或结单核销闭环的重量。", + metrics=[ + ("库外余额", _fmt_weight(ledger.outside_weight_kg)), + ("累计领料", _fmt_weight(ledger.total_issued_weight_kg)), + ("结单核销", _fmt_weight(ledger.writeoff_weight_kg)), + ], + raw={"production_ledger_id": ledger.id}, + ), + ) + if issue_node_ids: + for issue_node_id in issue_node_ids: + _append_edge(edges, edge_ids, node_ids, _edge(issue_node_id, outside_node_id, "形成库外余额")) + else: + _append_edge(edges, edge_ids, node_ids, _edge(ledger_node_id, outside_node_id, "库外余额")) + + settlement_groups: dict[tuple[str, str], list[ProductionBatchLedgerTxn]] = {} + for txn in txns_by_ledger_id.get(ledger_id, []): + if txn.txn_type in settlement_txn_types: + settlement_groups.setdefault(_settlement_group_key(txn), []).append(txn) + + for group_index, ((group_key, group_title), group_txns) in enumerate(settlement_groups.items(), start=1): + settlement_node_id = f"settlement-{ledger_id}-{group_key}" + total_weight_delta = sum(_num(txn.weight_delta_kg) for txn in group_txns) + _append_node( + nodes, + node_ids, + _node( + node_id=settlement_node_id, + node_type="production_ledger_settlement", + title=group_title, + subtitle=f"{len(group_txns)} 笔台账事务", + status="POSTED", + tone="success", + stage="入库结算", + level=4, + summary="按生产单据批次号聚合成品、余料、废料入库及结单核销事务；没有批次号的事务独立成组。", + metrics=[ + ("事务数", len(group_txns)), + ("重量变化", _fmt_weight(total_weight_delta)), + ], + details=[ + ("分组", group_title), + ("序号", group_index), + ], + raw={"production_ledger_id": ledger.id, "group_key": group_key}, + ), + ) + _append_edge(edges, edge_ids, node_ids, _edge(outside_node_id, settlement_node_id, "入库结算")) + + txns_by_type: dict[str, list[ProductionBatchLedgerTxn]] = {} + for txn in group_txns: + txns_by_type.setdefault(str(txn.txn_type), []).append(txn) + for txn_type, type_txns in txns_by_type.items(): + child_node_id = f"{_txn_node_type(txn_type)}-{ledger_id}-{group_key}" + child_weight_delta = sum(_num(txn.weight_delta_kg) for txn in type_txns) + child_qty_delta = sum(_num(txn.qty_delta) for txn in type_txns) + _append_node( + nodes, + node_ids, + _node( + node_id=child_node_id, + node_type=_txn_node_type(txn_type), + title=txn_type, + subtitle=group_title, + status="POSTED", + tone="success", + stage=txn_type, + level=5, + summary="该节点来自生产台账事务，并参与材料批次库外闭环。", + metrics=[ + ("数量变化", _fmt_qty(child_qty_delta)), + ("重量变化", _fmt_weight(child_weight_delta)), + ("事务数", len(type_txns)), + ], + details=[ + ("最近库外余额", _fmt_weight(type_txns[-1].outside_weight_after_kg)), + ("业务时间", _fmt_datetime(type_txns[-1].biz_time)), + ], + raw={ + "production_ledger_id": ledger.id, + "txn_ids": [txn.id for txn in type_txns], + "group_key": group_key, + }, + ), + ) + _append_edge(edges, edge_ids, node_ids, _edge(settlement_node_id, child_node_id, txn_type)) + if _txn_node_type(txn_type) == "finished_inbound": + for txn in type_txns: + if txn.source_line_id: + finished_node_id_by_lot_id[int(txn.source_line_id)] = child_node_id + + if ledger_ids and get_smart_operation_report_enabled(db): + report_rows = db.execute( + select( + OperationReport, + Employee.employee_name.label("employee_name"), + ) + .join(Employee, Employee.id == OperationReport.employee_id) + .where(OperationReport.production_ledger_id.in_(ledger_ids)) + .order_by(OperationReport.start_time, OperationReport.id) + ).all() + for row in report_rows: + report = row.OperationReport + node_id = f"report-{report.id}" + _append_node( + nodes, + node_ids, + _node( + node_id=node_id, + node_type="operation_report", + title=report.report_no, + subtitle=row.employee_name, + status="ABNORMAL" if report.is_abnormal else "NORMAL", + tone="danger" if report.is_abnormal else "success", + stage="现场报工证据", + level=2, + summary="只作为现场报工证据展示，不参与材料批次闭环判定。", + metrics=[ + ("报工数", _fmt_qty(report.report_qty)), + ("合格数", _fmt_qty(report.good_qty)), + ("报废数", _fmt_qty(report.scrap_qty)), + ], + details=[ + ("员工", row.employee_name), + ("开始", _fmt_datetime(report.start_time)), + ("结束", _fmt_datetime(report.end_time)), + ("来源", report.report_source), + ], + raw={"report_id": report.id, "production_ledger_id": report.production_ledger_id}, + ), + ) + _append_edge(edges, edge_ids, node_ids, _edge(f"ledger-{report.production_ledger_id}", node_id, "报工证据")) + + finished_lot = aliased(StockLot) + order_customer = aliased(Customer) + direct_customer = aliased(Customer) + delivery_rows = db.execute( + select( + DeliveryItem.id.label("delivery_item_id"), + DeliveryItem.lot_id.label("delivery_lot_id"), + finished_lot.id.label("finished_lot_id"), + finished_lot.source_doc_id.label("production_ledger_id"), + finished_lot.lot_no.label("lot_no"), + Delivery.delivery_no.label("delivery_no"), + SalesOrder.order_no.label("order_no"), + direct_customer.customer_name.label("direct_customer_name"), + order_customer.customer_name.label("order_customer_name"), + Delivery.delivery_date.label("delivery_date"), + DeliveryItem.delivery_qty.label("delivery_qty"), + DeliveryItem.delivery_weight_kg.label("delivery_weight_kg"), + DeliveryItem.line_amount.label("line_amount"), + DeliveryItem.status.label("status"), + ) + .join(finished_lot, finished_lot.id == DeliveryItem.lot_id) + .join(Item, Item.id == finished_lot.item_id) + .join(Delivery, Delivery.id == DeliveryItem.delivery_id) + .outerjoin(SalesOrder, SalesOrder.id == Delivery.sales_order_id) + .outerjoin(order_customer, order_customer.id == SalesOrder.customer_id) + .outerjoin(direct_customer, direct_customer.id == Delivery.customer_id) + .where( + finished_lot.source_doc_id.in_(ledger_ids) if ledger_ids else false(), + finished_lot.source_doc_type.in_(["PRODUCTION_LEDGER_IN", "生产台账"]), + Item.item_type == "FINISHED_GOOD", + ) + .order_by(Delivery.delivery_date, DeliveryItem.id) + ).mappings().all() + for row in delivery_rows: + delivery_node_id = f"delivery-{row['delivery_item_id']}" + _append_node( + nodes, + node_ids, + _node( + node_id=delivery_node_id, + node_type="delivery", + title=row["delivery_no"], + subtitle=row["order_no"], + status=row["status"], + stage="销售去向", + level=6, + summary="销售出库从生产台账生成的成品库存批次追溯到材料批次。", + metrics=[ + ("发货数量", _fmt_qty(row["delivery_qty"])), + ("发货重量", _fmt_weight(row["delivery_weight_kg"])), + ], + details=[ + ("客户", row["direct_customer_name"] or row["order_customer_name"]), + ("库存批次号", row["lot_no"]), + ("发货时间", _fmt_datetime(row["delivery_date"])), + ], + raw=dict(row), + ), + ) + source_node_id = finished_node_id_by_lot_id.get(int(row["delivery_lot_id"] or 0)) + if not source_node_id: + source_node_id = f"ledger-{row['production_ledger_id']}" + _append_edge(edges, edge_ids, node_ids, _edge(source_node_id, delivery_node_id, "销售出库")) + + return nodes, edges diff --git a/backend/tests/test_dashboard_material_lot_lifecycle.py b/backend/tests/test_dashboard_material_lot_lifecycle.py new file mode 100644 index 0000000..ac2597a --- /dev/null +++ b/backend/tests/test_dashboard_material_lot_lifecycle.py @@ -0,0 +1,510 @@ +from __future__ import annotations + +import unittest +from datetime import datetime +from decimal import Decimal + +from sqlalchemy import BigInteger, create_engine +from sqlalchemy.ext.compiler import compiles +from sqlalchemy.orm import Session, sessionmaker + + +@compiles(BigInteger, "sqlite") +def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str: + _ = type_, compiler, kw + return "INTEGER" + + +import app.models.master_data # noqa: E402,F401 +import app.models.miniapp # noqa: E402,F401 +import app.models.operations # noqa: E402,F401 +import app.models.org # noqa: E402,F401 +import app.models.planning # noqa: E402,F401 +import app.models.sales # noqa: E402,F401 +from app.api.routes.dashboard import get_material_lot_lifecycle, list_dashboard_material_lots # noqa: E402 +from app.models.base import Base # noqa: E402 +from app.models.master_data import Item, Warehouse # noqa: E402 +from app.models.operations import Delivery, DeliveryItem, OperationReport, ProductionBatchLedger, ProductionBatchLedgerTxn, StockLot # noqa: E402 +from app.models.org import Department, Employee, SystemConfig # noqa: E402 +from app.schemas.dashboard import DashboardMaterialLotRead # noqa: E402 +from app.services.material_lifecycle_dag import build_material_lot_lifecycle_graph # noqa: E402 + + +class DashboardMaterialLotLifecycleTest(unittest.TestCase): + def setUp(self) -> None: + engine = create_engine("sqlite+pysqlite:///:memory:", future=True) + Base.metadata.create_all(engine) + self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) + self.db: Session = self.SessionLocal() + self.now = datetime(2026, 6, 14, 9, 0) + self._seed_lot_and_ledger() + + def tearDown(self) -> None: + self.db.close() + + def _seed_lot_and_ledger(self) -> None: + raw_item = Item( + id=1, + item_code="RM-001", + item_name="冷轧钢板", + item_type="RAW_MATERIAL", + unit_weight_kg=Decimal("1"), + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + product = Item( + id=2, + item_code="FG-001", + item_name="钢制碗", + item_type="FINISHED_GOOD", + unit_weight_kg=Decimal("0.5"), + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + warehouse = Warehouse( + id=1, + warehouse_code="WH-RAW", + warehouse_name="原材料库", + warehouse_type="RAW", + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + lot = StockLot( + id=1, + lot_no="YL0001", + lot_role="RAW_MATERIAL", + item_id=1, + warehouse_id=1, + source_doc_type="期初入库", + source_doc_id=1, + inbound_qty=Decimal("0"), + inbound_weight_kg=Decimal("500"), + remaining_qty=Decimal("0"), + remaining_weight_kg=Decimal("300"), + locked_qty=Decimal("0"), + locked_weight_kg=Decimal("0"), + unit_cost=Decimal("4"), + quality_status="PASS", + status="AVAILABLE", + created_at=self.now, + updated_at=self.now, + ) + ledger = ProductionBatchLedger( + id=1, + material_lot_id=1, + material_lot_no="YL0001", + material_item_id=1, + product_item_id=2, + status="在生产", + miniapp_selectable=1, + total_issued_weight_kg=Decimal("120"), + outside_weight_kg=Decimal("42"), + finished_inbound_qty=Decimal("100"), + surplus_inbound_weight_kg=Decimal("18"), + scrap_inbound_weight_kg=Decimal("6"), + writeoff_weight_kg=Decimal("54"), + first_issue_time=self.now, + last_issue_time=self.now, + created_at=self.now, + updated_at=self.now, + ) + dept = Department( + id=1, + dept_code="D-001", + dept_name="生产部", + dept_type="PRODUCTION", + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + employee = Employee( + id=1, + employee_code="EMP-001", + employee_name="张三", + dept_id=1, + is_operator=1, + is_workshop_staff=1, + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + txns = [ + ProductionBatchLedgerTxn( + id=1, + production_ledger_id=1, + txn_type="生产出库", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("120"), + outside_weight_after_kg=Decimal("120"), + source_doc_type="库存流水", + source_doc_id=101, + biz_time=self.now, + operator_user_id=1, + remark="生产出库累计到生产台账", + created_at=self.now, + updated_at=self.now, + ), + ProductionBatchLedgerTxn( + id=2, + production_ledger_id=1, + txn_type="成品入库", + qty_delta=Decimal("100"), + weight_delta_kg=Decimal("-50"), + outside_weight_after_kg=Decimal("70"), + source_doc_type="库存流水", + source_doc_id=102, + biz_time=self.now, + operator_user_id=1, + remark="单据批次号：PB20260614001", + created_at=self.now, + updated_at=self.now, + ), + ProductionBatchLedgerTxn( + id=3, + production_ledger_id=1, + txn_type="生产余料入库", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("-18"), + outside_weight_after_kg=Decimal("52"), + source_doc_type="库存流水", + source_doc_id=103, + biz_time=self.now, + operator_user_id=1, + remark="单据批次号：PB20260614001", + created_at=self.now, + updated_at=self.now, + ), + ProductionBatchLedgerTxn( + id=4, + production_ledger_id=1, + txn_type="生产废料入库", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("-6"), + outside_weight_after_kg=Decimal("46"), + source_doc_type="库存流水", + source_doc_id=104, + biz_time=self.now, + operator_user_id=1, + remark="单据批次号：PB20260614001", + created_at=self.now, + updated_at=self.now, + ), + ProductionBatchLedgerTxn( + id=5, + production_ledger_id=1, + txn_type="结单核销", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("-4"), + outside_weight_after_kg=Decimal("42"), + source_doc_type="结单", + source_doc_id=105, + biz_time=self.now, + operator_user_id=1, + remark="单据批次号：PB20260614001", + created_at=self.now, + updated_at=self.now, + ), + ] + report = OperationReport( + id=1, + report_no="BG20260614001", + production_ledger_id=1, + employee_id=1, + report_source="MINIAPP", + start_time=self.now, + end_time=self.now, + report_qty=Decimal("100"), + good_qty=Decimal("96"), + scrap_qty=Decimal("4"), + rework_qty=Decimal("0"), + downtime_minutes=Decimal("0"), + extra_cost_amount=Decimal("0"), + is_abnormal=0, + created_at=self.now, + updated_at=self.now, + ) + self.db.add_all([raw_item, product, warehouse, lot, ledger, dept, employee, *txns, report]) + self.db.commit() + + def _lot_read(self) -> DashboardMaterialLotRead: + return DashboardMaterialLotRead( + lot_id=1, + lot_no="YL0001", + material_item_id=1, + material_code="RM-001", + material_name="冷轧钢板", + warehouse_name="原材料库", + inbound_weight_kg=500, + remaining_weight_kg=300, + unit_cost=4, + inbound_amount=2000, + issued_batch_count=0, + issued_weight_kg=0, + finished_batch_count=0, + finished_qty=0, + delivery_count=0, + status="AVAILABLE", + quality_status="PASS", + created_at=self.now, + ) + + def test_production_ledger_graph_replaces_work_order_main_chain(self) -> None: + nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + + node_types = {node.type for node in nodes} + self.assertIn("production_ledger", node_types) + self.assertIn("production_issue_txn", node_types) + self.assertIn("ledger_outside_balance", node_types) + self.assertIn("production_ledger_settlement", node_types) + self.assertIn("finished_inbound", node_types) + self.assertIn("surplus_inbound", node_types) + self.assertIn("scrap_inbound", node_types) + self.assertNotIn("work_order", node_types) + + def test_production_ledger_title_stage_and_outside_balance_metric(self) -> None: + nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + + ledger_node = next(node for node in nodes if node.type == "production_ledger") + metrics = {item.label: item.value for item in ledger_node.metrics} + self.assertEqual(ledger_node.title, "YL0001 · 钢制碗") + self.assertEqual(ledger_node.stage, "生产台账") + self.assertEqual(metrics["库外余额"], "42 kg") + + def test_material_lot_dashboard_list_excludes_scrap_warehouse_lots(self) -> None: + scrap_warehouse = Warehouse( + id=3, + warehouse_code="WH-SCRAP", + warehouse_name="废料库", + warehouse_type="SCRAP", + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + scrap_lot = StockLot( + id=30, + lot_no="SCRAPI-YL0001-001", + lot_role="PRODUCTION_SCRAP", + item_id=1, + warehouse_id=3, + source_doc_type="PRODUCTION_SCRAP_IN", + source_doc_id=1, + inbound_qty=Decimal("0"), + inbound_weight_kg=Decimal("5"), + remaining_qty=Decimal("0"), + remaining_weight_kg=Decimal("5"), + locked_qty=Decimal("0"), + locked_weight_kg=Decimal("0"), + unit_cost=Decimal("4"), + quality_status="SCRAP", + status="AVAILABLE", + created_at=self.now, + updated_at=self.now, + ) + self.db.add_all([scrap_warehouse, scrap_lot]) + self.db.commit() + + rows = list_dashboard_material_lots(limit=100, db=self.db) + + lot_nos = {row.lot_no for row in rows} + self.assertIn("YL0001", lot_nos) + self.assertNotIn("SCRAPI-YL0001-001", lot_nos) + self.assertTrue(all(row.warehouse_name == "原材料库" for row in rows)) + + def test_operation_report_respects_smart_operation_report_switch(self) -> None: + nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + self.assertIn("operation_report", {node.type for node in nodes}) + + self.db.add( + SystemConfig( + config_code="SMART_OPERATION_REPORT_ENABLED", + config_name="对接智能报工小程序", + config_value="关闭", + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + ) + self.db.commit() + + nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + self.assertNotIn("operation_report", {node.type for node in nodes}) + + def test_route_returns_new_graph_when_production_ledger_exists(self) -> None: + response = get_material_lot_lifecycle(1, self.db) + + node_types = {node.type for node in response.nodes} + self.assertIn("production_ledger", node_types) + self.assertNotIn("work_order", node_types) + + def test_route_legacy_fallback_does_not_crash_without_production_ledger(self) -> None: + legacy_lot = StockLot( + id=2, + lot_no="YL9999", + lot_role="RAW_MATERIAL", + item_id=1, + warehouse_id=1, + source_doc_type="期初入库", + source_doc_id=2, + inbound_qty=Decimal("0"), + inbound_weight_kg=Decimal("80"), + remaining_qty=Decimal("0"), + remaining_weight_kg=Decimal("80"), + locked_qty=Decimal("0"), + locked_weight_kg=Decimal("0"), + unit_cost=Decimal("4"), + quality_status="PASS", + status="AVAILABLE", + created_at=self.now, + updated_at=self.now, + ) + self.db.add(legacy_lot) + self.db.commit() + + response = get_material_lot_lifecycle(2, self.db) + + self.assertEqual(response.lot.lot_no, "YL9999") + node_types = {node.type for node in response.nodes} + self.assertIn("raw_material_lot", node_types) + self.assertNotIn("production_ledger", node_types) + + def test_delivery_edges_target_finished_inbound_for_exact_finished_lot(self) -> None: + finished_lot_a = StockLot( + id=10, + lot_no="FGL-A", + lot_role="INVENTORY", + item_id=2, + warehouse_id=1, + source_doc_type="PRODUCTION_LEDGER_IN", + source_doc_id=1, + inbound_qty=Decimal("40"), + inbound_weight_kg=Decimal("20"), + remaining_qty=Decimal("40"), + remaining_weight_kg=Decimal("20"), + locked_qty=Decimal("0"), + locked_weight_kg=Decimal("0"), + unit_cost=Decimal("8"), + quality_status="PASS", + status="AVAILABLE", + created_at=self.now, + updated_at=self.now, + ) + finished_lot_b = StockLot( + id=11, + lot_no="FGL-B", + lot_role="INVENTORY", + item_id=2, + warehouse_id=1, + source_doc_type="PRODUCTION_LEDGER_IN", + source_doc_id=1, + inbound_qty=Decimal("60"), + inbound_weight_kg=Decimal("30"), + remaining_qty=Decimal("0"), + remaining_weight_kg=Decimal("0"), + locked_qty=Decimal("0"), + locked_weight_kg=Decimal("0"), + unit_cost=Decimal("8"), + quality_status="PASS", + status="AVAILABLE", + created_at=self.now, + updated_at=self.now, + ) + finished_txn_a = ProductionBatchLedgerTxn( + id=20, + production_ledger_id=1, + txn_type="成品入库", + qty_delta=Decimal("40"), + weight_delta_kg=Decimal("-20"), + outside_weight_after_kg=Decimal("70"), + source_doc_type="库存流水", + source_doc_id=201, + source_line_id=10, + biz_time=self.now, + operator_user_id=1, + remark="单据批次号：PB-FGL-A", + created_at=self.now, + updated_at=self.now, + ) + finished_txn_b = ProductionBatchLedgerTxn( + id=21, + production_ledger_id=1, + txn_type="成品入库", + qty_delta=Decimal("60"), + weight_delta_kg=Decimal("-30"), + outside_weight_after_kg=Decimal("42"), + source_doc_type="库存流水", + source_doc_id=202, + source_line_id=11, + biz_time=self.now, + operator_user_id=1, + remark="单据批次号：PB-FGL-B", + created_at=self.now, + updated_at=self.now, + ) + delivery = Delivery( + id=1, + delivery_no="FH20260614001", + warehouse_id=1, + delivery_date=self.now, + status="POSTED", + created_at=self.now, + updated_at=self.now, + ) + delivery_item = DeliveryItem( + id=1, + delivery_id=1, + line_no=1, + product_item_id=2, + lot_id=11, + delivery_qty=Decimal("10"), + delivery_weight_kg=Decimal("5"), + unit_price=Decimal("12"), + line_amount=Decimal("120"), + status="POSTED", + created_at=self.now, + updated_at=self.now, + ) + self.db.add_all([finished_lot_a, finished_lot_b, finished_txn_a, finished_txn_b, delivery, delivery_item]) + self.db.commit() + + nodes, edges = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + + delivery_node = next(node for node in nodes if node.type == "delivery") + finished_a = next(node for node in nodes if node.type == "finished_inbound" and node.subtitle == "PB-FGL-A") + finished_b = next(node for node in nodes if node.type == "finished_inbound" and node.subtitle == "PB-FGL-B") + delivery_source_ids = {edge.source for edge in edges if edge.target == delivery_node.id} + self.assertIn(finished_b.id, delivery_source_ids) + self.assertNotIn(finished_a.id, delivery_source_ids) + self.assertNotIn("ledger-1", delivery_source_ids) + + def test_reopen_txn_uses_reopen_node_type_not_writeoff(self) -> None: + self.db.add( + ProductionBatchLedgerTxn( + id=30, + production_ledger_id=1, + txn_type="重新开工", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("0"), + outside_weight_after_kg=Decimal("42"), + source_doc_type="生产台账", + source_doc_id=1, + biz_time=self.now, + operator_user_id=1, + remark="重新开工", + created_at=self.now, + updated_at=self.now, + ) + ) + self.db.commit() + + nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + + reopen_node = next(node for node in nodes if node.title == "重新开工") + self.assertEqual(reopen_node.type, "ledger_reopen") + self.assertEqual(reopen_node.stage, "重新开工") + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/superpowers/plans/2026-06-14-material-batch-lifecycle-dag.md b/docs/superpowers/plans/2026-06-14-material-batch-lifecycle-dag.md new file mode 100644 index 0000000..872b97b --- /dev/null +++ b/docs/superpowers/plans/2026-06-14-material-batch-lifecycle-dag.md @@ -0,0 +1,1749 @@ +# Material Batch Lifecycle DAG 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:** 把经营总览里的“库存批次号生命周期 DAG”从旧的生产工单主链路，改造成以“材料库存批次号 + 产品”的生产台账闭环为主链路的材料批次 DAG。 + +**Architecture:** 后端保持现有 `/dashboard/material-lots/{lot_id}/lifecycle` 响应结构不变，新增独立的材料批次 DAG 构建服务，优先读取 `ProductionBatchLedger` 与 `ProductionBatchLedgerTxn`。前端继续复用现有 DAG 抽屉、画布、节点详情和边高亮能力，只替换节点类型、阶段文案、布局分组和视觉图例。智能报工小程序只作为可选证据层，受现有 `SMART_OPERATION_REPORT_ENABLED` 开关控制，不再参与闭环判断。 + +**Tech Stack:** FastAPI, SQLAlchemy, Pydantic, Vue 3, Vite, Node test runner, Python unittest/pytest-compatible tests. + +--- + +## Business DAG Model + +新 DAG 的主链路必须表达下面这个业务事实： + +```mermaid +flowchart LR + A["原材料库存批次号"] --> B["生产台账：库存批次号 + 产品"] + B --> C["生产出库流水"] + B --> D["库外材料余额"] + B --> E["生产台账入库/分支入库"] + E --> F["成品入库"] + E --> G["生产余料入库"] + E --> H["生产废料入库"] + E --> I["差异/结单核销"] + F --> J["销售出库"] + J --> K["退货/售后"] +``` + +报工小程序开启时，在生产台账下面增加“工序报工证据”节点： + +```mermaid +flowchart LR + B["生产台账：库存批次号 + 产品"] --> R["小程序报工证据"] + R --> R1["工序/员工/合格数"] + R --> R2["报废数/异常原因"] +``` + +报工小程序关闭时，不展示任何 `operation_report` / `work_order_operation` 节点，也不把报工数量作为 DAG 闭环条件。 + +--- + +## File Structure + +- Create: `backend/app/services/material_lifecycle_dag.py` + - 专职构建材料库存批次 DAG。 + - 输入：`Session` 与 `DashboardMaterialLotRead`。 + - 输出：`tuple[list[LifecycleNode], list[LifecycleEdge]]`。 + - 只负责读数据和组装节点/边，不提交事务，不修改库存。 + +- Modify: `backend/app/api/routes/dashboard.py` + - 保留销售订单生命周期接口。 + - 原材料批次生命周期接口改为调用 `build_material_lot_lifecycle_graph()`。 + - 保留旧工单图谱作为无生产台账历史数据的兼容降级路径。 + +- Create: `backend/tests/test_dashboard_material_lot_lifecycle.py` + - 覆盖新生产台账主链路。 + - 覆盖智能报工开关开启/关闭。 + - 覆盖旧工单数据降级展示。 + +- Modify: `frontend/src/views/DashboardView.vue` + - 标题从“原材料生命周期 DAG”调整为“材料库存批次闭环图”。 + - 阶段从旧 `生产工单/工序执行/工人报工` 改为 `原料入库/生产台账/生产出库/库外闭环/入库结果/销售去向`。 + - 新增图例和新节点类型布局。 + +- Modify: `frontend/src/styles/main.css` + - 增加新 DAG 节点 tone：`ledger`、`issue`、`balance`、`settlement`、`inbound`、`writeoff`、`evidence`。 + - 优化 DAG 图例、节点状态、长文本显示和连线层级。 + +- Create: `frontend/src/views/DashboardView.test.js` + - 使用现有 Node 源码断言测试风格，确认关键文案、节点类型、布局分类存在。 + +- No schema migration: + - 现有 `pp_production_batch_ledger`、`pp_production_batch_ledger_txn`、`pp_operation_report.production_ledger_id` 已满足本次 DAG 改造。 + +--- + +### Task 1: Backend Failing Tests For Production Ledger DAG + +**Files:** +- Create: `backend/tests/test_dashboard_material_lot_lifecycle.py` +- Read: `backend/tests/test_production_batch_ledger.py` +- Read: `backend/tests/test_selected_stock_lot_production_issue.py` + +- [ ] **Step 1: Create SQLite test fixture and seed helpers** + +Create `backend/tests/test_dashboard_material_lot_lifecycle.py` with this structure: + +```python +from __future__ import annotations + +import unittest +from datetime import datetime +from decimal import Decimal + +from sqlalchemy import BigInteger, create_engine, select +from sqlalchemy.ext.compiler import compiles +from sqlalchemy.orm import Session, sessionmaker + + +@compiles(BigInteger, "sqlite") +def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str: + _ = type_, compiler, kw + return "INTEGER" + + +import app.models.master_data # noqa: E402,F401 +import app.models.miniapp # noqa: E402,F401 +import app.models.operations # noqa: E402,F401 +import app.models.org # noqa: E402,F401 +import app.models.planning # noqa: E402,F401 +import app.models.sales # noqa: E402,F401 +from app.models.base import Base # noqa: E402 +from app.models.master_data import Item, Warehouse # noqa: E402 +from app.models.operations import ( # noqa: E402 + CompletionReceipt, + CompletionReceiptItem, + Delivery, + DeliveryItem, + InventoryTxn, + OperationReport, + ProductionBatchLedger, + ProductionBatchLedgerTxn, + StockLot, + WorkOrder, + WorkOrderMaterialIssue, + WorkOrderOperation, +) +from app.models.org import Employee, SystemConfig # noqa: E402 +from app.schemas.dashboard import DashboardMaterialLotRead # noqa: E402 +from app.services.material_lifecycle_dag import build_material_lot_lifecycle_graph # noqa: E402 + + +class MaterialLotLifecycleDagTest(unittest.TestCase): + def setUp(self) -> None: + engine = create_engine("sqlite+pysqlite:///:memory:", future=True) + Base.metadata.create_all(engine) + self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) + self.db: Session = self.SessionLocal() + self.now = datetime(2026, 6, 14, 9, 30, 0) + self._seed_common_master_data() + + def tearDown(self) -> None: + self.db.close() + + def _seed_common_master_data(self) -> None: + self.raw_item = Item( + id=1, + item_code="RM-001", + item_name="冷轧钢板", + item_type="RAW_MATERIAL", + unit_weight_kg=Decimal("1"), + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + self.product = Item( + id=2, + item_code="FG-001", + item_name="钢制碗", + item_type="FINISHED_GOOD", + unit_weight_kg=Decimal("0.45"), + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + self.raw_warehouse = Warehouse( + id=1, + warehouse_code="WH-RAW", + warehouse_name="原材料库", + warehouse_type="RAW", + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + self.finished_warehouse = Warehouse( + id=2, + warehouse_code="WH-FG", + warehouse_name="成品库", + warehouse_type="FINISHED", + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + self.raw_lot = StockLot( + id=1, + lot_no="YL0001", + lot_role="INBOUND_RAW", + item_id=self.raw_item.id, + warehouse_id=self.raw_warehouse.id, + source_doc_type="PURCHASE_RECEIPT", + source_doc_id=1, + inbound_qty=Decimal("0"), + inbound_weight_kg=Decimal("500"), + remaining_qty=Decimal("0"), + remaining_weight_kg=Decimal("300"), + locked_qty=Decimal("0"), + locked_weight_kg=Decimal("0"), + unit_cost=Decimal("4.2"), + quality_status="PASS", + status="AVAILABLE", + created_at=self.now, + updated_at=self.now, + ) + self.db.add_all([self.raw_item, self.product, self.raw_warehouse, self.finished_warehouse, self.raw_lot]) + self.db.flush() + + def _lot_read(self) -> DashboardMaterialLotRead: + return DashboardMaterialLotRead( + lot_id=self.raw_lot.id, + lot_no=self.raw_lot.lot_no, + material_item_id=self.raw_item.id, + material_code=self.raw_item.item_code, + material_name=self.raw_item.item_name, + specification=None, + warehouse_name=self.raw_warehouse.warehouse_name, + inbound_weight_kg=500, + remaining_weight_kg=300, + unit_cost=4.2, + inbound_amount=2100, + issued_batch_count=2, + issued_weight_kg=200, + finished_batch_count=1, + finished_qty=120, + delivery_count=1, + status="AVAILABLE", + quality_status="PASS", + created_at=self.now, + ) +``` + +- [ ] **Step 2: Add test for production ledger as the main chain** + +Append this test to `MaterialLotLifecycleDagTest`: + +```python + def test_material_lifecycle_uses_production_ledger_as_main_chain(self) -> None: + ledger = ProductionBatchLedger( + id=100, + material_lot_id=self.raw_lot.id, + material_lot_no=self.raw_lot.lot_no, + material_item_id=self.raw_item.id, + product_item_id=self.product.id, + status="在生产", + miniapp_selectable=1, + total_issued_weight_kg=Decimal("200"), + outside_weight_kg=Decimal("42"), + finished_inbound_qty=Decimal("120"), + surplus_inbound_weight_kg=Decimal("12"), + scrap_inbound_weight_kg=Decimal("18"), + writeoff_weight_kg=Decimal("0"), + first_issue_time=self.now, + last_issue_time=self.now, + created_at=self.now, + updated_at=self.now, + ) + self.db.add(ledger) + self.db.flush() + self.db.add_all( + [ + ProductionBatchLedgerTxn( + id=1001, + production_ledger_id=ledger.id, + txn_type="生产出库", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("200"), + outside_weight_after_kg=Decimal("200"), + source_doc_type="库存流水", + source_doc_id=501, + source_line_id=self.raw_lot.id, + biz_time=self.now, + operator_user_id=1, + remark="生产出库", + created_at=self.now, + updated_at=self.now, + ), + ProductionBatchLedgerTxn( + id=1002, + production_ledger_id=ledger.id, + txn_type="成品入库", + qty_delta=Decimal("120"), + weight_delta_kg=Decimal("-72"), + outside_weight_after_kg=Decimal("128"), + source_doc_type="库存流水", + source_doc_id=601, + source_line_id=701, + biz_time=self.now, + operator_user_id=1, + remark="生产台账入库；成品入库联动生产台账；单据批次号：PIN202606140001", + created_at=self.now, + updated_at=self.now, + ), + ProductionBatchLedgerTxn( + id=1003, + production_ledger_id=ledger.id, + txn_type="生产余料入库", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("-12"), + outside_weight_after_kg=Decimal("116"), + source_doc_type="库存流水", + source_doc_id=602, + source_line_id=self.raw_lot.id, + biz_time=self.now, + operator_user_id=1, + remark="生产台账入库；生产余料入库联动生产台账；单据批次号：PIN202606140001", + created_at=self.now, + updated_at=self.now, + ), + ProductionBatchLedgerTxn( + id=1004, + production_ledger_id=ledger.id, + txn_type="生产废料入库", + qty_delta=Decimal("0"), + weight_delta_kg=Decimal("-18"), + outside_weight_after_kg=Decimal("98"), + source_doc_type="库存流水", + source_doc_id=603, + source_line_id=801, + biz_time=self.now, + operator_user_id=1, + remark="生产台账入库；生产废料入库联动生产台账；单据批次号：PIN202606140001", + created_at=self.now, + updated_at=self.now, + ), + ] + ) + finished_lot = StockLot( + id=701, + lot_no="FGL202606140001", + lot_role="FINISHED_FROM_RAW_BATCH", + parent_lot_id=self.raw_lot.id, + item_id=self.product.id, + warehouse_id=self.finished_warehouse.id, + source_doc_type="PRODUCTION_LEDGER_IN", + source_doc_id=ledger.id, + source_line_id=None, + source_material_lot_id=self.raw_lot.id, + source_material_sub_batch_no=self.raw_lot.lot_no, + source_material_summary=f"{self.raw_item.item_name}：{self.raw_lot.lot_no}", + inbound_qty=Decimal("120"), + inbound_weight_kg=Decimal("54"), + remaining_qty=Decimal("70"), + remaining_weight_kg=Decimal("31.5"), + locked_qty=Decimal("0"), + locked_weight_kg=Decimal("0"), + unit_cost=Decimal("2.52"), + quality_status="PASS", + status="AVAILABLE", + created_at=self.now, + updated_at=self.now, + ) + self.db.add(finished_lot) + self.db.flush() + + nodes, edges = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + node_types = {node.type for node in nodes} + edge_labels = {edge.label for edge in edges} + + self.assertIn("raw_material_lot", node_types) + self.assertIn("production_ledger", node_types) + self.assertIn("production_issue_txn", node_types) + self.assertIn("ledger_outside_balance", node_types) + self.assertIn("production_ledger_settlement", node_types) + self.assertIn("finished_inbound", node_types) + self.assertIn("surplus_inbound", node_types) + self.assertIn("scrap_inbound", node_types) + self.assertNotIn("work_order", node_types) + self.assertIn("进入生产台账", edge_labels) + self.assertIn("入库闭环", edge_labels) + + ledger_node = next(node for node in nodes if node.type == "production_ledger") + self.assertEqual(ledger_node.title, "YL0001 · 钢制碗") + self.assertEqual(ledger_node.stage, "生产台账") + self.assertIn(("库外余额", "42 kg"), [(item.label, item.value) for item in ledger_node.metrics]) +``` + +- [ ] **Step 3: Add test for miniapp evidence switch** + +Append this test: + +```python + def test_material_lifecycle_hides_miniapp_evidence_when_switch_closed(self) -> None: + ledger = ProductionBatchLedger( + id=101, + material_lot_id=self.raw_lot.id, + material_lot_no=self.raw_lot.lot_no, + material_item_id=self.raw_item.id, + product_item_id=self.product.id, + status="在生产", + miniapp_selectable=1, + total_issued_weight_kg=Decimal("80"), + outside_weight_kg=Decimal("80"), + finished_inbound_qty=Decimal("0"), + surplus_inbound_weight_kg=Decimal("0"), + scrap_inbound_weight_kg=Decimal("0"), + writeoff_weight_kg=Decimal("0"), + first_issue_time=self.now, + last_issue_time=self.now, + created_at=self.now, + updated_at=self.now, + ) + employee = Employee( + id=1, + employee_code="EMP001", + employee_name="张三", + dept_id=1, + is_operator=1, + is_workshop_staff=1, + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + report = OperationReport( + id=1, + report_no="MOP-001", + work_order_id=None, + work_order_operation_id=None, + production_ledger_id=ledger.id, + employee_id=employee.id, + equipment_id=None, + report_source="MINIAPP", + shift_code=None, + start_time=self.now, + end_time=self.now, + report_qty=Decimal("30"), + good_qty=Decimal("28"), + scrap_qty=Decimal("2"), + rework_qty=Decimal("0"), + scrap_reason="调机废品", + downtime_minutes=Decimal("0"), + extra_cost_amount=Decimal("0"), + is_abnormal=1, + approved_by=None, + approved_at=None, + remark="小程序报工同步", + created_at=self.now, + updated_at=self.now, + ) + self.db.add_all([ledger, employee, report]) + self.db.flush() + + nodes_open, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + self.assertIn("operation_report", {node.type for node in nodes_open}) + + self.db.add( + SystemConfig( + id=1, + config_code="SMART_OPERATION_REPORT_ENABLED", + config_name="对接智能报工小程序", + config_value="关闭", + status="ACTIVE", + created_at=self.now, + updated_at=self.now, + ) + ) + self.db.flush() + + nodes_closed, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read()) + self.assertNotIn("operation_report", {node.type for node in nodes_closed}) +``` + +- [ ] **Step 4: Run the new test and verify it fails for missing service** + +Run: + +```bash +cd backend +python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q +``` + +Expected: + +```text +ModuleNotFoundError: No module named 'app.services.material_lifecycle_dag' +``` + +--- + +### Task 2: Backend DAG Builder Service + +**Files:** +- Create: `backend/app/services/material_lifecycle_dag.py` +- Read: `backend/app/api/routes/dashboard.py` +- Read: `backend/app/api/routes/production.py:1349-1368` +- Read: `backend/app/services/document_archives.py:168-179` + +- [ ] **Step 1: Create service with formatter helpers and add-node/add-edge utilities** + +Create `backend/app/services/material_lifecycle_dag.py`: + +```python +from __future__ import annotations + +from collections import defaultdict +from decimal import Decimal +from typing import Any + +from sqlalchemy import false, func, or_, select +from sqlalchemy.orm import Session, aliased + +from app.models.master_data import Item, Warehouse +from app.models.operations import ( + CompletionReceipt, + CompletionReceiptItem, + Delivery, + DeliveryItem, + OperationReport, + ProductionBatchLedger, + ProductionBatchLedgerTxn, + StockLot, + WorkOrder, + WorkOrderMaterialIssue, + WorkOrderOperation, +) +from app.models.org import Employee +from app.models.sales import Customer, SalesOrder +from app.schemas.dashboard import DashboardMaterialLotRead, LifecycleDetailItem, LifecycleEdge, LifecycleNode +from app.services.document_archives import production_document_batch_no_from_text +from app.services.system_config import get_smart_operation_report_enabled + + +PRODUCTION_LEDGER_INBOUND_TXN_TYPES = {"成品入库", "生产余料入库", "生产废料入库"} +PRODUCTION_LEDGER_CLOSING_TXN_TYPES = {"结单核销", "该批材料结单", "重新开工"} + + +def _num(value: Any) -> float: + if value is None: + return 0.0 + if isinstance(value, Decimal): + return float(value) + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + + +def _fmt_qty(value: Any, digits: int = 2) -> str: + numeric = _num(value) + if abs(numeric - round(numeric)) < 0.000001: + return f"{int(round(numeric))}" + return f"{numeric:.{digits}f}".rstrip("0").rstrip(".") + + +def _fmt_weight(value: Any) -> str: + return f"{_fmt_qty(value, 3)} kg" + + +def _fmt_money(value: Any) -> str: + return f"¥{_num(value):,.2f}" + + +def _fmt_datetime(value: Any) -> str: + if not value: + return "-" + return str(value).replace("T", " ")[:19] + + +def _detail(label: str, value: Any) -> LifecycleDetailItem: + text = "-" if value is None or value == "" else str(value) + return LifecycleDetailItem(label=label, value=text) + + +def _node( + *, + node_id: str, + node_type: str, + title: str, + stage: str, + level: int, + subtitle: str | None = None, + status: str | None = None, + tone: str = "normal", + summary: str | None = None, + metrics: list[tuple[str, Any]] | None = None, + details: list[tuple[str, Any]] | None = None, + raw: dict[str, Any] | None = None, +) -> LifecycleNode: + return LifecycleNode( + id=node_id, + type=node_type, + title=title, + subtitle=subtitle, + status=status, + stage=stage, + level=level, + tone=tone, + summary=summary, + metrics=[_detail(label, value) for label, value in (metrics or [])], + details=[_detail(label, value) for label, value in (details or [])], + raw=raw or {}, + ) + + +def _edge(source: str, target: str, label: str | None = None) -> LifecycleEdge: + return LifecycleEdge(id=f"{source}->{target}", source=source, target=target, label=label) +``` + +- [ ] **Step 2: Add graph accumulator** + +Append: + +```python +class _LifecycleGraph: + def __init__(self) -> None: + self.nodes: list[LifecycleNode] = [] + self.edges: list[LifecycleEdge] = [] + self.node_ids: set[str] = set() + self.edge_ids: set[str] = set() + + def add_node(self, node: LifecycleNode) -> None: + if node.id not in self.node_ids: + self.node_ids.add(node.id) + self.nodes.append(node) + + def add_edge(self, edge: LifecycleEdge) -> None: + if edge.source in self.node_ids and edge.target in self.node_ids and edge.id not in self.edge_ids: + self.edge_ids.add(edge.id) + self.edges.append(edge) +``` + +- [ ] **Step 3: Add raw lot node builder** + +Append: + +```python +def _add_raw_lot_node(graph: _LifecycleGraph, lot: DashboardMaterialLotRead) -> str: + node_id = f"raw-lot-{lot.lot_id}" + graph.add_node( + _node( + node_id=node_id, + node_type="raw_material_lot", + title=lot.lot_no, + subtitle=f"{lot.material_code} · {lot.material_name}", + status=lot.status, + tone="success" if lot.status in {"AVAILABLE", "PASS", "PASSED"} else "normal", + stage="原料入库", + level=0, + summary="材料库存批次闭环的起点。这里记录该批材料的入库成本、剩余库存、质检状态，以及后续生产台账和销售去向。", + metrics=[ + ("入库重量", _fmt_weight(lot.inbound_weight_kg)), + ("剩余重量", _fmt_weight(lot.remaining_weight_kg)), + ("入库金额", _fmt_money(lot.inbound_amount)), + ], + details=[ + ("库存批次号", lot.lot_no), + ("原材料编码", lot.material_code), + ("原材料名称", lot.material_name), + ("仓库", lot.warehouse_name), + ("入库单价", f"{_fmt_money(lot.unit_cost)}/kg"), + ("已领用重量", _fmt_weight(lot.issued_weight_kg)), + ], + raw=lot.model_dump(mode="json"), + ) + ) + return node_id +``` + +- [ ] **Step 4: Add production ledger query helper** + +Append: + +```python +def _production_ledger_rows(db: Session, lot_id: int): + material_item = aliased(Item) + product_item = aliased(Item) + return db.execute( + select( + ProductionBatchLedger, + material_item.item_code.label("material_code"), + material_item.item_name.label("material_name"), + product_item.item_code.label("product_code"), + product_item.item_name.label("product_name"), + ) + .join(material_item, material_item.id == ProductionBatchLedger.material_item_id, isouter=True) + .join(product_item, product_item.id == ProductionBatchLedger.product_item_id, isouter=True) + .where(ProductionBatchLedger.material_lot_id == lot_id) + .order_by(ProductionBatchLedger.first_issue_time.asc(), ProductionBatchLedger.id.asc()) + ).all() +``` + +- [ ] **Step 5: Add production ledger node and balance node builders** + +Append: + +```python +def _ledger_tone(ledger: ProductionBatchLedger) -> str: + status = str(ledger.status or "") + outside = _num(ledger.outside_weight_kg) + if status == "锁单": + return "success" if abs(outside) < 0.000001 else "warning" + if status == "在生产" and outside > 0: + return "warning" + return "normal" + + +def _ledger_visual_status(ledger: ProductionBatchLedger) -> str: + status = str(ledger.status or "在生产") + if status == "在生产" and _num(ledger.total_issued_weight_kg) > 0 and _num(ledger.outside_weight_kg) > 0: + return "部分闭环" if ( + _num(ledger.finished_inbound_qty) > 0 + or _num(ledger.surplus_inbound_weight_kg) > 0 + or _num(ledger.scrap_inbound_weight_kg) > 0 + ) else "在生产" + return status + + +def _add_ledger_node(graph: _LifecycleGraph, raw_node_id: str, row: Any) -> str: + ledger: ProductionBatchLedger = row.ProductionBatchLedger + product_name = row.product_name or f"产品#{ledger.product_item_id}" + product_code = row.product_code or "-" + node_id = f"production-ledger-{ledger.id}" + graph.add_node( + _node( + node_id=node_id, + node_type="production_ledger", + title=f"{ledger.material_lot_no} · {product_name}", + subtitle=f"生产台账 #{ledger.id}", + status=_ledger_visual_status(ledger), + tone=_ledger_tone(ledger), + stage="生产台账", + level=1, + summary="这一行代表该材料库存批次为了生产该产品而处于仓库外的总量，生产出库累加，成品/余料/废料/核销入库逐步闭环。", + metrics=[ + ("累计领料", _fmt_weight(ledger.total_issued_weight_kg)), + ("库外余额", _fmt_weight(ledger.outside_weight_kg)), + ("成品入库", _fmt_qty(ledger.finished_inbound_qty)), + ("余料入库", _fmt_weight(ledger.surplus_inbound_weight_kg)), + ("废料入库", _fmt_weight(ledger.scrap_inbound_weight_kg)), + ], + details=[ + ("材料库存批次号", ledger.material_lot_no), + ("产品编码", product_code), + ("产品名称", product_name), + ("小程序可选", "是" if int(ledger.miniapp_selectable or 0) == 1 else "否"), + ("首次出库", _fmt_datetime(ledger.first_issue_time)), + ("最近出库", _fmt_datetime(ledger.last_issue_time)), + ("锁单时间", _fmt_datetime(ledger.locked_at)), + ("备注", ledger.remark), + ], + raw={ + "production_ledger_id": ledger.id, + "material_lot_id": ledger.material_lot_id, + "material_lot_no": ledger.material_lot_no, + "product_item_id": ledger.product_item_id, + "product_name": product_name, + }, + ) + ) + graph.add_edge(_edge(raw_node_id, node_id, "进入生产台账")) + return node_id + + +def _add_outside_balance_node(graph: _LifecycleGraph, ledger: ProductionBatchLedger, ledger_node_id: str) -> str: + node_id = f"outside-balance-{ledger.id}" + graph.add_node( + _node( + node_id=node_id, + node_type="ledger_outside_balance", + title="库外材料余额", + subtitle=ledger.material_lot_no, + status=_ledger_visual_status(ledger), + tone=_ledger_tone(ledger), + stage="库外闭环", + level=3, + summary="库外余额 = 累计生产出库 - 成品耗用折算 - 余料入库 - 废料入库 - 结单核销。它表示现场仍未闭环的这批材料重量。", + metrics=[ + ("库外余额", _fmt_weight(ledger.outside_weight_kg)), + ("结单核销", _fmt_weight(ledger.writeoff_weight_kg)), + ], + details=[ + ("状态", _ledger_visual_status(ledger)), + ("锁单次数", int(ledger.lock_count or 0)), + ("重新开工时间", _fmt_datetime(ledger.reopened_at)), + ], + raw={"production_ledger_id": ledger.id}, + ) + ) + graph.add_edge(_edge(ledger_node_id, node_id, "库外余额")) + return node_id +``` + +- [ ] **Step 6: Add transaction grouping and transaction node builders** + +Append: + +```python +def _ledger_txns_by_ledger_id(db: Session, ledger_ids: list[int]) -> dict[int, list[ProductionBatchLedgerTxn]]: + if not ledger_ids: + return {} + rows = db.scalars( + select(ProductionBatchLedgerTxn) + .where(ProductionBatchLedgerTxn.production_ledger_id.in_(ledger_ids)) + .order_by(ProductionBatchLedgerTxn.biz_time.asc(), ProductionBatchLedgerTxn.id.asc()) + ).all() + grouped: dict[int, list[ProductionBatchLedgerTxn]] = defaultdict(list) + for txn in rows: + grouped[int(txn.production_ledger_id)].append(txn) + return grouped + + +def _settlement_key(txn: ProductionBatchLedgerTxn) -> str: + batch_no = production_document_batch_no_from_text(txn.remark) + if batch_no: + return f"batch:{batch_no}" + return f"txn:{txn.id}" + + +def _add_issue_txn_nodes( + graph: _LifecycleGraph, + ledger_node_id: str, + txns: list[ProductionBatchLedgerTxn], +) -> list[str]: + issue_node_ids: list[str] = [] + for txn in txns: + if txn.txn_type != "生产出库": + continue + node_id = f"production-issue-txn-{txn.id}" + graph.add_node( + _node( + node_id=node_id, + node_type="production_issue_txn", + title=f"生产出库 #{txn.source_doc_id or txn.id}", + subtitle=txn.remark or "生产出库", + status="已出库", + tone="warning", + stage="生产出库", + level=2, + summary="本节点表示一次从该材料库存批次领料出库到生产现场。同一材料批次和同一产品的多次出库会汇总到同一生产台账。", + metrics=[ + ("出库重量", _fmt_weight(abs(_num(txn.weight_delta_kg)))), + ("出库后库外余额", _fmt_weight(txn.outside_weight_after_kg)), + ], + details=[ + ("业务时间", _fmt_datetime(txn.biz_time)), + ("来源单据", f"{txn.source_doc_type or '-'} #{txn.source_doc_id or '-'}"), + ("说明", txn.remark), + ], + raw={"production_ledger_txn_id": txn.id, "production_ledger_id": txn.production_ledger_id}, + ) + ) + graph.add_edge(_edge(ledger_node_id, node_id, "生产出库")) + issue_node_ids.append(node_id) + return issue_node_ids +``` + +- [ ] **Step 7: Add inbound settlement nodes** + +Append: + +```python +def _add_inbound_settlement_nodes( + graph: _LifecycleGraph, + ledger_node_id: str, + balance_node_id: str, + txns: list[ProductionBatchLedgerTxn], +) -> None: + grouped: dict[str, list[ProductionBatchLedgerTxn]] = defaultdict(list) + for txn in txns: + if txn.txn_type in PRODUCTION_LEDGER_INBOUND_TXN_TYPES or txn.txn_type in PRODUCTION_LEDGER_CLOSING_TXN_TYPES: + grouped[_settlement_key(txn)].append(txn) + + txn_order = {"成品入库": 1, "生产余料入库": 2, "生产废料入库": 3, "结单核销": 4, "该批材料结单": 5, "重新开工": 6} + for group_index, group_txns in enumerate(grouped.values(), start=1): + ordered = sorted(group_txns, key=lambda item: (txn_order.get(str(item.txn_type or ""), 99), item.id)) + batch_no = production_document_batch_no_from_text(ordered[0].remark) + settlement_id = f"ledger-settlement-{ordered[0].production_ledger_id}-{batch_no or ordered[0].id}" + graph.add_node( + _node( + node_id=settlement_id, + node_type="production_ledger_settlement", + title=f"入库闭环 #{batch_no or group_index}", + subtitle="生产台账入库" if batch_no else "分支入库/结单事件", + status="已记录", + tone="success", + stage="入库结算", + level=4, + summary="一次生产台账入库可能同时生成成品、余料、废料和结单核销；单独分支入库也会在这里形成闭环事件。", + metrics=[ + ("成品数量", _fmt_qty(sum(_num(txn.qty_delta) for txn in ordered if txn.txn_type == "成品入库"))), + ("余料重量", _fmt_weight(sum(abs(_num(txn.weight_delta_kg)) for txn in ordered if txn.txn_type == "生产余料入库"))), + ("废料重量", _fmt_weight(sum(abs(_num(txn.weight_delta_kg)) for txn in ordered if txn.txn_type == "生产废料入库"))), + ], + details=[ + ("业务时间", _fmt_datetime(ordered[0].biz_time)), + ("单据批次", batch_no or "-"), + ("说明", "；".join(str(txn.remark or "") for txn in ordered if txn.remark) or "-"), + ], + raw={"production_ledger_id": ordered[0].production_ledger_id, "txn_ids": [txn.id for txn in ordered]}, + ) + ) + graph.add_edge(_edge(ledger_node_id, settlement_id, "入库闭环")) + graph.add_edge(_edge(balance_node_id, settlement_id, "冲减库外余额")) + + for txn in ordered: + if txn.txn_type == "成品入库": + node_type = "finished_inbound" + title = f"成品入库 #{txn.source_doc_id or txn.id}" + stage = "成品入库" + metric = ("入库数量", _fmt_qty(txn.qty_delta)) + elif txn.txn_type == "生产余料入库": + node_type = "surplus_inbound" + title = f"余料入库 #{txn.source_doc_id or txn.id}" + stage = "余料入库" + metric = ("入库重量", _fmt_weight(abs(_num(txn.weight_delta_kg)))) + elif txn.txn_type == "生产废料入库": + node_type = "scrap_inbound" + title = f"废料入库 #{txn.source_doc_id or txn.id}" + stage = "废料入库" + metric = ("入库重量", _fmt_weight(abs(_num(txn.weight_delta_kg)))) + elif txn.txn_type in {"结单核销", "该批材料结单"}: + node_type = "ledger_writeoff" + title = txn.txn_type + stage = "结单核销" + metric = ("核销重量", _fmt_weight(abs(_num(txn.weight_delta_kg)))) + elif txn.txn_type == "重新开工": + node_type = "ledger_reopen" + title = "重新开工" + stage = "重新开工" + metric = ("新增库外重量", _fmt_weight(abs(_num(txn.weight_delta_kg)))) + else: + continue + + child_id = f"{node_type}-{txn.id}" + graph.add_node( + _node( + node_id=child_id, + node_type=node_type, + title=title, + subtitle=txn.remark or txn.txn_type, + status="已记录", + tone="success" if node_type != "ledger_writeoff" else "warning", + stage=stage, + level=5, + summary="该节点来自生产台账交易流水，表示该批材料的一项入库或结单结果。", + metrics=[ + metric, + ("处理后库外余额", _fmt_weight(txn.outside_weight_after_kg)), + ], + details=[ + ("业务时间", _fmt_datetime(txn.biz_time)), + ("来源单据", f"{txn.source_doc_type or '-'} #{txn.source_doc_id or '-'}"), + ("说明", txn.remark), + ], + raw={"production_ledger_txn_id": txn.id, "production_ledger_id": txn.production_ledger_id}, + ) + ) + graph.add_edge(_edge(settlement_id, child_id, txn.txn_type)) +``` + +- [ ] **Step 8: Add optional miniapp report evidence nodes** + +Append: + +```python +def _add_operation_report_evidence_nodes( + db: Session, + graph: _LifecycleGraph, + ledger_node_by_id: dict[int, str], +) -> None: + ledger_ids = sorted(ledger_node_by_id) + if not ledger_ids or not get_smart_operation_report_enabled(db): + return + + rows = db.execute( + select( + OperationReport.id.label("report_id"), + OperationReport.report_no.label("report_no"), + OperationReport.production_ledger_id.label("production_ledger_id"), + Employee.employee_name.label("employee_name"), + OperationReport.report_qty.label("report_qty"), + OperationReport.good_qty.label("good_qty"), + OperationReport.scrap_qty.label("scrap_qty"), + OperationReport.rework_qty.label("rework_qty"), + OperationReport.is_abnormal.label("is_abnormal"), + OperationReport.scrap_reason.label("scrap_reason"), + OperationReport.start_time.label("start_time"), + OperationReport.end_time.label("end_time"), + ) + .join(Employee, Employee.id == OperationReport.employee_id) + .where(OperationReport.production_ledger_id.in_(ledger_ids)) + .order_by(OperationReport.start_time.asc(), OperationReport.id.asc()) + ).mappings().all() + + for row in rows: + parent_id = ledger_node_by_id.get(int(row["production_ledger_id"])) + if not parent_id: + continue + node_id = f"operation-report-{row['report_id']}" + graph.add_node( + _node( + node_id=node_id, + node_type="operation_report", + title=row["report_no"], + subtitle=f"{row['employee_name']} · 小程序报工", + status="异常" if row["is_abnormal"] else "正常", + tone="danger" if row["is_abnormal"] else "normal", + stage="报工证据", + level=3, + summary="该节点只作为现场报工证据展示，不参与材料批次闭环判定。", + metrics=[ + ("报工数", _fmt_qty(row["report_qty"])), + ("合格数", _fmt_qty(row["good_qty"])), + ("报废数", _fmt_qty(row["scrap_qty"])), + ], + details=[ + ("员工", row["employee_name"]), + ("开始", _fmt_datetime(row["start_time"])), + ("结束", _fmt_datetime(row["end_time"])), + ("异常原因", row["scrap_reason"] or ("报工异常" if row["is_abnormal"] else "-")), + ], + raw=dict(row), + ) + ) + graph.add_edge(_edge(parent_id, node_id, "报工证据")) +``` + +- [ ] **Step 9: Add delivery nodes from finished stock lots** + +Append: + +```python +def _add_delivery_nodes_for_ledgers( + db: Session, + graph: _LifecycleGraph, + ledger_ids: list[int], +) -> None: + if not ledger_ids: + return + + finished_lots = db.execute( + select( + StockLot.id.label("lot_id"), + StockLot.lot_no.label("lot_no"), + StockLot.source_doc_id.label("production_ledger_id"), + ) + .join(Item, Item.id == StockLot.item_id) + .where( + StockLot.source_doc_id.in_(ledger_ids), + StockLot.source_doc_type.in_(["PRODUCTION_LEDGER_IN", "生产台账"]), + Item.item_type == "FINISHED_GOOD", + ) + ).mappings().all() + finished_lot_by_id = {int(row["lot_id"]): row for row in finished_lots} + if not finished_lot_by_id: + return + + order_customer = aliased(Customer) + direct_customer = aliased(Customer) + delivery_rows = db.execute( + select( + DeliveryItem.id.label("delivery_item_id"), + DeliveryItem.lot_id.label("lot_id"), + Delivery.delivery_no.label("delivery_no"), + SalesOrder.order_no.label("order_no"), + func.coalesce(direct_customer.customer_name, order_customer.customer_name).label("customer_name"), + Delivery.delivery_date.label("delivery_date"), + DeliveryItem.delivery_qty.label("delivery_qty"), + DeliveryItem.delivery_weight_kg.label("delivery_weight_kg"), + DeliveryItem.line_amount.label("line_amount"), + DeliveryItem.status.label("status"), + ) + .join(Delivery, Delivery.id == DeliveryItem.delivery_id) + .outerjoin(SalesOrder, SalesOrder.id == Delivery.sales_order_id) + .outerjoin(order_customer, order_customer.id == SalesOrder.customer_id) + .outerjoin(direct_customer, direct_customer.id == Delivery.customer_id) + .where(DeliveryItem.lot_id.in_(finished_lot_by_id)) + .order_by(Delivery.delivery_date.asc(), DeliveryItem.id.asc()) + ).mappings().all() + + for row in delivery_rows: + finished_lot = finished_lot_by_id.get(int(row["lot_id"])) + if not finished_lot: + continue + inbound_parent_id = f"finished-inbound-by-lot-{row['lot_id']}" + if inbound_parent_id not in graph.node_ids: + inbound_parent_id = f"production-ledger-{finished_lot['production_ledger_id']}" + node_id = f"delivery-{row['delivery_item_id']}" + graph.add_node( + _node( + node_id=node_id, + node_type="delivery", + title=row["delivery_no"], + subtitle=row["order_no"] or row["customer_name"] or "销售出库", + status=row["status"], + tone="success", + stage="销售去向", + level=6, + summary="该成品销售出库可向前追溯到材料库存批次号与生产台账。", + metrics=[ + ("发货数量", _fmt_qty(row["delivery_qty"])), + ("发货重量", _fmt_weight(row["delivery_weight_kg"])), + ("发货金额", _fmt_money(row["line_amount"])), + ], + details=[ + ("发货单号", row["delivery_no"]), + ("销售订单", row["order_no"]), + ("客户", row["customer_name"]), + ("成品库存批次号", finished_lot["lot_no"]), + ("发货时间", _fmt_datetime(row["delivery_date"])), + ], + raw=dict(row), + ) + ) + graph.add_edge(_edge(inbound_parent_id, node_id, "销售出库")) +``` + +- [ ] **Step 10: Add public graph function** + +Append: + +```python +def build_material_lot_lifecycle_graph( + db: Session, + lot: DashboardMaterialLotRead, +) -> tuple[list[LifecycleNode], list[LifecycleEdge]]: + graph = _LifecycleGraph() + raw_node_id = _add_raw_lot_node(graph, lot) + + ledger_rows = _production_ledger_rows(db, lot.lot_id) + ledger_ids = [int(row.ProductionBatchLedger.id) for row in ledger_rows] + txns_by_ledger_id = _ledger_txns_by_ledger_id(db, ledger_ids) + ledger_node_by_id: dict[int, str] = {} + + for row in ledger_rows: + ledger: ProductionBatchLedger = row.ProductionBatchLedger + ledger_node_id = _add_ledger_node(graph, raw_node_id, row) + ledger_node_by_id[int(ledger.id)] = ledger_node_id + balance_node_id = _add_outside_balance_node(graph, ledger, ledger_node_id) + txns = txns_by_ledger_id.get(int(ledger.id), []) + _add_issue_txn_nodes(graph, ledger_node_id, txns) + _add_inbound_settlement_nodes(graph, ledger_node_id, balance_node_id, txns) + + _add_operation_report_evidence_nodes(db, graph, ledger_node_by_id) + _add_delivery_nodes_for_ledgers(db, graph, ledger_ids) + return graph.nodes, graph.edges +``` + +- [ ] **Step 11: Run backend test and verify first behavior passes enough to expose remaining gaps** + +Run: + +```bash +cd backend +python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q +``` + +Expected after this task: + +```text +2 passed +``` + +If `Employee.dept_id` foreign key causes a missing department failure in SQLite, add a `Department` seed to the test fixture with `id=1` before adding `Employee`. + +--- + +### Task 3: Dashboard Route Integration And Legacy Fallback + +**Files:** +- Modify: `backend/app/api/routes/dashboard.py` +- Test: `backend/tests/test_dashboard_material_lot_lifecycle.py` + +- [ ] **Step 1: Import new service in dashboard route** + +Modify the imports in `backend/app/api/routes/dashboard.py`: + +```python +from app.services.material_lifecycle_dag import build_material_lot_lifecycle_graph +``` + +- [ ] **Step 2: Preserve old material lifecycle body as a legacy helper** + +In `backend/app/api/routes/dashboard.py`, before the current `@router.get("/material-lots/{lot_id}/lifecycle"...` route, create: + +```python +def _build_legacy_material_lot_lifecycle_graph( + db: Session, + lot: DashboardMaterialLotRead, +) -> tuple[list[LifecycleNode], list[LifecycleEdge]]: + nodes: list[LifecycleNode] = [] + edges: list[LifecycleEdge] = [] + node_ids: set[str] = set() + edge_ids: set[str] = set() + + def add_node(node: LifecycleNode) -> None: + if node.id not in node_ids: + node_ids.add(node.id) + nodes.append(node) + + def add_edge(edge: LifecycleEdge) -> None: + if edge.source in node_ids and edge.target in node_ids and edge.id not in edge_ids: + edge_ids.add(edge.id) + edges.append(edge) + + # Move the old route's graph-building code here unchanged, starting at: + # lot_node_id = f"raw-lot-{lot.lot_id}" + # and ending before: + # return MaterialLotLifecycleRead(lot=lot, nodes=nodes, edges=edges) + return nodes, edges +``` + +When moving code, keep all old node types unchanged inside this helper. This fallback is only for historical data that does not have `ProductionBatchLedger`. + +- [ ] **Step 3: Replace route body with new-first fallback logic** + +Replace `get_material_lot_lifecycle()` after the lot lookup with: + +```python + lot = DashboardMaterialLotRead.model_validate(dict(lot_row)) + nodes, edges = build_material_lot_lifecycle_graph(db, lot) + has_production_ledger = any(node.type == "production_ledger" for node in nodes) + if not has_production_ledger: + nodes, edges = _build_legacy_material_lot_lifecycle_graph(db, lot) + return MaterialLotLifecycleRead(lot=lot, nodes=nodes, edges=edges) +``` + +- [ ] **Step 4: Add direct route integration test** + +Append to `backend/tests/test_dashboard_material_lot_lifecycle.py`: + +```python + def test_dashboard_route_returns_new_graph_when_production_ledger_exists(self) -> None: + from app.api.routes.dashboard import get_material_lot_lifecycle + + ledger = ProductionBatchLedger( + id=102, + material_lot_id=self.raw_lot.id, + material_lot_no=self.raw_lot.lot_no, + material_item_id=self.raw_item.id, + product_item_id=self.product.id, + status="在生产", + miniapp_selectable=1, + total_issued_weight_kg=Decimal("60"), + outside_weight_kg=Decimal("60"), + finished_inbound_qty=Decimal("0"), + surplus_inbound_weight_kg=Decimal("0"), + scrap_inbound_weight_kg=Decimal("0"), + writeoff_weight_kg=Decimal("0"), + first_issue_time=self.now, + last_issue_time=self.now, + created_at=self.now, + updated_at=self.now, + ) + self.db.add(ledger) + self.db.commit() + + result = get_material_lot_lifecycle(self.raw_lot.id, self.db) + self.assertIn("production_ledger", {node.type for node in result.nodes}) + self.assertEqual(result.lot.lot_no, "YL0001") +``` + +- [ ] **Step 5: Run backend lifecycle test** + +Run: + +```bash +cd backend +python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q +``` + +Expected: + +```text +3 passed +``` + +- [ ] **Step 6: Run nearby backend regression tests** + +Run: + +```bash +cd backend +python -m pytest tests/test_production_batch_ledger.py tests/test_selected_stock_lot_production_issue.py tests/test_smart_operation_report_config.py -q +``` + +Expected: + +```text +passed +``` + +--- + +### Task 4: Frontend DAG Copy, Stage Model, And Layout Classification + +**Files:** +- Modify: `frontend/src/views/DashboardView.vue` +- Create: `frontend/src/views/DashboardView.test.js` + +- [ ] **Step 1: Add frontend source assertions** + +Create `frontend/src/views/DashboardView.test.js`: + +```javascript +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { dirname, join } from "node:path"; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const source = readFileSync(join(__dirname, "DashboardView.vue"), "utf8"); + +describe("DashboardView material batch lifecycle DAG", () => { + it("uses current material-batch closed-loop copy instead of legacy work-order wording", () => { + assert.match(source, /材料库存批次闭环图/); + assert.match(source, /生产台账/); + assert.match(source, /库外闭环/); + assert.match(source, /入库结果/); + assert.doesNotMatch(source, /原材料生命周期 DAG/); + }); + + it("classifies new production-ledger node types for layout and visual treatment", () => { + assert.match(source, /production_ledger/); + assert.match(source, /production_issue_txn/); + assert.match(source, /ledger_outside_balance/); + assert.match(source, /production_ledger_settlement/); + assert.match(source, /finished_inbound/); + assert.match(source, /surplus_inbound/); + assert.match(source, /scrap_inbound/); + assert.match(source, /ledger_writeoff/); + assert.match(source, /operation_report/); + }); + + it("keeps miniapp reports visually labeled as evidence, not closure control", () => { + assert.match(source, /报工证据/); + assert.match(source, /只作为现场报工证据/); + }); +}); +``` + +- [ ] **Step 2: Run frontend test and verify it fails** + +Run: + +```bash +cd frontend +node --test src/views/DashboardView.test.js +``` + +Expected: + +```text +not ok +``` + +The failure should mention missing `材料库存批次闭环图` or new node type strings. + +- [ ] **Step 3: Update DAG header copy** + +In `frontend/src/views/DashboardView.vue`, replace the lifecycle modal header: + +```vue +

材料库存批次闭环图

+``` + +Replace the header helper text: + +```vue +拖动画布 · 点击节点看详情 · 报工仅作证据 +``` + +Replace loading text: + +```vue +
正在加载材料库存批次闭环图...
+``` + +- [ ] **Step 4: Add DAG legend to modal header** + +In the header action area, before close button, add: + +```vue +
+ 生产台账 + 库外余额 + 入库结果 + 报工证据 +
+``` + +- [ ] **Step 5: Update node visual class helper** + +If `DashboardView.vue` has a helper like `dagNodeClass(node)` or inline `dag-node-${node.tone}`, extend it to include type classes: + +```javascript +function dagNodeTypeClass(node) { + const typeClassMap = { + raw_material_lot: "dag-node-raw", + production_ledger: "dag-node-ledger", + production_issue_txn: "dag-node-issue", + ledger_outside_balance: "dag-node-balance", + production_ledger_settlement: "dag-node-settlement", + finished_inbound: "dag-node-inbound", + surplus_inbound: "dag-node-inbound", + scrap_inbound: "dag-node-inbound", + ledger_writeoff: "dag-node-writeoff", + ledger_reopen: "dag-node-reopen", + operation_report: "dag-node-evidence", + delivery: "dag-node-delivery" + }; + return typeClassMap[node?.type] || "dag-node-default"; +} +``` + +Use it on each node card: + +```vue +:class="[ + `dag-node-${node.tone || 'normal'}`, + dagNodeTypeClass(node), + { + 'dag-node-selected': selectedNode?.id === node.id, + 'dag-node-highlighted': isNodeHighlighted(node), + 'dag-node-dimmed': isNodeDimmed(node) + } +]" +``` + +- [ ] **Step 6: Update layout priority for new nodes** + +Find `layoutNodes` and its special row handling. Replace legacy-only grouping with this priority function: + +```javascript +function dagRowPriority(node) { + const priority = { + raw_material_lot: 0, + production_ledger: 0, + production_issue_txn: 0, + ledger_outside_balance: 1, + operation_report: 2, + production_ledger_settlement: 0, + finished_inbound: 0, + surplus_inbound: 1, + scrap_inbound: 2, + ledger_writeoff: 3, + ledger_reopen: 4, + delivery: 0 + }; + return priority[node?.type] ?? 0; +} +``` + +Sort nodes inside the same level with: + +```javascript +.sort((left, right) => { + const levelDiff = Number(left.level ?? 0) - Number(right.level ?? 0); + if (levelDiff !== 0) { + return levelDiff; + } + const rowDiff = dagRowPriority(left) - dagRowPriority(right); + if (rowDiff !== 0) { + return rowDiff; + } + return String(left.id).localeCompare(String(right.id), "zh-Hans-CN"); +}) +``` + +- [ ] **Step 7: Add stage label normalization** + +Before `stageBands`, add: + +```javascript +function dagStageLabel(level, fallback) { + const labels = { + 0: "原料入库", + 1: "生产台账", + 2: "生产出库", + 3: "库外闭环", + 4: "入库结算", + 5: "入库结果", + 6: "销售去向" + }; + return labels[Number(level)] || fallback || "流转节点"; +} +``` + +In `stageBands`, use: + +```javascript +label: dagStageLabel(node.level, node.stage), +``` + +- [ ] **Step 8: Run frontend DAG test** + +Run: + +```bash +cd frontend +node --test src/views/DashboardView.test.js +``` + +Expected: + +```text +ok +``` + +--- + +### Task 5: DAG Visual Polish + +**Files:** +- Modify: `frontend/src/styles/main.css` +- Test: `frontend/src/views/DashboardView.test.js` + +- [ ] **Step 1: Add legend and node type styles** + +Append near existing DAG styles in `frontend/src/styles/main.css`: + +```css +.dag-legend { + display: inline-flex; + align-items: center; + gap: 10px; + padding: 7px 10px; + border: 1px solid rgba(148, 163, 184, 0.22); + border-radius: 999px; + background: rgba(15, 23, 42, 0.38); + color: rgba(226, 232, 240, 0.9); + font-size: 12px; + white-space: nowrap; +} + +.dag-legend span { + display: inline-flex; + align-items: center; + gap: 5px; +} + +.legend-dot { + width: 8px; + height: 8px; + border-radius: 999px; + box-shadow: 0 0 14px currentColor; +} + +.legend-ledger { + color: #7dd3fc; + background: #7dd3fc; +} + +.legend-balance { + color: #fbbf24; + background: #fbbf24; +} + +.legend-inbound { + color: #86efac; + background: #86efac; +} + +.legend-evidence { + color: #c4b5fd; + background: #c4b5fd; +} + +.dag-node-ledger { + border-color: rgba(125, 211, 252, 0.56); + background: + linear-gradient(135deg, rgba(14, 165, 233, 0.24), rgba(15, 23, 42, 0.92)), + radial-gradient(circle at top right, rgba(125, 211, 252, 0.22), transparent 42%); +} + +.dag-node-issue { + border-color: rgba(251, 191, 36, 0.48); + background: + linear-gradient(135deg, rgba(245, 158, 11, 0.22), rgba(15, 23, 42, 0.92)), + radial-gradient(circle at top left, rgba(251, 191, 36, 0.18), transparent 42%); +} + +.dag-node-balance { + border-color: rgba(251, 191, 36, 0.6); + background: + linear-gradient(135deg, rgba(120, 53, 15, 0.34), rgba(15, 23, 42, 0.9)), + repeating-linear-gradient(135deg, rgba(251, 191, 36, 0.08) 0 6px, transparent 6px 14px); +} + +.dag-node-settlement, +.dag-node-inbound { + border-color: rgba(134, 239, 172, 0.5); + background: + linear-gradient(135deg, rgba(22, 163, 74, 0.22), rgba(15, 23, 42, 0.92)), + radial-gradient(circle at top right, rgba(134, 239, 172, 0.18), transparent 44%); +} + +.dag-node-writeoff { + border-color: rgba(251, 146, 60, 0.58); + background: + linear-gradient(135deg, rgba(194, 65, 12, 0.24), rgba(15, 23, 42, 0.92)), + radial-gradient(circle at top right, rgba(251, 146, 60, 0.2), transparent 40%); +} + +.dag-node-evidence { + border-style: dashed; + border-color: rgba(196, 181, 253, 0.62); + background: + linear-gradient(135deg, rgba(109, 40, 217, 0.18), rgba(15, 23, 42, 0.9)), + radial-gradient(circle at top left, rgba(196, 181, 253, 0.16), transparent 40%); +} +``` + +- [ ] **Step 2: Add CSS source assertion** + +Extend `frontend/src/views/DashboardView.test.js`: + +```javascript +import { readFileSync } from "node:fs"; + +const cssSource = readFileSync(join(__dirname, "../styles/main.css"), "utf8"); + +describe("DashboardView DAG visual system", () => { + it("defines dedicated visual classes for material batch lifecycle nodes", () => { + assert.match(cssSource, /\.dag-node-ledger/); + assert.match(cssSource, /\.dag-node-balance/); + assert.match(cssSource, /\.dag-node-settlement/); + assert.match(cssSource, /\.dag-node-evidence/); + assert.match(cssSource, /\.dag-legend/); + }); +}); +``` + +If the duplicate `readFileSync` import conflicts, merge it with the existing import instead of adding a second import. + +- [ ] **Step 3: Run frontend DAG test** + +Run: + +```bash +cd frontend +node --test src/views/DashboardView.test.js +``` + +Expected: + +```text +ok +``` + +--- + +### Task 6: End-To-End Verification + +**Files:** +- No edits unless verification exposes defects. + +- [ ] **Step 1: Run backend lifecycle tests** + +Run: + +```bash +cd backend +python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q +``` + +Expected: + +```text +3 passed +``` + +- [ ] **Step 2: Run backend production ledger regressions** + +Run: + +```bash +cd backend +python -m pytest tests/test_production_batch_ledger.py tests/test_selected_stock_lot_production_issue.py tests/test_smart_operation_report_config.py -q +``` + +Expected: + +```text +passed +``` + +- [ ] **Step 3: Run frontend tests** + +Run: + +```bash +cd frontend +node --test src/views/DashboardView.test.js +node --test src/views/InventoryLedgerView.test.js +node --test src/components/documentForms/documentForms.test.js +``` + +Expected: + +```text +ok +``` + +- [ ] **Step 4: Run frontend production build** + +Run: + +```bash +cd frontend +npm run build +``` + +Expected: + +```text +✓ built +``` + +The existing Vite chunk-size warning is acceptable unless a new build error appears. + +- [ ] **Step 5: Browser smoke test** + +Open the already logged-in browser at: + +```text +http://localhost:5173/ +``` + +Manual checks: + +- The material-lot table still loads. +- Clicking `YL0001` or any material lot opens a modal titled `材料库存批次闭环图`. +- The graph has stage bands: `原料入库`、`生产台账`、`生产出库`、`库外闭环`、`入库结算`、`入库结果`、`销售去向`. +- The graph does not show old `生产工单` as the main chain when a `ProductionBatchLedger` exists. +- If system extension `对接智能报工小程序` is closed, the graph does not show `报工证据` nodes. +- If the switch is open and `OperationReport.production_ledger_id` has data, report nodes appear as dashed evidence nodes. + +- [ ] **Step 6: Commit** + +After all checks pass: + +```bash +git status --short +git add backend/app/services/material_lifecycle_dag.py backend/app/api/routes/dashboard.py backend/tests/test_dashboard_material_lot_lifecycle.py frontend/src/views/DashboardView.vue frontend/src/views/DashboardView.test.js frontend/src/styles/main.css +git commit -m "重构材料批次生命周期DAG" +``` + +Do not add `.impeccable/` unless explicitly requested. + +--- + +## Self-Review + +- Spec coverage: The plan replaces old work-order main chain with production-ledger main chain, supports material-lot/product ledger rows, production issue accumulation, production ledger inbound settlement, finished/surplus/scrap/writeoff outputs, optional miniapp evidence, delivery tracing, and legacy fallback. +- Placeholder scan: The plan contains no open implementation placeholders. The only compatibility note is an explicit legacy fallback task. +- Type consistency: Backend public function is consistently named `build_material_lot_lifecycle_graph(db, lot) -> tuple[list[LifecycleNode], list[LifecycleEdge]]`; frontend new node types are consistently `production_ledger`, `production_issue_txn`, `ledger_outside_balance`, `production_ledger_settlement`, `finished_inbound`, `surplus_inbound`, `scrap_inbound`, `ledger_writeoff`, `operation_report`, `delivery`. +- Risk note: Delivery tracing depends on finished stock lots carrying `source_doc_type` in `PRODUCTION_LEDGER_IN` or `生产台账` and `source_doc_id = production_ledger_id`; this matches current production台账入库 and单独成品生产入库路径. diff --git a/docs/superpowers/plans/2026-06-14-system-extension-workbench.md b/docs/superpowers/plans/2026-06-14-system-extension-workbench.md new file mode 100644 index 0000000..c46d832 --- /dev/null +++ b/docs/superpowers/plans/2026-06-14-system-extension-workbench.md @@ -0,0 +1,1410 @@ +# System Extension Workbench 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:** 前端单页内重构：保留 `SystemExtensionView.vue` 的现有 API 调用和表单状态，新增顶部状态条、左侧配置目录、右侧单模块编辑区、底部固定保存栏。CSS 只新增/替换系统拓展专属 class，避免影响 DAG、仓库单据、权限脑图等近期未提交改动。 + +**Tech Stack:** Vue 3 `