出库细胞来源链质检查看 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在「出库物料准备」出库细胞行提供「查看质检」，经聚合接口实时展示该产品整条来源链（祖先细胞产品 + 上游样本）上全部状态质检任务的检测参数、结论、备注与附件。

Architecture: 后端新增 GET /web/admin/cell-products/{id}/lineage-qc：复用 get_product_chain 展平 ancestors+current，按节点批量查质检工单与任务（含附件与检测参数结果），一次返回。前端新建只读弹窗，操作列入口接入 OutboundReleaseMaterialDialog；任务卡片复用 QcWorkOrderTaskResultPanel。

Tech Stack: FastAPI + SQLAlchemy（后端），Vue 3 + Element Plus + SCSS（前端），pytest（后端单测）

Design Spec: docs/superpowers/specs/2026-07-29-outbound-cell-lineage-qc-design.md

File Structure

Action	File	Responsibility
Modify	`backend/app/crud/cell_product.py`	`_flatten_lineage_nodes`、`get_product_lineage_qc` 编排
Modify	`backend/app/crud/qc_task_result.py`	批量按 task_id 查已保存检测参数
Modify	`backend/app/api/production.py`	注册 `lineage-qc` 路由
Create	`backend/tests/test_cell_product_lineage_qc.py`	展平与编排单测
Modify	`web-admin/src/api/cell-product.ts`	`fetchCellProductLineageQc`
Create	`web-admin/src/components/business/cell-product-lineage-qc-dialog/index.vue`	来源链质检只读弹窗
Modify	`web-admin/src/views/pilot-service/outbound-release-review/OutboundReleaseMaterialDialog.vue`	操作列「查看质检」入口

本期不改： 出库保存 payload、outbound-material-form-sections（除非实现时发现物料弹窗完全委托该组件——则以该组件为准补入口，默认改 OutboundReleaseMaterialDialog.vue）。

权限（闭合设计开放项）： 与 /cell-products/{id}/chain 一致，使用 require_admin_or_super_admin（出库物料保存已用同一守卫；能进物料准备即可调本只读接口）。租户：仅 tenant_no 过滤，不套用质检 data scope，避免放行角色看不到工单。

Task 1: 后端 — 来源链节点展平（TDD）

Files:

Modify: backend/app/crud/cell_product.py
Create: backend/tests/test_cell_product_lineage_qc.py
[ ] Step 1: 写失败单测（展平顺序与去重）

创建 backend/tests/test_cell_product_lineage_qc.py：

"""细胞产品来源链质检 — 节点展平与编排"""
import unittest
from unittest.mock import MagicMock, patch

from app.crud.cell_product import _flatten_lineage_nodes


