zkxw/backend/app/crud/pre_transit_sample.py

"""预产及在途样本 — 质检看板（协议单维度列表）"""
from __future__ import annotations

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

from sqlalchemy import and_, exists, func, not_, or_
from sqlalchemy.orm import Session

from app.crud.collection_protocol import (
    _collection_info_to_dict,
    _fmt_dt,
    _load_express_name_map,
    _load_sample_type_name_map,
    _load_site_name_map,
    _normalize_date_bound,
    _party_customer_to_dict,
    _protocol_to_dict,
    total_pages,
    urls_to_attachment_list,
)
from app.utils.sample_special import sample_special_to_dict
from app.crud.research_donation import (
    _collection_to_dict,
    _donation_to_dict,
    _party_to_dict,
)
from app.models.models import (
    BizCollectionHospital,
    BizCollectionProtocol,
    BizCollectionRegion,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizProtocolSampleTransport,
    BizSampleReception,
    SysUser,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationParty,
    BizResearchDonationSample,
    BizResearchDonationSampleTransport,
)

_CELL_IN_TRANSIT_EXCLUDED = ("yzf", "sdfq", "zky", "ygq")
_DONATION_IN_TRANSIT_EXCLUDED = ("yzf", "sdfq", "zckc", "ygq")
_PRE_DELIVERY_STATUSES = ("dsx", "sxz")


def _is_uncollected_status(status: Optional[str]) -> bool:
    return not (status or "").strip()


def _is_in_transit_sample(status: Optional[str]) -> bool:
    return (status or "").strip() == "ycj"


def _cell_received_exists(tenant_no: str):
    return (
        exists()
        .where(
            and_(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                BizSampleReception.reception_source == "cell_storage",
                BizSampleReception.sample_code.isnot(None),
                BizSampleReception.sample_code != "",
                BizSampleReception.sample_code == BizProtocolSample.sample_code,
            )
        )
        .correlate(BizProtocolSample)
    )


def _donation_received_exists(tenant_no: str):
    return (
        exists()
        .where(
            and_(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                BizSampleReception.reception_source == "research_donation",
                BizSampleReception.sample_code.isnot(None),
                BizSampleReception.sample_code != "",
                BizSampleReception.sample_code == BizResearchDonationSample.sample_code,
            )
        )
        .correlate(BizResearchDonationSample)
    )


def _apply_collection_keyword(q, db: Session, tenant_no: str, keyword: str):
    kw = (keyword or "").strip()
    if not kw:
        return q
    like = f"%{kw}%"
    party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
        BizProtocolPartyCustomer.del_flag == 0,
        or_(
            BizProtocolPartyCustomer.mother_name.like(like),
            BizProtocolPartyCustomer.mother_id_card.like(like),
            BizProtocolPartyCustomer.mother_phone.like(like),
        ),
    )
    if tenant_no:
        party_q = party_q.filter(BizProtocolPartyCustomer.tenant_no == tenant_no)
    kw_conds = [
        BizCollectionProtocol.id.in_(party_q),
        BizCollectionProtocol.protocol_no.like(like),
    ]
    from app.crud import sample_reception as reception_svc

    proto_ids = reception_svc.protocol_ids_matching_lab_storage_keyword(db, tenant_no, kw)
    if proto_ids:
        kw_conds.append(BizCollectionProtocol.id.in_(proto_ids))
    sample_proto_q = (
        db.query(BizProtocolSample.collection_protocol_id)
        .filter(
            BizProtocolSample.del_flag == 0,
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.sample_code.like(like),
        )
    )
    kw_conds.append(BizCollectionProtocol.id.in_(sample_proto_q))
    return q.filter(or_(*kw_conds))


def _apply_donation_keyword(q, db: Session, tenant_no: str, keyword: str):
    kw = (keyword or "").strip()
    if not kw:
        return q
    like = f"%{kw}%"
    party_q = db.query(BizResearchDonationParty.research_donation_id).filter(
        BizResearchDonationParty.del_flag == 0,
        or_(
            BizResearchDonationParty.mother_name.like(like),
            BizResearchDonationParty.mother_id_card.like(like),
            BizResearchDonationParty.mother_phone.like(like),
        ),
    )
    if tenant_no:
        party_q = party_q.filter(BizResearchDonationParty.tenant_no == tenant_no)
    kw_conds = [
        BizResearchDonation.id.in_(party_q),
        BizResearchDonation.donation_no.like(like),
    ]
    from app.crud import sample_reception as reception_svc

    donation_ids = reception_svc.donation_ids_matching_lab_storage_keyword(db, tenant_no, kw)
    if donation_ids:
        kw_conds.append(BizResearchDonation.id.in_(donation_ids))
    sample_donation_q = (
        db.query(BizResearchDonationSample.research_donation_id)
        .filter(
            BizResearchDonationSample.del_flag == 0,
            BizResearchDonationSample.tenant_no == tenant_no,
            BizResearchDonationSample.sample_code.like(like),
        )
    )
    kw_conds.append(BizResearchDonation.id.in_(sample_donation_q))
    return q.filter(or_(*kw_conds))


