2026-06-14 20:38:53 +08:00

60 KiB

Raw Blame History

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 的主链路必须表达下面这个业务事实：

flowchart LR
  A["原材料库存批次号"] --> B["生产台账：库存批次号 + 产品"]
  B --> C["生产出库流水"]
  B --> D["库外材料余额"]
  B --> E["生产台账入库/分支入库"]
  E --> F["成品入库"]
  E --> G["生产余料入库"]
  E --> H["生产废料入库"]
  E --> I["差异/结单核销"]
  F --> J["销售出库"]
  J --> K["退货/售后"]

报工小程序开启时，在生产台账下面增加“工序报工证据”节点：

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:

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:

    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:

    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:

cd backend
python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q

Expected:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

cd backend
python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q

Expected after this task:

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:

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:

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:

    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:

    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:

cd backend
python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q

Expected:

3 passed

Step 6: Run nearby backend regression tests

Run:

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:

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:

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:

cd frontend
node --test src/views/DashboardView.test.js

Expected:

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:

<p class="eyebrow">材料库存批次闭环图</p>

Replace the header helper text:

<span class="table-chip">拖动画布 · 点击节点看详情 · 报工仅作证据</span>

Replace loading text:

<div v-if="loadingLifecycle" class="dag-loading-card">正在加载材料库存批次闭环图...</div>

Step 4: Add DAG legend to modal header

In the header action area, before close button, add:

<div class="dag-legend" aria-label="材料批次闭环图例">
  <span><i class="legend-dot legend-ledger"></i>生产台账</span>
  <span><i class="legend-dot legend-balance"></i>库外余额</span>
  <span><i class="legend-dot legend-inbound"></i>入库结果</span>
  <span><i class="legend-dot legend-evidence"></i>报工证据</span>
</div>

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:

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:

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

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:

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

function dagStageLabel(level, fallback) {
  const labels = {
    0: "原料入库",
    1: "生产台账",
    2: "生产出库",
    3: "库外闭环",
    4: "入库结算",
    5: "入库结果",
    6: "销售去向"
  };
  return labels[Number(level)] || fallback || "流转节点";
}

In stageBands, use:

label: dagStageLabel(node.level, node.stage),

Step 8: Run frontend DAG test

Run:

cd frontend
node --test src/views/DashboardView.test.js

Expected:

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:

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

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:

cd frontend
node --test src/views/DashboardView.test.js

Expected:

ok

Task 6: End-To-End Verification

Files:

No edits unless verification exposes defects.
Step 1: Run backend lifecycle tests

Run:

cd backend
python -m pytest tests/test_dashboard_material_lot_lifecycle.py -q

Expected:

3 passed

Step 2: Run backend production ledger regressions

Run:

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:

passed

Step 3: Run frontend tests

Run:

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:

ok

Step 4: Run frontend production build

Run:

cd frontend
npm run build

Expected:

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

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:

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单独成品生产入库路径.

60 KiB Raw Blame History Unescape Escape

Material Batch Lifecycle DAG Implementation Plan

Business DAG Model

File Structure

Task 1: Backend Failing Tests For Production Ledger DAG

Task 2: Backend DAG Builder Service

Task 3: Dashboard Route Integration And Legacy Fallback

Task 4: Frontend DAG Copy, Stage Model, And Layout Classification

Task 5: DAG Visual Polish

Task 6: End-To-End Verification

Self-Review

60 KiB

Raw Blame History