class TestFlattenLineageNodes(unittest.TestCase):
    def test_current_only_when_no_ancestors(self):
        chain = {
            "current": {
                "id": "cp-cur",
                "productCode": "C-CUR",
                "productName": "当前",
                "nodeType": "cell_product",
                "passage": "P2",
                "isCurrent": True,
            },
            "ancestors": [],
        }
        nodes = _flatten_lineage_nodes(chain)
        self.assertEqual(len(nodes), 1)
        self.assertEqual(nodes[0]["id"], "cp-cur")
        self.assertTrue(nodes[0]["isCurrent"])

    def test_sample_then_parent_then_current_order(self):
        sample = {
            "id": "s1",
            "nodeType": "sample",
            "sampleCode": "YB001",
            "receptionSource": "cell_storage",
            "receptionSourceLabel": "细胞存储",
        }
        parent = {
            "id": "cp-p1",
            "productCode": "C-P1",
            "productName": "父代",
            "nodeType": "cell_product",
            "passage": "P1",
            "isCurrent": False,
        }
        current = {
            "id": "cp-cur",
            "productCode": "C-CUR",
            "productName": "当前",
            "nodeType": "cell_product",
            "passage": "P2",
            "isCurrent": True,
        }
        chain = {
            "current": current,
            "ancestors": [
                {
                    "node": parent,
                    "edge": {},
                    "parent": {"node": sample, "edge": {}, "parent": None},
                }
            ],
        }
        nodes = _flatten_lineage_nodes(chain)
        self.assertEqual([n["id"] for n in nodes], ["s1", "cp-p1", "cp-cur"])
        self.assertEqual(nodes[0]["nodeType"], "sample")
        self.assertTrue(nodes[-1]["isCurrent"])

    def test_dedupe_same_node_type_and_id(self):
        parent = {
            "id": "cp-p1",
            "nodeType": "cell_product",
            "productCode": "C-P1",
            "productName": "父",
            "passage": "P1",
            "isCurrent": False,
        }
        current = {
            "id": "cp-cur",
            "nodeType": "cell_product",
            "productCode": "C-CUR",
            "productName": "当前",
            "passage": "P2",
            "isCurrent": True,
        }
        # 多进一：两条边指向同一父产品
        chain = {
            "current": current,
            "ancestors": [
                {"node": parent, "edge": {}, "parent": None},
                {"node": dict(parent), "edge": {}, "parent": None},
            ],
        }
        nodes = _flatten_lineage_nodes(chain)
        ids = [n["id"] for n in nodes]
        self.assertEqual(ids.count("cp-p1"), 1)
        self.assertEqual(ids[-1], "cp-cur")

    def test_parent_array_multi_input(self):
        s1 = {"id": "s1", "nodeType": "sample", "sampleCode": "A", "receptionSource": "cell_storage"}
        s2 = {"id": "s2", "nodeType": "sample", "sampleCode": "B", "receptionSource": "cell_storage"}
        parent = {
            "id": "cp-p1",
            "nodeType": "cell_product",
            "productCode": "P1",
            "productName": "父",
            "passage": "P1",
            "isCurrent": False,
        }
        current = {
            "id": "cp-cur",
            "nodeType": "cell_product",
            "productCode": "CUR",
            "productName": "当前",
            "passage": "P2",
            "isCurrent": True,
        }
        chain = {
            "current": current,
            "ancestors": [
                {
                    "node": parent,
                    "edge": {},
                    "parent": [
                        {"node": s1, "edge": {}, "parent": None},
                        {"node": s2, "edge": {}, "parent": None},
                    ],
                }
            ],
        }
        nodes = _flatten_lineage_nodes(chain)
        ids = [n["id"] for n in nodes]
        self.assertIn("s1", ids)
        self.assertIn("s2", ids)
        self.assertEqual(ids[-1], "cp-cur")


if __name__ == "__main__":
    unittest.main()

Step 2: 跑测确认失败

cd backend && source venv/bin/activate && pytest tests/test_cell_product_lineage_qc.py::TestFlattenLineageNodes -v

Expected: FAIL（_flatten_lineage_nodes 未定义）

Step 3: 实现 _flatten_lineage_nodes

在 backend/app/crud/cell_product.py 靠近 get_product_chain 处新增：

def _lineage_node_key(node: Dict[str, Any]) -> str:
    nt = (node.get("nodeType") or "").strip()
    nid = (node.get("id") or "").strip()
    if nt == "sample":
        code = (node.get("sampleCode") or "").strip()
        return f"sample:{(nid or code)}"
    return f"{nt}:{nid}"


def _walk_ancestor_link(
    link: Any,
    out: List[Dict[str, Any]],
    seen: set,
) -> None:
    if not link:
        return
    if isinstance(link, list):
        for item in link:
            _walk_ancestor_link(item, out, seen)
        return
    if not isinstance(link, dict):
        return
    parent = link.get("parent")
    _walk_ancestor_link(parent, out, seen)
    node = link.get("node")
    if not isinstance(node, dict):
        return
    key = _lineage_node_key(node)
    if not key or key in seen:
        return
    seen.add(key)
    out.append(dict(node))