def _apply_create_time_filter(q, model, create_time_start: Optional[str], create_time_end: Optional[str]):
    if create_time_start:
        q = q.filter(model.create_time >= _normalize_date_bound(create_time_start))
    if create_time_end:
        q = q.filter(model.create_time <= _normalize_date_bound(create_time_end, end_of_day=True))
    return q


def _apply_collection_sample_collect_time_filter(
    q,
    db: Session,
    tenant_no: str,
    sample_collect_time_start: Optional[str],
    sample_collect_time_end: Optional[str],
):
    if not sample_collect_time_start and not sample_collect_time_end:
        return q
    ci_q = db.query(BizProtocolCollectionInfo.collection_protocol_id).filter(
        BizProtocolCollectionInfo.del_flag == 0,
    )
    if tenant_no:
        ci_q = ci_q.filter(BizProtocolCollectionInfo.tenant_no == tenant_no)
    if sample_collect_time_start:
        ci_q = ci_q.filter(
            BizProtocolCollectionInfo.sample_collect_time
            >= _normalize_date_bound(sample_collect_time_start)
        )
    if sample_collect_time_end:
        ci_q = ci_q.filter(
            BizProtocolCollectionInfo.sample_collect_time
            <= _normalize_date_bound(sample_collect_time_end, end_of_day=True)
        )
    return q.filter(BizCollectionProtocol.id.in_(ci_q))


def _apply_donation_sample_collect_time_filter(
    q,
    db: Session,
    tenant_no: str,
    sample_collect_time_start: Optional[str],
    sample_collect_time_end: Optional[str],
):
    if not sample_collect_time_start and not sample_collect_time_end:
        return q
    coll_q = db.query(BizResearchDonationCollection.research_donation_id).filter(
        BizResearchDonationCollection.del_flag == 0,
    )
    if tenant_no:
        coll_q = coll_q.filter(BizResearchDonationCollection.tenant_no == tenant_no)
    if sample_collect_time_start:
        coll_q = coll_q.filter(
            BizResearchDonationCollection.sample_collect_time
            >= _normalize_date_bound(sample_collect_time_start)
        )
    if sample_collect_time_end:
        coll_q = coll_q.filter(
            BizResearchDonationCollection.sample_collect_time
            <= _normalize_date_bound(sample_collect_time_end, end_of_day=True)
        )
    return q.filter(BizResearchDonation.id.in_(coll_q))


def _earliest_eta(eta_values: List[str]) -> str:
    cleaned = [v.strip() for v in eta_values if (v or "").strip()]
    if not cleaned:
        return ""
    return min(cleaned)


def _build_sample_items(
    samples: List[Any],
    *,
    sample_type_name_map: Dict[str, str],
    lab_code_map: Dict[str, str],
    transport_by_sample_id: Dict[str, Any],
    site_name_map: Dict[str, str],
    express_name_map: Dict[str, str],
    sample_filter,
) -> Tuple[List[Dict[str, Any]], List[str]]:
    items: List[Dict[str, Any]] = []
    eta_values: List[str] = []
    for s in samples:
        if not sample_filter(s):
            continue
        type_name = sample_type_name_map.get(getattr(s, "sample_type_id", None) or "", "")
        code = (getattr(s, "sample_code", None) or "").strip()
        transport_obj = transport_by_sample_id.get(s.id)
        transport_dict: Optional[Dict[str, Any]] = None
        if transport_obj and (transport_obj.transport_type or "").strip():
            dest_site_id = transport_obj.dest_site_id or ""
            express_company_id = transport_obj.express_company_id or ""
            eta = _fmt_dt(transport_obj.eta_time) if transport_obj.eta_time else ""
            if eta:
                eta_values.append(eta)
            transport_dict = {
                "transportType": transport_obj.transport_type or "",
                "destSiteId": dest_site_id,
                "destSiteName": site_name_map.get(dest_site_id, ""),
                "expressCompanyId": express_company_id,
                "expressCompanyName": express_name_map.get(express_company_id, ""),
                "waybillNo": transport_obj.waybill_no or "",
                "carrierName": transport_obj.carrier_name or "",
                "carrierPhone": transport_obj.carrier_phone or "",
                "etaTime": eta,
            }
        items.append(
            {
                "sampleTypeName": type_name,
                "sampleCode": code,
                "sampleStatus": (getattr(s, "sample_status", None) or "").strip(),
                "labStorageCode": lab_code_map.get(code, "") if code else "",
                "transport": transport_dict,
                **sample_special_to_dict(s),
            }
        )
    return items, eta_values


def _summarize_in_transit_items(items: List[Dict[str, Any]]) -> str:
    n = len(items)
    if n <= 0:
        return ""
    return f"{n}个已采集（待接收）"


def _summarize_pre_delivery(samples: List[Any], pre_kind: str) -> str:
    if pre_kind == "no_sample":
        return "样本未录入"
    n = sum(1 for s in samples if _is_uncollected_status(getattr(s, "sample_status", None)))
    return f"{n}个未采集" if n else "样本未采集"


