from __future__ import annotations

from decimal import Decimal
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import false, func, select
from sqlalchemy.orm import Session, aliased

from app.db.session import get_db
from app.models.master_data import Item, Warehouse
from app.models.operations import (
    CompletionReceipt,
    CompletionReceiptItem,
    Delivery,
    DeliveryItem,
    OperationReport,
    PurchaseOrder,
    PurchaseOrderItem,
    PurchaseReceipt,
    PurchaseReceiptItem,
    ReturnDisposition,
    ReturnItem,
    ReturnOrder,
    ScrapRecord,
    StockLot,
    Supplier,
    WorkOrder,
    WorkOrderMaterialIssue,
    WorkOrderOperation,
)
from app.models.org import Employee
from app.models.planning import MaterialDemand
from app.models.sales import Customer, SalesOrder, SalesOrderItem
from app.schemas.dashboard import (
    DashboardPayload,
    DashboardMaterialLotRead,
    DashboardSalesOrderRead,
    LifecycleDetailItem,
    LifecycleEdge,
    LifecycleNode,
    MaterialLotLifecycleRead,
    SalesOrderLifecycleRead,
)
from app.services.auth import require_authenticated_user
from app.services.material_lifecycle_dag import build_material_lot_lifecycle_graph
from app.services.mock_data import get_dashboard_payload

router = APIRouter(dependencies=[Depends(require_authenticated_user)])

STATUS_LABELS = {
    "ACTIVE": "启用",
    "INACTIVE": "停用",
    "OPEN": "未开始",
    "PARTIAL": "部分完成",
    "CLOSED": "已关闭",
    "ORDERED": "已下单",
    "RECEIVED": "已收货",
    "RELEASED": "已下达",
    "IN_PROGRESS": "进行中",
    "READY_TO_RECEIPT": "待成品入库",
    "COMPLETED": "已完工",
    "DONE": "已完成",
    "PENDING": "待处理",
    "PENDING_QC": "待质检",
    "PASSED": "已放行",
    "PASS": "已放行",
    "REJECTED": "已拒收",
    "REJECT": "已拒收",
    "POSTED": "已过账",
    "AVAILABLE": "可用",
    "LOCKED": "锁定",
    "DEPLETED": "已耗尽",
    "CONFIRMED": "已确认",
    "OVERRIDDEN": "已覆盖",
    "CREATED": "已创建",
    "IDLE": "空闲",
    "SUPERSEDED": "已替代",
    "SCRAP": "已报废",
    "SCRAPPED": "已报废",
    "NORMAL": "正常",
    "ABNORMAL": "异常",
}

WORK_ORDER_TYPE_LABELS = {
    "NORMAL": "常规生产",
    "REWORK": "返工生产",
}

DISPOSAL_TYPE_LABELS = {
    "SCRAP": "报废",
    "REWORK": "返工",
}


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_date(value: Any) -> str:
    if not value:
        return "-"
    return str(value)[:10]


def _fmt_datetime(value: Any) -> str:
    if not value:
        return "-"
    return str(value).replace("T", " ")[:19]


def _label(mapping: dict[str, str], value: Any) -> str:
    if value is None or value == "":
        return "-"
    key = str(value).upper()
    return mapping.get(key, str(value))


def _status_label(value: Any) -> str:
    return _label(STATUS_LABELS, value)


def _disposal_label(value: Any) -> str:
    return _label(DISPOSAL_TYPE_LABELS, value)


def _detail(label: str, value: Any) -> LifecycleDetailItem:
    text = "-" if value is None or value == "" else str(value)
    return LifecycleDetailItem(label=label, value=text)


def _node(
    *,
    node_id: str,
    node_type: str,
    title: str,
    stage: str,
    level: int,
    subtitle: str | None = None,
    status: str | None = None,
    tone: str = "normal",
    summary: str | None = None,
    metrics: list[tuple[str, Any]] | None = None,
    details: list[tuple[str, Any]] | None = None,
    raw: dict[str, Any] | None = None,
) -> LifecycleNode:
    return LifecycleNode(
        id=node_id,
        type=node_type,
        title=title,
        subtitle=subtitle,
        status=status,
        stage=stage,
        level=level,
        tone=tone,
        summary=summary,
        metrics=[_detail(label, value) for label, value in (metrics or [])],
        details=[_detail(label, value) for label, value in (details or [])],
        raw=raw or {},
    )


def _edge(source: str, target: str, label: str | None = None) -> LifecycleEdge:
    return LifecycleEdge(id=f"{source}->{target}", source=source, target=target, label=label)


def _tone_for_status(status: str | None) -> str:
    value = str(status or "").upper()
    if value in {"REJECTED", "SCRAP", "SCRAPPED", "LOCKED"}:
        return "danger"
    if value in {"PENDING", "PENDING_QC", "OPEN", "PARTIAL", "IN_PROGRESS", "RELEASED", "NEW", "CREATED"}:
        return "warning"
    if value in {"CLOSED", "COMPLETED", "DONE", "POSTED", "PASSED", "RECEIVED", "CONFIRMED"}:
        return "success"
    return "normal"