def _flatten_lineage_nodes(chain: Dict[str, Any]) -> List[Dict[str, Any]]:
    """将 get_product_chain 结果展平为上游→下游节点列表（含 current，不含 descendants）。"""
    out: List[Dict[str, Any]] = []
    seen: set = set()
    for link in chain.get("ancestors") or []:
        _walk_ancestor_link(link, out, seen)
    current = chain.get("current")
    if isinstance(current, dict):
        key = _lineage_node_key(current)
        if key and key not in seen:
            seen.add(key)
            node = dict(current)
            node["isCurrent"] = True
            out.append(node)
        elif key:
            for n in out:
                if _lineage_node_key(n) == key:
                    n["isCurrent"] = True
                    break
    return out

Step 4: 跑测确认通过

cd backend && source venv/bin/activate && pytest tests/test_cell_product_lineage_qc.py::TestFlattenLineageNodes -v

Expected: PASS

Step 5: Commit（若本会话要求提交）

git add backend/app/crud/cell_product.py backend/tests/test_cell_product_lineage_qc.py
git commit -m "$(cat <<'EOF'
feat: add lineage node flatten helper for cell product QC view

EOF
)"

Task 2: 后端 — 批量检测参数 + `get_product_lineage_qc`

Files:

Modify: backend/app/crud/qc_task_result.py
Modify: backend/app/crud/cell_product.py
Modify: backend/tests/test_cell_product_lineage_qc.py
[ ] Step 1: 写失败单测（编排挂载工单与任务字段）

在 test_cell_product_lineage_qc.py 追加：

from app.crud.cell_product import get_product_lineage_qc


class TestGetProductLineageQc(unittest.TestCase):
    @patch("app.crud.cell_product.get_product_chain")
    @patch("app.crud.cell_product._load_lineage_qc_for_nodes")
    def test_missing_product_returns_none(self, _load, get_chain):
        get_chain.return_value = None
        db = MagicMock()
        self.assertIsNone(get_product_lineage_qc(db, "missing", "zkxw"))
        _load.assert_not_called()

    @patch("app.crud.cell_product.get_product_chain")
    @patch("app.crud.cell_product._load_lineage_qc_for_nodes")
    def test_returns_product_header_and_nodes(self, load_nodes, get_chain):
        get_chain.return_value = {
            "current": {
                "id": "cp1",
                "productCode": "C1",
                "productName": "产品1",
                "nodeType": "cell_product",
                "passage": "P1",
                "isCurrent": True,
            },
            "ancestors": [],
        }
        load_nodes.return_value = [
            {
                "nodeType": "cell_product",
                "id": "cp1",
                "label": "C1",
                "productName": "产品1",
                "passage": "P1",
                "receptionSource": None,
                "isCurrent": True,
                "workOrders": [
                    {
                        "id": "wo1",
                        "workOrderNo": "WO1",
                        "workOrderType": "temporary",
                        "workOrderStatus": "ywc",
                        "workOrderStatusLabel": "已完成",
                        "tasks": [
                            {
                                "id": "t1",
                                "taskNo": "T1",
                                "itemName": "无菌",
                                "results": [{"paramName": "结果", "paramValue": "阴性"}],
                                "attachmentFiles": [],
                                "remark": "",
                                "conclusion": "tg",
                            }
                        ],
                    }
                ],
            }
        ]
        db = MagicMock()
        out = get_product_lineage_qc(db, "cp1", "zkxw")
        self.assertEqual(out["productId"], "cp1")
        self.assertEqual(out["productCode"], "C1")
        self.assertEqual(len(out["nodes"]), 1)
        self.assertEqual(out["nodes"][0]["workOrders"][0]["tasks"][0]["results"][0]["paramValue"], "阴性")

Step 2: 跑测确认失败

cd backend && source venv/bin/activate && pytest tests/test_cell_product_lineage_qc.py::TestGetProductLineageQc -v

Expected: FAIL（函数未定义）

Step 3: 在 qc_task_result.py 增加批量已保存结果