def _list_in_transit_collection_ids(
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str],
    create_time_start: Optional[str],
    create_time_end: Optional[str],
    sample_collect_time_start: Optional[str],
    sample_collect_time_end: Optional[str],
    user_info,
) -> List[str]:
    q = (
        db.query(BizCollectionProtocol.id)
        .join(
            BizProtocolSample,
            and_(
                BizProtocolSample.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.sample_status == "ycj",
                not_(_cell_received_exists(tenant_no)),
            ),
        )
        .filter(
            BizCollectionProtocol.del_flag == 0,
            BizCollectionProtocol.tenant_no == tenant_no,
            BizCollectionProtocol.protocol_status.notin_(_CELL_IN_TRANSIT_EXCLUDED),
        )
    )
    q = _apply_collection_keyword(q, db, tenant_no, keyword or "")
    q = _apply_create_time_filter(q, BizCollectionProtocol, create_time_start, create_time_end)
    q = _apply_collection_sample_collect_time_filter(
        q, db, tenant_no, sample_collect_time_start, sample_collect_time_end
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizCollectionProtocol, user_info, db, tenant_no)
    rows = q.distinct().all()
    return [r[0] for r in rows]


def _list_in_transit_donation_ids(
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str],
    create_time_start: Optional[str],
    create_time_end: Optional[str],
    sample_collect_time_start: Optional[str],
    sample_collect_time_end: Optional[str],
    user_info,
) -> List[str]:
    q = (
        db.query(BizResearchDonation.id)
        .join(
            BizResearchDonationSample,
            and_(
                BizResearchDonationSample.research_donation_id == BizResearchDonation.id,
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.sample_status == "ycj",
                not_(_donation_received_exists(tenant_no)),
            ),
        )
        .filter(
            BizResearchDonation.del_flag == 0,
            BizResearchDonation.tenant_no == tenant_no,
            BizResearchDonation.donation_status.notin_(_DONATION_IN_TRANSIT_EXCLUDED),
        )
    )
    q = _apply_donation_keyword(q, db, tenant_no, keyword or "")
    q = _apply_create_time_filter(q, BizResearchDonation, create_time_start, create_time_end)
    q = _apply_donation_sample_collect_time_filter(
        q, db, tenant_no, sample_collect_time_start, sample_collect_time_end
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizResearchDonation, user_info, db, tenant_no)
    rows = q.distinct().all()
    return [r[0] for r in rows]


def _list_pre_delivery_collection_ids(
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str],
    create_time_start: Optional[str],
    create_time_end: Optional[str],
    sample_collect_time_start: Optional[str],
    sample_collect_time_end: Optional[str],
    user_info,
) -> List[Tuple[str, str]]:
    has_sample = (
        exists()
        .where(
            and_(
                BizProtocolSample.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
            )
        )
        .correlate(BizCollectionProtocol)
    )
    has_uncollected = (
        exists()
        .where(
            and_(
                BizProtocolSample.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
                or_(
                    BizProtocolSample.sample_status.is_(None),
                    BizProtocolSample.sample_status == "",
                ),
            )
        )
        .correlate(BizCollectionProtocol)
    )
    q = db.query(BizCollectionProtocol.id).filter(
        BizCollectionProtocol.del_flag == 0,
        BizCollectionProtocol.tenant_no == tenant_no,
        BizCollectionProtocol.protocol_status.in_(_PRE_DELIVERY_STATUSES),
        or_(not_(has_sample), has_uncollected),
    )
    q = _apply_collection_keyword(q, db, tenant_no, keyword or "")
    q = _apply_create_time_filter(q, BizCollectionProtocol, create_time_start, create_time_end)
    q = _apply_collection_sample_collect_time_filter(
        q, db, tenant_no, sample_collect_time_start, sample_collect_time_end
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizCollectionProtocol, user_info, db, tenant_no)
    out: List[Tuple[str, str]] = []
    for (pid,) in q.all():
        sample_count = (
            db.query(func.count(BizProtocolSample.id))
            .filter(
                BizProtocolSample.collection_protocol_id == pid,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
            )
            .scalar()
            or 0
        )
        kind = "no_sample" if sample_count == 0 else "uncollected"
        out.append((pid, kind))
    return out


def _list_pre_delivery_donation_ids(
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str],
    create_time_start: Optional[str],
    create_time_end: Optional[str],
    sample_collect_time_start: Optional[str],
    sample_collect_time_end: Optional[str],
    user_info,
) -> List[Tuple[str, str]]:
    has_sample = (
        exists()
        .where(
            and_(
                BizResearchDonationSample.research_donation_id == BizResearchDonation.id,
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.tenant_no == tenant_no,
            )
        )
        .correlate(BizResearchDonation)
    )
    has_uncollected = (
        exists()
        .where(
            and_(
                BizResearchDonationSample.research_donation_id == BizResearchDonation.id,
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.tenant_no == tenant_no,
                or_(
                    BizResearchDonationSample.sample_status.is_(None),
                    BizResearchDonationSample.sample_status == "",
                ),
            )
        )
        .correlate(BizResearchDonation)
    )
    q = db.query(BizResearchDonation.id).filter(
        BizResearchDonation.del_flag == 0,
        BizResearchDonation.tenant_no == tenant_no,
        BizResearchDonation.donation_status.in_(_PRE_DELIVERY_STATUSES),
        or_(not_(has_sample), has_uncollected),
    )
    q = _apply_donation_keyword(q, db, tenant_no, keyword or "")
    q = _apply_create_time_filter(q, BizResearchDonation, create_time_start, create_time_end)
    q = _apply_donation_sample_collect_time_filter(
        q, db, tenant_no, sample_collect_time_start, sample_collect_time_end
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizResearchDonation, user_info, db, tenant_no)
    out: List[Tuple[str, str]] = []
    for (did,) in q.all():
        sample_count = (
            db.query(func.count(BizResearchDonationSample.id))
            .filter(
                BizResearchDonationSample.research_donation_id == did,
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.tenant_no == tenant_no,
            )
            .scalar()
            or 0
        )
        kind = "no_sample" if sample_count == 0 else "uncollected"
        out.append((did, kind))
    return out


