zkxw/backend/app/crud/lab_sample_process.py

"""实验室样本处理台账列表与补录 CRUD。"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import and_
from sqlalchemy.orm import Session

from app.core.data_scope import apply_data_scope
from app.core.utils import generate_uuid
from app.crud import sample_production_reception as prod_svc
from app.models.models import (
    BizCellSampleType,
    BizCollectionHospital,
    BizCollectionProtocol,
    BizLabSampleProcessSupplement,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizSampleProductionReception,
    BizSampleReception,
    BizSampleReceptionCheckResult,
    BizStorageReleaseReview,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationParty,
    BizResearchDonationSample,
)
from app.services import lab_sample_process as ledger


def _fmt_raw(dt: Optional[datetime]) -> str:
    return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else ""


def _normalize_bound(value: Optional[str], *, end_of_day: bool = False) -> Optional[datetime]:
    text = (value or "").strip()
    if not text:
        return None
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
        try:
            dt = datetime.strptime(text[:19] if " " in text else text[:10], fmt)
            if end_of_day and fmt == "%Y-%m-%d":
                return dt.replace(hour=23, minute=59, second=59)
            return dt
        except ValueError:
            continue
    return None


def _type_ids_by_name(db: Session, tenant_no: str, type_name: str) -> List[str]:
    rows = (
        db.query(BizCellSampleType.id)
        .filter(
            BizCellSampleType.tenant_no == tenant_no,
            BizCellSampleType.del_flag == 0,
            BizCellSampleType.type_name == type_name,
        )
        .all()
    )
    return [r.id for r in rows if r.id]


def _get_supplement_map(
    db: Session, tenant_no: str, codes: List[str]
) -> Dict[str, Dict[str, Any]]:
    codes = [c for c in {(x or "").strip() for x in codes} if c]
    if not codes:
        return {}
    rows = (
        db.query(BizLabSampleProcessSupplement)
        .filter(
            BizLabSampleProcessSupplement.tenant_no == tenant_no,
            BizLabSampleProcessSupplement.del_flag == 0,
            BizLabSampleProcessSupplement.lab_storage_code.in_(codes),
        )
        .all()
    )
    out: Dict[str, Dict[str, Any]] = {}
    for row in rows:
        out[row.lab_storage_code] = {
            "id": row.id,
            "payload": ledger.parse_payload(row.payload),
            "remark": row.remark or "",
            "updateTime": _fmt_raw(row.update_time),
        }
    return out


def get_supplement(
    db: Session, *, tenant_no: str, lab_storage_code: str
) -> Optional[Dict[str, Any]]:
    code = (lab_storage_code or "").strip()
    if not code:
        return None
    row = (
        db.query(BizLabSampleProcessSupplement)
        .filter(
            BizLabSampleProcessSupplement.tenant_no == tenant_no,
            BizLabSampleProcessSupplement.lab_storage_code == code,
            BizLabSampleProcessSupplement.del_flag == 0,
        )
        .first()
    )
    if not row:
        return None
    return {
        "id": row.id,
        "labStorageCode": row.lab_storage_code,
        "receptionSource": row.reception_source,
        "sampleTypeName": row.sample_type_name or "",
        "payload": ledger.parse_payload(row.payload),
        "remark": row.remark or "",
        "createTime": _fmt_raw(row.create_time),
        "updateTime": _fmt_raw(row.update_time),
    }


def upsert_supplement(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    dept_id: str,
    lab_storage_code: str,
    reception_source: str,
    sample_type_name: str,
    payload: Dict[str, Any],
    remark: Optional[str] = None,
) -> Dict[str, Any]:
    code = (lab_storage_code or "").strip()
    if not code:
        raise ValueError("实验室入库编码不能为空")
    src = (reception_source or "").strip() or "cell_storage"
    if src not in ("cell_storage", "research_donation"):
        raise ValueError("来源类型无效")
    type_name = (sample_type_name or "").strip()
    clean_payload = {
        str(k): ("" if v is None else str(v).strip())
        for k, v in (payload or {}).items()
        if str(k).strip()
    }

    row = (
        db.query(BizLabSampleProcessSupplement)
        .filter(
            BizLabSampleProcessSupplement.tenant_no == tenant_no,
            BizLabSampleProcessSupplement.lab_storage_code == code,
            BizLabSampleProcessSupplement.del_flag == 0,
        )
        .first()
    )
    now = datetime.now()
    if row:
        row.reception_source = src
        row.sample_type_name = type_name or row.sample_type_name
        row.payload = ledger.dumps_payload(clean_payload)
        if remark is not None:
            row.remark = (remark or "").strip() or None
        row.update_by = user_id
        row.update_time = now
    else:
        row = BizLabSampleProcessSupplement(
            id=generate_uuid(),
            lab_storage_code=code,
            reception_source=src,
            sample_type_name=type_name or None,
            payload=ledger.dumps_payload(clean_payload),
            tenant_no=tenant_no,
            remark=(remark or "").strip() or None,
            create_by=user_id,
            create_dept_id=dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(row)
    db.commit()
    db.refresh(row)
    return get_supplement(db, tenant_no=tenant_no, lab_storage_code=code) or {}


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 _storage_status_map(db: Session, tenant_no: str, sample_codes: List[str]) -> Dict[str, str]:
    codes = [c for c in sample_codes if c]
    if not codes:
        return {}
    rows = (
        db.query(BizStorageReleaseReview.sample_code, BizStorageReleaseReview.storage_status)
        .filter(
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
            BizStorageReleaseReview.sample_code.in_(codes),
        )
        .all()
    )
    labels = {"drk": "待入库", "yrk": "已入库"}
    return {r.sample_code: labels.get(r.storage_status or "", r.storage_status or "") for r in rows}


def _hospital_name_map(db: Session, hospital_ids: List[str]) -> Dict[str, str]:
    ids = [i for i in hospital_ids if i]
    if not ids:
        return {}
    rows = (
        db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name)
        .filter(BizCollectionHospital.id.in_(ids), BizCollectionHospital.del_flag == 0)
        .all()
    )
    return {r.id: r.hospital_name or "" for r in rows}


def _build_cell_system(
    *,
    tab_key: str,
    rec: BizSampleReception,
    sample: BizProtocolSample,
    party: Optional[BizProtocolPartyCustomer],
    coll: Optional[BizProtocolCollectionInfo],
    protocol: Optional[BizCollectionProtocol],
    prod: Optional[Dict[str, Any]],
    hospital_name: str,
    check_flags: Dict[str, bool],
    storage_status: str,
    qc_mother: Dict[str, str],
    qc_process: Dict[str, str],
) -> Dict[str, Any]:
    prod = prod or {}
    maternal_time = coll.maternal_blood_collect_time if coll else None
    clear_flag = None
    for name, checked in check_flags.items():
        if "标识清晰" in name or "样本信息正确" in name:
            clear_flag = checked
            break

    base: Dict[str, Any] = {
        "receptionId": rec.id,
        "sampleCode": (rec.sample_code or sample.sample_code or "").strip(),
        "labStorageCode": (rec.lab_storage_code or "").strip(),
        "motherName": (party.mother_name if party else "") or "",
        "collectionTime": ledger.format_cn_datetime(coll.sample_collect_time if coll else None),
        "collectionTimeRaw": _fmt_raw(coll.sample_collect_time if coll else None),
        "hospitalName": hospital_name,
        "receptionTime": ledger.format_cn_datetime(rec.reception_time),
        "receptionTimeRaw": _fmt_raw(rec.reception_time),
        "receiveTemperature": rec.receive_temperature or "",
        "handlerName": rec.handler_name or "",
        "remarkReception": rec.remark or "",
        "createTime": ledger.format_cn_datetime(rec.create_time),
        "createTimeRaw": _fmt_raw(rec.create_time),
        "operatorName": prod.get("handlerName") or "",
        "remarkProcess": prod.get("remark") or "",
        "storageStatus": storage_status,
    }

    if tab_key == "cell_hsc":
        base.update(
            {
                "cellCode": (rec.lab_storage_code or "").strip(),
                "hasMaternalBlood": "是" if maternal_time else "否",
                "remeasureMaternalBlood": "",
                "consentForm": ledger.yes_no_from_urls(
                    protocol.consent_form_urls if protocol else None
                ),
                "transportPackingList": "",
                "cordBloodReceiveConfirm": "",
                "healthSurvey": ledger.yes_no_from_urls(
                    protocol.health_survey_form_urls if protocol else None
                ),
                "labelClear": ledger.yes_no_from_bool(clear_flag) if clear_flag is not None else "",
                "receiveCheckText": "",
                "originalBloodWeightG": prod.get("originalBloodWeightG"),
                "hesVolumeMl": prod.get("hesVolumeMl"),
                "bloodType": prod.get("bloodType") or "",
                "infectiousFourTestLabel": prod.get("infectiousFourTestResultLabel")
                or ledger.infectious_label(prod.get("infectiousFourTestResult")),
                "infectiousFourExternal": "",
                "preEnrichNucleatedCount": "",
                "enrichedBloodWeightG": prod.get("enrichedBloodWeightG"),
                "postEnrichNucleatedCount": "",
                "cryopreservationVolumeMl": prod.get("cryopreservationVolumeMl"),
                "cd34Count1e4": "",
            }
        )
    else:  # cell_msc
        base.update(
            {
                "labFlowCode": (rec.lab_storage_code or "").strip(),
                "cordCollectTime": ledger.format_cn_datetime(
                    coll.sample_collect_time if coll else None
                ),
                "transportTempC": rec.receive_temperature or "",
                "cordReceiveTime": ledger.format_cn_datetime(rec.reception_time),
                "inoculationAmount": prod.get("inoculationAmount") or "",
                "cryopreservationTime": ledger.format_cn_datetime(prod.get("cryopreservationTime")),
                "storageTotalTubes": prod.get("storageTotalTubes"),
                "availableCellCount1e7": prod.get("availableCellCount1e7"),
                "remark": rec.remark or prod.get("remark") or "",
                "storageDateP2": "",
            }
        )
        base.update(qc_process)
        base.update(qc_mother)
    return base


def _build_donation_system(
    *,
    tab_key: str,
    rec: BizSampleReception,
    sample: BizResearchDonationSample,
    party: Optional[BizResearchDonationParty],
    coll: Optional[BizResearchDonationCollection],
    prod: Optional[Dict[str, Any]],
    storage_status: str,
    qc_mother: Dict[str, str],
    qc_process: Dict[str, str],
) -> Dict[str, Any]:
    prod = prod or {}
    base: Dict[str, Any] = {
        "receptionId": rec.id,
        "sampleCode": (rec.sample_code or sample.sample_code or "").strip(),
        "labStorageCode": (rec.lab_storage_code or "").strip(),
        "motherName": (party.mother_name if party else "") or "",
        "receptionTime": ledger.format_cn_datetime(rec.reception_time),
        "receptionTimeRaw": _fmt_raw(rec.reception_time),
        "receiveTemperature": rec.receive_temperature or "",
        "handlerName": rec.handler_name or "",
        "remarkReception": rec.remark or "",
        "createTime": ledger.format_cn_datetime(rec.create_time),
        "createTimeRaw": _fmt_raw(rec.create_time),
        "operatorName": prod.get("handlerName") or "",
        "remarkProcess": prod.get("remark") or "",
        "storageStatus": storage_status,
    }
    if tab_key == "donation_hsc":
        base.update(
            {
                "cellCode": (rec.lab_storage_code or "").strip(),
                "labMaterialCode": "",
                "inboundCode": (rec.lab_storage_code or "").strip(),
                "discardReason": "",
                "receiveCheckText": "",
                "originalBloodVolumeMl": prod.get("originalBloodVolumeMl"),
                "nucleatedCellCount1e8": prod.get("nucleatedCellCount1e8"),
                "preOpNucleatedCount": "",
                "infectiousFourTestLabel": prod.get("infectiousFourTestResultLabel")
                or ledger.infectious_label(prod.get("infectiousFourTestResult")),
            }
        )
        # 科研捐赠母血 5 项（不含核酸）
        for k in (
            "motherHbsag",
            "motherAntiHcv",
            "motherAntiTp",
            "motherAntiHiv",
            "motherCmvIgm",
        ):
            if qc_mother.get(k):
                base[k] = qc_mother[k]
    else:
        base.update(
            {
                "labFlowCode": (rec.lab_storage_code or "").strip(),
                "cordCollectTime": ledger.format_cn_datetime(
                    coll.sample_collect_time if coll else None
                ),
                "transportTempC": rec.receive_temperature or "",
                "cordReceiveTime": ledger.format_cn_datetime(rec.reception_time),
                "inoculationAmount": prod.get("inoculationAmount") or "",
                "cryopreservationTime": ledger.format_cn_datetime(prod.get("cryopreservationTime")),
                "storageTotalTubes": prod.get("storageTotalTubes"),
                "availableCellCount1e7": prod.get("availableCellCount1e7"),
                "remark": rec.remark or prod.get("remark") or "",
                "storageDateP2": "",
            }
        )
        base.update(qc_process)
        base.update(qc_mother)
    return base


def list_ledger(
    db: Session,
    *,
    tab_key: str,
    page: int = 1,
    page_size: int = 10,
    tenant_no: str,
    lab_storage_code: Optional[str] = None,
    mother_name: Optional[str] = None,
    create_time_start: Optional[str] = None,
    create_time_end: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    meta = ledger.TAB_META.get(tab_key)
    if not meta:
        raise ValueError("无效的 Tab")
    source = meta["source"]
    type_name = meta["sampleTypeName"]
    type_ids = _type_ids_by_name(db, tenant_no, type_name)
    if not type_ids:
        return [], 0

    q = db.query(BizSampleReception).filter(
        BizSampleReception.tenant_no == tenant_no,
        BizSampleReception.del_flag == 0,
        BizSampleReception.reception_source == source,
        BizSampleReception.lab_storage_code.isnot(None),
        BizSampleReception.lab_storage_code != "",
        BizSampleReception.accept_conclusion == "js",
    )
    code_kw = (lab_storage_code or "").strip()
    if code_kw:
        q = q.filter(BizSampleReception.lab_storage_code.like(f"%{code_kw}%"))
    start = _normalize_bound(create_time_start)
    end = _normalize_bound(create_time_end, end_of_day=True)
    if start:
        q = q.filter(BizSampleReception.create_time >= start)
    if end:
        q = q.filter(BizSampleReception.create_time <= end)
    if user_info:
        q = apply_data_scope(q, BizSampleReception, user_info, db, tenant_no)

    mother_kw = (mother_name or "").strip()

    # 按来源限制样本类型，并可选母亲姓名
    if source == "cell_storage":
        sample_q = (
            db.query(BizProtocolSample.id)
            .join(
                BizCollectionProtocol,
                and_(
                    BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id,
                    BizCollectionProtocol.del_flag == 0,
                    BizCollectionProtocol.tenant_no == tenant_no,
                ),
            )
            .filter(
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.sample_type_id.in_(type_ids),
            )
        )
        if mother_kw:
            sample_q = sample_q.join(
                BizProtocolPartyCustomer,
                and_(
                    BizProtocolPartyCustomer.collection_protocol_id
                    == BizProtocolSample.collection_protocol_id,
                    BizProtocolPartyCustomer.del_flag == 0,
                    BizProtocolPartyCustomer.tenant_no == tenant_no,
                ),
            ).filter(BizProtocolPartyCustomer.mother_name.like(f"%{mother_kw}%"))
        sample_ids = [r.id for r in sample_q.all()]
        if not sample_ids:
            return [], 0
        q = q.filter(BizSampleReception.protocol_sample_id.in_(sample_ids))
    else:
        sample_q = (
            db.query(BizResearchDonationSample.id)
            .join(
                BizResearchDonation,
                and_(
                    BizResearchDonation.id == BizResearchDonationSample.research_donation_id,
                    BizResearchDonation.del_flag == 0,
                    BizResearchDonation.tenant_no == tenant_no,
                ),
            )
            .filter(
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.sample_type_id.in_(type_ids),
            )
        )
        if mother_kw:
            sample_q = sample_q.join(
                BizResearchDonationParty,
                and_(
                    BizResearchDonationParty.research_donation_id
                    == BizResearchDonationSample.research_donation_id,
                    BizResearchDonationParty.del_flag == 0,
                    BizResearchDonationParty.tenant_no == tenant_no,
                ),
            ).filter(BizResearchDonationParty.mother_name.like(f"%{mother_kw}%"))
        sample_ids = [r.id for r in sample_q.all()]
        if not sample_ids:
            return [], 0
        q = q.filter(BizSampleReception.donation_sample_id.in_(sample_ids))

    total = q.count()
    page = max(1, int(page or 1))
    page_size = min(max(1, int(page_size or 10)), 200)
    receptions = (
        q.order_by(BizSampleReception.create_time.desc(), BizSampleReception.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    if not receptions:
        return [], int(total)

    reception_ids = [r.id for r in receptions]
    codes = [(r.lab_storage_code or "").strip() for r in receptions]
    sample_codes = [(r.sample_code or "").strip() for r in receptions]
    prod_map = _prod_map(db, tenant_no, reception_ids)
    check_map = _check_flag_map(db, tenant_no, reception_ids)
    storage_map = _storage_status_map(db, tenant_no, [c for c in sample_codes if c])
    sup_map = _get_supplement_map(db, tenant_no, codes)

    items: List[Dict[str, Any]] = []
    if source == "cell_storage":
        items = _assemble_cell_rows(
            db,
            tenant_no=tenant_no,
            tab_key=tab_key,
            receptions=receptions,
            prod_map=prod_map,
            check_map=check_map,
            storage_map=storage_map,
            sup_map=sup_map,
        )
    else:
        items = _assemble_donation_rows(
            db,
            tenant_no=tenant_no,
            tab_key=tab_key,
            receptions=receptions,
            prod_map=prod_map,
            storage_map=storage_map,
            sup_map=sup_map,
        )
    return items, int(total)


def _prod_map(db: Session, tenant_no: str, reception_ids: List[str]) -> Dict[str, Dict[str, Any]]:
    rows = (
        db.query(BizSampleProductionReception)
        .filter(
            BizSampleProductionReception.tenant_no == tenant_no,
            BizSampleProductionReception.del_flag == 0,
            BizSampleProductionReception.reception_id.in_(reception_ids),
        )
        .all()
    )
    return {r.reception_id: prod_svc._row_to_dict(r) for r in rows}


def _assemble_cell_rows(
    db: Session,
    *,
    tenant_no: str,
    tab_key: str,
    receptions: List[BizSampleReception],
    prod_map: Dict[str, Dict[str, Any]],
    check_map: Dict[str, Dict[str, bool]],
    storage_map: Dict[str, str],
    sup_map: Dict[str, Dict[str, Any]],
) -> List[Dict[str, Any]]:
    sample_ids = [r.protocol_sample_id for r in receptions if r.protocol_sample_id]
    samples = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.del_flag == 0,
            BizProtocolSample.id.in_(sample_ids),
        )
        .all()
        if sample_ids
        else []
    )
    sample_map = {s.id: s for s in samples}
    protocol_ids = list({s.collection_protocol_id for s in samples if s.collection_protocol_id})
    parties = (
        db.query(BizProtocolPartyCustomer)
        .filter(
            BizProtocolPartyCustomer.tenant_no == tenant_no,
            BizProtocolPartyCustomer.del_flag == 0,
            BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
        )
        .all()
        if protocol_ids
        else []
    )
    party_map = {p.collection_protocol_id: p for p in parties}
    colls = (
        db.query(BizProtocolCollectionInfo)
        .filter(
            BizProtocolCollectionInfo.tenant_no == tenant_no,
            BizProtocolCollectionInfo.del_flag == 0,
            BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids),
        )
        .all()
        if protocol_ids
        else []
    )
    coll_map = {c.collection_protocol_id: c for c in colls}
    protocols = (
        db.query(BizCollectionProtocol)
        .filter(
            BizCollectionProtocol.tenant_no == tenant_no,
            BizCollectionProtocol.del_flag == 0,
            BizCollectionProtocol.id.in_(protocol_ids),
        )
        .all()
        if protocol_ids
        else []
    )
    protocol_map = {p.id: p for p in protocols}
    hospital_ids = [c.hospital_id for c in colls if c.hospital_id]
    hospital_map = _hospital_name_map(db, hospital_ids)

    out: List[Dict[str, Any]] = []
    for rec in receptions:
        sample = sample_map.get(rec.protocol_sample_id or "")
        if not sample:
            continue
        pid = sample.collection_protocol_id
        sample_code = (rec.sample_code or sample.sample_code or "").strip()
        tasks = ledger.load_qc_tasks_for_sample(db, tenant_no, sample_code)
        qc_mother = ledger.extract_mother_blood(tasks, include_nucleic=True)
        qc_process = ledger.extract_process_tests(tasks)
        system = _build_cell_system(
            tab_key=tab_key,
            rec=rec,
            sample=sample,
            party=party_map.get(pid),
            coll=coll_map.get(pid),
            protocol=protocol_map.get(pid),
            prod=prod_map.get(rec.id),
            hospital_name=hospital_map.get((coll_map.get(pid).hospital_id if coll_map.get(pid) else "") or "", ""),
            check_flags=check_map.get(rec.id, {}),
            storage_status=storage_map.get(sample_code, ""),
            qc_mother=qc_mother,
            qc_process=qc_process,
        )
        lab_code = system.get("labStorageCode") or ""
        merged = ledger.merge_row(
            tab_key=tab_key,
            system_values=system,
            supplement=(sup_map.get(lab_code) or {}).get("payload"),
        )
        merged["supplementId"] = (sup_map.get(lab_code) or {}).get("id") or ""
        merged["hasSupplement"] = bool((sup_map.get(lab_code) or {}).get("payload"))
        out.append(merged)
    return out


def _assemble_donation_rows(
    db: Session,
    *,
    tenant_no: str,
    tab_key: str,
    receptions: List[BizSampleReception],
    prod_map: Dict[str, Dict[str, Any]],
    storage_map: Dict[str, str],
    sup_map: Dict[str, Dict[str, Any]],
) -> List[Dict[str, Any]]:
    sample_ids = [r.donation_sample_id for r in receptions if r.donation_sample_id]
    samples = (
        db.query(BizResearchDonationSample)
        .filter(
            BizResearchDonationSample.tenant_no == tenant_no,
            BizResearchDonationSample.del_flag == 0,
            BizResearchDonationSample.id.in_(sample_ids),
        )
        .all()
        if sample_ids
        else []
    )
    sample_map = {s.id: s for s in samples}
    donation_ids = list({s.research_donation_id for s in samples if s.research_donation_id})
    parties = (
        db.query(BizResearchDonationParty)
        .filter(
            BizResearchDonationParty.tenant_no == tenant_no,
            BizResearchDonationParty.del_flag == 0,
            BizResearchDonationParty.research_donation_id.in_(donation_ids),
        )
        .all()
        if donation_ids
        else []
    )
    party_map = {p.research_donation_id: p for p in parties}
    colls = (
        db.query(BizResearchDonationCollection)
        .filter(
            BizResearchDonationCollection.tenant_no == tenant_no,
            BizResearchDonationCollection.del_flag == 0,
            BizResearchDonationCollection.research_donation_id.in_(donation_ids),
        )
        .all()
        if donation_ids
        else []
    )
    coll_map = {c.research_donation_id: c for c in colls}

    out: List[Dict[str, Any]] = []
    for rec in receptions:
        sample = sample_map.get(rec.donation_sample_id or "")
        if not sample:
            continue
        did = sample.research_donation_id
        sample_code = (rec.sample_code or sample.sample_code or "").strip()
        tasks = ledger.load_qc_tasks_for_sample(db, tenant_no, sample_code)
        qc_mother = ledger.extract_mother_blood(tasks, include_nucleic=True)
        qc_process = ledger.extract_process_tests(tasks)
        system = _build_donation_system(
            tab_key=tab_key,
            rec=rec,
            sample=sample,
            party=party_map.get(did),
            coll=coll_map.get(did),
            prod=prod_map.get(rec.id),
            storage_status=storage_map.get(sample_code, ""),
            qc_mother=qc_mother,
            qc_process=qc_process,
        )
        lab_code = system.get("labStorageCode") or ""
        merged = ledger.merge_row(
            tab_key=tab_key,
            system_values=system,
            supplement=(sup_map.get(lab_code) or {}).get("payload"),
        )
        merged["supplementId"] = (sup_map.get(lab_code) or {}).get("id") or ""
        merged["hasSupplement"] = bool((sup_map.get(lab_code) or {}).get("payload"))
        out.append(merged)
    return out