"""实验室样本处理台账：字段合并、质检参数匹配、中文时间格式。""" from __future__ import annotations import json from datetime import datetime from typing import Any, Dict, Iterable, List, Optional, Set from sqlalchemy.orm import Session from app.models.models import ( BizQcTask, BizQcWorkOrder, BizQcWorkOrderInspectionBatch, ) from app.models.qc_param import BizQcTaskResult from app.services.sample_storage_cert_qc_prefill import ( _MOTHER_SEROLOGY_RULES, _process_qual_from_tasks, _serology_from_tasks, _stamp, batch_kind_from_name, normalize_negative_positive, qual_yin_yang, ) SAMPLE_TYPE_PERINATAL_HSC = "围产期造血干细胞" SAMPLE_TYPE_UC_MSC = "脐带间充质干细胞" TAB_KEYS = ( "cell_hsc", "donation_hsc", "cell_msc", "donation_msc", ) TAB_META = { "cell_hsc": { "source": "cell_storage", "sampleTypeName": SAMPLE_TYPE_PERINATAL_HSC, "label": "细胞存储·围产期造血干细胞", }, "donation_hsc": { "source": "research_donation", "sampleTypeName": SAMPLE_TYPE_PERINATAL_HSC, "label": "科研捐赠·围产期造血干细胞", }, "cell_msc": { "source": "cell_storage", "sampleTypeName": SAMPLE_TYPE_UC_MSC, "label": "细胞存储·脐带间充质干细胞", }, "donation_msc": { "source": "research_donation", "sampleTypeName": SAMPLE_TYPE_UC_MSC, "label": "科研捐赠·脐带间充质干细胞", }, } # 各 Tab 可补录字段（系统空或口径需覆盖时写入） SUPPLEMENT_KEYS: Dict[str, List[str]] = { "cell_hsc": [ "receiveCheckText", "hasMaternalBlood", "remeasureMaternalBlood", "transportPackingList", "cordBloodReceiveConfirm", "infectiousFourExternal", "preEnrichNucleatedCount", "postEnrichNucleatedCount", "cd34Count1e4", "remarkProcess", ], "donation_hsc": [ "labMaterialCode", "storageStatus", "discardReason", "receiveCheckText", "preOpNucleatedCount", "infectiousFourTestLabel", "motherHbsag", "motherAntiHcv", "motherAntiTp", "motherAntiHiv", "motherCmvIgm", "remarkProcess", ], "cell_msc": [ "storageStatus", "endotoxinTest", "bacteriaTest", "fungusTest", "mycoplasmaTest", "motherHbsag", "motherHbvDna", "motherAntiHcv", "motherHcvRna", "motherAntiTp", "motherCmvIgm", "motherAntiHiv", "motherHiv1Rna", "storageDateP2", "remark", ], "donation_msc": [ "storageStatus", "storageTotalTubes", "availableCellCount1e7", "endotoxinTest", "bacteriaTest", "fungusTest", "mycoplasmaTest", "motherHbsag", "motherHbvDna", "motherAntiHcv", "motherHcvRna", "motherAntiTp", "motherCmvIgm", "motherAntiHiv", "motherHiv1Rna", "storageDateP2", "remark", ], } # 补录优先覆盖系统推导值的字段 SUPPLEMENT_OVERRIDE_KEYS: Dict[str, Set[str]] = { "cell_hsc": {"receiveCheckText"}, "donation_hsc": { "receiveCheckText", "storageStatus", "infectiousFourTestLabel", "motherHbsag", "motherAntiHcv", "motherAntiTp", "motherAntiHiv", "motherCmvIgm", }, "cell_msc": { "storageStatus", "endotoxinTest", "bacteriaTest", "fungusTest", "mycoplasmaTest", "motherHbsag", "motherHbvDna", "motherAntiHcv", "motherHcvRna", "motherAntiTp", "motherCmvIgm", "motherAntiHiv", "motherHiv1Rna", "storageDateP2", }, "donation_msc": { "storageStatus", "storageTotalTubes", "availableCellCount1e7", "endotoxinTest", "bacteriaTest", "fungusTest", "mycoplasmaTest", "motherHbsag", "motherHbvDna", "motherAntiHcv", "motherHcvRna", "motherAntiTp", "motherCmvIgm", "motherAntiHiv", "motherHiv1Rna", "storageDateP2", }, } def format_cn_datetime(value: Any) -> str: """统一展示：2026年11月12日 00时00分00秒""" if value is None or value == "": return "" if isinstance(value, datetime): dt = value else: text = str(value).strip().replace("T", " ") if not text: return "" dt = None for candidate, fmt in ( (text[:19], "%Y-%m-%d %H:%M:%S"), (text[:16], "%Y-%m-%d %H:%M"), (text[:10], "%Y-%m-%d"), ): try: dt = datetime.strptime(candidate, fmt) break except ValueError: continue if dt is None: return text return ( f"{dt.year:04d}年{dt.month:02d}月{dt.day:02d}日 " f"{dt.hour:02d}时{dt.minute:02d}分{dt.second:02d}秒" ) def parse_payload(raw: Any) -> Dict[str, Any]: if isinstance(raw, dict): return dict(raw) if not raw: return {} try: data = json.loads(raw) except (TypeError, ValueError, json.JSONDecodeError): return {} return data if isinstance(data, dict) else {} def dumps_payload(payload: Dict[str, Any]) -> str: return json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":")) def yes_no_from_urls(urls: Optional[str]) -> str: return "是" if (urls or "").strip() else "否" def yes_no_from_bool(flag: Optional[bool]) -> str: if flag is None: return "" return "是" if flag else "否" def merge_row( *, tab_key: str, system_values: Dict[str, Any], supplement: Optional[Dict[str, Any]], ) -> Dict[str, Any]: """系统值优先；override 键补录优先；空用补录填。""" out = dict(system_values or {}) sup = supplement or {} overrides = SUPPLEMENT_OVERRIDE_KEYS.get(tab_key, set()) for key, val in sup.items(): text = "" if val is None else str(val).strip() if not text: continue cur = out.get(key) cur_empty = cur is None or (isinstance(cur, str) and not cur.strip()) if key in overrides or cur_empty: out[key] = text return out def load_qc_tasks_for_sample(db: Session, tenant_no: str, sample_code: str) -> List[Dict[str, Any]]: code = (sample_code or "").strip() if not code: return [] rows = ( db.query( BizQcTask.id, BizQcTask.item_name, BizQcTask.completed_time, BizQcTask.update_time, BizQcWorkOrderInspectionBatch.batch_name, BizQcTaskResult.param_name, BizQcTaskResult.param_value, ) .join(BizQcWorkOrder, BizQcWorkOrder.id == BizQcTask.work_order_id) .outerjoin( BizQcWorkOrderInspectionBatch, BizQcWorkOrderInspectionBatch.id == BizQcTask.batch_id, ) .outerjoin( BizQcTaskResult, (BizQcTaskResult.task_id == BizQcTask.id) & (BizQcTaskResult.del_flag == 0), ) .filter( BizQcWorkOrder.tenant_no == tenant_no, BizQcWorkOrder.sample_code == code, BizQcWorkOrder.del_flag == 0, BizQcTask.tenant_no == tenant_no, BizQcTask.del_flag == 0, BizQcTask.task_status == "ywc", ) .all() ) grouped: Dict[str, Dict[str, Any]] = {} for row in rows: task = grouped.setdefault( row.id, { "item_name": row.item_name or "", "batch_kind": batch_kind_from_name(row.batch_name), "stamp": _stamp(row.completed_time, row.update_time), "params": [], "values": [], }, ) if (row.param_name or "").strip() or (row.param_value or "").strip(): task["params"].append((row.param_name or "", row.param_value or "")) if (row.param_value or "").strip(): task["values"].append(row.param_value or "") return list(grouped.values()) def extract_mother_blood(tasks: List[Dict[str, Any]], *, include_nucleic: bool = True) -> Dict[str, str]: rules = _MOTHER_SEROLOGY_RULES if not include_nucleic: rules = tuple( r for r in rules if r[0] not in {"mother_hbv_dna", "mother_hcv_rna", "mother_hiv1_rna"} ) raw = _serology_from_tasks(tasks, rules) mapping = { "mother_hbsag": "motherHbsag", "mother_hbv_dna": "motherHbvDna", "mother_anti_hcv": "motherAntiHcv", "mother_hcv_rna": "motherHcvRna", "mother_anti_tp": "motherAntiTp", "mother_cmv_igm": "motherCmvIgm", "mother_anti_hiv": "motherAntiHiv", "mother_hiv1_rna": "motherHiv1Rna", } out: Dict[str, str] = {} for src, dst in mapping.items(): val = normalize_negative_positive(raw.get(src) or "") or (raw.get(src) or "") if val: # 台账 Excel 多用「阴」 if val in ("阴性", "阴"): out[dst] = "阴" elif val in ("阳性", "阳"): out[dst] = "阳" else: out[dst] = val return out def extract_process_tests(tasks: List[Dict[str, Any]]) -> Dict[str, str]: raw = _process_qual_from_tasks(tasks) bacteria_values: List[str] = [] for task in tasks: name = task.get("item_name") or "" params = task.get("params") or [] hit = "细菌" in name or any("细菌" in (pn or "") for pn, _ in params) if not hit: continue for _, pv in params: bacteria_values.append(pv) bacteria_values.extend(task.get("values") or []) bacteria = qual_yin_yang(bacteria_values) out = { "endotoxinTest": _to_yin(raw.get("endotoxin_test")), "fungusTest": _to_yin(raw.get("fungus_test")), "mycoplasmaTest": _to_yin(raw.get("mycoplasma_test")), } if bacteria: out["bacteriaTest"] = _to_yin(bacteria) return {k: v for k, v in out.items() if v} def _to_yin(value: Optional[str]) -> str: text = (value or "").strip() if text in ("阴性", "阴"): return "阴" if text in ("阳性", "阳"): return "阳" return text def infectious_label(code: Optional[str]) -> str: c = (code or "").strip() if c == "zc": return "正常" if c == "bzc": return "不正常" return c