def _sort_key_in_transit(eta: str, create_time: Optional[datetime]) -> Tuple[int, str, datetime]:
    eta_clean = (eta or "").strip()
    return (0 if eta_clean else 1, eta_clean or "9999-99-99 99:99:99", create_time or datetime.min)


def _sort_key_pre_delivery(expected: str, create_time: Optional[datetime]) -> Tuple[int, str, datetime]:
    exp = (expected or "").strip()
    return (0 if exp else 1, exp or "", create_time or datetime.min)


def _hydrate_collection_rows(
    db: Session,
    *,
    tenant_no: str,
    protocol_ids: List[str],
    mode: str,
    pre_kinds: Optional[Dict[str, str]] = None,
) -> List[Dict[str, Any]]:
    if not protocol_ids:
        return []
    rows = (
        db.query(BizCollectionProtocol)
        .filter(
            BizCollectionProtocol.id.in_(protocol_ids),
            BizCollectionProtocol.del_flag == 0,
            BizCollectionProtocol.tenant_no == tenant_no,
        )
        .all()
    )
    row_map = {r.id: r for r in rows}

    pc_map: Dict[str, Dict[str, Any]] = {}
    for pc in (
        db.query(BizProtocolPartyCustomer)
        .filter(
            BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
            BizProtocolPartyCustomer.del_flag == 0,
        )
        .all()
    ):
        pc_map[pc.collection_protocol_id] = _party_customer_to_dict(pc)

    ci_map: Dict[str, Dict[str, Any]] = {}
    for ci in (
        db.query(BizProtocolCollectionInfo)
        .filter(
            BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids),
            BizProtocolCollectionInfo.del_flag == 0,
        )
        .all()
    ):
        ci_map[ci.collection_protocol_id] = _collection_info_to_dict(ci)

    region_ids = list({ci_map.get(pid, {}).get("regionId", "") for pid in protocol_ids} - {""})
    hospital_ids = list({ci_map.get(pid, {}).get("hospitalId", "") for pid in protocol_ids} - {""})
    region_name_map: Dict[str, str] = {}
    hospital_name_map: Dict[str, str] = {}
    if region_ids:
        for r in (
            db.query(BizCollectionRegion.id, BizCollectionRegion.region_name)
            .filter(BizCollectionRegion.id.in_(region_ids))
            .all()
        ):
            region_name_map[r.id] = r.region_name or ""
    if hospital_ids:
        for h in (
            db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name)
            .filter(BizCollectionHospital.id.in_(hospital_ids))
            .all()
        ):
            hospital_name_map[h.id] = h.hospital_name or ""

    sample_map: Dict[str, List[BizProtocolSample]] = {}
    s_rows = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.collection_protocol_id.in_(protocol_ids),
            BizProtocolSample.del_flag == 0,
        )
        .order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc())
        .all()
    )
    sample_type_name_map = _load_sample_type_name_map(db, s_rows)
    for s in s_rows:
        sample_map.setdefault(s.collection_protocol_id, []).append(s)

    all_codes: List[str] = []
    for samples in sample_map.values():
        for s in samples:
            c = (s.sample_code or "").strip()
            if c:
                all_codes.append(c)
    lab_code_map: Dict[str, str] = {}
    if all_codes:
        from app.crud import sample_reception as reception_svc

        lab_code_map = reception_svc.get_lab_storage_code_map_by_sample_codes(
            db, tenant_no, all_codes, reception_source="cell_storage"
        )

    transport_by_sample_id: Dict[str, BizProtocolSampleTransport] = {}
    for t in (
        db.query(BizProtocolSampleTransport)
        .filter(
            BizProtocolSampleTransport.collection_protocol_id.in_(protocol_ids),
            BizProtocolSampleTransport.tenant_no == tenant_no,
            BizProtocolSampleTransport.del_flag == 0,
        )
        .all()
    ):
        transport_by_sample_id[t.sample_id] = t
    site_name_map = _load_site_name_map(
        db, [t.dest_site_id for t in transport_by_sample_id.values() if t.dest_site_id]
    )
    express_name_map = _load_express_name_map(
        db,
        [t.express_company_id for t in transport_by_sample_id.values() if t.express_company_id],
    )

    received_codes: set = set()
    if mode == "in_transit" and all_codes:
        for rec in (
            db.query(BizSampleReception.sample_code)
            .filter(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                BizSampleReception.reception_source == "cell_storage",
                BizSampleReception.sample_code.in_(all_codes),
            )
            .all()
        ):
            if rec[0]:
                received_codes.add(rec[0])

    user_ids = list(
        {
            *(ci_map.get(pid, {}).get("salesUserId", "") for pid in protocol_ids),
            *(_protocol_to_dict(row_map[pid]).get("collectorUserId", "") for pid in protocol_ids if pid in row_map),
        }
        - {""}
    )
    user_name_map: Dict[str, str] = {}
    if user_ids:
        for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(user_ids)).all():
            user_name_map[u.id] = u.nick_name or ""

    result: List[Dict[str, Any]] = []
    for pid in protocol_ids:
        r = row_map.get(pid)
        if not r:
            continue
        d = _protocol_to_dict(r)
        d["healthSurveyAttachmentFiles"] = urls_to_attachment_list(
            r.health_survey_form_urls, db, tenant_no
        )
        d["sourceType"] = "cell_storage"
        d["sourceTypeLabel"] = "细胞存储"
        d["collectionProtocolId"] = r.id
        d["researchDonationId"] = ""
        pc = pc_map.get(pid, {})
        ci = ci_map.get(pid, {})
        d["motherName"] = pc.get("motherName", "")
        d["motherPhone"] = pc.get("motherPhone", "")
        d["motherIdCard"] = pc.get("motherIdCard", "")
        d["fatherName"] = pc.get("fatherName", "")
        d["newbornName"] = pc.get("newbornName", "")
        d["expectedDeliveryDate"] = pc.get("expectedDeliveryDate", "")
        d["newbornDeliveryTime"] = pc.get("newbornDeliveryTime", "")
        d["newbornSex"] = pc.get("newbornSex", "")
        d["deliveryMode"] = pc.get("deliveryMode", "")
        d["newbornWeightGram"] = pc.get("newbornWeightGram")
        d["regionName"] = region_name_map.get(ci.get("regionId", ""), "")
        d["hospitalName"] = hospital_name_map.get(ci.get("hospitalId", ""), "")
        d["salesUserName"] = user_name_map.get(ci.get("salesUserId", ""), "")
        d["sampleCollectTime"] = ci.get("sampleCollectTime", "")
        d["maternalBloodCollectTime"] = ci.get("maternalBloodCollectTime", "")
        d["collectorUserName"] = user_name_map.get(d.get("collectorUserId", ""), "")

        samples = sample_map.get(pid, [])

        if mode == "in_transit":

            def _flt(s):
                code = (s.sample_code or "").strip()
                return _is_in_transit_sample(s.sample_status) and code not in received_codes

        else:
            pre_kind = (pre_kinds or {}).get(pid, "no_sample")

            def _flt(s):
                if pre_kind == "no_sample":
                    return False
                return _is_uncollected_status(s.sample_status)

        items, eta_values = _build_sample_items(
            samples,
            sample_type_name_map=sample_type_name_map,
            lab_code_map=lab_code_map,
            transport_by_sample_id=transport_by_sample_id,
            site_name_map=site_name_map,
            express_name_map=express_name_map,
            sample_filter=_flt,
        )
        type_names = [it["sampleTypeName"] for it in items if it.get("sampleTypeName")]
        codes = [it["sampleCode"] for it in items if it.get("sampleCode")]
        d["sampleItems"] = items
        d["sampleTypeNames"] = ",".join(type_names)
        d["sampleCodes"] = ",".join(codes)
        d["labStorageCodes"] = ",".join(
            (it.get("labStorageCode") or "") for it in items if it.get("labStorageCode")
        )
        if mode == "in_transit":
            d["sampleCollectSummary"] = _summarize_in_transit_items(items)
            d["etaTimeSummary"] = _earliest_eta(eta_values)
        else:
            pre_kind = (pre_kinds or {}).get(pid, "no_sample")
            d["preDeliveryKind"] = pre_kind
            d["preDeliveryKindLabel"] = "样本未录入" if pre_kind == "no_sample" else "样本未采集"
            d["sampleCollectSummary"] = _summarize_pre_delivery(samples, pre_kind)
            d["etaTimeSummary"] = ""
        result.append(d)
    return result