def _sales_order_rows(limit: int = 100):
    line_subquery = (
        select(
            SalesOrderItem.sales_order_id.label("sales_order_id"),
            func.count(SalesOrderItem.id).label("line_count"),
            func.coalesce(func.sum(SalesOrderItem.order_qty), 0).label("total_order_qty"),
            func.coalesce(func.sum(SalesOrderItem.delivered_qty), 0).label("total_delivered_qty"),
        )
        .group_by(SalesOrderItem.sales_order_id)
        .subquery()
    )
    mrp_subquery = (
        select(
            SalesOrderItem.sales_order_id.label("sales_order_id"),
            func.count(MaterialDemand.id).label("mrp_count"),
        )
        .join(MaterialDemand, MaterialDemand.sales_order_item_id == SalesOrderItem.id)
        .where(MaterialDemand.status != "SUPERSEDED")
        .group_by(SalesOrderItem.sales_order_id)
        .subquery()
    )
    purchase_subquery = (
        select(
            SalesOrderItem.sales_order_id.label("sales_order_id"),
            func.count(func.distinct(PurchaseOrderItem.purchase_order_id)).label("purchase_order_count"),
        )
        .join(MaterialDemand, MaterialDemand.sales_order_item_id == SalesOrderItem.id)
        .join(PurchaseOrderItem, PurchaseOrderItem.source_demand_id == MaterialDemand.id)
        .group_by(SalesOrderItem.sales_order_id)
        .subquery()
    )
    work_order_subquery = (
        select(
            SalesOrderItem.sales_order_id.label("sales_order_id"),
            func.count(WorkOrder.id).label("work_order_count"),
        )
        .join(WorkOrder, WorkOrder.source_sales_order_item_id == SalesOrderItem.id)
        .group_by(SalesOrderItem.sales_order_id)
        .subquery()
    )
    completion_subquery = (
        select(
            SalesOrderItem.sales_order_id.label("sales_order_id"),
            func.count(func.distinct(CompletionReceipt.id)).label("completion_receipt_count"),
        )
        .join(WorkOrder, WorkOrder.source_sales_order_item_id == SalesOrderItem.id)
        .join(CompletionReceipt, CompletionReceipt.work_order_id == WorkOrder.id)
        .group_by(SalesOrderItem.sales_order_id)
        .subquery()
    )
    delivery_subquery = (
        select(
            Delivery.sales_order_id.label("sales_order_id"),
            func.count(Delivery.id).label("delivery_count"),
        )
        .group_by(Delivery.sales_order_id)
        .subquery()
    )
    return_subquery = (
        select(
            ReturnOrder.sales_order_id.label("sales_order_id"),
            func.count(ReturnOrder.id).label("return_count"),
        )
        .where(ReturnOrder.sales_order_id.is_not(None))
        .group_by(ReturnOrder.sales_order_id)
        .subquery()
    )

    return (
        select(
            SalesOrder.id.label("order_id"),
            SalesOrder.order_no.label("order_no"),
            Customer.customer_code.label("customer_code"),
            Customer.customer_name.label("customer_name"),
            SalesOrder.order_date.label("order_date"),
            SalesOrder.promised_date.label("promised_date"),
            SalesOrder.delivery_address.label("delivery_address"),
            SalesOrder.total_amount.label("total_amount"),
            SalesOrder.status.label("status"),
            func.coalesce(line_subquery.c.line_count, 0).label("line_count"),
            func.coalesce(line_subquery.c.total_order_qty, 0).label("total_order_qty"),
            func.coalesce(line_subquery.c.total_delivered_qty, 0).label("total_delivered_qty"),
            func.coalesce(mrp_subquery.c.mrp_count, 0).label("mrp_count"),
            func.coalesce(purchase_subquery.c.purchase_order_count, 0).label("purchase_order_count"),
            func.coalesce(work_order_subquery.c.work_order_count, 0).label("work_order_count"),
            func.coalesce(completion_subquery.c.completion_receipt_count, 0).label("completion_receipt_count"),
            func.coalesce(delivery_subquery.c.delivery_count, 0).label("delivery_count"),
            func.coalesce(return_subquery.c.return_count, 0).label("return_count"),
            SalesOrder.updated_at.label("updated_at"),
        )
        .join(Customer, Customer.id == SalesOrder.customer_id)
        .outerjoin(line_subquery, line_subquery.c.sales_order_id == SalesOrder.id)
        .outerjoin(mrp_subquery, mrp_subquery.c.sales_order_id == SalesOrder.id)
        .outerjoin(purchase_subquery, purchase_subquery.c.sales_order_id == SalesOrder.id)
        .outerjoin(work_order_subquery, work_order_subquery.c.sales_order_id == SalesOrder.id)
        .outerjoin(completion_subquery, completion_subquery.c.sales_order_id == SalesOrder.id)
        .outerjoin(delivery_subquery, delivery_subquery.c.sales_order_id == SalesOrder.id)
        .outerjoin(return_subquery, return_subquery.c.sales_order_id == SalesOrder.id)
        .order_by(SalesOrder.id.desc())
        .limit(limit)
    )


def _material_lot_rows(limit: int = 100):
    issue_subquery = (
        select(
            WorkOrderMaterialIssue.source_lot_id.label("source_lot_id"),
            func.count(func.distinct(WorkOrderMaterialIssue.material_sub_batch_no)).label("issued_batch_count"),
            func.coalesce(func.sum(WorkOrderMaterialIssue.issued_weight_kg), 0).label("issued_weight_kg"),
        )
        .group_by(WorkOrderMaterialIssue.source_lot_id)
        .subquery()
    )
    finished_subquery = (
        select(
            CompletionReceiptItem.source_material_lot_id.label("source_material_lot_id"),
            func.count(func.distinct(CompletionReceiptItem.id)).label("finished_batch_count"),
            func.coalesce(func.sum(CompletionReceiptItem.receipt_qty), 0).label("finished_qty"),
        )
        .where(CompletionReceiptItem.source_material_lot_id.is_not(None))
        .group_by(CompletionReceiptItem.source_material_lot_id)
        .subquery()
    )
    finished_lot = aliased(StockLot)
    delivery_subquery = (
        select(
            finished_lot.source_material_lot_id.label("source_material_lot_id"),
            func.count(func.distinct(Delivery.id)).label("delivery_count"),
        )
        .join(DeliveryItem, DeliveryItem.lot_id == finished_lot.id)
        .join(Delivery, Delivery.id == DeliveryItem.delivery_id)
        .where(finished_lot.source_material_lot_id.is_not(None))
        .group_by(finished_lot.source_material_lot_id)
        .subquery()
    )
    return (
        select(
            StockLot.id.label("lot_id"),
            StockLot.lot_no.label("lot_no"),
            StockLot.item_id.label("material_item_id"),
            Item.item_code.label("material_code"),
            Item.item_name.label("material_name"),
            Item.specification.label("specification"),
            Warehouse.warehouse_name.label("warehouse_name"),
            StockLot.inbound_weight_kg.label("inbound_weight_kg"),
            StockLot.remaining_weight_kg.label("remaining_weight_kg"),
            StockLot.unit_cost.label("unit_cost"),
            (StockLot.inbound_weight_kg * StockLot.unit_cost).label("inbound_amount"),
            func.coalesce(issue_subquery.c.issued_batch_count, 0).label("issued_batch_count"),
            func.coalesce(issue_subquery.c.issued_weight_kg, 0).label("issued_weight_kg"),
            func.coalesce(finished_subquery.c.finished_batch_count, 0).label("finished_batch_count"),
            func.coalesce(finished_subquery.c.finished_qty, 0).label("finished_qty"),
            func.coalesce(delivery_subquery.c.delivery_count, 0).label("delivery_count"),
            StockLot.status.label("status"),
            StockLot.quality_status.label("quality_status"),
            StockLot.created_at.label("created_at"),
        )
        .join(Item, Item.id == StockLot.item_id)
        .join(Warehouse, Warehouse.id == StockLot.warehouse_id)
        .outerjoin(issue_subquery, issue_subquery.c.source_lot_id == StockLot.id)
        .outerjoin(finished_subquery, finished_subquery.c.source_material_lot_id == StockLot.id)
        .outerjoin(delivery_subquery, delivery_subquery.c.source_material_lot_id == StockLot.id)
        .where(Item.item_type == "RAW_MATERIAL", Warehouse.warehouse_type == "RAW")
        .order_by(StockLot.created_at.desc(), StockLot.id.desc())
        .limit(limit)
    )


@router.get("/overview", response_model=DashboardPayload)
def overview() -> DashboardPayload:
    return get_dashboard_payload()


