tangxh/ForgeFlow-ERP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
main
ForgeFlow-ERP/docs/superpowers/plans/2026-06-14-material-batch-lifecycle-dag.md

1750 lines
60 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# 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
<p class="eyebrow">材料库存批次闭环图</p>
```
Replace the header helper text:
```vue
<span class="table-chip">拖动画布 · 点击节点看详情 · 报工仅作证据</span>
```
Replace loading text:
```vue
<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:
```vue
<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:
```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单独成品生产入库路径.
Reference in New Issue View Git Blame Copy Permalink