fix(lab-sample): 新旧实验室样本台账补录双向联动

新模块读取旧台账补录并映射字段；旧台账保存时同步到新补充表与生产传染四项，新模块保存时回写旧补录。

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

backend/app/crud/lab_sample_process.py

@@ -179,6 +179,17 @@ def upsert_supplement(
             del_flag=0,
         )
         db.add(row)
+    # 同步到新模块补充表/生产接收，保证实验室样本处理数据联动
+    from app.services.lab_sample_processing import sync_legacy_supplement_to_processing
+
+    sync_legacy_supplement_to_processing(
+        db,
+        tenant_no=tenant_no,
+        lab_storage_code=code,
+        sample_type_name=type_name,
+        payload=clean_payload,
+        operator_id=user_id,
+    )
     db.commit()
     db.refresh(row)
     return get_supplement(db, tenant_no=tenant_no, lab_storage_code=code) or {}

backend/app/services/lab_sample_processing.py

@@ -1,4 +1,8 @@
-"""实验室样本处理数据 — 聚合读取、Excel 导入与部门字段回写。"""
+"""实验室样本处理数据 — 聚合市场/生产/质检已发生业务数据与部门字段回写。
+
+Excel 仅作为字段目录示例；列表以协议样本为主表，关联样本接收与生产接收，
+补充表只承载源系统没有的列，且不得覆盖源表字段。
+"""
 from __future__ import annotations
 
 import json
@@ -7,17 +11,21 @@ from datetime import datetime
 from decimal import Decimal
 from typing import Any, Dict, List, Optional, Tuple
 
