# 细胞产品库 — 来源链路展示 实施计划 > **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:** 后端在现有列表 API 中批量富化来源字段，新增 `/cell-products/{id}/chain` 链路 API。前端列表新增 4 列 + 1 个筛选，查看弹窗新增 `SourceChainGraph.vue` 组件（CSS 绘制横向 DAG）。 **Tech Stack:** FastAPI + SQLAlchemy + Pydantic v2（后端），Vue 3 + Element Plus + SCSS（前端） **Design Spec:** `docs/superpowers/specs/2026-05-24-cell-product-source-chain-design.md` **Demo:** `docs/demo-source-chain-graph.html` --- ## File Structure | Action | File | Responsibility | |--------|------|----------------| | Modify | `backend/app/crud/cell_product.py` | 来源富化查询 + 链路递归函数 | | Modify | `backend/app/schemas/cell_product.py` | CellProductListQuery 新增 source_type | | Modify | `backend/app/api/production.py` | 列表路由传 source_type + 新增 chain 路由 | | Modify | `web-admin/src/api/cell-product.ts` | 新增 fetchCellProductChain + list 参数扩展 | | Modify | `web-admin/src/views/production/cell-product/index.vue` | 列表新增列/筛选 + 查看弹窗集成链路图 | | Create | `web-admin/src/views/production/cell-product/SourceChainGraph.vue` | 链路图组件 | --- ### Task 1: 后端 — Schema 新增 source_type 筛选参数 **Files:** - Modify: `backend/app/schemas/cell_product.py:127-137` - [ ] **Step 1: 在 CellProductListQuery 中新增 source_type 字段** 在 `backend/app/schemas/cell_product.py` 的 `CellProductListQuery` 类末尾追加： ```python source_type: Optional[str] = Field(None, alias="sourceType") ``` 完整的 `CellProductListQuery` 应变为： ```python class CellProductListQuery(BaseModel): model_config = ConfigDict(populate_by_name=True) page: int = 1 page_size: int = Field(20, alias="pageSize") keyword: Optional[str] = None product_code: Optional[str] = Field(None, alias="productCode") product_name: Optional[str] = Field(None, alias="productName") product_type_id: Optional[str] = Field(None, alias="productTypeId") product_status: Optional[str] = Field(None, alias="productStatus") passage: Optional[str] = None source_type: Optional[str] = Field(None, alias="sourceType") ``` - [ ] **Step 2: 验证无语法错误** Run: `cd /Users/wuzilong/work/00code/zkxw/backend && source venv/bin/activate && python -c "from app.schemas.cell_product import CellProductListQuery; print('OK')"` Expected: `OK` --- ### Task 2: 后端 — 列表 API 增加 source_type 筛选 **Files:** - Modify: `backend/app/api/production.py:1941-1966` - Modify: `backend/app/crud/cell_product.py:150-208` - [ ] **Step 1: API 路由层新增 sourceType 参数** 在 `backend/app/api/production.py` 的 `list_cell_products` 函数中： 1. 在参数列表 `passage` 之后新增：`sourceType: Optional[str] = None,` 2. 在 `cell_product_svc.list_cell_products(...)` 调用中新增：`source_type=sourceType,` 修改后的函数签名和调用： ```python @router.get("/cell-products") async def list_cell_products( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100, alias="pageSize"), keyword: Optional[str] = None, productCode: Optional[str] = None, productName: Optional[str] = None, productTypeId: Optional[str] = None, productStatus: Optional[str] = None, passage: Optional[str] = None, sourceType: Optional[str] = None, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _get_tenant_no(db, user) result = cell_product_svc.list_cell_products( db, tenant_no=tenant_no, page=page, page_size=pageSize, keyword=keyword, product_code=productCode, product_name=productName, product_type_id=productTypeId, product_status=productStatus, passage=passage, source_type=sourceType, user_info=user, ) items = result.get("items", []) total = result.get("total", 0) tp = cell_product_svc.total_pages(total, pageSize) return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp}) ``` - [ ] **Step 2: CRUD 层 list_cell_products 新增 source_type 过滤** 在 `backend/app/crud/cell_product.py` 的 `list_cell_products` 函数中： 1. 在参数列表 `passage` 之后新增：`source_type: Optional[str] = None,` 2. 在 `_filtered()` 内部、`if passage:` 块之后新增： ```python if source_type: q0 = q0.filter(BizCellProduct.source_type == source_type) ``` - [ ] **Step 3: 验证无语法错误** Run: `cd /Users/wuzilong/work/00code/zkxw/backend && source venv/bin/activate && python -c "from app.crud.cell_product import list_cell_products; print('OK')"` Expected: `OK` --- ### Task 3: 后端 — 列表富化查询新增来源字段 **Files:** - Modify: `backend/app/crud/cell_product.py:72-142`（`_enrich_product_dicts` 函数） - [ ] **Step 1: 在 `_enrich_product_dicts` 中新增来源信息批量富化** 在 `_enrich_product_dicts` 函数中，在 `slot_map` 查询之后、`for item in items:` 循环之前，新增以下代码块： ```python # --- 来源信息富化 --- from app.models.models import BizProductionTask, BizProductionWorkOrder task_ids = {i.get("productionTaskId") for i in items if i.get("productionTaskId")} task_map: Dict[str, Any] = {} task_to_wo: Dict[str, str] = {} wo_map: Dict[str, Any] = {} if task_ids: tasks = ( db.query(BizProductionTask) .filter( BizProductionTask.id.in_(task_ids), BizProductionTask.del_flag == 0, ) .all() ) for t in tasks: task_map[t.id] = t if t.work_order_id: task_to_wo[t.id] = t.work_order_id wo_ids = set(task_to_wo.values()) if wo_ids: wos = ( db.query(BizProductionWorkOrder) .filter( BizProductionWorkOrder.id.in_(wo_ids), BizProductionWorkOrder.del_flag == 0, ) .all() ) wo_map = {w.id: w for w in wos} source_type_labels = {"production": "生产入库", "manual": "手动录入"} ``` 然后在 `for item in items:` 循环内，在 `item["storageSlotName"] = ...` 之后追加： ```python # 来源信息 src_type = item.get("sourceType", "") item["sourceTypeLabel"] = source_type_labels.get(src_type, src_type or "") tid = item.get("productionTaskId") or "" if tid and tid in task_map: t_obj = task_map[tid] item["sourceTaskNo"] = t_obj.task_no or "" wo_id = task_to_wo.get(tid) if wo_id and wo_id in wo_map: wo_obj = wo_map[wo_id] item["sourceWorkOrderNo"] = wo_obj.work_order_no or "" item["sourceProtocolNo"] = wo_obj.protocol_no or "" item["sourceReceptionSource"] = wo_obj.reception_source or "" else: item["sourceWorkOrderNo"] = "" item["sourceProtocolNo"] = "" item["sourceReceptionSource"] = "" else: item["sourceTaskNo"] = "" item["sourceWorkOrderNo"] = "" item["sourceProtocolNo"] = "" item["sourceReceptionSource"] = "" ``` 注意：需要确保 `_row_to_dict` 不需要修改 — 它已经返回 `sourceType` 和 `productionTaskId`，新的富化字段是在 `_enrich_product_dicts` 中动态追加的。 - [ ] **Step 2: 验证无语法错误** Run: `cd /Users/wuzilong/work/00code/zkxw/backend && source venv/bin/activate && python -c "from app.crud.cell_product import _enrich_product_dicts; print('OK')"` Expected: `OK` - [ ] **Step 3: 启动后端验证列表 API** Run: `cd /Users/wuzilong/work/00code/zkxw/backend && source venv/bin/activate && python -c " from app.core.database import SessionLocal from app.crud.cell_product import list_cell_products db = SessionLocal() try: result = list_cell_products(db, tenant_no='zkxw', page=1, page_size=3) for item in result.get('items', []): print(f\"{item.get('productCode')} | src={item.get('sourceTypeLabel')} | proto={item.get('sourceProtocolNo')} | wo={item.get('sourceWorkOrderNo')} | task={item.get('sourceTaskNo')}\") finally: db.close() "` Expected: 每行输出产品编码及来源字段（可能为空字符串） --- ### Task 4: 后端 — 新增链路 API **Files:** - Modify: `backend/app/crud/cell_product.py`（文件末尾追加） - Modify: `backend/app/api/production.py`（在 `GET /cell-products/{product_id}` 路由之后追加） - [ ] **Step 1: 在 `cell_product.py` 末尾新增链路递归查询函数** 在 `backend/app/crud/cell_product.py` 文件末尾追加： ```python # --------------------------------------------------------------------------- # 来源链路 API # --------------------------------------------------------------------------- _MAX_CHAIN_DEPTH = 10 _SOURCE_TYPE_LABELS = {"production": "生产入库", "manual": "手动录入"} def _product_node(obj: BizCellProduct, is_current: bool = False) -> Dict[str, Any]: return { "id": obj.id, "productCode": obj.product_code or "", "productName": obj.product_name or "", "nodeType": "cell_product", "productStatus": obj.product_status or "", "passage": obj.passage or "", "isCurrent": is_current, } def _build_ancestor_chain( db: Session, product: BizCellProduct, depth: int ) -> List[Dict[str, Any]]: if depth >= _MAX_CHAIN_DEPTH: return [] task_id = product.production_task_id if not task_id: return [] from app.models.models import BizProductionTask, BizProductionWorkOrder from app.models.models import BizProductionTaskInput task = ( db.query(BizProductionTask) .filter(BizProductionTask.id == task_id, BizProductionTask.del_flag == 0) .first() ) if not task: return [] wo = None edge: Dict[str, Any] = {"workOrderNo": "", "workOrderType": "", "taskNo": task.task_no or "", "taskName": task.task_name or ""} if task.work_order_id: wo = ( db.query(BizProductionWorkOrder) .filter(BizProductionWorkOrder.id == task.work_order_id, BizProductionWorkOrder.del_flag == 0) .first() ) if wo: edge["workOrderNo"] = wo.work_order_no or "" edge["workOrderType"] = wo.reception_source or "" # 检查投入物中是否有父细胞产品 parent_input = ( db.query(BizProductionTaskInput) .filter( BizProductionTaskInput.task_id == task_id, BizProductionTaskInput.input_type == "cell_product", BizProductionTaskInput.del_flag == 0, ) .first() ) if parent_input and parent_input.input_id: parent_product = ( db.query(BizCellProduct) .filter(BizCellProduct.id == parent_input.input_id, BizCellProduct.del_flag == 0) .first() ) if parent_product: return [{"node": _product_node(parent_product), "edge": edge, "parent": _build_ancestor_chain(db, parent_product, depth + 1)}] # 检查投入物中是否有样本（终止节点） sample_input = ( db.query(BizProductionTaskInput) .filter( BizProductionTaskInput.task_id == task_id, BizProductionTaskInput.input_type == "sample", BizProductionTaskInput.del_flag == 0, ) .first() ) if sample_input and sample_input.input_id: from app.models.models import BizProtocolSample sample = ( db.query(BizProtocolSample) .filter(BizProtocolSample.id == sample_input.input_id, BizProtocolSample.del_flag == 0) .first() ) if sample: sample_node = { "id": sample.id, "productCode": sample.sample_code or "", "productName": "", "nodeType": "sample", "sampleCode": sample.sample_code or "", "sampleType": sample.sample_type or "", "isCurrent": False, } return [{"node": sample_node, "edge": edge, "parent": None}] return [] def _build_descendant_chain( db: Session, product_id: str, depth: int ) -> List[Dict[str, Any]]: if depth >= _MAX_CHAIN_DEPTH: return [] from app.models.models import BizProductionTaskInput, BizProductionTaskOutput, BizProductionTask, BizProductionWorkOrder # 找到引用此产品作为投入物的所有任务 input_rows = ( db.query(BizProductionTaskInput) .filter( BizProductionTaskInput.input_type == "cell_product", BizProductionTaskInput.input_id == product_id, BizProductionTaskInput.del_flag == 0, ) .all() ) if not input_rows: return [] task_ids = list({r.task_id for r in input_rows}) tasks = ( db.query(BizProductionTask) .filter(BizProductionTask.id.in_(task_ids), BizProductionTask.del_flag == 0) .all() ) task_by_id = {t.id: t for t in tasks} wo_ids = list({t.work_order_id for t in tasks if t.work_order_id}) wo_by_id: Dict[str, Any] = {} if wo_ids: wos = ( db.query(BizProductionWorkOrder) .filter(BizProductionWorkOrder.id.in_(wo_ids), BizProductionWorkOrder.del_flag == 0) .all() ) wo_by_id = {w.id: w for w in wos} # 对每个任务，找产出 result: List[Dict[str, Any]] = [] output_rows = ( db.query(BizProductionTaskOutput) .filter( BizProductionTaskOutput.task_id.in_(task_ids), BizProductionTaskOutput.del_flag == 0, ) .all() ) child_product_ids = list({o.cell_product_id for o in output_rows if o.cell_product_id}) child_products = {} if child_product_ids: cps = ( db.query(BizCellProduct) .filter(BizCellProduct.id.in_(child_product_ids), BizCellProduct.del_flag == 0) .all() ) child_products = {cp.id: cp for cp in cps} for o in output_rows: cp = child_products.get(o.cell_product_id) if not cp: continue t = task_by_id.get(o.task_id) edge: Dict[str, Any] = {"workOrderNo": "", "workOrderType": "", "taskNo": "", "taskName": ""} if t: edge["taskNo"] = t.task_no or "" edge["taskName"] = t.task_name or "" if t.work_order_id and t.work_order_id in wo_by_id: wo_obj = wo_by_id[t.work_order_id] edge["workOrderNo"] = wo_obj.work_order_no or "" edge["workOrderType"] = wo_obj.reception_source or "" result.append({ "node": _product_node(cp), "edge": edge, "children": _build_descendant_chain(db, cp.id, depth + 1), }) return result def get_product_chain(db: Session, product_id: str, tenant_no: str) -> Optional[Dict[str, Any]]: product = get_cell_product(db, product_id, tenant_no) if not product: return None return { "current": _product_node(product, is_current=True), "ancestors": _build_ancestor_chain(db, product, 0), "descendants": _build_descendant_chain(db, product.id, 0), } ``` - [ ] **Step 2: 在 production.py 新增链路路由** 在 `backend/app/api/production.py` 的 `@router.get("/cell-products/{product_id}")` 路由之后（约 line 1979 之后）追加新路由： ```python @router.get("/cell-products/{product_id}/chain") async def get_cell_product_chain( product_id: str, user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): tenant_no = _get_tenant_no(db, user) chain = cell_product_svc.get_product_chain(db, product_id, tenant_no) if not chain: raise HTTPException(status_code=404, detail="细胞产品不存在") return success_response(chain) ``` 注意：此路由必须放在 `@router.get("/cell-products/{product_id}")` 之前或之后都可以，因为 FastAPI 按注册顺序匹配，`/cell-products/next-code` 已经在前面。但 `/chain` 是固定路径不会与 `{product_id}` 冲突 — 实际上 `/cell-products/{product_id}/chain` 和 `/cell-products/{product_id}` 是不同的路径模式，不存在冲突。 - [ ] **Step 3: 验证链路 API** Run: `cd /Users/wuzilong/work/00code/zkxw/backend && source venv/bin/activate && python -c " from app.core.database import SessionLocal from app.crud.cell_product import get_product_chain db = SessionLocal() try: from app.crud.cell_product import list_cell_products prods = list_cell_products(db, tenant_no='zkxw', page=1, page_size=1) items = prods.get('items', []) if items: pid = items[0]['id'] chain = get_product_chain(db, pid, 'zkxw') if chain: cur = chain['current'] print(f'Current: {cur[\"productCode\"]}') print(f'Ancestors: {len(chain[\"ancestors\"])} levels') print(f'Descendants: {len(chain[\"descendants\"])} branches') else: print('No chain data') else: print('No products found') finally: db.close() "` Expected: 输出当前产品的链路信息（祖先层数和子代分支数） --- ### Task 5: 前端 — API 层扩展 **Files:** - Modify: `web-admin/src/api/cell-product.ts` - [ ] **Step 1: 新增 fetchCellProductChain 函数 + fetchCellProductList 参数扩展** 将 `web-admin/src/api/cell-product.ts` 完整替换为： ```typescript import request from '@/utils/http' export function fetchCellProductList(params?: { page?: number pageSize?: number keyword?: string productStatus?: string passage?: string sourceType?: string }) { return request.get<{ list: any[] total: number page: number pageSize: number totalPages: number }>({ url: '/web/admin/cell-products', params }) } export function fetchCellProductDetail(id: string) { return request.get({ url: `/web/admin/cell-products/${id}` }) } export function fetchCreateCellProduct(data: Record) { return request.post({ url: '/web/admin/cell-products', params: data }) } export function fetchUpdateCellProduct(id: string, data: Record) { return request.put({ url: `/web/admin/cell-products/${id}`, params: data }) } export function fetchDeleteCellProduct(id: string) { return request.del({ url: `/web/admin/cell-products/${id}` }) } export function fetchNextCellProductCode(count = 1) { return request.get<{ code: string; codes?: string[] }>({ url: '/web/admin/cell-products/next-code', params: count > 1 ? { count } : undefined }) } export function fetchCellProductChain(productId: string) { return request.get({ url: `/web/admin/cell-products/${productId}/chain` }) } ``` - [ ] **Step 2: 验证无 TS 错误** Run: `cd /Users/wuzilong/work/00code/zkxw/web-admin && npx vue-tsc --noEmit --pretty 2>&1 | head -20` Expected: 无新增错误（可能有预存的） --- ### Task 6: 前端 — 列表页新增列和筛选 **Files:** - Modify: `web-admin/src/views/production/cell-product/index.vue` - [ ] **Step 1: 搜索表单新增来源类型筛选** 在 `index.vue` 的搜索表单中找到 `productStatus` 的 `el-select` 之后，新增一个筛选下拉： ```html ``` 同时在 `searchForm` 的响应式定义中增加 `sourceType: ''` 字段。 在调用 `fetchCellProductList` 时传入 `sourceType: searchForm.sourceType || undefined`。 - [ ] **Step 2: 表格新增 4 列** 在 `el-table` 中 `productStatus`（状态）列之后，新增： ```html ``` - [ ] **Step 3: 验证列表页渲染** Run: `cd /Users/wuzilong/work/00code/zkxw/web-admin && pnpm dev` 浏览器访问 `http://localhost:3006/#/production/cell-product`，确认： 1. 列表新增 4 列可见 2. 搜索栏有「来源类型」下拉 3. 选择「生产入库」筛选后列表正确过滤 --- ### Task 7: 前端 — 新增 SourceChainGraph 链路图组件 **Files:** - Create: `web-admin/src/views/production/cell-product/SourceChainGraph.vue` - [ ] **Step 1: 创建链路图组件** 新建 `web-admin/src/views/production/cell-product/SourceChainGraph.vue`，实现以下功能： - 接收 `productId` 和 `visible` props - `visible` 变为 true 时调用 `fetchCellProductChain(productId)` 获取链路数据 - 渲染横向 DAG：祖先节点 → 当前节点 → 子代节点 - 节点卡片样式区分：样本（浅蓝）、细胞产品（浅绿）、当前产品（深绿边框 + 角标） - 节点之间用 CSS border + 伪元素画水平连线和箭头 - 连线上方显示工单号和任务号标签 - 1对多分支使用竖线 + 横线分叉 - 手动录入产品（起源节点）无祖先但可能有子代，只有**既无祖先又无子代的孤立产品**才显示空状态文字「该产品暂无来源及衍生链路信息」 - 节点 hover 时用 `el-popover` 显示详情 组件核心模板结构： ```vue ``` 注意：实际实现中 `ChainNodeCard`、`ChainConnector`、`ForkBranch` 可以是同文件内的子组件或直接内联模板。核心样式参考 `docs/demo-source-chain-graph.html`。 - [ ] **Step 2: 验证组件无 TS 错误** Run: `cd /Users/wuzilong/work/00code/zkxw/web-admin && npx vue-tsc --noEmit --pretty 2>&1 | head -20` Expected: 无新增错误 --- ### Task 8: 前端 — 查看弹窗集成链路图 **Files:** - Modify: `web-admin/src/views/production/cell-product/index.vue` - [ ] **Step 1: 在查看弹窗中集成 SourceChainGraph** 在 `index.vue` 中： 1. 导入组件： ```typescript import SourceChainGraph from './SourceChainGraph.vue' ``` 2. 在 `SystemViewDialog` 的 sections 配置中新增第三个 section（在产品图片 section 之后）： ```typescript { label: '来源链路', slot: 'sourceChain', } ``` 3. 在 `SystemViewDialog` 的模板中新增 slot： ```html ``` 注意：需要确认 `SystemViewDialog` 组件是否支持 `slot` 类型的 section。如果不支持，需要查看该组件源码并添加 slot 支持（通常是在 section 渲染时判断 `slot` 字段，用 `` 渲染自定义内容）。 - [ ] **Step 2: 端到端验证** 1. 启动后端：`cd backend && source venv/bin/activate && ./start-local.sh` 2. 启动前端：`cd web-admin && pnpm dev` 3. 浏览器访问 `http://localhost:3006/#/production/cell-product` 4. 点击任意产品的「查看」按钮 5. 确认查看弹窗中有「来源链路」tab/section 6. 确认链路图正确展示节点和连线 7. 确认节点点击可显示详情 Popover 8. 确认手动录入产品显示空状态 --- ### Task 9: 提交 - [ ] **Step 1: 检查变更范围** Run: `cd /Users/wuzilong/work/00code/zkxw && git status` Expected: 仅涉及以下文件： - `backend/app/crud/cell_product.py` - `backend/app/schemas/cell_product.py` - `backend/app/api/production.py` - `web-admin/src/api/cell-product.ts` - `web-admin/src/views/production/cell-product/index.vue` - `web-admin/src/views/production/cell-product/SourceChainGraph.vue`（新建） - `docs/demo-source-chain-graph.html` - `docs/superpowers/specs/2026-05-24-cell-product-source-chain-design.md` - `docs/superpowers/plans/2026-05-24-cell-product-source-chain.md` - [ ] **Step 2: 提交** ```bash git add backend/app/crud/cell_product.py backend/app/schemas/cell_product.py backend/app/api/production.py web-admin/src/api/cell-product.ts web-admin/src/views/production/cell-product/ docs/ git commit -m "feat(cell-product): 添加细胞产品来源链路展示功能 列表新增来源类型、来源协议号、来源工单号、来源任务号四列， 支持来源类型筛选。查看弹窗新增来源链路图，展示祖先-当前-子代 双向追溯链路（横向 DAG，CSS 绘制连线），支持1对多分叉。" ```