From 40d93934a99c6371f7674080ee5904c56765e839 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Wed, 29 Jul 2026 09:54:28 +0800 Subject: [PATCH] feat: add cell product lineage QC aggregate loader Co-authored-by: Cursor --- backend/app/crud/cell_product.py | 205 ++++++++++++++++++ backend/app/crud/qc_task_result.py | 25 +++ backend/tests/test_cell_product_lineage_qc.py | 67 +++++- 3 files changed, 296 insertions(+), 1 deletion(-) diff --git a/backend/app/crud/cell_product.py b/backend/app/crud/cell_product.py index 3006b4e6..45ab18da 100644 --- a/backend/app/crud/cell_product.py +++ b/backend/app/crud/cell_product.py @@ -1478,3 +1478,208 @@ def get_product_chain(db: Session, product_id: str, tenant_no: str) -> Optional[ db, product.id, product.product_code or "", 0, tenant_no ), } + + +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 "", + "taskStatusLabel": task.get("taskStatusLabel") 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 + + 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]]] = {} + + 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_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] + + pending_wo_ids: List[str] = [] + 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) + + unique_wo_ids = list({(x or "").strip() for x in pending_wo_ids if (x or "").strip()}) + tasks_by_wo: Dict[str, List[Dict[str, Any]]] = {} + for i in range(0, len(unique_wo_ids), 100): + chunk = unique_wo_ids[i : i + 100] + chunk_tasks = task_svc.list_tasks_by_work_order_ids(db, chunk, tenant_no) + for wo_id, tasks in chunk_tasks.items(): + tasks_by_wo.setdefault(wo_id, []).extend(tasks) + + 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 + + out_nodes: List[Dict[str, Any]] = [] + 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, + } diff --git a/backend/app/crud/qc_task_result.py b/backend/app/crud/qc_task_result.py index a93a36c3..102dbff8 100644 --- a/backend/app/crud/qc_task_result.py +++ b/backend/app/crud/qc_task_result.py @@ -97,6 +97,31 @@ def _merge_saved_into_template( return out +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 + + def list_results_by_task( db: Session, task_id: str, tenant_no: str, *, user_info=None ) -> List[Dict[str, Any]]: diff --git a/backend/tests/test_cell_product_lineage_qc.py b/backend/tests/test_cell_product_lineage_qc.py index cd8a883f..04c94125 100644 --- a/backend/tests/test_cell_product_lineage_qc.py +++ b/backend/tests/test_cell_product_lineage_qc.py @@ -2,7 +2,7 @@ import unittest from unittest.mock import MagicMock, patch -from app.crud.cell_product import _flatten_lineage_nodes +from app.crud.cell_product import _flatten_lineage_nodes, get_product_lineage_qc class TestFlattenLineageNodes(unittest.TestCase): @@ -130,5 +130,70 @@ class TestFlattenLineageNodes(unittest.TestCase): self.assertEqual(ids[-1], "cp-cur") +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"], + "阴性", + ) + + if __name__ == "__main__": unittest.main()