-from sqlalchemy import or_
+from types import SimpleNamespace
+
+from sqlalchemy import or_, text
 from sqlalchemy.orm import Session
 
 from app.models.models import (
     BizCellSampleType,
     BizCollectionHospital,
+    BizCollectionProtocol,
     BizProtocolCollectionInfo,
     BizProtocolPartyCustomer,
     BizProtocolSample,
     BizSampleProductionReception,
     BizSampleReception,
+    BizSampleReceptionCheckResult,
 )
 from app.services.lab_sample_processing_fields import (
     DATASET_HSC,
@@ -106,10 +114,332 @@ def _code_prefix_filters(dataset: str) -> List[str]:
 def _match_keyword(row: Dict[str, Any], kw: str) -> bool:
     if not kw:
         return True
-    text = f"{row.get('sampleCode') or ''} {row.get('motherName') or ''}"
+    text = (
+        f"{row.get('sampleCode') or ''} "
+        f"{row.get('protocolSampleCode') or ''} "
+        f"{row.get('motherName') or ''}"
+    )
     return kw.lower() in text.lower()
 
 
+def _infectious_label(code: Optional[str]) -> Optional[str]:
+    c = ("" if code is None else str(code)).strip()
+    if not c:
+        return None
+    if c == "zc":
+        return "正常"
+    if c == "bzc":
+        return "不正常"
+    return c
+
+
+def _infectious_code(label: Optional[Any]) -> Optional[str]:
+    if label is None:
+        return None
+    s = str(label).strip()
+    if not s:
+        return None
+    if s == "正常":
+        return "zc"
+    if s == "不正常":
+        return "bzc"
+    return s
+
+
+def _yes_no_from_bool(flag: Optional[bool]) -> Optional[str]:
+    if flag is None:
+        return None
+    return "是" if flag else "否"
+
+
+def _derive_receive_check(flags: Dict[str, bool]) -> Optional[str]:
+    if not flags:
+        return None
+    skip = {"知情同意书", "健康调查表", "样本信息正确，标识清晰"}
+    preferred = [
+        "针路密闭",
+        "容器无破损",
+        "外观正常",
+        "样本无破损",
+        "针卡关紧",
+        "两端系紧",
+        "温度4-25",
+        "采集至到库时间≤24小时",
+    ]
+    parts = [name for name in preferred if flags.get(name)]
+    if not parts:
+        parts = [name for name, checked in flags.items() if checked and name not in skip]
+    return "、".join(parts) if parts else None
+
+
+def _check_flag_map(
+    db: Session, tenant_no: str, reception_ids: List[str]
+) -> Dict[str, Dict[str, bool]]:
+    ids = [i for i in reception_ids if i]
+    if not ids:
+        return {}
+    rows = (
+        db.query(BizSampleReceptionCheckResult)
+        .filter(
+            BizSampleReceptionCheckResult.tenant_no == tenant_no,
+            BizSampleReceptionCheckResult.del_flag == 0,
+            BizSampleReceptionCheckResult.reception_id.in_(ids),
+        )
+        .all()
+    )
+    out: Dict[str, Dict[str, bool]] = {}
+    for r in rows:
+        bucket = out.setdefault(r.reception_id, {})
+        name = (r.item_name or "").strip()
+        if name:
+            bucket[name] = bool(r.checked)
+    return out
+
+
+def _pick_extra(
+    extras: Dict[str, Any], *codes: Optional[str]
+) -> Dict[str, Any]:
+    for code in codes:
+        key = (code or "").strip()
+        if key and key in extras:
+            return _parse_extra(extras[key].payload_json)
+    return {}
+
+
+# 旧台账补录字段名 → 新模块字段名（同名键保持不变）
+_LEGACY_SUPPLEMENT_TO_NEW: Dict[str, str] = {
+    "receiveCheckText": "receiveCheck",
+    "infectiousFourTestLabel": "infectiousFourCard",
+    "preOpNucleatedCount": "preOpTnc",
+    "preEnrichNucleatedCount": "preEnrichTnc",
+    "postEnrichNucleatedCount": "postEnrichTnc",
+    "cd34Count1e4": "cd34Count",
+    "transportPackingList": "hasPackingList",
+    "cordBloodReceiveConfirm": "hasReceiveConfirm",
+    "remarkProcess": "processRemark",
+    "motherHbsag": "hbsag",
+    "motherAntiHcv": "antiHcv",
+    "motherAntiTp": "antiTp",
+    "motherAntiHiv": "antiHiv",
+    "motherCmvIgm": "cmvIgm",
+}
+
+_NEW_TO_LEGACY_SUPPLEMENT: Dict[str, str] = {
+    v: k for k, v in _LEGACY_SUPPLEMENT_TO_NEW.items()
+}
+# MSC 等仍使用 mother* 键名，双向都保留
+for _same in (
+    "labMaterialCode",
+    "storageStatus",
+    "discardReason",
+    "hasMaternalBlood",
+    "remeasureMaternalBlood",
+    "infectiousFourExternal",
+    "endotoxinTest",
+    "bacteriaTest",
+    "fungusTest",
+    "mycoplasmaTest",
+    "motherHbsag",
+    "motherHbvDna",
+    "motherAntiHcv",
+    "motherHcvRna",
+    "motherAntiTp",
+    "motherCmvIgm",
+    "motherAntiHiv",
+    "motherHiv1Rna",
+    "storageDateP2",
+    "storageTotalTubes",
+    "availableCellCount1e7",
+):
+    _LEGACY_SUPPLEMENT_TO_NEW.setdefault(_same, _same)
+    _NEW_TO_LEGACY_SUPPLEMENT.setdefault(_same, _same)
+
+
+def _is_blank(val: Any) -> bool:
+    return val is None or (isinstance(val, str) and not val.strip())
+
+
+def _map_legacy_supplement(payload: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    out: Dict[str, Any] = {}
+    for key, value in (payload or {}).items():
+        if _is_blank(value):
+            continue
+        new_key = _LEGACY_SUPPLEMENT_TO_NEW.get(str(key), str(key))
+        out[new_key] = value
+        # MSC 场景保留 mother* 原键，避免被 UBMC 别名覆盖后丢失
+        if str(key).startswith("mother") and new_key != str(key):
+            out[str(key)] = value
+    return out
+
+
+def _map_new_to_legacy_supplement(payload: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    out: Dict[str, Any] = {}
+    for key, value in (payload or {}).items():
+        if _is_blank(value):
+            continue
+        legacy_key = _NEW_TO_LEGACY_SUPPLEMENT.get(str(key), str(key))
+        out[legacy_key] = value
+    return out
+
+
+def _merge_extra_with_legacy(
+    extra: Optional[Dict[str, Any]], legacy: Optional[Dict[str, Any]]
+) -> Dict[str, Any]:
+    """新补充表优先；旧台账补录填空（含字段名映射）。"""
+    mapped = _map_legacy_supplement(legacy)
+    merged = dict(mapped)
+    for key, value in (extra or {}).items():
+        if not _is_blank(value):
+            merged[key] = value
+    return merged
+
+
+def _load_legacy_supplements(
+    db: Session, tenant_no: str, codes: List[str]
+) -> Dict[str, Dict[str, Any]]:
+    from app.models.models import BizLabSampleProcessSupplement
+    from app.services import lab_sample_process as legacy_ledger
+
+    uniq = [c for c in {(x or "").strip() for x in codes} if c]
+    if not uniq:
+        return {}
+    rows = (
+        db.query(BizLabSampleProcessSupplement)
+        .filter(
+            BizLabSampleProcessSupplement.tenant_no == tenant_no,
+            BizLabSampleProcessSupplement.del_flag == 0,
+            BizLabSampleProcessSupplement.lab_storage_code.in_(uniq),
+        )
+        .all()
+    )
+    out: Dict[str, Dict[str, Any]] = {}
+    for row in rows:
+        out[row.lab_storage_code] = legacy_ledger.parse_payload(row.payload)
+    return out
+
+
+def sync_legacy_supplement_to_processing(
+    db: Session,
+    *,
+    tenant_no: str,
+    lab_storage_code: str,
+    sample_type_name: Optional[str] = None,
+    payload: Optional[Dict[str, Any]] = None,
+    operator_id: str = "legacy-sync",
+) -> None:
+    """旧台账补录保存后，同步到新模块补充表，并把传染四项写回生产接收。"""
+    code = (lab_storage_code or "").strip()
+    if not code:
+        return
+    mapped = _map_legacy_supplement(payload)
+    if not mapped:
+        return
+    dataset = resolve_dataset_by_type_name(sample_type_name or "", code)
+    if not dataset:
+        if code.upper().startswith("UBMC"):
+            dataset = DATASET_UBMC
+        elif code.upper().startswith(("BC", "UC")):
+            dataset = DATASET_MSC
+        else:
+            dataset = DATASET_HSC
+
+    upsert_extra_payload(
+        db,
+        tenant_no=tenant_no,
+        dataset=dataset,
+        sample_code=code,
+        payload=mapped,
+        operator_id=operator_id,
+        replace=False,
+    )
+
+    infectious = mapped.get("infectiousFourCard")
+    if infectious not in (None, ""):
+        reception = (
+            db.query(BizSampleReception)
+            .filter(
+                BizSampleReception.tenant_no == tenant_no,
+                BizSampleReception.del_flag == 0,
+                or_(
+                    BizSampleReception.lab_storage_code == code,
+                    BizSampleReception.sample_code == code,
+                ),
+            )
+            .order_by(BizSampleReception.reception_time.desc())
+            .first()
+        )
+        if reception:
+            production = (
+                db.query(BizSampleProductionReception)
+                .filter(
+                    BizSampleProductionReception.tenant_no == tenant_no,
+                    BizSampleProductionReception.reception_id == reception.id,
+                    BizSampleProductionReception.del_flag == 0,
+                )
+                .order_by(BizSampleProductionReception.process_time.desc())
+                .first()
+            )
+            if production is not None:
+                production.infectious_four_test_result = _infectious_code(infectious)
+                production.update_by = operator_id
+                production.update_time = datetime.now()
+
+
+def sync_processing_to_legacy_supplement(
+    db: Session,
+    *,
+    tenant_no: str,
+    lab_storage_code: str,
+    sample_type_name: Optional[str] = None,
+    payload: Optional[Dict[str, Any]] = None,
+    operator_id: str = "processing-sync",
+) -> None:
+    """新模块保存后，回写旧台账补录，保持两边可见字段一致。"""
+    from app.models.models import BizLabSampleProcessSupplement
+    from app.services import lab_sample_process as legacy_ledger
+
+    code = (lab_storage_code or "").strip()
+    mapped = _map_new_to_legacy_supplement(payload)
+    if not code or not mapped:
+        return
+    now = datetime.now()
+    row = (
+        db.query(BizLabSampleProcessSupplement)
+        .filter(
+            BizLabSampleProcessSupplement.tenant_no == tenant_no,
+            BizLabSampleProcessSupplement.lab_storage_code == code,
+            BizLabSampleProcessSupplement.del_flag == 0,
+        )
+        .first()
+    )
+    current = legacy_ledger.parse_payload(row.payload) if row else {}
+    current.update(mapped)
+    src = "research_donation" if code.upper().startswith("UBMC") else "cell_storage"
+    if row is None:
+        db.add(
+            BizLabSampleProcessSupplement(
+                id=_new_id(),
+                lab_storage_code=code,
+                reception_source=src,
+                sample_type_name=(sample_type_name or None),
+                payload=legacy_ledger.dumps_payload(current),
+                tenant_no=tenant_no,
+                create_by=operator_id,
+                create_dept_id="",
+                create_time=now,
+                update_by=operator_id,
+                update_time=now,
+                del_flag=0,
+            )
+        )
+    else:
+        if sample_type_name:
+            row.sample_type_name = sample_type_name
+        row.payload = legacy_ledger.dumps_payload(current)
+        row.update_by = operator_id
+        row.update_time = now
+
+
 def _match_date_range(row: Dict[str, Any], date_from: Optional[str], date_to: Optional[str]) -> bool:
     if not date_from and not date_to:
         return True
@@ -215,23 +545,39 @@ def list_rows(
         .all()
     }
 
-    codes = list({*(s.sample_code for s, *_ in filtered if s.sample_code), *extras.keys()})
-    receptions: Dict[str, BizSampleReception] = {}
+    # 仅聚合真实协议样本；补充表只挂到已有样本上，不再单独展示 Excel 演示行
+    codes = [s.sample_code for s, *_ in filtered if s.sample_code]
+    sample_ids = [s.id for s, *_ in filtered if s.id]
+    receptions_by_sample_id: Dict[str, BizSampleReception] = {}
+    receptions_by_code: Dict[str, BizSampleReception] = {}
     productions: Dict[str, BizSampleProductionReception] = {}
-    if codes:
+    if sample_ids or codes:
+        recv_filters = []
+        if sample_ids:
+            recv_filters.append(BizSampleReception.protocol_sample_id.in_(sample_ids))
+        if codes:
+            recv_filters.append(BizSampleReception.sample_code.in_(codes))
         for r in (
             db.query(BizSampleReception)
             .filter(
                 BizSampleReception.tenant_no == tenant_no,
                 BizSampleReception.del_flag == 0,
-                BizSampleReception.sample_code.in_(codes),
+                or_(*recv_filters) if recv_filters else False,
             )
             .order_by(BizSampleReception.reception_time.desc())
             .all()
         ):
-            receptions.setdefault(r.sample_code, r)
-        if receptions:
-            rids = [r.id for r in receptions.values()]
+            if r.protocol_sample_id:
+                receptions_by_sample_id.setdefault(r.protocol_sample_id, r)
+            if r.sample_code:
+                receptions_by_code.setdefault(r.sample_code, r)
+        if receptions_by_sample_id or receptions_by_code:
+            rids = list(
+                {
+                    *(r.id for r in receptions_by_sample_id.values()),
+                    *(r.id for r in receptions_by_code.values()),
+                }
+            )
             for p in (
                 db.query(BizSampleProductionReception)
                 .filter(
@@ -244,13 +590,55 @@ def list_rows(
             ):
                 productions.setdefault(p.reception_id, p)
 
+    protocol_ids = sorted(
+        {
+            s.collection_protocol_id
+            for s, *_ in filtered
+            if getattr(s, "collection_protocol_id", None)
+        }
+    )
+    protocols: Dict[str, BizCollectionProtocol] = {}
+    if protocol_ids:
+        for p in (
+            db.query(BizCollectionProtocol)
+            .filter(
+                BizCollectionProtocol.tenant_no == tenant_no,
+                BizCollectionProtocol.del_flag == 0,
+                BizCollectionProtocol.id.in_(protocol_ids),
+            )
+            .all()
+        ):
+            protocols[p.id] = p
+
+    reception_ids_for_flags = list(
+        {
+            *(r.id for r in receptions_by_sample_id.values() if r and r.id),
+            *(r.id for r in receptions_by_code.values() if r and r.id),
+        }
+    )
+    check_flags_by_reception = _check_flag_map(db, tenant_no, reception_ids_for_flags)
+
+    legacy_codes: List[str] = []
+    for sample, *_ in filtered:
+        legacy_codes.append(sample.sample_code or "")
+        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)
+    legacy_supplements = _load_legacy_supplements(db, tenant_no, legacy_codes)
+
     built_map: Dict[str, Dict[str, Any]] = {}
     for sample, stype, party, coll, hospital in filtered:
         code = sample.sample_code
-        reception = receptions.get(code)
+        reception = receptions_by_sample_id.get(sample.id) or receptions_by_code.get(code)
         production = productions.get(reception.id) if reception else None
-        extra = _parse_extra(extras[code].payload_json if code in extras else None)
-        built_map[code] = _build_row(
+        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 ""),
+        )
+        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(
             dataset=dataset,
             sample=sample,
             type_name=(stype.type_name if stype else None),
@@ -260,26 +648,109 @@ def list_rows(
             reception=reception,
             production=production,
             extra=extra,
+            protocol=protocol,
+            check_flags=check_flags,
         )
+        built_map[row.get("sampleCode") or code] = row
 
-    # Excel/补充表独有样本也要展示
-    for code, extra_row in extras.items():
-        if code in built_map:
-            continue
-        extra = _parse_extra(extra_row.payload_json)
-        reception = receptions.get(code)
-        production = productions.get(reception.id) if reception else None
-        built_map[code] = _build_row(
-            dataset=dataset,
-            sample=None,
-            type_name=None,
-            party=None,
-            coll=None,
-            hospital=None,
-            reception=reception,
-            production=production,
-            extra=extra,
+    # 单个核：业务已发生数据主要在科研捐献接收（lab_storage_code=UBMC*），无细胞储存协议样本时仍需展示
+    if dataset == DATASET_UBMC:
+        ubmc_recs = (
+            db.query(BizSampleReception)
+            .filter(
+                BizSampleReception.tenant_no == tenant_no,
+                BizSampleReception.del_flag == 0,
+                BizSampleReception.lab_storage_code.like("UBMC%"),
+            )
+            .order_by(BizSampleReception.reception_time.desc())
+            .all()
         )
+        mother_by_donation: Dict[str, str] = {}
+        donation_ids = sorted(
+            {r.research_donation_id for r in ubmc_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 ubmc_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)
+
+        ubmc_flags = _check_flag_map(
+            db,
+            tenant_no,
+            [r.id for r in ubmc_recs if r.id and r.id not in check_flags_by_reception],
+        )
+        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
+            ],
+        )
+        legacy_supplements.update(ubmc_legacy)
+
+        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()),
+            )
+            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 = []
@@ -311,9 +782,23 @@ def _build_row(
     reception: Optional[BizSampleReception],
     production: Optional[BizSampleProductionReception],
     extra: Dict[str, Any],
+    protocol: Optional[BizCollectionProtocol] = None,
+    check_flags: Optional[Dict[str, bool]] = None,
 ) -> Dict[str, Any]:
+    flags = check_flags or {}
+    protocol_code = (
+        (sample.sample_code if sample else None)
+        or extra.get("protocolSampleCode")
+        or ""
+    )
+    lab_code = ((reception.lab_storage_code if reception else None) or "").strip()
+    # 台账「细胞编码」展示实验室入库编码（B000…），协议样本号保留检索
+    display_code = lab_code if lab_code and len(lab_code) > 1 else (
+        protocol_code or extra.get("sampleCode") or ""
+    )
+
     protocol_vals = {
-        "sample_code": sample.sample_code if sample else extra.get("sampleCode"),
+        "sample_code": display_code,
         "mother_name": party.mother_name if party else None,
         "sample_collect_time": coll.sample_collect_time if coll else None,
         "hospital_name": hospital.hospital_name if hospital else None,
@@ -346,15 +831,11 @@ def _build_row(
             "available_cell_count_1e7": production.available_cell_count_1e7,
         }
 
-    sample_code = (
-        (sample.sample_code if sample else None)
-        or extra.get("sampleCode")
-        or ""
-    )
     data: Dict[str, Any] = {
-        "id": (sample.id if sample else f"extra:{dataset}:{sample_code}"),
+        "id": (sample.id if sample else f"extra:{dataset}:{display_code}"),
         "dataset": dataset,
-        "sampleCode": sample_code,
+        "sampleCode": display_code,
+        "protocolSampleCode": protocol_code or None,
         "sampleTypeName": type_name,
         "protocolSampleId": sample.id if sample else None,
         "collectionProtocolId": sample.collection_protocol_id if sample else None,
@@ -371,19 +852,57 @@ def _build_row(
         elif f.source == SOURCE_PRODUCTION:
             val = production_vals.get(f.attr)
         elif f.source == SOURCE_EXTRA:
+            # 仅源系统没有的列走补充表；业务源表字段不被补充表覆盖
             val = extra.get(f.key)
 
-        # 导入/补充表优先：有值则覆盖源表，保证 Excel 录入可见
-        if f.key in extra and extra.get(f.key) not in (None, ""):
-            val = extra.get(f.key)
+        # 源表为空时，允许用新/旧补充表填空（旧台账联动）
+        if f.source != SOURCE_EXTRA and _is_blank(val):
+            fallback = extra.get(f.key)
+            if not _is_blank(fallback):
+                val = fallback
 
         if isinstance(val, datetime):
             val = _fmt_dt(val)
         elif isinstance(val, Decimal):
             val = _fmt_num(val)
         data[f.key] = val
-    if not data.get("sampleCode"):
-        data["sampleCode"] = sample_code
+
+    data["sampleCode"] = display_code
+
+    # 从协议采集信息 / 附件 / 收样检查项推导补充列（补充表有值时不覆盖）
+    if not data.get("hasMaternalBlood"):
+        if coll is not None:
+            data["hasMaternalBlood"] = "是" if coll.maternal_blood_collect_time else "否"
+
+    if not data.get("hasConsent"):
+        if protocol and (protocol.consent_form_urls or "").strip():
+            data["hasConsent"] = "是"
+        else:
+            flagged = _yes_no_from_bool(flags.get("知情同意书"))
+            if flagged is not None:
+                data["hasConsent"] = flagged
+            elif protocol is not None:
+                data["hasConsent"] = "否"
+
+    if not data.get("hasHealthSurvey"):
+        if protocol and (protocol.health_survey_form_urls or "").strip():
+            data["hasHealthSurvey"] = "是"
+        else:
+            flagged = _yes_no_from_bool(flags.get("健康调查表"))
+            if flagged is not None:
+                data["hasHealthSurvey"] = flagged
+            elif protocol is not None:
+                data["hasHealthSurvey"] = "否"
+
+    if not data.get("labelClear"):
+        data["labelClear"] = _yes_no_from_bool(flags.get("样本信息正确，标识清晰"))
+
+    if not data.get("receiveCheck"):
+        data["receiveCheck"] = _derive_receive_check(flags)
+
+    if data.get("infectiousFourCard") is not None:
+        data["infectiousFourCard"] = _infectious_label(data.get("infectiousFourCard"))
+
     return data
 
 
@@ -473,6 +992,19 @@ def update_row(
     if not accepted:
         raise ValueError("没有可保存的字段")
 
+    reception = (
+        db.query(BizSampleReception)
+        .filter(
+            BizSampleReception.tenant_no == tenant_no,
+            BizSampleReception.del_flag == 0,
+            or_(
+                BizSampleReception.lab_storage_code == code,
+                BizSampleReception.sample_code == code,
+            ),
+        )
+        .order_by(BizSampleReception.reception_time.desc())
+        .first()
+    )
     sample = (
         db.query(BizProtocolSample)
         .filter(
@@ -482,16 +1014,30 @@ def update_row(
         )
         .first()
     )
-    reception = (
-        db.query(BizSampleReception)
-        .filter(
-            BizSampleReception.tenant_no == tenant_no,
-            BizSampleReception.sample_code == code,
-            BizSampleReception.del_flag == 0,
+    if sample is None and reception is not None and reception.protocol_sample_id:
+        sample = (
+            db.query(BizProtocolSample)
+            .filter(
+                BizProtocolSample.tenant_no == tenant_no,
+                BizProtocolSample.id == reception.protocol_sample_id,
+                BizProtocolSample.del_flag == 0,
+            )
+            .first()
+        )
+    if reception is None and sample is not None:
+        reception = (
+            db.query(BizSampleReception)
+            .filter(
+                BizSampleReception.tenant_no == tenant_no,
+                BizSampleReception.del_flag == 0,
+                or_(
+                    BizSampleReception.protocol_sample_id == sample.id,
+                    BizSampleReception.sample_code == sample.sample_code,
+                ),
+            )
+            .order_by(BizSampleReception.reception_time.desc())
+            .first()
         )
-        .order_by(BizSampleReception.reception_time.desc())
-        .first()
-    )
     production = None
     if reception:
         production = (
@@ -585,9 +1131,12 @@ def update_row(
             elif f.attr in numeric_attrs:
                 production_val = None if value in (None, "") else Decimal(str(value))
                 setattr(production, f.attr, production_val)
+            elif f.attr == "infectious_four_test_result":
+                production.infectious_four_test_result = (
+                    None if value in (None, "") else _infectious_code(value)
+                )
             elif f.attr in {
                 "blood_type",
-                "infectious_four_test_result",
                 "handler_name",
                 "remark",
                 "inoculation_amount",
@@ -610,11 +1159,24 @@ def update_row(
         operator_id=operator_id,
         replace=False,
     )
+    legacy_code = (
+        (reception.lab_storage_code if reception else None)
+        or (sample.sample_code if sample else None)
+        or code
+    )
+    sync_processing_to_legacy_supplement(
+        db,
+        tenant_no=tenant_no,
+        lab_storage_code=str(legacy_code),
+        sample_type_name=None,
+        payload=extra_payload,
+        operator_id=operator_id,
+    )
     db.commit()
 
     rows, _ = list_rows(db, tenant_no=tenant_no, dataset=dataset, keyword=code, page=1, page_size=20)
     for row in rows:
-        if row.get("sampleCode") == code:
+        if row.get("sampleCode") == code or row.get("protocolSampleCode") == code:
             return row
     raise LookupError("保存成功但读取失败")
 

backend/tests/test_lab_sample_processing.py

@@ -48,6 +48,139 @@ class LabSampleProcessingFieldsTest(unittest.TestCase):
         self.assertEqual(fmap["cd34Count"].dept, DEPT_QC)
         self.assertEqual(fmap["originalBloodWeightG"].dept, DEPT_PRODUCTION)
 
+    def test_source_fields_not_overridden_by_extra(self):
+        from datetime import datetime
+        from app.services.lab_sample_processing import _build_row
+        from types import SimpleNamespace
+
+        sample = SimpleNamespace(id="S1", sample_code="ZKZX34000865", collection_protocol_id="P1")
+        party = SimpleNamespace(mother_name="真实妈妈")
+        coll = SimpleNamespace(
+            sample_collect_time=None,
+            maternal_blood_collect_time=datetime(2026, 11, 4, 5, 20),
+        )
+        hospital = SimpleNamespace(hospital_name="真实医院")
+        protocol = SimpleNamespace(
+            consent_form_urls="protocol-consent/a",
+            health_survey_form_urls="",
+        )
+        reception = SimpleNamespace(
+            id="R1",
+            reception_time=None,
+            receive_temperature=None,
+            handler_name="接收人",
+            remark=None,
+            lab_storage_code="B0000103",
+        )
+        production = SimpleNamespace(
+            id="PR1",
+            original_blood_weight_g=72.08,
+            hes_volume_ml=None,
+            blood_type="B",
+            infectious_four_test_result="zc",
+            enriched_blood_weight_g=None,
+            cryopreservation_volume_ml=None,
+            handler_name="操作人",
+            remark=None,
+            original_blood_volume_ml=None,
+            nucleated_cell_count_1e8=None,
+            inoculation_amount=None,
+            cryopreservation_time=None,
+            storage_total_tubes=None,
+            available_cell_count_1e7=None,
+        )
+        row = _build_row(
+            dataset=DATASET_HSC,
+            sample=sample,
+            type_name="围产期造血干细胞",
+            party=party,
+            coll=coll,
+            hospital=hospital,
+            reception=reception,
+            production=production,
+            extra={
+                "motherName": "Excel妈妈",
+                "originalBloodWeightG": 1,
+                "cd34Count": "11.4",
+            },
+            protocol=protocol,
+            check_flags={
+                "健康调查表": True,
+                "样本信息正确，标识清晰": True,
+                "针路密闭": True,
+                "容器无破损": True,
+            },
+        )
+        self.assertEqual(row["sampleCode"], "B0000103")
+        self.assertEqual(row["protocolSampleCode"], "ZKZX34000865")
+        self.assertEqual(row["motherName"], "真实妈妈")
+        self.assertEqual(row["originalBloodWeightG"], 72.08)
+        self.assertEqual(row["bloodType"], "B")
+        self.assertEqual(row["cd34Count"], "11.4")
+        self.assertEqual(row["hasMaternalBlood"], "是")
+        self.assertEqual(row["hasConsent"], "是")
+        self.assertEqual(row["hasHealthSurvey"], "是")
+        self.assertEqual(row["labelClear"], "是")
+        self.assertEqual(row["infectiousFourCard"], "正常")
+        self.assertIn("针路密闭", row["receiveCheck"] or "")
+        self.assertFalse(row["fromExtraOnly"])
+
+    def test_infectious_code_roundtrip(self):
+        from app.services.lab_sample_processing import _infectious_code, _infectious_label
+
+        self.assertEqual(_infectious_label("zc"), "正常")
+        self.assertEqual(_infectious_label("bzc"), "不正常")
+        self.assertEqual(_infectious_code("正常"), "zc")
+        self.assertEqual(_infectious_code("不正常"), "bzc")
+
+    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
+
+        extra = _merge_extra_with_legacy(
+            {},
+            {"infectiousFourTestLabel": "正常", "labMaterialCode": "202611006"},
+        )
+        sample = SimpleNamespace(id="S1", sample_code="UBMC202611006", collection_protocol_id=None)
+        reception = SimpleNamespace(
+            id="R1",
+            reception_time=None,
+            receive_temperature=None,
+            handler_name=None,
+            remark=None,
+            lab_storage_code="UBMC202611006",
+        )
+        production = SimpleNamespace(
+            id="PR1",
+            original_blood_weight_g=None,
+            hes_volume_ml=None,
+            blood_type=None,
+            infectious_four_test_result=None,
+            enriched_blood_weight_g=None,
+            cryopreservation_volume_ml=None,
+            handler_name=None,
+            remark=None,
+            original_blood_volume_ml=None,
+            nucleated_cell_count_1e8=None,
+            inoculation_amount=None,
+            cryopreservation_time=None,
+            storage_total_tubes=None,
+            available_cell_count_1e7=None,
+        )
+        row = _build_row(
+            dataset=DATASET_UBMC,
+            sample=sample,
+            type_name="脐血单个核细胞",
+            party=None,
+            coll=None,
+            hospital=None,
+            reception=reception,
+            production=production,
+            extra=extra,
+        )
+        self.assertEqual(row["infectiousFourCard"], "正常")
+        self.assertEqual(row["labMaterialCode"], "202611006")
+
 
 if __name__ == "__main__":
     unittest.main()

docs/sql/20261112_lab_sample_processing_normalize_acl.sql

@@ -0,0 +1,21 @@
+-- 实验室样本处理 ACL 归一化（可重复执行）
+-- 1) 撤销所有 lab:sample-processing* / menu:lab-sample-processing 角色菜单
+-- 2) 仅按下列角色重授：
+--    市场编辑：市场专员/临床销售/销售助理/客服/采集专员
+--    生产编辑：生产专员/实验专员/细胞实验员/研发专员
+--    质检编辑：质检专员/质量安全员
+--    主管：生产主管/质量主管/采集主管/采集经理/研发主管/研发总监/总经理/副总经理/董事长/超级管理员
+--    只读：采集接受人
+-- 说明：正式执行请用 Agent 脚本 tmp_normalize_lab_acl.py（含幂等 undelete/insert）。
+-- 本文件作为规则记录；勿在未核对角色名时直接全量套用到其他环境。
+
+UPDATE sys_role_menu rm
+JOIN sys_menu m ON m.id = rm.menu_id
+SET rm.del_flag = 1,
+    rm.update_by = 'normalize-acl',
+    rm.update_time = NOW()
+WHERE rm.del_flag = 0
+  AND (
+    m.perms LIKE 'lab:sample-processing%'
+    OR m.perms = 'menu:lab-sample-processing'
+  );

docs/sql/20261112_lab_sample_processing_retire_excel_demo.sql

@@ -0,0 +1,14 @@
+-- 软删除实验室样本处理 Excel 演示补充行：
+-- 仅删除在协议样本表中不存在对应 sample_code 的补充记录。
+-- Excel 列结构仍由代码字段目录维护；真实列表只聚合协议/接收/生产源数据。
+
+UPDATE biz_lab_sample_processing_extra e
+LEFT JOIN biz_protocol_sample s
+  ON s.sample_code = e.sample_code
+ AND s.tenant_no = e.tenant_no
+ AND s.del_flag = 0
+SET e.del_flag = 1,
+    e.update_by = 'retire-excel-demo',
+    e.update_time = NOW()
+WHERE e.del_flag = 0
+  AND s.id IS NULL;

web-admin/src/router/modules/sqxp.ts

@@ -1335,6 +1335,59 @@ export const sqxpAdminRoutes: AppRouteRecord[] = [
     ]
   },
 
+  // ═══════════════════════════════════════════════════════════════════════════
+  // 3b. 实验室样本处理数据 (menu:lab-sample-processing)
+  // ═══════════════════════════════════════════════════════════════════════════
+  {
+    id: 'TOP0000000000000000000000000000L1',
+    name: 'LabSampleProcessing',
+    path: '/lab-sample-processing',
+    component: '/index/index',
+    meta: {
+      title: '实验室样本处理数据',
+      icon: 'ri:test-tube-line',
+      keepAlive: false,
+      isTopMenu: true,
+      menuCode: 'menu:lab-sample-processing',
+      roles: ['admin']
+    },
+    children: [
+      {
+        id: 'LAB00000000000000000000000000001',
+        name: 'LabSampleProcessingHsc',
+        path: 'hsc',
+        component: '/lab-sample-processing/hsc/index',
+        meta: {
+          title: '造血干细胞数据',
+          keepAlive: true,
+          menuCode: 'lab:sample-processing:list'
+        }
+      },
+      {
+        id: 'LAB00000000000000000000000000002',
+        name: 'LabSampleProcessingUbmc',
+        path: 'ubmc',
+        component: '/lab-sample-processing/ubmc/index',
+        meta: {
+          title: '单个核数据',
+          keepAlive: true,
+          menuCode: 'lab:sample-processing:list'
+        }
+      },
+      {
+        id: 'LAB00000000000000000000000000003',
+        name: 'LabSampleProcessingMsc',
+        path: 'msc',
+        component: '/lab-sample-processing/msc/index',
+        meta: {
+          title: '脐带间充质干细胞数据',
+          keepAlive: true,
+          menuCode: 'lab:sample-processing:list'
+        }
+      }
+    ]
+  },
+
   // ═══════════════════════════════════════════════════════════════════════════
   // 4. 运营支撑 (menu:operations)
   // 二级菜单：系统基础信息、运营支撑

web-admin/src/views/lab-sample-processing/LabSampleProcessingPage.vue

@@ -275,15 +275,28 @@
   async function handleExport() {
     exporting.value = true
     try {
-      const res = await fetchLabSampleList(props.dataset, {
-        keyword: searchForm.keyword || undefined,
-        dateFrom: dateRange.value?.[0],
-        dateTo: dateRange.value?.[1],
-        page: 1,
-        pageSize: 5000
-      })
-      const cols = res?.fields?.length ? res.fields : fields.value
-      const rows = res?.list || []
+      // 列表接口 pageSize 上限为 200，导出需分页拉全量后再生成 xlsx
+      const pageSize = 200
+      let page = 1
+      let total = Number.POSITIVE_INFINITY
+      const rows: Record<string, any>[] = []
+      let cols = fields.value
+      while (rows.length < total) {
+        const res = await fetchLabSampleList(props.dataset, {
+          keyword: searchForm.keyword || undefined,
+          dateFrom: dateRange.value?.[0],
+          dateTo: dateRange.value?.[1],
+          page,
+          pageSize
+        })
+        if (res?.fields?.length) cols = res.fields
+        total = Number(res?.total ?? 0)
+        const batch = res?.list || []
+        rows.push(...batch)
+        if (!batch.length || batch.length < pageSize) break
+        page += 1
+        if (page > 500) break
+      }
       const ExcelJS = (await import('exceljs')).default
       const workbook = new ExcelJS.Workbook()
       const sheet = workbook.addWorksheet(props.title || props.dataset)