def list_saved_results_by_task_ids(
    db: Session, task_ids: List[str], tenant_no: str
) -> Dict[str, List[Dict[str, Any]]]:
    """按任务 ID 批量返回已保存检测参数（只读展示用，不合并模板空项）。"""
    ids = list({(x or "").strip() for x in task_ids if (x or "").strip()})
    if not ids or not tenant_no:
        return {}
    rows = (
        db.query(BizQcTaskResult)
        .filter(
            BizQcTaskResult.task_id.in_(ids),
            BizQcTaskResult.tenant_no == tenant_no,
            BizQcTaskResult.del_flag == 0,
        )
        .order_by(BizQcTaskResult.create_time.asc())
        .all()
    )
    grouped: Dict[str, List[Dict[str, Any]]] = {i: [] for i in ids}
    for r in rows:
        tid = (r.task_id or "").strip()
        if tid in grouped:
            grouped[tid].append(_row_to_dict(r))
    return grouped

Step 4: 实现 _load_lineage_qc_for_nodes 与 get_product_lineage_qc

在 cell_product.py 实现（要点，实现时保持与仓库 import 风格一致）：

def _wo_status_label(status: str) -> str:
    from app.crud.qc_work_order import _WORK_ORDER_STATUS_LABELS
    return _WORK_ORDER_STATUS_LABELS.get(status or "", status or "")


def _slim_task_for_lineage(task: Dict[str, Any], results: List[Dict[str, Any]]) -> Dict[str, Any]:
    return {
        "id": task.get("id") or "",
        "taskNo": task.get("taskNo") or "",
        "itemName": task.get("itemName") or "",
        "itemCode": task.get("itemCode") or "",
        "taskType": task.get("taskType") or "",
        "taskStatus": task.get("taskStatus") or "",
        "conclusion": task.get("conclusion") or "",
        "conclusionLabel": task.get("conclusionLabel") or "",
        "remark": task.get("remark") or "",
        "completedTime": task.get("completedTime") or "",
        "companyName": task.get("companyName") or "",
        "results": [
            {
                "paramName": r.get("paramName") or "",
                "paramValue": r.get("paramValue") or "",
                "paramUnit": r.get("paramUnit") or "",
            }
            for r in (results or [])
        ],
        "attachmentFiles": task.get("attachmentFiles") or [],
    }


