fix(lab-sample): 恢复 Excel 真菌等补充字段并同步质检

纠正入库编码被误软删问题，恢复可关联样本的补充字段；列表接入质检填空，并补充间充质科研捐献接收行。

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/lab_sample_processing.py

@@ -385,6 +385,196 @@ def sync_legacy_supplement_to_processing(
                 production.update_time = datetime.now()
 
 
+def _extra_field_keys(dataset: str) -> set:
+    return {f.key for f in fields_for(dataset) if f.source == SOURCE_EXTRA}
+
+
+def _live_sample_codes(db: Session, tenant_no: str) -> set:
+    codes = set()
+    for (code,) in (
+        db.query(BizProtocolSample.sample_code)
+        .filter(
+            BizProtocolSample.tenant_no == tenant_no,
+            BizProtocolSample.del_flag == 0,
+            BizProtocolSample.sample_code.isnot(None),
+            BizProtocolSample.sample_code != "",
+        )
+        .all()
+    ):
+        if code:
+            codes.add(str(code).strip())
+    for lab_code, sample_code in (
+        db.query(BizSampleReception.lab_storage_code, BizSampleReception.sample_code)
+        .filter(
+            BizSampleReception.tenant_no == tenant_no,
+            BizSampleReception.del_flag == 0,
+        )
+        .all()
+    ):
+        if lab_code:
+            codes.add(str(lab_code).strip())
+        if sample_code:
+            codes.add(str(sample_code).strip())
+    return codes
+
+
+def restore_retired_excel_extras(
+    db: Session,
+    *,
+    tenant_no: str,
+    operator_id: str = "restore-excel-extra",
+) -> Dict[str, Any]:
+    """恢复被误删的 Excel 补充字段（真菌/内毒素等），并同步旧台账。
+
+    当初按协议样本号匹配软删，但 Excel 多用实验室入库编码，导致真实样本补录被删。
+    仅恢复仍能关联到当前有效协议/接收样本编码的补充字段。
+    """
+    from app.models.models import BizLabSampleProcessingExtra
+
+    live_codes = _live_sample_codes(db, tenant_no)
+    retired = (
+        db.query(BizLabSampleProcessingExtra)
+        .filter(
+            BizLabSampleProcessingExtra.tenant_no == tenant_no,
+            BizLabSampleProcessingExtra.del_flag == 1,
+            BizLabSampleProcessingExtra.update_by == "retire-excel-demo",
+        )
+        .all()
+    )
+    restored = 0
+    skipped = 0
+    field_hits = 0
+    for row in retired:
+        code = (row.sample_code or "").strip()
+        if not code or code not in live_codes:
+            skipped += 1
+            continue
+        payload = _parse_extra(row.payload_json)
+        allowed = _extra_field_keys(row.dataset)
+        extra_only = {
+            k: v
+            for k, v in payload.items()
+            if k in allowed and not _is_blank(v)
+        }
+        if not extra_only:
+            skipped += 1
+            continue
+        upsert_extra_payload(
+            db,
+            tenant_no=tenant_no,
+            dataset=row.dataset,
+            sample_code=code,
+            payload=extra_only,
+            operator_id=operator_id,
+            replace=False,
+        )
+        sync_processing_to_legacy_supplement(
+            db,
+            tenant_no=tenant_no,
+            lab_storage_code=code,
+            sample_type_name=None,
+            payload=extra_only,
+            operator_id=operator_id,
+        )
+        restored += 1
+        field_hits += len(extra_only)
+    db.commit()
+    return {
+        "liveCodes": len(live_codes),
+        "retiredCandidates": len(retired),
+        "restoredRows": restored,
+        "skippedRows": skipped,
+        "restoredFields": field_hits,
+    }
+
+
+def _load_qc_tasks_map(
+    db: Session, tenant_no: str, codes: List[str]
+) -> Dict[str, List[Dict[str, Any]]]:
+    from app.models.models import BizQcTask, BizQcWorkOrder, BizQcWorkOrderInspectionBatch
+    from app.models.qc_param import BizQcTaskResult
+    from app.services.sample_storage_cert_qc_prefill import _stamp, batch_kind_from_name
+
+    uniq = [c for c in {(x or "").strip() for x in codes} if c]
+    if not uniq:
+        return {}
+    rows = (
+        db.query(
+            BizQcWorkOrder.sample_code,
+            BizQcTask.id,
+            BizQcTask.item_name,
+            BizQcTask.completed_time,
+            BizQcTask.update_time,
+            BizQcWorkOrderInspectionBatch.batch_name,
+            BizQcTaskResult.param_name,
+            BizQcTaskResult.param_value,
+        )
+        .join(BizQcTask, BizQcTask.work_order_id == BizQcWorkOrder.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.in_(uniq),
+            BizQcWorkOrder.del_flag == 0,
+            BizQcTask.tenant_no == tenant_no,
+            BizQcTask.del_flag == 0,
+            BizQcTask.task_status == "ywc",
+        )
+        .all()
+    )
+    by_code: Dict[str, Dict[str, Dict[str, Any]]] = {}
+    for row in rows:
+        code = (row.sample_code or "").strip()
+        if not code:
+            continue
+        grouped = by_code.setdefault(code, {})
+        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 {code: list(tasks.values()) for code, tasks in by_code.items()}
+
+
+def _qc_fields_for_dataset(dataset: str, tasks: List[Dict[str, Any]]) -> Dict[str, Any]:
+    from app.services import lab_sample_process as legacy_ledger
+
+    if not tasks:
+        return {}
+    out: Dict[str, Any] = {}
+    out.update(legacy_ledger.extract_process_tests(tasks))
+    mother = legacy_ledger.extract_mother_blood(tasks)
+    if dataset == DATASET_UBMC:
+        remap = {
+            "motherHbsag": "hbsag",
+            "motherAntiHcv": "antiHcv",
+            "motherAntiTp": "antiTp",
+            "motherAntiHiv": "antiHiv",
+            "motherCmvIgm": "cmvIgm",
+        }
+        for key, value in mother.items():
+            out[remap.get(key, key)] = value
+    else:
+        out.update(mother)
+    allowed = _extra_field_keys(dataset)
+    return {k: v for k, v in out.items() if k in allowed and not _is_blank(v)}
+
+
 def sync_processing_to_legacy_supplement(
     db: Session,
     *,
@@ -624,7 +814,28 @@ def list_rows(
         reception = receptions_by_sample_id.get(sample.id) or receptions_by_code.get(sample.sample_code or "")
         if reception and reception.lab_storage_code:
             legacy_codes.append(reception.lab_storage_code)
+        if reception and reception.sample_code:
+            legacy_codes.append(reception.sample_code)
     legacy_supplements = _load_legacy_supplements(db, tenant_no, legacy_codes)
+    qc_tasks_map = _load_qc_tasks_map(db, tenant_no, legacy_codes)
+
+    def _compose_extra(*codes: Optional[str]) -> Dict[str, Any]:
+        picked = _pick_extra(extras, *codes)
+        legacy = {}
+        for c in codes:
+            key = (c or "").strip()
+            if key and key in legacy_supplements:
+                legacy = legacy_supplements[key]
+                break
+        merged = _merge_extra_with_legacy(picked, legacy)
+        for c in codes:
+            key = (c or "").strip()
+            if not key:
+                continue
+            for qk, qv in _qc_fields_for_dataset(dataset, qc_tasks_map.get(key) or []).items():
+                if qk not in merged or _is_blank(merged.get(qk)):
+                    merged[qk] = qv
+        return merged
 
     built_map: Dict[str, Dict[str, Any]] = {}
     for sample, stype, party, coll, hospital in filtered:
@@ -632,10 +843,7 @@ def list_rows(
         reception = receptions_by_sample_id.get(sample.id) or receptions_by_code.get(code)
         production = productions.get(reception.id) if reception else None
         lab_code = (reception.lab_storage_code if reception else None) or ""
-        extra = _merge_extra_with_legacy(
-            _pick_extra(extras, lab_code, code),
-            legacy_supplements.get(lab_code) or legacy_supplements.get(code or ""),
-        )
+        extra = _compose_extra(lab_code, code, reception.sample_code if reception else None)
         protocol = protocols.get(sample.collection_protocol_id) if sample.collection_protocol_id else None
         check_flags = check_flags_by_reception.get(reception.id, {}) if reception else {}
         row = _build_row(
@@ -711,26 +919,20 @@ def list_rows(
         )
         check_flags_by_reception.update(ubmc_flags)
 
-        ubmc_legacy = _load_legacy_supplements(
-            db,
-            tenant_no,
-            [
-                (r.lab_storage_code or r.sample_code or "")
-                for r in ubmc_recs
-            ],
-        )
+        ubmc_legacy_codes = [
+            (r.lab_storage_code or r.sample_code or "")
+            for r in ubmc_recs
+        ]
+        ubmc_legacy = _load_legacy_supplements(db, tenant_no, ubmc_legacy_codes)
         legacy_supplements.update(ubmc_legacy)
+        qc_tasks_map.update(_load_qc_tasks_map(db, tenant_no, ubmc_legacy_codes))
 
         for reception in ubmc_recs:
             code = (reception.lab_storage_code or reception.sample_code or "").strip()
             if not code or code in built_map:
                 continue
             production = productions.get(reception.id)
-            extra = _merge_extra_with_legacy(
-                _pick_extra(extras, code, reception.sample_code),
-                legacy_supplements.get(code)
-                or legacy_supplements.get((reception.sample_code or "").strip()),
-            )
+            extra = _compose_extra(code, reception.sample_code)
             mother = mother_by_donation.get(str(reception.research_donation_id or ""))
             fake_sample = SimpleNamespace(
                 id=reception.donation_sample_id or reception.id,
@@ -752,6 +954,95 @@ def list_rows(
                 check_flags=check_flags_by_reception.get(reception.id, {}),
             )
 
+    # 间充质：科研捐献接收的 UC*/BC* 入库编码也需展示（无协议样本时）
+    if dataset == DATASET_MSC:
+        msc_recs = (
+            db.query(BizSampleReception)
+            .filter(
+                BizSampleReception.tenant_no == tenant_no,
+                BizSampleReception.del_flag == 0,
+                or_(
+                    BizSampleReception.lab_storage_code.like("UC%"),
+                    BizSampleReception.lab_storage_code.like("BC%"),
+                ),
+            )
+            .order_by(BizSampleReception.reception_time.desc())
+            .all()
+        )
+        mother_by_donation: Dict[str, str] = {}
+        donation_ids = sorted(
+            {r.research_donation_id for r in msc_recs if r.research_donation_id}
+        )
+        if donation_ids:
+            rows = db.execute(
+                text(
+                    "SELECT research_donation_id, mother_name FROM biz_research_donation_party "
+                    "WHERE del_flag=0 AND tenant_no=:tenant_no AND research_donation_id IN ("
+                    + ",".join(f":d{i}" for i in range(len(donation_ids)))
+                    + ")"
+                ),
+                {
+                    "tenant_no": tenant_no,
+                    **{f"d{i}": did for i, did in enumerate(donation_ids)},
+                },
+            ).fetchall()
+            for did, mother in rows:
+                if did and mother and did not in mother_by_donation:
+                    mother_by_donation[str(did)] = mother
+
+        missing_rids = [r.id for r in msc_recs if r.id and r.id not in productions]
+        if missing_rids:
+            for p in (
+                db.query(BizSampleProductionReception)
+                .filter(
+                    BizSampleProductionReception.tenant_no == tenant_no,
+                    BizSampleProductionReception.del_flag == 0,
+                    BizSampleProductionReception.reception_id.in_(missing_rids),
+                )
+                .order_by(BizSampleProductionReception.process_time.desc())
+                .all()
+            ):
+                productions.setdefault(p.reception_id, p)
+
+        msc_flags = _check_flag_map(
+            db,
+            tenant_no,
+            [r.id for r in msc_recs if r.id and r.id not in check_flags_by_reception],
+        )
+        check_flags_by_reception.update(msc_flags)
+        msc_legacy_codes = [
+            (r.lab_storage_code or r.sample_code or "") for r in msc_recs
+        ]
+        legacy_supplements.update(_load_legacy_supplements(db, tenant_no, msc_legacy_codes))
+        qc_tasks_map.update(_load_qc_tasks_map(db, tenant_no, msc_legacy_codes))
+
+        for reception in msc_recs:
+            code = (reception.lab_storage_code or reception.sample_code or "").strip()
+            if not code or code in built_map:
+                continue
+            production = productions.get(reception.id)
+            extra = _compose_extra(code, reception.sample_code)
+            mother = mother_by_donation.get(str(reception.research_donation_id or ""))
+            fake_sample = SimpleNamespace(
+                id=reception.donation_sample_id or reception.id,
+                sample_code=reception.sample_code or code,
+                collection_protocol_id=None,
+            )
+            fake_party = SimpleNamespace(mother_name=mother) if mother else None
+            built_map[code] = _build_row(
+                dataset=dataset,
+                sample=fake_sample,
+                type_name="脐带间充质干细胞",
+                party=fake_party,
+                coll=None,
+                hospital=None,
+                reception=reception,
+                production=production,
+                extra=extra,
+                protocol=None,
+                check_flags=check_flags_by_reception.get(reception.id, {}),
+            )
+
     kw = (keyword or "").strip()
     built = []
     for row in built_map.values():

backend/scripts/restore_lab_sample_processing_excel_extras.py

@@ -0,0 +1,30 @@
+"""恢复被误软删的 Excel 补充字段（真菌/内毒素/CD34 等）并同步旧台账。
+
+用法:
+  ENV_FILE=.../backend/.env python backend/scripts/restore_lab_sample_processing_excel_extras.py
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+BACKEND = ROOT / "backend"
+if str(BACKEND) not in sys.path:
+    sys.path.insert(0, str(BACKEND))
+
+from app.core.database import SessionLocal  # noqa: E402
+from app.services.lab_sample_processing import restore_retired_excel_extras  # noqa: E402
+
+
+def main() -> None:
+    db = SessionLocal()
+    try:
+        result = restore_retired_excel_extras(db, tenant_no="zkxw")
+        print(result)
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    main()

backend/tests/test_lab_sample_processing.py

@@ -133,6 +133,15 @@ class LabSampleProcessingFieldsTest(unittest.TestCase):
         self.assertEqual(_infectious_code("正常"), "zc")
         self.assertEqual(_infectious_code("不正常"), "bzc")
 
+    def test_extra_field_keys_include_fungus(self):
+        from app.services.lab_sample_processing import _extra_field_keys
+
+        keys = _extra_field_keys(DATASET_MSC)
+        self.assertIn("fungusTest", keys)
+        self.assertIn("endotoxinTest", keys)
+        self.assertIn("bacteriaTest", keys)
+        self.assertIn("mycoplasmaTest", keys)
+
     def test_legacy_supplement_fills_empty_source_fields(self):
         from app.services.lab_sample_processing import _build_row, _merge_extra_with_legacy
         from types import SimpleNamespace

docs/sql/20261112_lab_sample_processing_restore_excel_extras.md

@@ -0,0 +1,23 @@
+# 恢复 Excel 补充字段（真菌等）
+
+## 原因
+
+`20261112_lab_sample_processing_retire_excel_demo.sql` 按「补充表 sample_code = 协议样本号」软删演示行。
+会议 Excel 实际多用实验室入库编码（`B000*` / `BC*` / `UBMC*`），与协议号不一致，导致真实样本的真菌/内毒素/CD34 等补充字段被误删。
+
+## 处理
+
+运行：
+
+```bash
+ENV_FILE=backend/.env python backend/scripts/restore_lab_sample_processing_excel_extras.py
+```
+
+脚本会：
+
+1. 找出 `update_by=retire-excel-demo` 且 `del_flag=1` 的补充行
+2. 仅恢复仍能关联到当前有效协议/接收编码的行
+3. 只写回 `SOURCE_EXTRA` 字段到有效补充表
+4. 同步旧台账 `biz_lab_sample_process_supplement`
+
+列表侧同时会从质检完成任务填空（内毒素/真菌/支原体/细菌及母血相关）。