def _hydrate_donation_rows(
    db: Session,
    *,
    tenant_no: str,
    donation_ids: List[str],
    mode: str,
    pre_kinds: Optional[Dict[str, str]] = None,
) -> List[Dict[str, Any]]:
    if not donation_ids:
        return []
    rows = (
        db.query(BizResearchDonation)
        .filter(
            BizResearchDonation.id.in_(donation_ids),
            BizResearchDonation.del_flag == 0,
            BizResearchDonation.tenant_no == tenant_no,
        )
        .all()
    )
    row_map = {r.id: r for r in rows}

    party_map: Dict[str, Dict[str, Any]] = {}
    for p in (
        db.query(BizResearchDonationParty)
        .filter(
            BizResearchDonationParty.research_donation_id.in_(donation_ids),
            BizResearchDonationParty.del_flag == 0,
        )
        .all()
    ):
        party_map[p.research_donation_id] = _party_to_dict(p)

    coll_map: Dict[str, Dict[str, Any]] = {}
    for c in (
        db.query(BizResearchDonationCollection)
        .filter(
            BizResearchDonationCollection.research_donation_id.in_(donation_ids),
            BizResearchDonationCollection.del_flag == 0,
        )
        .all()
    ):
        coll_map[c.research_donation_id] = _collection_to_dict(c)

    region_ids = list({coll_map.get(did, {}).get("regionId", "") for did in donation_ids} - {""})
    hospital_ids = list({coll_map.get(did, {}).get("hospitalId", "") for did in donation_ids} - {""})
    region_name_map: Dict[str, str] = {}
    hospital_name_map: Dict[str, str] = {}
    if region_ids:
        for r in (
            db.query(BizCollectionRegion.id, BizCollectionRegion.region_name)
            .filter(BizCollectionRegion.id.in_(region_ids))
            .all()
        ):
            region_name_map[r.id] = r.region_name or ""
    if hospital_ids:
        for h in (
            db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name)
            .filter(BizCollectionHospital.id.in_(hospital_ids))
            .all()
        ):
            hospital_name_map[h.id] = h.hospital_name or ""

    sample_map: Dict[str, List[BizResearchDonationSample]] = {}
    s_rows = (
        db.query(BizResearchDonationSample)
        .filter(
            BizResearchDonationSample.research_donation_id.in_(donation_ids),
            BizResearchDonationSample.del_flag == 0,
        )
        .order_by(BizResearchDonationSample.order_num.asc(), BizResearchDonationSample.create_time.asc())
        .all()
    )
    sample_type_name_map = _load_sample_type_name_map(db, s_rows)
    for s in s_rows:
        sample_map.setdefault(s.research_donation_id, []).append(s)

    all_codes: List[str] = []
    for samples in sample_map.values():
        for s in samples:
            c = (s.sample_code or "").strip()
            if c:
                all_codes.append(c)
    lab_code_map: Dict[str, str] = {}
    if all_codes:
        from app.crud import sample_reception as reception_svc

        lab_code_map = reception_svc.get_lab_storage_code_map_by_sample_codes(
            db, tenant_no, all_codes, reception_source="research_donation"
        )

    transport_by_sample_id: Dict[str, BizResearchDonationSampleTransport] = {}
    for t in (
        db.query(BizResearchDonationSampleTransport)
        .filter(
            BizResearchDonationSampleTransport.research_donation_id.in_(donation_ids),
            BizResearchDonationSampleTransport.tenant_no == tenant_no,
            BizResearchDonationSampleTransport.del_flag == 0,
        )
        .all()
    ):
        transport_by_sample_id[t.sample_id] = t
    site_name_map = _load_site_name_map(
        db, [t.dest_site_id for t in transport_by_sample_id.values() if t.dest_site_id]
    )
    express_name_map = _load_express_name_map(
        db,
        [t.express_company_id for t in transport_by_sample_id.values() if t.express_company_id],
    )

    received_codes: set = set()
    if mode == "in_transit" and all_codes:
        for rec in (
            db.query(BizSampleReception.sample_code)
            .filter(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                BizSampleReception.reception_source == "research_donation",
                BizSampleReception.sample_code.in_(all_codes),
            )
            .all()
        ):
            if rec[0]:
                received_codes.add(rec[0])

    user_ids = list(
        {
            *(coll_map.get(did, {}).get("salesUserId", "") for did in donation_ids),
            *(row_map[did].collector_user_id or "" for did in donation_ids if did in row_map),
        }
        - {""}
    )
    user_name_map: Dict[str, str] = {}
    if user_ids:
        for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(user_ids)).all():
            user_name_map[u.id] = u.nick_name or ""

    result: List[Dict[str, Any]] = []
    for did in donation_ids:
        r = row_map.get(did)
        if not r:
            continue
        d = _donation_to_dict(r, db, tenant_no)
        d["sourceType"] = "research_donation"
        d["sourceTypeLabel"] = "科研捐赠"
        d["collectionProtocolId"] = ""
        d["researchDonationId"] = r.id
        d["protocolNo"] = d.get("donationNo", "")
        d["protocolStatus"] = d.get("donationStatus", "")
        d["registerDate"] = d.get("donationDate", "")
        party = party_map.get(did, {})
        coll = coll_map.get(did, {})
        d["motherName"] = party.get("motherName", "")
        d["motherPhone"] = party.get("motherPhone", "")
        d["motherIdCard"] = party.get("motherIdCard", "")
        d["fatherName"] = party.get("fatherName", "")
        d["newbornName"] = party.get("newbornName", "")
        d["expectedDeliveryDate"] = party.get("expectedDeliveryDate", "")
        d["newbornDeliveryTime"] = party.get("newbornDeliveryTime", "")
        d["newbornSex"] = party.get("newbornSex", "")
        d["deliveryMode"] = party.get("deliveryMode", "")
        d["newbornWeightGram"] = party.get("newbornWeightGram")
        d["regionName"] = region_name_map.get(coll.get("regionId", ""), "")
        d["hospitalName"] = hospital_name_map.get(coll.get("hospitalId", ""), "")
        d["salesUserName"] = user_name_map.get(coll.get("salesUserId", ""), "")
        d["sampleCollectTime"] = coll.get("sampleCollectTime", "")
        d["maternalBloodCollectTime"] = coll.get("maternalBloodCollectTime", "")
        d["collectorUserName"] = user_name_map.get(r.collector_user_id or "", "")

        samples = sample_map.get(did, [])

        if mode == "in_transit":

            def _flt(s):
                code = (s.sample_code or "").strip()
                return _is_in_transit_sample(s.sample_status) and code not in received_codes

        else:
            pre_kind = (pre_kinds or {}).get(did, "no_sample")

            def _flt(s):
                if pre_kind == "no_sample":
                    return False
                return _is_uncollected_status(s.sample_status)

        items, eta_values = _build_sample_items(
            samples,
            sample_type_name_map=sample_type_name_map,
            lab_code_map=lab_code_map,
            transport_by_sample_id=transport_by_sample_id,
            site_name_map=site_name_map,
            express_name_map=express_name_map,
            sample_filter=_flt,
        )
        type_names = [it["sampleTypeName"] for it in items if it.get("sampleTypeName")]
        codes = [it["sampleCode"] for it in items if it.get("sampleCode")]
        d["sampleItems"] = items
        d["sampleTypeNames"] = ",".join(type_names)
        d["sampleCodes"] = ",".join(codes)
        d["labStorageCodes"] = ",".join(
            (it.get("labStorageCode") or "") for it in items if it.get("labStorageCode")
        )
        if mode == "in_transit":
            d["sampleCollectSummary"] = _summarize_in_transit_items(items)
            d["etaTimeSummary"] = _earliest_eta(eta_values)
        else:
            pre_kind = (pre_kinds or {}).get(did, "no_sample")
            d["preDeliveryKind"] = pre_kind
            d["preDeliveryKindLabel"] = "样本未录入" if pre_kind == "no_sample" else "样本未采集"
            d["sampleCollectSummary"] = _summarize_pre_delivery(samples, pre_kind)
            d["etaTimeSummary"] = ""
        result.append(d)
    return result