def _load_lineage_qc_for_nodes(
    db: Session, nodes: List[Dict[str, Any]], tenant_no: str
) -> List[Dict[str, Any]]:
    from app.models.models import BizQcWorkOrder
    from app.crud import qc_task as task_svc
    from app.crud import qc_task_result as result_svc
    from app.crud.qc_work_order import list_work_orders_by_sample, _wo_to_dict

    product_ids = [
        (n.get("id") or "").strip()
        for n in nodes
        if (n.get("nodeType") or "") == "cell_product" and (n.get("id") or "").strip()
    ]
    wo_by_product: Dict[str, List[BizQcWorkOrder]] = {pid: [] for pid in product_ids}
    if product_ids:
        rows = (
            db.query(BizQcWorkOrder)
            .filter(
                BizQcWorkOrder.cell_product_id.in_(product_ids),
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
            )
            .order_by(BizQcWorkOrder.create_time.desc())
            .all()
        )
        for wo in rows:
            pid = (wo.cell_product_id or "").strip()
            if pid in wo_by_product:
                wo_by_product[pid].append(wo)

    sample_wo_cache: Dict[Tuple[str, str], List[Dict[str, Any]]] = {}

    all_wo_ids: List[str] = []
    node_wo_objs: List[List[Any]] = []  # parallel to nodes: list of wo dict or ORM

    out_nodes: List[Dict[str, Any]] = []
    pending_wo_ids: List[str] = []

    # 先收集全部 wo id，再批量拉 tasks
    node_work_orders_orm: List[List[BizQcWorkOrder]] = []
    for n in nodes:
        if (n.get("nodeType") or "") == "cell_product":
            node_work_orders_orm.append(wo_by_product.get((n.get("id") or "").strip(), []))
        else:
            node_work_orders_orm.append([])  # sample filled below as dicts

    # sample：对齐 list_work_orders_by_sample（返回 dict 列表）
    sample_wo_dicts: Dict[int, List[Dict[str, Any]]] = {}
    for idx, n in enumerate(nodes):
        if (n.get("nodeType") or "") != "sample":
            continue
        code = (n.get("sampleCode") or "").strip()
        rs = (n.get("receptionSource") or "").strip()
        if not code or not rs:
            sample_wo_dicts[idx] = []
            continue
        key = (code, rs)
        if key not in sample_wo_cache:
            sample_wo_cache[key] = list_work_orders_by_sample(
                db,
                sample_code=code,
                reception_source=rs,
                tenant_no=tenant_no,
                user_info=None,
            )
        sample_wo_dicts[idx] = sample_wo_cache[key]

    for wos in node_work_orders_orm:
        for wo in wos:
            pending_wo_ids.append(wo.id)
    for wos in sample_wo_dicts.values():
        for wo in wos:
            wid = (wo.get("id") or "").strip()
            if wid:
                pending_wo_ids.append(wid)

    tasks_by_wo = task_svc.list_tasks_by_work_order_ids(db, pending_wo_ids, tenant_no)
    all_task_ids = [
        (t.get("id") or "").strip()
        for tasks in tasks_by_wo.values()
        for t in tasks
        if (t.get("id") or "").strip()
    ]
    results_by_task = result_svc.list_saved_results_by_task_ids(db, all_task_ids, tenant_no)

    def pack_orm_wos(wos: List[BizQcWorkOrder]) -> List[Dict[str, Any]]:
        packed = []
        for wo in wos:
            tasks = tasks_by_wo.get(wo.id, [])
            packed.append(
                {
                    "id": wo.id,
                    "workOrderNo": wo.work_order_no or "",
                    "workOrderType": wo.work_order_type or "",
                    "workOrderStatus": wo.work_order_status or "",
                    "workOrderStatusLabel": _wo_status_label(wo.work_order_status or ""),
                    "tasks": [
                        _slim_task_for_lineage(t, results_by_task.get(t.get("id") or "", []))
                        for t in tasks
                    ],
                }
            )
        return packed

    def pack_dict_wos(wos: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        packed = []
        for wo in wos:
            wid = (wo.get("id") or "").strip()
            tasks = tasks_by_wo.get(wid, [])
            packed.append(
                {
                    "id": wid,
                    "workOrderNo": wo.get("workOrderNo") or "",
                    "workOrderType": wo.get("workOrderType") or "",
                    "workOrderStatus": wo.get("workOrderStatus") or "",
                    "workOrderStatusLabel": wo.get("workOrderStatusLabel")
                    or _wo_status_label(wo.get("workOrderStatus") or ""),
                    "tasks": [
                        _slim_task_for_lineage(t, results_by_task.get(t.get("id") or "", []))
                        for t in tasks
                    ],
                }
            )
        return packed

    for idx, n in enumerate(nodes):
        nt = n.get("nodeType") or ""
        if nt == "sample":
            label = (n.get("sampleCode") or n.get("id") or "").strip()
            work_orders = pack_dict_wos(sample_wo_dicts.get(idx, []))
        else:
            label = (n.get("productCode") or n.get("id") or "").strip()
            work_orders = pack_orm_wos(node_work_orders_orm[idx])
        out_nodes.append(
            {
                "nodeType": nt,
                "id": n.get("id") or "",
                "label": label,
                "productName": n.get("productName") or "",
                "passage": n.get("passage") or None,
                "receptionSource": n.get("receptionSource") or None,
                "receptionSourceLabel": n.get("receptionSourceLabel") or None,
                "isCurrent": bool(n.get("isCurrent")),
                "workOrders": work_orders,
            }
        )
    return out_nodes


def get_product_lineage_qc(
    db: Session, product_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    chain = get_product_chain(db, product_id, tenant_no)
    if not chain:
        return None
    current = chain.get("current") or {}
    flat = _flatten_lineage_nodes(chain)
    nodes = _load_lineage_qc_for_nodes(db, flat, tenant_no)
    return {
        "productId": current.get("id") or product_id,
        "productCode": current.get("productCode") or "",
        "productName": current.get("productName") or "",
        "nodes": nodes,
    }

实现时注意：

工单状态文案：from app.crud.qc_work_order import _WORK_ORDER_STATUS_LABELS，然后
def _wo_status_label(status: str) -> str: return _WORK_ORDER_STATUS_LABELS.get(status or "", status or "")
list_tasks_by_work_order_ids 有 ids[:100] 上限；若 pending_wo_ids 超过 100，按 100 分块合并。
样本节点无 sampleCode/receptionSource 时 workOrders: []，不抛错。
删除未使用的 import。
Task 字段若列表接口已有 taskStatusLabel，slim 时一并带上，供前端 Tag 使用（可选，面板会用 registry 回退）。

Step 5: 跑全文件测试

cd backend && source venv/bin/activate && pytest tests/test_cell_product_lineage_qc.py -v

Expected: PASS

Step 6: Commit（若要求提交）

git add backend/app/crud/cell_product.py backend/app/crud/qc_task_result.py backend/tests/test_cell_product_lineage_qc.py
git commit -m "$(cat <<'EOF'
feat: add cell product lineage QC aggregate loader

EOF
)"

Task 3: 后端 — 注册路由

Files:

Modify: backend/app/api/production.py（紧挨 /cell-products/{product_id}/chain）
[ ] Step 1: 增加路由

在 get_cell_product_chain 之后插入：

@router.get("/cell-products/{product_id}/lineage-qc")
async def get_cell_product_lineage_qc(
    product_id: str,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    data = cell_product_svc.get_product_lineage_qc(db, product_id, tenant_no)
    if not data:
        raise HTTPException(status_code=404, detail="细胞产品不存在")
    return success_response(data)

确认文件顶部已有 HTTPException、require_admin_or_super_admin、cell_product_svc 导入（与 chain 路由相同）。

Step 2: 手工烟测（有本地服务时）

# 替换 TOKEN 与 PRODUCT_ID
curl -s -H "Authorization: Bearer $TOKEN" \
  "http://127.0.0.1:8000/api/web/admin/cell-products/$PRODUCT_ID/lineage-qc" | head -c 500

Expected: JSON code:200，含 nodes 数组；错误 id 返回 404。

Step 3: Commit（若要求提交）

git add backend/app/api/production.py
git commit -m "$(cat <<'EOF'
feat: expose cell product lineage-qc API endpoint

EOF
)"

Task 4: 前端 — API 封装

Files:

Modify: web-admin/src/api/cell-product.ts
[ ] Step 1: 追加类型与方法

在 fetchCellProductChain 附近增加：

export type CellProductLineageQcTask = {
  id?: string
  taskNo?: string
  itemName?: string
  itemCode?: string
  taskType?: string
  taskStatus?: string
  conclusion?: string
  conclusionLabel?: string
  remark?: string
  completedTime?: string
  companyName?: string
  results?: Array<{ paramName?: string; paramValue?: string; paramUnit?: string }>
  attachmentFiles?: Array<{ fileName?: string; fileUrl?: string; fileType?: string }>
}

export type CellProductLineageQcWorkOrder = {
  id?: string
  workOrderNo?: string
  workOrderType?: string
  workOrderStatus?: string
  workOrderStatusLabel?: string
  tasks?: CellProductLineageQcTask[]
}

export type CellProductLineageQcNode = {
  nodeType?: 'sample' | 'cell_product' | string
  id?: string
  label?: string
  productName?: string
  passage?: string | null
  receptionSource?: string | null
  receptionSourceLabel?: string | null
  isCurrent?: boolean
  workOrders?: CellProductLineageQcWorkOrder[]
}

export type CellProductLineageQc = {
  productId?: string
  productCode?: string
  productName?: string
  nodes?: CellProductLineageQcNode[]
}

export function fetchCellProductLineageQc(productId: string) {
  return request.get<CellProductLineageQc>({
    url: `/web/admin/cell-products/${productId}/lineage-qc`
  })
}

Step 2: Commit（若要求提交）

git add web-admin/src/api/cell-product.ts
git commit -m "$(cat <<'EOF'
feat: add fetchCellProductLineageQc API client

EOF
)"

