feat: add cell product lineage QC aggregate loader
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
f2e582b4dc
commit
40d93934a9
3 files changed
+296
-1
No files matched your search
@@ -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,
|
||||
}
|
||||
@@ -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]]:
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user