diff --git a/backend/app/crud/cell_product.py b/backend/app/crud/cell_product.py index bb5a3d52..3006b4e6 100644 --- a/backend/app/crud/cell_product.py +++ b/backend/app/crud/cell_product.py @@ -1410,6 +1410,62 @@ def _build_descendant_chain( return result +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 + + 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: diff --git a/backend/tests/test_cell_product_lineage_qc.py b/backend/tests/test_cell_product_lineage_qc.py new file mode 100644 index 00000000..cd8a883f --- /dev/null +++ b/backend/tests/test_cell_product_lineage_qc.py @@ -0,0 +1,134 @@ +"""细胞产品来源链质检 — 节点展平与编排""" +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()