Task 5: 前端 — `CellProductLineageQcDialog`

Files:

Create: web-admin/src/components/business/cell-product-lineage-qc-dialog/index.vue
[ ] Step 1: 创建弹窗组件

实现要点：

<!-- 细胞产品 — 来源链质检只读 -->
<template>
  <el-dialog
    :model-value="modelValue"
    title="来源链质检"
    width="1040px"
    class="dialog-form-layout cell-lineage-qc-dialog"
    align-center
    append-to-body
    modal-class="protocol-dialog-mask"
    destroy-on-close
    :close-on-click-modal="false"
    style="backdrop-filter: none"
    @update:model-value="emit('update:modelValue', $event)"
    @closed="resetState"
  >
    <div v-loading="loading" class="lineage-qc-body">
      <div v-if="data" class="lineage-qc-summary">
        <span class="lineage-qc-summary__code">{{ data.productCode || '—' }}</span>
        <span class="lineage-qc-summary__name">{{ data.productName || '—' }}</span>
      </div>

      <el-empty v-if="!loading && (!data?.nodes || !data.nodes.length)" description="暂无来源链数据" />

      <el-collapse v-else v-model="activeNames" class="lineage-qc-collapse">
        <el-collapse-item
          v-for="(node, idx) in data?.nodes || []"
          :key="`${node.nodeType}-${node.id}-${idx}`"
          :name="String(idx)"
        >
          <template #title>
            <div class="lineage-qc-node-title">
              <el-tag v-if="node.isCurrent" type="success" size="small">当前</el-tag>
              <el-tag v-if="node.nodeType === 'sample'" type="info" size="small">样本</el-tag>
              <el-tag v-else type="primary" size="small">细胞产品</el-tag>
              <span class="lineage-qc-node-title__label">{{ node.label || '—' }}</span>
              <span v-if="node.productName" class="lineage-qc-node-title__sub">{{ node.productName }}</span>
              <span v-if="node.passage" class="lineage-qc-node-title__sub">代次 {{ node.passage }}</span>
              <span v-if="node.receptionSourceLabel" class="lineage-qc-node-title__sub">
                {{ node.receptionSourceLabel }}
              </span>
            </div>
          </template>

          <div v-if="!(node.workOrders || []).length" class="lineage-qc-empty">暂无质检工单</div>
          <div v-for="wo in node.workOrders || []" :key="wo.id" class="lineage-qc-wo">
            <div class="lineage-qc-wo__head">
              <span>{{ wo.workOrderNo || '—' }}</span>
              <el-tag v-if="wo.workOrderStatus" size="small">{{ wo.workOrderStatusLabel || wo.workOrderStatus }}</el-tag>
            </div>
            <QcWorkOrderTaskResultPanel :tasks="mapTasks(wo.tasks)" />
          </div>
        </el-collapse-item>
      </el-collapse>
    </div>

    <template #footer>
      <el-button class="btn-ghost" @click="emit('update:modelValue', false)">
        <el-icon><Close /></el-icon>
        关闭
      </el-button>
    </template>
  </el-dialog>