@router.get("/sales-orders", response_model=list[DashboardSalesOrderRead])
def list_dashboard_sales_orders(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[DashboardSalesOrderRead]:
    rows = db.execute(_sales_order_rows(limit=limit)).mappings().all()
    return [DashboardSalesOrderRead.model_validate(dict(row)) for row in rows]


@router.get("/material-lots", response_model=list[DashboardMaterialLotRead])
def list_dashboard_material_lots(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[DashboardMaterialLotRead]:
    rows = db.execute(_material_lot_rows(limit=limit)).mappings().all()
    return [DashboardMaterialLotRead.model_validate(dict(row)) for row in rows]


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)

    lot_node_id = f"raw-lot-{lot.lot_id}"
    add_node(
        _node(
            node_id=lot_node_id,
            node_type="raw_material_lot",
            title=lot.lot_no,
            subtitle=f"{lot.material_code} · {lot.material_name}",
            status=lot.status,
            tone=_tone_for_status(lot.status),
            stage="原料入库",
            level=0,
            summary="原材料驱动的起点。该批次记录入库成本、剩余库存、质检状态，并向后追踪所有生产出库、成品入库和发货。",
            metrics=[
                ("入库重量", _fmt_weight(lot.inbound_weight_kg)),
                ("剩余重量", _fmt_weight(lot.remaining_weight_kg)),
                ("入库金额", _fmt_money(lot.inbound_amount)),
            ],
            details=[
                ("原材料编码", lot.material_code),
                ("原材料名称", lot.material_name),
                ("规格", lot.specification),
                ("仓库", lot.warehouse_name),
                ("入库单价", f"{_fmt_money(lot.unit_cost)}/kg"),
                ("质检状态", _status_label(lot.quality_status)),
                ("生产出库数", lot.issued_batch_count),
                ("已领用重量", _fmt_weight(lot.issued_weight_kg)),
                ("成品批次数", lot.finished_batch_count),
                ("已发货次数", lot.delivery_count),
            ],
            raw=lot.model_dump(mode="json"),
        )
    )

    issues = db.execute(
        select(
            WorkOrderMaterialIssue.id.label("issue_id"),
            WorkOrderMaterialIssue.material_sub_batch_no.label("material_sub_batch_no"),
            StockLot.lot_no.label("source_lot_no"),
            WorkOrderMaterialIssue.work_order_id.label("work_order_id"),
            WorkOrder.work_order_no.label("work_order_no"),
            WorkOrder.product_item_id.label("product_item_id"),
            Item.item_code.label("product_code"),
            Item.item_name.label("product_name"),
            WorkOrder.planned_qty.label("planned_qty"),
            WorkOrder.finished_qty.label("finished_qty"),
            WorkOrder.status.label("work_order_status"),
            WorkOrderMaterialIssue.issued_qty.label("issued_qty"),
            WorkOrderMaterialIssue.issued_weight_kg.label("issued_weight_kg"),
            WorkOrderMaterialIssue.unit_cost.label("unit_cost"),
            WorkOrderMaterialIssue.amount.label("amount"),
            WorkOrderMaterialIssue.issue_time.label("issue_time"),
            WorkOrderMaterialIssue.remark.label("remark"),
        )
        .join(WorkOrder, WorkOrder.id == WorkOrderMaterialIssue.work_order_id)
        .join(Item, Item.id == WorkOrder.product_item_id)
        .outerjoin(StockLot, StockLot.id == WorkOrderMaterialIssue.source_lot_id)
        .where(WorkOrderMaterialIssue.source_lot_id == lot.lot_id)
        .order_by(WorkOrderMaterialIssue.issue_time, WorkOrderMaterialIssue.id)
    ).mappings().all()
    work_order_ids = sorted({row["work_order_id"] for row in issues})

    for row in issues:
        source_lot_no = row["source_lot_no"] or row["material_sub_batch_no"] or "未记录库存批次号"
        issue_node_id = f"issue-{row['issue_id']}"
        add_node(
            _node(
                node_id=issue_node_id,
                node_type="material_issue_batch",
                title=source_lot_no,
                subtitle=row["work_order_no"],
                status="POSTED",
                tone="success",
                stage="生产出库",
                level=1,
                summary="原材料按库存批次生产出库，记录领用重量、单价和成本。",
                metrics=[
                    ("领用重量", _fmt_weight(row["issued_weight_kg"])),
                    ("领用金额", _fmt_money(row["amount"])),
                    ("单价", f"{_fmt_money(row['unit_cost'])}/kg"),
                ],
                details=[
                    ("库存批次号", source_lot_no),
                    ("工单号", row["work_order_no"]),
                    ("产品", f"{row['product_code']} · {row['product_name']}"),
                    ("领用时间", _fmt_datetime(row["issue_time"])),
                    ("领用重量", _fmt_weight(row["issued_weight_kg"])),
                    ("备注", row["remark"]),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(lot_node_id, issue_node_id, "生产出库"))

        work_order_node_id = f"wo-{row['work_order_id']}"
        add_node(
            _node(
                node_id=work_order_node_id,
                node_type="work_order",
                title=row["work_order_no"],
                subtitle=f"{row['product_code']} · {row['product_name']}",
                status=row["work_order_status"],
                tone=_tone_for_status(row["work_order_status"]),
                stage="生产工单",
                level=2,
                summary="原材料按库存批次生产出库进入生产工单，后续工序报工、报废返工和完工入库均挂在该工单下。",
                metrics=[
                    ("计划数", _fmt_qty(row["planned_qty"])),
                    ("完工数", _fmt_qty(row["finished_qty"])),
                    ("本批领用", _fmt_weight(row["issued_weight_kg"])),
                ],
                details=[
                    ("工单号", row["work_order_no"]),
                    ("产品编码", row["product_code"]),
                    ("产品名称", row["product_name"]),
                    ("状态", _status_label(row["work_order_status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(issue_node_id, work_order_node_id, "投入生产"))

    operations = db.execute(
        select(
            WorkOrderOperation.id.label("operation_id"),
            WorkOrderOperation.work_order_id.label("work_order_id"),
            WorkOrderOperation.seq_no.label("seq_no"),
            WorkOrderOperation.operation_name.label("operation_name"),
            WorkOrderOperation.planned_qty.label("planned_qty"),
            WorkOrderOperation.report_qty.label("report_qty"),
            WorkOrderOperation.good_qty.label("good_qty"),
            WorkOrderOperation.scrap_qty.label("scrap_qty"),
            WorkOrderOperation.rework_qty.label("rework_qty"),
            WorkOrderOperation.is_abnormal.label("is_abnormal"),
            WorkOrderOperation.status.label("status"),
        )
        .where(WorkOrderOperation.work_order_id.in_(work_order_ids) if work_order_ids else false())
        .order_by(WorkOrderOperation.work_order_id, WorkOrderOperation.seq_no)
    ).mappings().all()

    for row in operations:
        node_id = f"op-{row['operation_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="work_order_operation",
                title=f"OP{row['seq_no']:02d} {row['operation_name']}",
                subtitle="异常工序" if row["is_abnormal"] else "工序执行",
                status=row["status"],
                tone="danger" if row["is_abnormal"] else _tone_for_status(row["status"]),
                stage="工序执行",
                level=3,
                summary="工序节点展示该库存批次对应工单的加工进度、报废返工和异常情况。",
                metrics=[
                    ("计划数", _fmt_qty(row["planned_qty"])),
                    ("已报工", _fmt_qty(row["report_qty"])),
                    ("报废数", _fmt_qty(row["scrap_qty"])),
                ],
                details=[
                    ("合格数", _fmt_qty(row["good_qty"])),
                    ("返工数", _fmt_qty(row["rework_qty"])),
                    ("是否异常", "是" if row["is_abnormal"] else "否"),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(f"wo-{row['work_order_id']}", node_id, "工艺路线"))

    reports = db.execute(
        select(
            OperationReport.id.label("report_id"),
            OperationReport.report_no.label("report_no"),
            OperationReport.work_order_id.label("work_order_id"),
            OperationReport.work_order_operation_id.label("work_order_operation_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.work_order_id.in_(work_order_ids) if work_order_ids else false())
        .order_by(OperationReport.start_time, OperationReport.id)
    ).mappings().all()

    for row in reports:
        node_id = f"report-{row['report_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="operation_report",
                title=row["report_no"],
                subtitle=row["employee_name"],
                status="ABNORMAL" if row["is_abnormal"] else "NORMAL",
                tone="danger" if row["is_abnormal"] else "success",
                stage="工人报工",
                level=4,
                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"])),
                    ("返工数", _fmt_qty(row["rework_qty"])),
                    ("异常原因", row["scrap_reason"] or ("报工异常" if row["is_abnormal"] else "-")),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(f"op-{row['work_order_operation_id']}", node_id, "扫码报工"))

    source_material_lot = aliased(StockLot)
    completion_items = db.execute(
        select(
            CompletionReceiptItem.id.label("completion_item_id"),
            CompletionReceipt.work_order_id.label("work_order_id"),
            CompletionReceipt.receipt_no.label("receipt_no"),
            CompletionReceipt.receipt_time.label("receipt_time"),
            CompletionReceiptItem.lot_no.label("lot_no"),
            CompletionReceiptItem.receipt_qty.label("receipt_qty"),
            CompletionReceiptItem.receipt_weight_kg.label("receipt_weight_kg"),
            CompletionReceiptItem.unit_cost.label("unit_cost"),
            CompletionReceiptItem.source_material_sub_batch_no.label("source_material_sub_batch_no"),
            func.coalesce(
                source_material_lot.lot_no,
                CompletionReceiptItem.source_material_sub_batch_no,
            ).label("source_material_lot_no"),
            CompletionReceiptItem.source_material_summary.label("source_material_summary"),
            Warehouse.warehouse_name.label("warehouse_name"),
        )
        .join(CompletionReceipt, CompletionReceipt.id == CompletionReceiptItem.completion_receipt_id)
        .join(Warehouse, Warehouse.id == CompletionReceipt.warehouse_id)
        .outerjoin(source_material_lot, source_material_lot.id == CompletionReceiptItem.source_material_lot_id)
        .where(CompletionReceipt.work_order_id.in_(work_order_ids) if work_order_ids else false())
        .order_by(CompletionReceipt.receipt_time, CompletionReceiptItem.id)
    ).mappings().all()
    completion_item_ids = [row["completion_item_id"] for row in completion_items]

    for row in completion_items:
        node_id = f"fg-{row['completion_item_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="completion_receipt",
                title=row["receipt_no"],
                subtitle=row["lot_no"],
                status="POSTED",
                tone="success",
                stage="成品入库",
                level=5,
                summary="成品入库批次已记录来源库存批次号，后续发货可以从成品批次倒查到原材料入库成本。",
                metrics=[
                    ("入库数量", _fmt_qty(row["receipt_qty"])),
                    ("入库重量", _fmt_weight(row["receipt_weight_kg"])),
                    ("单位成本", _fmt_money(row["unit_cost"])),
                ],
                details=[
                    ("成品批次", row["lot_no"]),
                    ("仓库", row["warehouse_name"]),
                    ("入库时间", _fmt_datetime(row["receipt_time"])),
                    ("来源库存批次号", row["source_material_lot_no"]),
                    ("来源摘要", row["source_material_summary"]),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(f"wo-{row['work_order_id']}", node_id, "完工入库"))

    finished_lot = aliased(StockLot)
    order_customer = aliased(Customer)
    direct_customer = aliased(Customer)
    deliveries = db.execute(
        select(
            DeliveryItem.id.label("delivery_item_id"),
            finished_lot.source_line_id.label("completion_item_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"),
            finished_lot.lot_no.label("lot_no"),
        )
        .join(finished_lot, finished_lot.id == DeliveryItem.lot_id)
        .join(Delivery, Delivery.id == DeliveryItem.delivery_id)
        .outerjoin(SalesOrder, SalesOrder.id == Delivery.sales_order_id)
        .outerjoin(order_customer, order_customer.id == SalesOrder.customer_id)
        .outerjoin(direct_customer, direct_customer.id == Delivery.customer_id)
        .where(finished_lot.source_line_id.in_(completion_item_ids) if completion_item_ids else false())
        .order_by(Delivery.delivery_date, DeliveryItem.id)
    ).mappings().all()

    for row in deliveries:
        node_id = f"delivery-{row['delivery_item_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="delivery",
                title=row["delivery_no"],
                subtitle=row["order_no"],
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="发货",
                level=6,
                summary="成品库存发给客户。即使订单晚于生产发生，也能从发货批次倒查到原材料入库批次和成本。",
                metrics=[
                    ("发货数量", _fmt_qty(row["delivery_qty"])),
                    ("发货重量", _fmt_weight(row["delivery_weight_kg"])),
                    ("发货金额", _fmt_money(row["line_amount"])),
                ],
                details=[
                    ("订单", row["order_no"]),
                    ("客户", row["customer_name"]),
                    ("库存批次号", row["lot_no"]),
                    ("发货时间", _fmt_datetime(row["delivery_date"])),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(f"fg-{row['completion_item_id']}", node_id, "库存发货"))

    return nodes, edges


@router.get("/material-lots/{lot_id}/lifecycle", response_model=MaterialLotLifecycleRead)
def get_material_lot_lifecycle(lot_id: int, db: Session = Depends(get_db)) -> MaterialLotLifecycleRead:
    lot_row = db.execute(_material_lot_rows(limit=1).where(StockLot.id == lot_id)).mappings().first()
    if not lot_row:
        raise HTTPException(status_code=404, detail="原材料批次不存在")

    lot = DashboardMaterialLotRead.model_validate(dict(lot_row))
    nodes, edges = build_material_lot_lifecycle_graph(db, lot)
    if not any(node.type == "production_ledger" for node in nodes):
        nodes, edges = _build_legacy_material_lot_lifecycle_graph(db, lot)
    return MaterialLotLifecycleRead(lot=lot, nodes=nodes, edges=edges)


@router.get("/sales-orders/{sales_order_id}/lifecycle", response_model=SalesOrderLifecycleRead)
def get_sales_order_lifecycle(
    sales_order_id: int,
    db: Session = Depends(get_db),
) -> SalesOrderLifecycleRead:
    order_row = db.execute(_sales_order_rows(limit=1).where(SalesOrder.id == sales_order_id)).mappings().first()
    if not order_row:
        raise HTTPException(status_code=404, detail="订单不存在")

    order = DashboardSalesOrderRead.model_validate(dict(order_row))
    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)

    order_node_id = f"order-{order.order_id}"
    add_node(
        _node(
            node_id=order_node_id,
            node_type="sales_order",
            title=order.order_no,
            subtitle=order.customer_name,
            status=order.status,
            tone=_tone_for_status(order.status),
            stage="订单",
            level=0,
            summary="订单生命周期起点，后续 MRP、采购、生产、入库、发货和售后均围绕此单展开。",
            metrics=[
                ("订单金额", _fmt_money(order.total_amount)),
                ("产品行数", order.line_count),
                ("订单数量", _fmt_qty(order.total_order_qty)),
                ("已发数量", _fmt_qty(order.total_delivered_qty)),
            ],
            details=[
                ("订单号", order.order_no),
                ("客户编码", order.customer_code),
                ("客户名称", order.customer_name),
                ("订单日期", _fmt_date(order.order_date)),
                ("承诺交期", _fmt_date(order.promised_date)),
                ("送货地址", order.delivery_address),
                ("MRP需求数", order.mrp_count),
                ("采购单数", order.purchase_order_count),
                ("生产工单数", order.work_order_count),
                ("成品入库单数", order.completion_receipt_count),
                ("发货单数", order.delivery_count),
                ("退货单数", order.return_count),
                ("最近更新", _fmt_datetime(order.updated_at)),
            ],
            raw=order.model_dump(mode="json"),
        )
    )

    order_items = db.execute(
        select(
            SalesOrderItem.id.label("sales_order_item_id"),
            SalesOrderItem.line_no.label("line_no"),
            SalesOrderItem.product_item_id.label("product_item_id"),
            Item.item_code.label("product_code"),
            Item.item_name.label("product_name"),
            Item.specification.label("specification"),
            Item.unit_weight_kg.label("unit_weight_kg"),
            SalesOrderItem.customer_part_no.label("customer_part_no"),
            SalesOrderItem.order_qty.label("order_qty"),
            SalesOrderItem.delivered_qty.label("delivered_qty"),
            SalesOrderItem.unit_price.label("unit_price"),
            SalesOrderItem.line_amount.label("line_amount"),
            SalesOrderItem.promised_date.label("promised_date"),
            SalesOrderItem.status.label("status"),
        )
        .join(Item, Item.id == SalesOrderItem.product_item_id)
        .where(SalesOrderItem.sales_order_id == sales_order_id)
        .order_by(SalesOrderItem.line_no)
    ).mappings().all()

    item_ids = [row["sales_order_item_id"] for row in order_items]
    product_ids = [row["product_item_id"] for row in order_items]

    demands = db.execute(
        select(
            MaterialDemand.id.label("demand_id"),
            MaterialDemand.demand_no.label("demand_no"),
            MaterialDemand.sales_order_item_id.label("sales_order_item_id"),
            MaterialDemand.material_item_id.label("material_item_id"),
            Item.item_code.label("material_code"),
            Item.item_name.label("material_name"),
            Item.specification.label("specification"),
            MaterialDemand.order_qty.label("order_qty"),
            MaterialDemand.theoretical_weight_kg.label("theoretical_weight_kg"),
            MaterialDemand.planned_weight_kg.label("planned_weight_kg"),
            MaterialDemand.shortage_weight_kg.label("shortage_weight_kg"),
            MaterialDemand.suggested_purchase_weight_kg.label("suggested_purchase_weight_kg"),
            MaterialDemand.status.label("status"),
            MaterialDemand.generated_at.label("generated_at"),
            MaterialDemand.remark.label("remark"),
        )
        .join(Item, Item.id == MaterialDemand.material_item_id)
        .where(MaterialDemand.sales_order_item_id.in_(item_ids) if item_ids else false())
        .where(MaterialDemand.status != "SUPERSEDED")
        .order_by(MaterialDemand.id)
    ).mappings().all()
    demand_ids = [row["demand_id"] for row in demands]

    po_items = db.execute(
        select(
            PurchaseOrderItem.id.label("purchase_order_item_id"),
            PurchaseOrderItem.purchase_order_id.label("purchase_order_id"),
            PurchaseOrderItem.source_demand_id.label("source_demand_id"),
            PurchaseOrderItem.line_no.label("line_no"),
            PurchaseOrderItem.order_weight_kg.label("order_weight_kg"),
            PurchaseOrderItem.received_weight_kg.label("received_weight_kg"),
            PurchaseOrderItem.unit_price.label("unit_price"),
            PurchaseOrderItem.line_amount.label("line_amount"),
            PurchaseOrderItem.status.label("status"),
            PurchaseOrder.po_no.label("po_no"),
            PurchaseOrder.order_date.label("order_date"),
            PurchaseOrder.expected_date.label("expected_date"),
            PurchaseOrder.total_amount.label("total_amount"),
            Supplier.supplier_name.label("supplier_name"),
        )
        .join(PurchaseOrder, PurchaseOrder.id == PurchaseOrderItem.purchase_order_id)
        .join(Supplier, Supplier.id == PurchaseOrder.supplier_id)
        .where(PurchaseOrderItem.source_demand_id.in_(demand_ids) if demand_ids else false())
        .order_by(PurchaseOrder.id, PurchaseOrderItem.line_no)
    ).mappings().all()
    po_item_ids = [row["purchase_order_item_id"] for row in po_items]

    receipt_items = db.execute(
        select(
            PurchaseReceiptItem.id.label("receipt_item_id"),
            PurchaseReceiptItem.purchase_order_item_id.label("purchase_order_item_id"),
            PurchaseReceiptItem.line_no.label("line_no"),
            PurchaseReceiptItem.lot_no.label("lot_no"),
            PurchaseReceiptItem.received_weight_kg.label("received_weight_kg"),
            PurchaseReceiptItem.accepted_weight_kg.label("accepted_weight_kg"),
            PurchaseReceiptItem.unit_cost.label("unit_cost"),
            PurchaseReceiptItem.status.label("status"),
            PurchaseReceipt.receipt_no.label("receipt_no"),
            PurchaseReceipt.receipt_date.label("receipt_date"),
            Warehouse.warehouse_name.label("warehouse_name"),
        )
        .join(PurchaseReceipt, PurchaseReceipt.id == PurchaseReceiptItem.receipt_id)
        .join(Warehouse, Warehouse.id == PurchaseReceipt.warehouse_id)
        .where(PurchaseReceiptItem.purchase_order_item_id.in_(po_item_ids) if po_item_ids else false())
        .order_by(PurchaseReceipt.id, PurchaseReceiptItem.line_no)
    ).mappings().all()

    work_orders = db.execute(
        select(
            WorkOrder.id.label("work_order_id"),
            WorkOrder.work_order_no.label("work_order_no"),
            WorkOrder.source_sales_order_item_id.label("source_sales_order_item_id"),
            WorkOrder.product_item_id.label("product_item_id"),
            Item.item_code.label("product_code"),
            Item.item_name.label("product_name"),
            WorkOrder.work_order_type.label("work_order_type"),
            WorkOrder.planned_qty.label("planned_qty"),
            WorkOrder.finished_qty.label("finished_qty"),
            WorkOrder.scrap_qty.label("scrap_qty"),
            WorkOrder.planned_start_time.label("planned_start_time"),
            WorkOrder.planned_end_time.label("planned_end_time"),
            WorkOrder.actual_start_time.label("actual_start_time"),
            WorkOrder.actual_end_time.label("actual_end_time"),
            WorkOrder.priority_level.label("priority_level"),
            WorkOrder.status.label("status"),
            WorkOrder.remark.label("remark"),
        )
        .join(Item, Item.id == WorkOrder.product_item_id)
        .where(WorkOrder.source_sales_order_item_id.in_(item_ids) if item_ids else false())
        .order_by(WorkOrder.id)
    ).mappings().all()
    work_order_ids = [row["work_order_id"] for row in work_orders]

    operations = db.execute(
        select(
            WorkOrderOperation.id.label("work_order_operation_id"),
            WorkOrderOperation.work_order_id.label("work_order_id"),
            WorkOrderOperation.seq_no.label("seq_no"),
            WorkOrderOperation.operation_name.label("operation_name"),
            WorkOrderOperation.planned_qty.label("planned_qty"),
            WorkOrderOperation.report_qty.label("report_qty"),
            WorkOrderOperation.good_qty.label("good_qty"),
            WorkOrderOperation.scrap_qty.label("scrap_qty"),
            WorkOrderOperation.rework_qty.label("rework_qty"),
            WorkOrderOperation.allowed_scrap_rate.label("allowed_scrap_rate"),
            WorkOrderOperation.is_abnormal.label("is_abnormal"),
            WorkOrderOperation.status.label("status"),
            WorkOrderOperation.remark.label("remark"),
        )
        .where(WorkOrderOperation.work_order_id.in_(work_order_ids) if work_order_ids else false())
        .order_by(WorkOrderOperation.work_order_id, WorkOrderOperation.seq_no)
    ).mappings().all()
    operation_ids = [row["work_order_operation_id"] for row in operations]

    reports = db.execute(
        select(
            OperationReport.id.label("report_id"),
            OperationReport.report_no.label("report_no"),
            OperationReport.work_order_id.label("work_order_id"),
            OperationReport.work_order_operation_id.label("work_order_operation_id"),
            Employee.employee_name.label("employee_name"),
            OperationReport.start_time.label("start_time"),
            OperationReport.end_time.label("end_time"),
            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.scrap_reason.label("scrap_reason"),
            OperationReport.downtime_minutes.label("downtime_minutes"),
            OperationReport.extra_cost_amount.label("extra_cost_amount"),
            OperationReport.is_abnormal.label("is_abnormal"),
        )
        .join(Employee, Employee.id == OperationReport.employee_id)
        .where(OperationReport.work_order_operation_id.in_(operation_ids) if operation_ids else false())
        .order_by(OperationReport.id)
    ).mappings().all()
    report_ids = [row["report_id"] for row in reports]

    scrap_records = db.execute(
        select(
            ScrapRecord.id.label("scrap_record_id"),
            ScrapRecord.scrap_no.label("scrap_no"),
            ScrapRecord.report_id.label("report_id"),
            ScrapRecord.work_order_id.label("work_order_id"),
            ScrapRecord.work_order_operation_id.label("work_order_operation_id"),
            ScrapRecord.scrap_qty.label("scrap_qty"),
            ScrapRecord.scrap_weight_kg.label("scrap_weight_kg"),
            ScrapRecord.disposal_type.label("disposal_type"),
            ScrapRecord.saleable_scrap_weight_kg.label("saleable_scrap_weight_kg"),
            ScrapRecord.scrap_cost_amount.label("scrap_cost_amount"),
            ScrapRecord.scrap_reason.label("scrap_reason"),
            ScrapRecord.status.label("status"),
        )
        .where(ScrapRecord.report_id.in_(report_ids) if report_ids else false())
        .order_by(ScrapRecord.id)
    ).mappings().all()

    source_material_lot = aliased(StockLot)
    completion_items = db.execute(
        select(
            CompletionReceiptItem.id.label("completion_item_id"),
            CompletionReceiptItem.completion_receipt_id.label("completion_receipt_id"),
            CompletionReceipt.work_order_id.label("work_order_id"),
            CompletionReceipt.receipt_no.label("receipt_no"),
            CompletionReceipt.receipt_time.label("receipt_time"),
            CompletionReceiptItem.lot_no.label("lot_no"),
            CompletionReceiptItem.receipt_qty.label("receipt_qty"),
            CompletionReceiptItem.receipt_weight_kg.label("receipt_weight_kg"),
            CompletionReceiptItem.unit_cost.label("unit_cost"),
            CompletionReceiptItem.source_material_lot_id.label("source_material_lot_id"),
            source_material_lot.lot_no.label("source_material_lot_no"),
            CompletionReceiptItem.source_material_sub_batch_no.label("source_material_sub_batch_no"),
            CompletionReceiptItem.status.label("status"),
            Warehouse.warehouse_name.label("warehouse_name"),
        )
        .join(CompletionReceipt, CompletionReceipt.id == CompletionReceiptItem.completion_receipt_id)
        .join(Warehouse, Warehouse.id == CompletionReceipt.warehouse_id)
        .outerjoin(source_material_lot, source_material_lot.id == CompletionReceiptItem.source_material_lot_id)
        .where(CompletionReceipt.work_order_id.in_(work_order_ids) if work_order_ids else false())
        .order_by(CompletionReceipt.id, CompletionReceiptItem.line_no)
    ).mappings().all()

    deliveries = db.execute(
        select(
            DeliveryItem.id.label("delivery_item_id"),
            DeliveryItem.delivery_id.label("delivery_id"),
            Delivery.delivery_no.label("delivery_no"),
            DeliveryItem.sales_order_item_id.label("sales_order_item_id"),
            DeliveryItem.product_item_id.label("product_item_id"),
            DeliveryItem.lot_id.label("lot_id"),
            StockLot.lot_no.label("lot_no"),
            DeliveryItem.delivery_qty.label("delivery_qty"),
            DeliveryItem.delivery_weight_kg.label("delivery_weight_kg"),
            DeliveryItem.unit_price.label("unit_price"),
            DeliveryItem.line_amount.label("line_amount"),
            DeliveryItem.status.label("status"),
            Delivery.delivery_date.label("delivery_date"),
            Delivery.consignee_name.label("consignee_name"),
            Delivery.consignee_phone.label("consignee_phone"),
            Delivery.delivery_address.label("delivery_address"),
        )
        .join(Delivery, Delivery.id == DeliveryItem.delivery_id)
        .outerjoin(StockLot, StockLot.id == DeliveryItem.lot_id)
        .where(Delivery.sales_order_id == sales_order_id)
        .order_by(Delivery.id, DeliveryItem.line_no)
    ).mappings().all()

    return_items = db.execute(
        select(
            ReturnItem.id.label("return_item_id"),
            ReturnItem.return_order_id.label("return_order_id"),
            ReturnOrder.return_no.label("return_no"),
            ReturnOrder.return_date.label("return_date"),
            ReturnOrder.return_reason.label("return_reason"),
            ReturnOrder.status.label("return_status"),
            ReturnItem.sales_order_item_id.label("sales_order_item_id"),
            ReturnItem.product_item_id.label("product_item_id"),
            ReturnItem.return_qty.label("return_qty"),
            ReturnItem.return_weight_kg.label("return_weight_kg"),
            ReturnItem.disposition_status.label("disposition_status"),
            ReturnItem.responsibility_type.label("responsibility_type"),
            ReturnItem.remark.label("remark"),
        )
        .join(ReturnOrder, ReturnOrder.id == ReturnItem.return_order_id)
        .where(
            (ReturnOrder.sales_order_id == sales_order_id)
            | (ReturnItem.sales_order_item_id.in_(item_ids) if item_ids else false())
        )
        .order_by(ReturnOrder.id, ReturnItem.line_no)
    ).mappings().all()
    return_item_ids = [row["return_item_id"] for row in return_items]

    rework_wo = aliased(WorkOrder)
    return_dispositions = db.execute(
        select(
            ReturnDisposition.id.label("return_disposition_id"),
            ReturnDisposition.disposition_no.label("disposition_no"),
            ReturnDisposition.return_item_id.label("return_item_id"),
            ReturnDisposition.disposition_type.label("disposition_type"),
            ReturnDisposition.scrap_weight_kg.label("scrap_weight_kg"),
            ReturnDisposition.extra_cost_amount.label("extra_cost_amount"),
            ReturnDisposition.scrap_sale_amount.label("scrap_sale_amount"),
            ReturnDisposition.closed_at.label("closed_at"),
            ReturnDisposition.status.label("status"),
            ReturnDisposition.remark.label("remark"),
            rework_wo.work_order_no.label("rework_work_order_no"),
        )
        .outerjoin(rework_wo, rework_wo.id == ReturnDisposition.rework_work_order_id)
        .where(ReturnDisposition.return_item_id.in_(return_item_ids) if return_item_ids else false())
        .order_by(ReturnDisposition.id)
    ).mappings().all()

    # Build nodes and edges from left to right.
    for row in order_items:
        node_id = f"item-{row['sales_order_item_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="sales_order_item",
                title=f"L{row['line_no']:02d} {row['product_name']}",
                subtitle=row["product_code"],
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="产品明细",
                level=1,
                summary="订单中的一个产品行，后续 MRP、工单、发货和退货都可追溯到该行。",
                metrics=[
                    ("订单数", _fmt_qty(row["order_qty"])),
                    ("已发数", _fmt_qty(row["delivered_qty"])),
                    ("行金额", _fmt_money(row["line_amount"])),
                ],
                details=[
                    ("产品编码", row["product_code"]),
                    ("产品名称", row["product_name"]),
                    ("规格", row["specification"]),
                    ("单重", _fmt_weight(row["unit_weight_kg"])),
                    ("客户料号", row["customer_part_no"]),
                    ("承诺交期", _fmt_date(row["promised_date"])),
                    ("订单单价", _fmt_money(row["unit_price"])),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(order_node_id, node_id, "订单产品"))

    for row in demands:
        node_id = f"mrp-{row['demand_id']}"
        source_id = f"item-{row['sales_order_item_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="mrp_demand",
                title=row["demand_no"],
                subtitle=f"{row['material_code']} · {row['material_name']}",
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="MRP需求",
                level=2,
                summary="按产品 BOM、订单数量、损耗率和库存余额生成的原材料需求。",
                metrics=[
                    ("计划重量", _fmt_weight(row["planned_weight_kg"])),
                    ("缺口重量", _fmt_weight(row["shortage_weight_kg"])),
                    ("建议采购", _fmt_weight(row["suggested_purchase_weight_kg"])),
                ],
                details=[
                    ("需求单号", row["demand_no"]),
                    ("原材料编码", row["material_code"]),
                    ("原材料名称", row["material_name"]),
                    ("规格", row["specification"]),
                    ("理论重量", _fmt_weight(row["theoretical_weight_kg"])),
                    ("订单数量", _fmt_qty(row["order_qty"])),
                    ("生成时间", _fmt_datetime(row["generated_at"])),
                    ("备注", row["remark"]),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "BOM展开"))

    for row in po_items:
        node_id = f"po-item-{row['purchase_order_item_id']}"
        source_id = f"mrp-{row['source_demand_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="purchase_order_item",
                title=f"{row['po_no']} L{row['line_no']:02d}",
                subtitle=row["supplier_name"],
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="采购订单",
                level=3,
                summary="MRP需求转为采购执行，记录供应商、采购重量、到货进度和采购金额。",
                metrics=[
                    ("采购重量", _fmt_weight(row["order_weight_kg"])),
                    ("已收重量", _fmt_weight(row["received_weight_kg"])),
                    ("行金额", _fmt_money(row["line_amount"])),
                ],
                details=[
                    ("采购单号", row["po_no"]),
                    ("供应商", row["supplier_name"]),
                    ("采购日期", _fmt_date(row["order_date"])),
                    ("预计到货", _fmt_date(row["expected_date"])),
                    ("采购单价", f"{_fmt_money(row['unit_price'])}/kg"),
                    ("采购总额", _fmt_money(row["total_amount"])),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "采购下单"))

    for row in receipt_items:
        node_id = f"receipt-item-{row['receipt_item_id']}"
        source_id = f"po-item-{row['purchase_order_item_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="purchase_receipt_item",
                title=f"{row['receipt_no']} L{row['line_no']:02d}",
                subtitle=row["lot_no"],
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="到货质检",
                level=4,
                summary="采购到货后进入仓库待检/放行，批次号成为后续库存追溯的关键。",
                metrics=[
                    ("收货重量", _fmt_weight(row["received_weight_kg"])),
                    ("合格重量", _fmt_weight(row["accepted_weight_kg"])),
                    ("入库单价", f"{_fmt_money(row['unit_cost'])}/kg"),
                ],
                details=[
                    ("入库单号", row["receipt_no"]),
                    ("批次号", row["lot_no"]),
                    ("仓库", row["warehouse_name"]),
                    ("收货时间", _fmt_datetime(row["receipt_date"])),
                    ("质检/入库状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "到货入库"))

    for row in work_orders:
        node_id = f"wo-{row['work_order_id']}"
        source_id = f"item-{row['source_sales_order_item_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="work_order",
                title=row["work_order_no"],
                subtitle=row["product_name"],
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="生产工单",
                level=5,
                summary="订单产品进入车间执行，工单承接 BOM 用料、工艺路线、报工和完工入库。",
                metrics=[
                    ("计划数", _fmt_qty(row["planned_qty"])),
                    ("完工数", _fmt_qty(row["finished_qty"])),
                    ("报废数", _fmt_qty(row["scrap_qty"])),
                ],
                details=[
                    ("工单号", row["work_order_no"]),
                    ("工单类型", _label(WORK_ORDER_TYPE_LABELS, row["work_order_type"])),
                    ("产品编码", row["product_code"]),
                    ("产品名称", row["product_name"]),
                    ("计划开始", _fmt_datetime(row["planned_start_time"])),
                    ("计划完工", _fmt_datetime(row["planned_end_time"])),
                    ("实际开始", _fmt_datetime(row["actual_start_time"])),
                    ("实际完工", _fmt_datetime(row["actual_end_time"])),
                    ("优先级", row["priority_level"]),
                    ("备注", row["remark"]),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "工单下达"))

    operation_by_work_order: dict[int, list[dict[str, Any]]] = {}
    for row in operations:
        operation_by_work_order.setdefault(row["work_order_id"], []).append(dict(row))
        node_id = f"op-{row['work_order_operation_id']}"
        source_id = f"wo-{row['work_order_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="work_order_operation",
                title=f"OP{row['seq_no']:02d} {row['operation_name']}",
                subtitle="关键异常" if row["is_abnormal"] else "工序执行",
                status=row["status"],
                tone="danger" if row["is_abnormal"] else _tone_for_status(row["status"]),
                stage="工序执行",
                level=6,
                summary="工序节点反映计划数、报工数、合格数、报废返工数和异常情况。",
                metrics=[
                    ("计划数", _fmt_qty(row["planned_qty"])),
                    ("已报工", _fmt_qty(row["report_qty"])),
                    ("报废数", _fmt_qty(row["scrap_qty"])),
                ],
                details=[
                    ("工序序号", row["seq_no"]),
                    ("工序名称", row["operation_name"]),
                    ("合格数", _fmt_qty(row["good_qty"])),
                    ("返工数", _fmt_qty(row["rework_qty"])),
                    ("允许报废率", f"{_num(row['allowed_scrap_rate']) * 100:.2f}%"),
                    ("是否异常", "是" if row["is_abnormal"] else "否"),
                    ("备注", row["remark"]),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "工艺路线"))

    for row in reports:
        node_id = f"report-{row['report_id']}"
        source_id = f"op-{row['work_order_operation_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="operation_report",
                title=row["report_no"],
                subtitle=row["employee_name"],
                status="ABNORMAL" if row["is_abnormal"] else "NORMAL",
                tone="danger" if row["is_abnormal"] else "success",
                stage="工人报工",
                level=7,
                summary="一线员工扫码报工记录，直接影响工序进度、节拍、报废返工和成本。",
                metrics=[
                    ("报工数", _fmt_qty(row["report_qty"])),
                    ("合格数", _fmt_qty(row["good_qty"])),
                    ("报废数", _fmt_qty(row["scrap_qty"])),
                ],
                details=[
                    ("报工单号", row["report_no"]),
                    ("员工", row["employee_name"]),
                    ("开始时间", _fmt_datetime(row["start_time"])),
                    ("结束时间", _fmt_datetime(row["end_time"])),
                    ("返工数", _fmt_qty(row["rework_qty"])),
                    ("停机分钟", _fmt_qty(row["downtime_minutes"])),
                    ("额外成本", _fmt_money(row["extra_cost_amount"])),
                    ("异常原因", row["scrap_reason"] or ("节拍/报废异常" if row["is_abnormal"] else "-")),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "扫码报工"))

    for row in scrap_records:
        node_id = f"scrap-{row['scrap_record_id']}"
        source_id = f"report-{row['report_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="scrap_rework",
                title=row["scrap_no"],
                subtitle="返工" if row["disposal_type"] == "REWORK" else "报废",
                status=row["status"],
                tone="warning" if row["disposal_type"] == "REWORK" else "danger",
                stage="报废/返工",
                level=8,
                summary="对报工异常、不良品进行处置，返工进入再加工，报废进入废料成本/收益口径。",
                metrics=[
                    ("处置数", _fmt_qty(row["scrap_qty"])),
                    ("处置重量", _fmt_weight(row["scrap_weight_kg"])),
                    ("成本", _fmt_money(row["scrap_cost_amount"])),
                ],
                details=[
                    ("处置单号", row["scrap_no"]),
                    ("处置方式", _disposal_label(row["disposal_type"])),
                    ("可售废料重量", _fmt_weight(row["saleable_scrap_weight_kg"])),
                    ("处置原因", row["scrap_reason"]),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "异常处置"))

    for row in completion_items:
        node_id = f"fg-{row['completion_item_id']}"
        source_id = f"wo-{row['work_order_id']}"
        source_material_lot_no = row["source_material_lot_no"] or row["source_material_sub_batch_no"] or "-"
        add_node(
            _node(
                node_id=node_id,
                node_type="completion_receipt",
                title=row["receipt_no"],
                subtitle=row["lot_no"],
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="成品入库",
                level=9,
                summary="工单全部工序完成后形成成品入库批次，后续发货从该批次扣减。",
                metrics=[
                    ("入库数量", _fmt_qty(row["receipt_qty"])),
                    ("入库重量", _fmt_weight(row["receipt_weight_kg"])),
                    ("单位成本", _fmt_money(row["unit_cost"])),
                ],
                details=[
                    ("成品入库单", row["receipt_no"]),
                    ("成品批次", row["lot_no"]),
                    ("来源库存批次号", source_material_lot_no),
                    ("仓库", row["warehouse_name"]),
                    ("入库时间", _fmt_datetime(row["receipt_time"])),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "完工入库"))

    for row in deliveries:
        node_id = f"delivery-{row['delivery_item_id']}"
        source_id = f"item-{row['sales_order_item_id']}"
        lot_node = next((node for node in nodes if node.type == "completion_receipt" and node.subtitle == row["lot_no"]), None)
        if lot_node:
            source_id = lot_node.id
        add_node(
            _node(
                node_id=node_id,
                node_type="delivery",
                title=row["delivery_no"],
                subtitle=row["lot_no"] or "发货明细",
                status=row["status"],
                tone=_tone_for_status(row["status"]),
                stage="发货",
                level=10,
                summary="成品从库存批次发出，记录发货数量、金额、收货人与送货地址。",
                metrics=[
                    ("发货数量", _fmt_qty(row["delivery_qty"])),
                    ("发货重量", _fmt_weight(row["delivery_weight_kg"])),
                    ("金额", _fmt_money(row["line_amount"])),
                ],
                details=[
                    ("发货单号", row["delivery_no"]),
                    ("库存批次号", row["lot_no"]),
                    ("发货时间", _fmt_datetime(row["delivery_date"])),
                    ("收货人", row["consignee_name"]),
                    ("联系电话", row["consignee_phone"]),
                    ("送货地址", row["delivery_address"]),
                    ("单价", _fmt_money(row["unit_price"])),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "发货出库"))

    for row in return_items:
        node_id = f"return-{row['return_item_id']}"
        source_id = f"item-{row['sales_order_item_id']}" if row["sales_order_item_id"] else order_node_id
        add_node(
            _node(
                node_id=node_id,
                node_type="return",
                title=row["return_no"],
                subtitle=row["return_reason"] or "退货",
                status=row["disposition_status"],
                tone=_tone_for_status(row["disposition_status"]),
                stage="退货售后",
                level=11,
                summary="客户退货登记，进入待处置状态后可选择返工或报废。",
                metrics=[
                    ("退货数量", _fmt_qty(row["return_qty"])),
                    ("退货重量", _fmt_weight(row["return_weight_kg"])),
                    ("处置状态", _status_label(row["disposition_status"])),
                ],
                details=[
                    ("退货单号", row["return_no"]),
                    ("退货时间", _fmt_datetime(row["return_date"])),
                    ("退货原因", row["return_reason"]),
                    ("责任归属", row["responsibility_type"]),
                    ("退货单状态", _status_label(row["return_status"])),
                    ("备注", row["remark"]),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "退货登记"))

    for row in return_dispositions:
        node_id = f"return-disposition-{row['return_disposition_id']}"
        source_id = f"return-{row['return_item_id']}"
        add_node(
            _node(
                node_id=node_id,
                node_type="return_disposition",
                title=row["disposition_no"],
                subtitle="返工" if row["disposition_type"] == "REWORK" else "报废",
                status=row["status"],
                tone="warning" if row["disposition_type"] == "REWORK" else "danger",
                stage="退货处置",
                level=12,
                summary="退货最终处置结果。返工会产生返工工单和额外人工成本，报废会形成废料销售金额。",
                metrics=[
                    ("报废重量", _fmt_weight(row["scrap_weight_kg"])),
                    ("额外人工", _fmt_money(row["extra_cost_amount"])),
                    ("废料销售", _fmt_money(row["scrap_sale_amount"])),
                ],
                details=[
                    ("处置单号", row["disposition_no"]),
                    ("处置方式", _disposal_label(row["disposition_type"])),
                    ("返工工单", row["rework_work_order_no"]),
                    ("完成时间", _fmt_datetime(row["closed_at"])),
                    ("备注", row["remark"]),
                    ("状态", _status_label(row["status"])),
                ],
                raw=dict(row),
            )
        )
        add_edge(_edge(source_id, node_id, "售后处置"))

    return SalesOrderLifecycleRead(order=order, nodes=nodes, edges=edges)