feat: add lineage node flatten helper for cell product QC view
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
b569643581
commit
f2e582b4dc
2 files changed
+190
No files matched your search
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user