</template>

Script 要点：

props: modelValue: boolean, productId: string
watch [modelValue, productId] → 打开且有 id 时 fetchCellProductLineageQc
activeNames 默认 nodes.map((_, i) => String(i))（全部展开）
mapTasks：把 API task 转成 QcWorkOrderTaskResultPanel 所需形状（含 results、attachmentFiles；materialUsages 不传）
导入：Close、fetchCellProductLineageQc、QcWorkOrderTaskResultPanel（路径 @/views/quality-inspection/qc-work-order/QcWorkOrderTaskResultPanel.vue）
样式：Fresh Mint；节点标题 flex + gap；工单卡片边框圆角与物料弹窗 section-card 一致
[ ] Step 2: Commit（若要求提交）

git add web-admin/src/components/business/cell-product-lineage-qc-dialog/index.vue
git commit -m "$(cat <<'EOF'
feat: add CellProductLineageQcDialog readonly panel

EOF
)"

Task 6: 接入出库物料准备弹窗

Files:

Modify: web-admin/src/views/pilot-service/outbound-release-review/OutboundReleaseMaterialDialog.vue
[ ] Step 1: 操作列增加「查看质检」

在出库细胞表格操作列模板中，删除按钮前增加：

<el-button type="primary" link @click="openLineageQc(row)">
  <el-icon><View /></el-icon>
  查看质检