def list_in_transit_protocols(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    tenant_no: str,
    keyword: Optional[str] = None,
    source_type: Optional[str] = None,
    create_time_start: Optional[str] = None,
    create_time_end: Optional[str] = None,
    sample_collect_time_start: Optional[str] = None,
    sample_collect_time_end: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0

    candidates: List[Tuple[str, str, str, Optional[datetime]]] = []

    if not source_type or source_type == "cell_storage":
        for pid in _list_in_transit_collection_ids(
            db,
            tenant_no=tenant_no,
            keyword=keyword,
            create_time_start=create_time_start,
            create_time_end=create_time_end,
            sample_collect_time_start=sample_collect_time_start,
            sample_collect_time_end=sample_collect_time_end,
            user_info=user_info,
        ):
            row = (
                db.query(BizCollectionProtocol.create_time)
                .filter(BizCollectionProtocol.id == pid)
                .first()
            )
            candidates.append(("cell_storage", pid, "", row[0] if row else None))

    if not source_type or source_type == "research_donation":
        for did in _list_in_transit_donation_ids(
            db,
            tenant_no=tenant_no,
            keyword=keyword,
            create_time_start=create_time_start,
            create_time_end=create_time_end,
            sample_collect_time_start=sample_collect_time_start,
            sample_collect_time_end=sample_collect_time_end,
            user_info=user_info,
        ):
            row = (
                db.query(BizResearchDonation.create_time)
                .filter(BizResearchDonation.id == did)
                .first()
            )
            candidates.append(("research_donation", did, "", row[0] if row else None))

    if not candidates:
        return [], 0

    # 预计算 eta 用于排序
    cell_ids = [pid for st, pid, _, _ in candidates if st == "cell_storage"]
    donation_ids = [did for st, did, _, _ in candidates if st == "research_donation"]
    eta_by_protocol: Dict[str, str] = {}

    if cell_ids:
        for pid, eta in (
            db.query(
                BizProtocolSample.collection_protocol_id,
                func.min(BizProtocolSampleTransport.eta_time),
            )
            .join(
                BizProtocolSampleTransport,
                and_(
                    BizProtocolSampleTransport.sample_id == BizProtocolSample.id,
                    BizProtocolSampleTransport.del_flag == 0,
                ),
            )
            .filter(
                BizProtocolSample.collection_protocol_id.in_(cell_ids),
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.sample_status == "ycj",
                not_(_cell_received_exists(tenant_no)),
            )
            .group_by(BizProtocolSample.collection_protocol_id)
            .all()
        ):
            eta_by_protocol[f"cell_storage:{pid}"] = _fmt_dt(eta) if eta else ""

    if donation_ids:
        for did, eta in (
            db.query(
                BizResearchDonationSample.research_donation_id,
                func.min(BizResearchDonationSampleTransport.eta_time),
            )
            .join(
                BizResearchDonationSampleTransport,
                and_(
                    BizResearchDonationSampleTransport.sample_id == BizResearchDonationSample.id,
                    BizResearchDonationSampleTransport.del_flag == 0,
                ),
            )
            .filter(
                BizResearchDonationSample.research_donation_id.in_(donation_ids),
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.sample_status == "ycj",
                not_(_donation_received_exists(tenant_no)),
            )
            .group_by(BizResearchDonationSample.research_donation_id)
            .all()
        ):
            eta_by_protocol[f"research_donation:{did}"] = _fmt_dt(eta) if eta else ""

    enriched: List[Tuple[str, str, str, Optional[datetime]]] = []
    for st, pid, _, ct in candidates:
        eta = eta_by_protocol.get(f"{st}:{pid}", "")
        enriched.append((st, pid, eta, ct))

    enriched.sort(key=lambda x: _sort_key_in_transit(x[2], x[3]))
    total = len(enriched)
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    page_slice = enriched[(page - 1) * page_size : page * page_size]

    cell_page = [pid for st, pid, _, _ in page_slice if st == "cell_storage"]
    donation_page = [pid for st, pid, _, _ in page_slice if st == "research_donation"]

    items: List[Dict[str, Any]] = []
    items.extend(
        _hydrate_collection_rows(db, tenant_no=tenant_no, protocol_ids=cell_page, mode="in_transit")
    )
    items.extend(
        _hydrate_donation_rows(db, tenant_no=tenant_no, donation_ids=donation_page, mode="in_transit")
    )

    order_keys = [f"{st}:{pid}" for st, pid, _, _ in page_slice]
    item_map = {f"{it['sourceType']}:{it.get('collectionProtocolId') or it.get('researchDonationId')}": it for it in items}
    ordered = [item_map[k] for k in order_keys if k in item_map]
    return ordered, total


def list_pre_delivery_protocols(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    tenant_no: str,
    keyword: Optional[str] = None,
    source_type: Optional[str] = None,
    create_time_start: Optional[str] = None,
    create_time_end: Optional[str] = None,
    sample_collect_time_start: Optional[str] = None,
    sample_collect_time_end: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0

    candidates: List[Tuple[str, str, str, Optional[datetime], str]] = []

    if not source_type or source_type == "cell_storage":
        for pid, kind in _list_pre_delivery_collection_ids(
            db,
            tenant_no=tenant_no,
            keyword=keyword,
            create_time_start=create_time_start,
            create_time_end=create_time_end,
            sample_collect_time_start=sample_collect_time_start,
            sample_collect_time_end=sample_collect_time_end,
            user_info=user_info,
        ):
            proto = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == pid).first()
            party = (
                db.query(BizProtocolPartyCustomer.expected_delivery_date)
                .filter(
                    BizProtocolPartyCustomer.collection_protocol_id == pid,
                    BizProtocolPartyCustomer.del_flag == 0,
                )
                .first()
            )
            exp = ""
            if party and party[0]:
                from app.core.utils import format_date

                exp = format_date(party[0])
            candidates.append(
                (
                    "cell_storage",
                    pid,
                    exp,
                    proto.create_time if proto else None,
                    kind,
                )
            )

    if not source_type or source_type == "research_donation":
        for did, kind in _list_pre_delivery_donation_ids(
            db,
            tenant_no=tenant_no,
            keyword=keyword,
            create_time_start=create_time_start,
            create_time_end=create_time_end,
            sample_collect_time_start=sample_collect_time_start,
            sample_collect_time_end=sample_collect_time_end,
            user_info=user_info,
        ):
            donation = db.query(BizResearchDonation).filter(BizResearchDonation.id == did).first()
            party = (
                db.query(BizResearchDonationParty.expected_delivery_date)
                .filter(
                    BizResearchDonationParty.research_donation_id == did,
                    BizResearchDonationParty.del_flag == 0,
                )
                .first()
            )
            exp = ""
            if party and party[0]:
                from app.core.utils import format_date

                exp = format_date(party[0])
            candidates.append(
                (
                    "research_donation",
                    did,
                    exp,
                    donation.create_time if donation else None,
                    kind,
                )
            )

    if not candidates:
        return [], 0

    candidates.sort(
        key=lambda x: (
            (x[2] or "").strip() or "0000-00-00",
            x[3] or datetime.min,
        ),
        reverse=True,
    )
    total = len(candidates)
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    page_slice = candidates[(page - 1) * page_size : page * page_size]

    cell_page = [pid for st, pid, _, _, _ in page_slice if st == "cell_storage"]
    donation_page = [did for st, did, _, _, _ in page_slice if st == "research_donation"]
    pre_kinds_cell = {pid: kind for st, pid, _, _, kind in page_slice if st == "cell_storage"}
    pre_kinds_donation = {did: kind for st, did, _, _, kind in page_slice if st == "research_donation"}

    items: List[Dict[str, Any]] = []
    items.extend(
        _hydrate_collection_rows(
            db,
            tenant_no=tenant_no,
            protocol_ids=cell_page,
            mode="pre_delivery",
            pre_kinds=pre_kinds_cell,
        )
    )
    items.extend(
        _hydrate_donation_rows(
            db,
            tenant_no=tenant_no,
            donation_ids=donation_page,
            mode="pre_delivery",
            pre_kinds=pre_kinds_donation,
        )
    )

    order_keys = [f"{st}:{pid}" for st, pid, _, _, _ in page_slice]
    item_map = {f"{it['sourceType']}:{it.get('collectionProtocolId') or it.get('researchDonationId')}": it for it in items}
    ordered = [item_map[k] for k in order_keys if k in item_map]
    return ordered, total


__all__ = ["list_in_transit_protocols", "list_pre_delivery_protocols", "total_pages"]