</el-button>

操作列 min-width 调到 180 或 200
View 已从 @element-plus/icons-vue 导入则复用；否则补 import
状态：

const lineageQcVisible = ref(false)
const lineageQcProductId = ref('')

function openLineageQc(row: OutboundReleaseCellRow) {
  const id = String(row.cellProductId || '')
  if (!id) {
    ElMessage.warning('缺少细胞产品信息')
    return
  }
  lineageQcProductId.value = id
  lineageQcVisible.value = true
}

模板根内（主 dialog 外层已有 <div>）追加：

<CellProductLineageQcDialog
  v-model="lineageQcVisible"
  :product-id="lineageQcProductId"
/>

import：import CellProductLineageQcDialog from '@/components/business/cell-product-lineage-qc-dialog/index.vue'
[ ] Step 2: 浏览器验收

打开 /#/market/basic-experiment/outbound-release-review → 出库物料准备
添加有来源链的细胞产品 → 点「查看质检」
确认：独立弹窗、节点上游→当前、任务参数/结论/备注/附件、未完成状态可见
关闭质检弹窗后物料准备仍可继续编辑并保存

Step 3: 更新知识图谱

./scripts/update-graph.sh frontend
./scripts/update-graph.sh backend

Step 4: Commit（若要求提交）

git add web-admin/src/views/pilot-service/outbound-release-review/OutboundReleaseMaterialDialog.vue web-admin/graphify-out backend/graphify-out
git commit -m "$(cat <<'EOF'
feat: wire lineage QC viewer into outbound material dialog

EOF
)"

Spec Coverage Checklist

Spec 要求	Task
整条来源链（产品+样本）	Task 1–2
全部任务状态	Task 2（不过滤）
参数/结论/备注/附件	Task 2 slim task + Task 5 panel
行按钮独立弹窗	Task 5–6
实时查询不落库	全程 GET，不改 save
不展示物料使用	Task 5 不传 usages；全局 flag 已为 false
批量避免 N+1	Task 2 `list_tasks_by_work_order_ids` + `list_saved_results_by_task_ids`
权限可进物料准备即可	Task 3 `require_admin_or_super_admin`

Self-Review Notes

检测参数：现有工单列表任务不含 results，必须走 list_saved_results_by_task_ids（已写入 Task 2），避免前端再 N 次 fetchQcTaskResults。
list_tasks_by_work_order_ids 的 100 上限已要求分块。
未改 outbound-material-form-sections：当前放行页使用 OutboundReleaseMaterialDialog；若联调发现实际渲染走共用组件，在 Task 6 改为改共用组件并两边验证。

2026-07-29-outbound-cell-lineage-qc.md 31 KB تاريخچه خام

出库细胞来源链质检查看 Implementation Plan

File Structure

Task 1: 后端 — 来源链节点展平（TDD）

Task 2: 后端 — 批量检测参数 + get_product_lineage_qc

Task 3: 后端 — 注册路由

Task 4: 前端 — API 封装

Task 5: 前端 — CellProductLineageQcDialog

Task 6: 接入出库物料准备弹窗

Spec Coverage Checklist

Self-Review Notes

2026-07-29-outbound-cell-lineage-qc.md 31 KB

تاريخچه خام

Task 2: 后端 — 批量检测参数 + `get_product_lineage_qc`

Task 5: 前端 — `CellProductLineageQcDialog`