zkxw/backend/app/crud/sample_reception.py

"""样本接收 CRUD（质量检测）"""
from __future__ import annotations

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

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

from app.core.utils import generate_uuid
from app.crud.collection_protocol import urls_to_attachment_list
from app.utils.sample_special import sample_special_to_dict
from app.crud import sample_reception_check_result as check_result_svc
from app.crud import sample_production_reception as prod_reception_svc
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizExpressCompany,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizProtocolSampleTransport,
    BizSampleReception,
    SysUser,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationParty,
    BizResearchDonationSample,
    BizResearchDonationSampleTransport,
)
from app.core.protocol_status_display import INACTIVE_PROTOCOL_STATUSES
from app.services.web_auth_roles import get_user_role_names

SAMPLE_RECEPTION_EDIT_ROLE = "质量主管"

# 科研捐赠表默认 general_ci，样本接收等为 unicode_ci，跨表比较需统一
_UNICODE_CI = "utf8mb4_unicode_ci"


def _donation_sample_id_col(col):
    """捐赠样本 ID 与 biz_sample_reception.donation_sample_id 比较时统一排序规则。"""
    return col.collate(_UNICODE_CI)


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


def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
    val = (value or "").strip()
    if len(val) == 10:
        return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
    return val


def _dec(v: Any) -> Optional[float]:
    if v is None:
        return None
    return float(v)


def _reception_to_dict(
    r: BizSampleReception,
    *,
    check_results: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    return {
        "id": r.id,
        "protocolSampleId": r.protocol_sample_id,
        "collectionProtocolId": r.collection_protocol_id,
        "receptionTime": _fmt_dt(r.reception_time),
        "handlerUserId": r.handler_user_id or "",
        "handlerName": r.handler_name or "",
        "receiveTemperature": r.receive_temperature or "",
        "checkResults": check_results or [],
        "acceptConclusion": r.accept_conclusion or "",
        "labStorageCode": r.lab_storage_code or "",
        "receiveAttachments": r.receive_attachments or "",
        "remark": r.remark or "",
        "createTime": _fmt_dt(r.create_time),
        "updateTime": _fmt_dt(r.update_time),
    }


def _sample_detail_rows_query(db: Session):
    """协议样本 + 协议单 + 客户 + 运输（显式 select_from 避免多表 join 歧义）。"""
    return (
        db.query(
            BizProtocolSample,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
            BizProtocolSampleTransport,
        )
        .select_from(BizProtocolSample)
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolSampleTransport,
            and_(
                BizProtocolSampleTransport.sample_id == BizProtocolSample.id,
                BizProtocolSampleTransport.del_flag == 0,
            ),
        )
    )


def _donation_sample_detail_rows_query(db: Session):
    """捐赠样本 + 捐赠单 + 当事人 + 运输（四表 JOIN）。"""
    return (
        db.query(
            BizResearchDonationSample,
            BizResearchDonation,
            BizResearchDonationParty,
            BizResearchDonationSampleTransport,
        )
        .select_from(BizResearchDonationSample)
        .join(
            BizResearchDonation,
            and_(
                BizResearchDonation.id == BizResearchDonationSample.research_donation_id,
                BizResearchDonation.del_flag == 0,
            ),
        )
        .outerjoin(
            BizResearchDonationParty,
            and_(
                BizResearchDonationParty.research_donation_id == BizResearchDonation.id,
                BizResearchDonationParty.del_flag == 0,
            ),
        )
        .outerjoin(
            BizResearchDonationSampleTransport,
            and_(
                BizResearchDonationSampleTransport.sample_id == BizResearchDonationSample.id,
                BizResearchDonationSampleTransport.del_flag == 0,
            ),
        )
    )


def _donation_sample_pending_query(db: Session, tenant_no: str, keyword: Optional[str] = None):
    """待接收：捐赠样本且尚无有效接收记录（按租户 + 样本编码判定，与细胞存储一致）；排除已作废捐赠单。"""
    rec = BizSampleReception
    received_exists = (
        db.query(rec.id)
        .filter(
            rec.tenant_no == tenant_no,
            rec.del_flag == 0,
            rec.reception_source == "research_donation",
            rec.sample_code.isnot(None),
            rec.sample_code != "",
            rec.sample_code == BizResearchDonationSample.sample_code,
        )
        .exists()
    )
    q = _donation_sample_detail_rows_query(db).filter(
        BizResearchDonationSample.del_flag == 0,
        BizResearchDonationSample.tenant_no == tenant_no,
        BizResearchDonation.tenant_no == tenant_no,
        BizResearchDonation.donation_status.notin_(INACTIVE_PROTOCOL_STATUSES),
        ~received_exists,
    )
    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        kw_conds = [
            BizResearchDonationSample.sample_code.like(kw),
            BizResearchDonationParty.mother_name.like(kw),
            BizResearchDonationParty.newborn_name.like(kw),
        ]
        lab_codes = sample_codes_matching_lab_storage_keyword(
            db, tenant_no, keyword, reception_source="research_donation"
        )
        if lab_codes:
            kw_conds.append(BizResearchDonationSample.sample_code.in_(lab_codes))
        q = q.filter(or_(*kw_conds))
    return q


def _donation_rows_to_sample_list_items(
    db: Session,
    rows: list,
    tenant_no: str,
    *,
    reception_map: Optional[Dict[str, BizSampleReception]] = None,
) -> List[Dict[str, Any]]:
    """将捐赠样本行转为统一列表项字典（字段名与细胞存储保持一致）。"""
    type_ids = list({s.sample_type_id for s, *_ in rows if s.sample_type_id})
    type_map: Dict[str, str] = {}
    if type_ids:
        for t in (
            db.query(BizCellSampleType)
            .filter(
                BizCellSampleType.id.in_(type_ids),
                BizCellSampleType.del_flag == 0,
            )
            .all()
        ):
            type_map[t.id] = t.type_name or ""

    express_ids = list(
        {tr.express_company_id for *_, tr in rows if tr and tr.express_company_id}
    )
    express_map: Dict[str, str] = {}
    if express_ids:
        for e in (
            db.query(BizExpressCompany)
            .filter(BizExpressCompany.id.in_(express_ids), BizExpressCompany.del_flag == 0)
            .all()
        ):
            express_map[e.id] = e.express_name or ""

    donation_ids = list({donation.id for _, donation, _, _ in rows if donation and donation.id})
    collect_info_map: Dict[str, Dict[str, str]] = {}
    if donation_ids:
        for coll in (
            db.query(BizResearchDonationCollection)
            .filter(
                BizResearchDonationCollection.research_donation_id.in_(donation_ids),
                BizResearchDonationCollection.del_flag == 0,
            )
            .all()
        ):
            collect_info_map[coll.research_donation_id] = {
                "sampleCollectTime": _fmt_dt(coll.sample_collect_time),
                "maternalBloodCollectTime": _fmt_dt(coll.maternal_blood_collect_time),
            }

    reception_ids = (
        [reception_map[s.id].id for s, *_ in rows if s.id in reception_map]
        if reception_map
        else []
    )
    results_map = (
        check_result_svc.list_results_by_reception_ids(db, reception_ids, tenant_no)
        if reception_ids
        else {}
    )

    items: List[Dict[str, Any]] = []
    for sample, donation, party, transport in rows:
        item: Dict[str, Any] = {
            "protocolSampleId": sample.id,
            "collectionProtocolId": donation.id,
            "protocolNo": donation.donation_no or "",
            "protocolStatus": donation.donation_status or "",
            "sampleCode": sample.sample_code or "",
            "sampleTypeId": sample.sample_type_id or "",
            "sampleTypeName": type_map.get(sample.sample_type_id or "", ""),
            "collectAmount": _dec(sample.collect_amount),
            "motherName": (party.mother_name if party else "") or "",
            "newbornName": (party.newborn_name if party else "") or "",
            "sampleCollectTime": (
                collect_info_map.get(donation.id, {}).get("sampleCollectTime", "")
                if donation
                else ""
            ),
            "maternalBloodCollectTime": (
                collect_info_map.get(donation.id, {}).get("maternalBloodCollectTime", "")
                if donation
                else ""
            ),
            "sourceType": "research_donation",
            "sourceTypeLabel": "科研捐赠",
            "infectiousDiseaseReportAttachmentFiles": urls_to_attachment_list(
                donation.infectious_disease_report_urls if donation else None, db, tenant_no
            ),
            "etaTime": _fmt_dt(transport.eta_time) if transport and transport.eta_time else "",
            "transportType": (transport.transport_type if transport else "") or "",
            "waybillNo": (transport.waybill_no if transport else "") or "",
            "expressCompanyId": (transport.express_company_id if transport else "") or "",
            "expressCompanyName": express_map.get(
                (transport.express_company_id if transport else "") or "", ""
            ),
            "carrierName": (transport.carrier_name if transport else "") or "",
            "carrierPhone": (transport.carrier_phone if transport else "") or "",
            **sample_special_to_dict(sample),
        }
        if reception_map and sample.id in reception_map:
            rec = reception_map[sample.id]
            cr = results_map.get(rec.id)
            if not cr:
                cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
            item.update(_reception_to_dict(rec, check_results=cr))
            item["receiveAttachmentFiles"] = urls_to_attachment_list(
                rec.receive_attachments, db, tenant_no
            )
        items.append(item)
    rec_ids = [str(item.get("id") or "") for item in items if item.get("id")]
    if rec_ids:
        production_map = prod_reception_svc.get_map_by_reception_ids(db, rec_ids, tenant_no)
        for item in items:
            rid = str(item.get("id") or "")
            item["productionReception"] = production_map.get(rid) if rid else None
    return items


def _sample_pending_query(db: Session, tenant_no: str, keyword: Optional[str] = None):
    """待接收：协议样本且尚无有效接收记录（仅按租户 + 样本编码判定）；排除已作废采集协议单。"""
    rec = BizSampleReception
    received_exists = (
        db.query(rec.id)
        .filter(
            rec.tenant_no == tenant_no,
            rec.del_flag == 0,
            rec.reception_source == "cell_storage",
            rec.sample_code.isnot(None),
            rec.sample_code != "",
            rec.sample_code == BizProtocolSample.sample_code,
        )
        .exists()
    )
    q = _sample_detail_rows_query(db).filter(
        BizProtocolSample.del_flag == 0,
        BizProtocolSample.tenant_no == tenant_no,
        BizCollectionProtocol.tenant_no == tenant_no,
        BizCollectionProtocol.protocol_status.notin_(INACTIVE_PROTOCOL_STATUSES),
        ~received_exists,
    )
    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        kw_conds = [
            BizProtocolSample.sample_code.like(kw),
            BizProtocolPartyCustomer.mother_name.like(kw),
            BizProtocolPartyCustomer.newborn_name.like(kw),
        ]
        lab_codes = sample_codes_matching_lab_storage_keyword(
            db, tenant_no, keyword, reception_source="cell_storage"
        )
        if lab_codes:
            kw_conds.append(BizProtocolSample.sample_code.in_(lab_codes))
        q = q.filter(or_(*kw_conds))
    return q


def _rows_to_sample_list_items(
    db: Session,
    rows: list,
    tenant_no: str,
    *,
    reception_map: Optional[Dict[str, BizSampleReception]] = None,
) -> List[Dict[str, Any]]:
    type_ids = list({s.sample_type_id for s, *_ in rows if s.sample_type_id})
    type_map: Dict[str, str] = {}
    if type_ids:
        for t in (
            db.query(BizCellSampleType)
            .filter(
                BizCellSampleType.id.in_(type_ids),
                BizCellSampleType.del_flag == 0,
            )
            .all()
        ):
            type_map[t.id] = t.type_name or ""

    express_ids = list(
        {tr.express_company_id for *_, tr in rows if tr and tr.express_company_id}
    )
    express_map: Dict[str, str] = {}
    if express_ids:
        for e in (
            db.query(BizExpressCompany)
            .filter(BizExpressCompany.id.in_(express_ids), BizExpressCompany.del_flag == 0)
            .all()
        ):
            express_map[e.id] = e.express_name or ""

    protocol_ids = list({protocol.id for _, protocol, _, _ in rows if protocol and protocol.id})
    collect_info_map: Dict[str, Dict[str, str]] = {}
    payment_summary_map: Dict[str, Dict[str, Any]] = {}
    if protocol_ids:
        from app.services.qc_work_order_fee_summary import build_protocol_payment_summaries

        payment_summary_map = build_protocol_payment_summaries(db, protocol_ids, tenant_no)
        for coll in (
            db.query(BizProtocolCollectionInfo)
            .filter(
                BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids),
                BizProtocolCollectionInfo.del_flag == 0,
            )
            .all()
        ):
            collect_info_map[coll.collection_protocol_id] = {
                "sampleCollectTime": _fmt_dt(coll.sample_collect_time),
                "maternalBloodCollectTime": _fmt_dt(coll.maternal_blood_collect_time),
            }

    def _map_rec(sample: BizProtocolSample) -> Optional[BizSampleReception]:
        if not reception_map:
            return None
        code = (sample.sample_code or "").strip()
        if code and code in reception_map:
            return reception_map[code]
        return None

    matched_recs = [_map_rec(s) for s, *_ in rows]
    reception_ids = [r.id for r in matched_recs if r]
    results_map = (
        check_result_svc.list_results_by_reception_ids(db, reception_ids, tenant_no)
        if reception_ids
        else {}
    )

    items: List[Dict[str, Any]] = []
    for sample, protocol, party, transport in rows:
        item: Dict[str, Any] = {
            "protocolSampleId": sample.id,
            "collectionProtocolId": protocol.id,
            "protocolNo": protocol.protocol_no or "",
            "protocolStatus": protocol.protocol_status or "",
            "sampleCode": sample.sample_code or "",
            "sampleTypeId": sample.sample_type_id or "",
            "sampleTypeName": type_map.get(sample.sample_type_id or "", ""),
            "collectAmount": _dec(sample.collect_amount),
            "sampleStatus": sample.sample_status or "",
            "motherName": (party.mother_name if party else "") or "",
            "newbornName": (party.newborn_name if party else "") or "",
            "sampleCollectTime": (
                collect_info_map.get(protocol.id, {}).get("sampleCollectTime", "")
                if protocol
                else ""
            ),
            "maternalBloodCollectTime": (
                collect_info_map.get(protocol.id, {}).get("maternalBloodCollectTime", "")
                if protocol
                else ""
            ),
            "etaTime": _fmt_dt(transport.eta_time) if transport and transport.eta_time else "",
            "transportType": (transport.transport_type if transport else "") or "",
            "waybillNo": (transport.waybill_no if transport else "") or "",
            "expressCompanyId": (transport.express_company_id if transport else "") or "",
            "expressCompanyName": express_map.get(
                (transport.express_company_id if transport else "") or "", ""
            ),
            "carrierName": (transport.carrier_name if transport else "") or "",
            "carrierPhone": (transport.carrier_phone if transport else "") or "",
            **sample_special_to_dict(sample),
            "sourceType": "cell_storage",
            "sourceTypeLabel": "细胞存储",
            "healthSurveyAttachmentFiles": urls_to_attachment_list(
                protocol.health_survey_form_urls if protocol else None, db, tenant_no
            ),
            "paymentSummary": payment_summary_map.get(protocol.id) if protocol else None,
        }
        rec = _map_rec(sample)
        if rec:
            cr = results_map.get(rec.id)
            if not cr:
                cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
            item.update(_reception_to_dict(rec, check_results=cr))
            item["receiveAttachmentFiles"] = urls_to_attachment_list(
                rec.receive_attachments, db, tenant_no
            )
        items.append(item)
    reception_ids = [str(item.get("id") or "") for item in items if item.get("id")]
    if reception_ids:
        production_map = prod_reception_svc.get_map_by_reception_ids(db, reception_ids, tenant_no)
        for item in items:
            rid = str(item.get("id") or "")
            item["productionReception"] = production_map.get(rid) if rid else None
    return items


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

    all_items: List[Dict[str, Any]] = []

    if not source_type or source_type == "cell_storage":
        q_base = _sample_pending_query(db, tenant_no, keyword)
        if user_info:
            from app.core.data_scope import apply_data_scope
            q_base = apply_data_scope(q_base, BizProtocolSample, user_info, db, tenant_no)
        cell_rows = (
            q_base.order_by(
                BizProtocolSampleTransport.eta_time.is_(None),
                BizProtocolSampleTransport.eta_time.asc(),
                BizProtocolSample.create_time.asc(),
            )
            .all()
        )
        all_items.extend(_rows_to_sample_list_items(db, cell_rows, tenant_no))

    if not source_type or source_type == "research_donation":
        q_donation = _donation_sample_pending_query(db, tenant_no, keyword)
        if user_info:
            from app.core.data_scope import apply_data_scope
            q_donation = apply_data_scope(q_donation, BizResearchDonationSample, user_info, db, tenant_no)
        donation_rows = (
            q_donation.order_by(
                BizResearchDonationSampleTransport.eta_time.is_(None),
                BizResearchDonationSampleTransport.eta_time.asc(),
                BizResearchDonationSample.create_time.asc(),
            )
            .all()
        )
        all_items.extend(_donation_rows_to_sample_list_items(db, donation_rows, tenant_no))

    # 统一按 etaTime 排序（空值排后）
    all_items.sort(key=lambda x: (not x.get("etaTime"), x.get("etaTime") or ""))

    start_eta = _normalize_date_bound(eta_time_start) if eta_time_start else ""
    end_eta = _normalize_date_bound(eta_time_end, end_of_day=True) if eta_time_end else ""
    if start_eta or end_eta:
        filtered: List[Dict[str, Any]] = []
        for item in all_items:
            eta = (item.get("etaTime") or "").strip()
            if not eta:
                continue
            if start_eta and eta < start_eta:
                continue
            if end_eta and eta > end_eta:
                continue
            filtered.append(item)
        all_items = filtered

    total = len(all_items)
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    start = (page - 1) * page_size
    page_items = all_items[start:start + page_size]
    return page_items, int(total)


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

    q_rec = db.query(BizSampleReception).filter(
        BizSampleReception.tenant_no == tenant_no,
        BizSampleReception.del_flag == 0,
    )
    if source_type:
        q_rec = q_rec.filter(BizSampleReception.reception_source == source_type)

    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        cell_sample_ids_kw = (
            _sample_detail_rows_query(db)
            .with_entities(BizProtocolSample.id)
            .filter(
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
                BizCollectionProtocol.tenant_no == tenant_no,
                or_(
                    BizProtocolSample.sample_code.like(kw),
                    BizProtocolPartyCustomer.mother_name.like(kw),
                    BizProtocolPartyCustomer.newborn_name.like(kw),
                ),
            )
            .subquery()
        )
        donation_sample_ids_kw = (
            _donation_sample_detail_rows_query(db)
            .with_entities(_donation_sample_id_col(BizResearchDonationSample.id))
            .filter(
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonation.tenant_no == tenant_no,
                or_(
                    BizResearchDonationSample.sample_code.like(kw),
                    BizResearchDonationParty.mother_name.like(kw),
                    BizResearchDonationParty.newborn_name.like(kw),
                ),
            )
            .subquery()
        )
        q_rec = q_rec.filter(
            or_(
                BizSampleReception.handler_name.like(kw),
                BizSampleReception.lab_storage_code.like(kw),
                BizSampleReception.protocol_sample_id.in_(cell_sample_ids_kw),
                BizSampleReception.donation_sample_id.in_(donation_sample_ids_kw),
            )
        )

    if user_info:
        from app.core.data_scope import apply_data_scope
        q_rec = apply_data_scope(q_rec, BizSampleReception, user_info, db, tenant_no)

    if reception_time_start:
        q_rec = q_rec.filter(
            BizSampleReception.reception_time >= _normalize_date_bound(reception_time_start)
        )
    if reception_time_end:
        q_rec = q_rec.filter(
            BizSampleReception.reception_time
            <= _normalize_date_bound(reception_time_end, end_of_day=True)
        )

    total = q_rec.count()
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    receptions = (
        q_rec.order_by(BizSampleReception.reception_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    if not receptions:
        return [], int(total)

    # 按 reception_source 分组查询详情
    cell_receptions = [r for r in receptions if r.reception_source == "cell_storage"]
    donation_receptions = [r for r in receptions if r.reception_source == "research_donation"]

    items: List[Dict[str, Any]] = []

    if cell_receptions:
        cell_reception_map: Dict[str, BizSampleReception] = {}
        seen_codes: set = set()
        cell_rows: list = []
        for rec in cell_receptions:
            code = _reception_sample_code(db, rec)
            if not code:
                continue
            cell_reception_map[code] = rec
            if code in seen_codes:
                continue
            seen_codes.add(code)
            rows = (
                _sample_detail_rows_query(db)
                .filter(
                    BizProtocolSample.sample_code == code,
                    BizProtocolSample.del_flag == 0,
                    BizProtocolSample.tenant_no == tenant_no,
                    BizCollectionProtocol.tenant_no == tenant_no,
                )
                .all()
            )
            cell_rows.extend(rows)
        items.extend(
            _rows_to_sample_list_items(db, cell_rows, tenant_no, reception_map=cell_reception_map)
        )

    if donation_receptions:
        donation_sample_ids = [r.donation_sample_id for r in donation_receptions]
        donation_reception_map = {r.donation_sample_id: r for r in donation_receptions}
        donation_rows = (
            _donation_sample_detail_rows_query(db)
            .filter(_donation_sample_id_col(BizResearchDonationSample.id).in_(donation_sample_ids))
            .all()
        )
        donation_row_by_sample = {row[0].id: row for row in donation_rows}
        donation_ordered = [donation_row_by_sample[sid] for sid in donation_sample_ids if sid in donation_row_by_sample]
        items.extend(_donation_rows_to_sample_list_items(db, donation_ordered, tenant_no, reception_map=donation_reception_map))

    # 按 reception_time DESC 排序
    items.sort(key=lambda x: x.get("receptionTime") or "", reverse=True)
    return items, int(total)


def _finalize_reception_detail(db: Session, item: Optional[Dict[str, Any]], tenant_no: str) -> Optional[Dict[str, Any]]:
    if not item:
        return item
    return _attach_production_reception(db, item, tenant_no)


def get_reception_by_id(
    db: Session, reception_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    rec = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.id == reception_id,
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
        )
        .first()
    )
    if not rec:
        return None
    if (rec.reception_source or "cell_storage") == "research_donation":
        dsid = rec.donation_sample_id or ""
        if not dsid:
            cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
            out = _reception_to_dict(rec, check_results=cr)
            out["sourceType"] = "research_donation"
            out["sourceTypeLabel"] = "科研捐赠"
            out["receiveAttachmentFiles"] = urls_to_attachment_list(
                rec.receive_attachments, db, tenant_no
            )
            return _finalize_reception_detail(db, out, tenant_no)
        rows = (
            _donation_sample_detail_rows_query(db)
            .filter(_donation_sample_id_col(BizResearchDonationSample.id) == dsid)
            .all()
        )
        if not rows:
            cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
            out = _reception_to_dict(rec, check_results=cr)
            out["sourceType"] = "research_donation"
            out["sourceTypeLabel"] = "科研捐赠"
            out["receiveAttachmentFiles"] = urls_to_attachment_list(
                rec.receive_attachments, db, tenant_no
            )
            return _finalize_reception_detail(db, out, tenant_no)
        items = _donation_rows_to_sample_list_items(
            db, rows, tenant_no, reception_map={dsid: rec}
        )
        return items[0] if items else None

    code = _reception_sample_code(db, rec)
    if code:
        rows = (
            _sample_detail_rows_query(db)
            .filter(
                BizProtocolSample.sample_code == code,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
            )
            .all()
        )
        if rows:
            items = _rows_to_sample_list_items(
                db, rows, tenant_no, reception_map={code: rec}
            )
            return items[0] if items else None
    cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
    out = _reception_to_dict(rec, check_results=cr)
    out["receiveAttachmentFiles"] = urls_to_attachment_list(
        rec.receive_attachments, db, tenant_no
    )
    return _finalize_reception_detail(db, out, tenant_no)


def create_sample_reception(
    db: Session,
    *,
    reception_source: str = "cell_storage",
    protocol_sample_id: Optional[str] = None,
    donation_sample_id: Optional[str] = None,
    research_donation_id: Optional[str] = None,
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str = "",
    reception_time: datetime,
    handler_user_id: Optional[str],
    handler_name: Optional[str],
    receive_temperature: Optional[str],
    volume_ml: Optional[float] = None,
    net_weight_g: Optional[float] = None,
    checks: List[Dict[str, Any]],
    accept_conclusion: str,
    receive_attachments: Optional[str],
    remark: Optional[str],
    lab_storage_code: Optional[str] = None,
) -> BizSampleReception:
    source = (reception_source or "").strip()
    if source not in ("cell_storage", "research_donation"):
        raise ValueError("来源类型无效")

    # 按来源校验样本并获取关联 ID
    if source == "cell_storage":
        if not protocol_sample_id:
            raise ValueError("细胞存储来源必须传入 protocolSampleId")
        sample = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.id == protocol_sample_id,
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
            .first()
        )
        if not sample:
            raise ValueError("样本不存在")
        _sample_code = (sample.sample_code or "").strip()
        if not _sample_code:
            raise ValueError("样本编码不能为空")
        exists = (
            db.query(BizSampleReception)
            .filter(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                BizSampleReception.reception_source == "cell_storage",
                BizSampleReception.sample_code == _sample_code,
            )
            .first()
        )
        if exists:
            raise ValueError("该样本已登记接收，请勿重复操作")
        collection_protocol_id = sample.collection_protocol_id
        _donation_sample_id = None
        _research_donation_id = None
        _sample_code = sample.sample_code or ""
    else:
        # research_donation
        if not donation_sample_id:
            raise ValueError("科研捐赠来源必须传入 donationSampleId")
        dsample = (
            db.query(BizResearchDonationSample)
            .filter(
                _donation_sample_id_col(BizResearchDonationSample.id) == donation_sample_id,
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
            )
            .first()
        )
        if not dsample:
            raise ValueError("捐赠样本不存在")
        _sample_code = (dsample.sample_code or "").strip()
        if not _sample_code:
            raise ValueError("样本编码不能为空")
        exists = (
            db.query(BizSampleReception)
            .filter(
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.del_flag == 0,
                BizSampleReception.reception_source == "research_donation",
                BizSampleReception.sample_code == _sample_code,
            )
            .first()
        )
        if exists:
            raise ValueError("该样本已登记接收，请勿重复操作")
        collection_protocol_id = ""
        protocol_sample_id = ""
        _donation_sample_id = donation_sample_id
        _research_donation_id = dsample.research_donation_id

    conclusion = (accept_conclusion or "").strip()
    if conclusion not in ("js", "bj"):
        raise ValueError("接收结论无效")

    from app.crud import sample_reception_lab_code as lab_code_svc

    resolved_lab_code: Optional[str] = None
    code_raw = (lab_storage_code or "").strip()
    if conclusion == "js":
        resolved_lab_code = lab_code_svc.validate_lab_storage_code(code_raw)
        lab_code_svc.assert_lab_storage_code_available(
            db, tenant_no=tenant_no, code=resolved_lab_code
        )
    elif code_raw:
        resolved_lab_code = lab_code_svc.validate_lab_storage_code(code_raw)
        lab_code_svc.assert_lab_storage_code_available(
            db, tenant_no=tenant_no, code=resolved_lab_code
        )

    handler_display = (handler_name or "").strip()
    if handler_user_id and not handler_display:
        u = db.query(SysUser).filter(SysUser.id == handler_user_id).first()
        if u:
            handler_display = u.nick_name or ""

    obj = BizSampleReception(
        id=generate_uuid(),
        protocol_sample_id=protocol_sample_id or "",
        collection_protocol_id=collection_protocol_id or "",
        reception_source=source,
        donation_sample_id=_donation_sample_id,
        research_donation_id=_research_donation_id,
        sample_code=_sample_code or None,
        lab_storage_code=resolved_lab_code,
        reception_time=reception_time,
        handler_user_id=handler_user_id,
        handler_name=handler_display or None,
        receive_temperature=(receive_temperature or "").strip() or None,
        check_packaging=0,
        check_sealing=0,
        check_no_damage=0,
        check_no_leakage=0,
        check_color=0,
        check_tying_rope=0,
        check_no_stratification=0,
        accept_conclusion=conclusion,
        receive_attachments=(receive_attachments or "").strip() or None,
        tenant_no=tenant_no,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=create_dept_id or "",
        update_by=user_id,
    )
    db.add(obj)
    db.flush()
    check_result_svc.save_reception_check_results(
        db,
        reception_id=obj.id,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=create_dept_id or "",
        checks=checks,
    )
    from app.models.models import BizSampleReceptionCheckResult

    saved_rows = (
        db.query(BizSampleReceptionCheckResult)
        .filter(
            BizSampleReceptionCheckResult.reception_id == obj.id,
            BizSampleReceptionCheckResult.del_flag == 0,
        )
        .all()
    )
    check_result_svc.sync_legacy_columns_from_results(db, obj, saved_rows)
    db.add(obj)
    if conclusion == "js":
        from app.crud import qc_work_order as qc_wo_svc

        qc_wo_svc.create_work_order_from_reception(
            db,
            reception=obj,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id or "",
        )
        from app.crud import production_work_order as prod_wo_svc

        prod_wo_svc.create_production_work_order_from_reception(
            db,
            reception=obj,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id or "",
        )
        from app.crud import storage_release_review as srr_svc

        _srr_protocol_no = ""
        _srr_sample_name = ""
        _src = getattr(obj, "reception_source", "cell_storage") or "cell_storage"
        if _src == "research_donation":
            _d = (
                db.query(BizResearchDonation)
                .filter(
                    BizResearchDonation.id == (obj.research_donation_id or ""),
                    BizResearchDonation.del_flag == 0,
                )
                .first()
            )
            if _d:
                _srr_protocol_no = _d.donation_no or ""
            _ds = (
                db.query(BizResearchDonationSample)
                .filter(
                    BizResearchDonationSample.id == (obj.donation_sample_id or ""),
                    BizResearchDonationSample.del_flag == 0,
                )
                .first()
            )
            if _ds:
                _srr_sample_name = _ds.sample_code or ""
        else:
            _p = (
                db.query(BizCollectionProtocol)
                .filter(
                    BizCollectionProtocol.id == obj.collection_protocol_id,
                    BizCollectionProtocol.del_flag == 0,
                )
                .first()
            )
            if _p:
                _srr_protocol_no = _p.protocol_no or ""
            _s = (
                db.query(BizProtocolSample)
                .filter(
                    BizProtocolSample.id == obj.protocol_sample_id,
                    BizProtocolSample.del_flag == 0,
                )
                .first()
            )
            if _s:
                _srr_sample_name = _s.sample_code or ""
        srr_svc.create_review_from_reception(
            db,
            reception=obj,
            protocol_no=_srr_protocol_no,
            sample_name=_srr_sample_name,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id or "",
        )
    db.commit()
    db.refresh(obj)
    return obj


def _can_edit_sample_reception(
    db: Session,
    user_id: Optional[str],
    rec: BizSampleReception,
) -> bool:
    """已接收记录：仅经手人或「质量主管」角色可编辑。"""
    uid = (user_id or "").strip()
    if not uid:
        return False
    handler_id = (rec.handler_user_id or "").strip()
    if handler_id and handler_id == uid:
        return True
    return SAMPLE_RECEPTION_EDIT_ROLE in get_user_role_names(db, uid)


def update_sample_reception(
    db: Session,
    *,
    reception_id: str,
    tenant_no: str,
    user_id: Optional[str],
    reception_time: datetime,
    handler_user_id: Optional[str],
    handler_name: Optional[str],
    receive_temperature: Optional[str],
    volume_ml: Optional[float] = None,
    net_weight_g: Optional[float] = None,
    checks: List[Dict[str, Any]],
    accept_conclusion: str,
    receive_attachments: Optional[str],
    remark: Optional[str],
    lab_storage_code: Optional[str] = None,
) -> BizSampleReception:
    """更新已登记样本接收信息（不重复创建质检/生产工单）。"""
    rec = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.id == reception_id,
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
        )
        .first()
    )
    if not rec:
        raise ValueError("接收记录不存在")

    if not _can_edit_sample_reception(db, user_id, rec):
        raise PermissionError("仅该接收记录的经手人或质量主管可编辑")

    conclusion = (accept_conclusion or "").strip()
    if conclusion not in ("js", "bj"):
        raise ValueError("接收结论无效")

    from app.crud import sample_reception_lab_code as lab_code_svc

    code_raw = (lab_storage_code or "").strip()
    if conclusion == "js":
        parsed_code = lab_code_svc.validate_lab_storage_code(code_raw)
        lab_code_svc.assert_lab_storage_code_available(
            db,
            tenant_no=tenant_no,
            code=parsed_code,
            exclude_reception_id=reception_id,
        )
        rec.lab_storage_code = parsed_code
    elif code_raw:
        parsed_code = lab_code_svc.validate_lab_storage_code(code_raw)
        lab_code_svc.assert_lab_storage_code_available(
            db,
            tenant_no=tenant_no,
            code=parsed_code,
            exclude_reception_id=reception_id,
        )
        rec.lab_storage_code = parsed_code
    else:
        rec.lab_storage_code = None

    handler_display = (handler_name or "").strip()
    if handler_user_id and not handler_display:
        u = db.query(SysUser).filter(SysUser.id == handler_user_id).first()
        if u:
            handler_display = u.nick_name or ""

    rec.reception_time = reception_time
    rec.handler_user_id = handler_user_id
    rec.handler_name = handler_display or None
    rec.receive_temperature = (receive_temperature or "").strip() or None
    rec.accept_conclusion = conclusion
    rec.receive_attachments = (receive_attachments or "").strip() or None
    rec.remark = (remark or "").strip() or None
    rec.update_by = user_id
    rec.update_time = datetime.now()

    check_result_svc.soft_delete_reception_check_results(
        db,
        reception_id=reception_id,
        tenant_no=tenant_no,
        user_id=user_id,
    )
    check_result_svc.save_reception_check_results(
        db,
        reception_id=reception_id,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=rec.create_dept_id or "",
        checks=checks,
    )
    from app.models.models import BizSampleReceptionCheckResult

    saved_rows = (
        db.query(BizSampleReceptionCheckResult)
        .filter(
            BizSampleReceptionCheckResult.reception_id == reception_id,
            BizSampleReceptionCheckResult.del_flag == 0,
        )
        .all()
    )
    check_result_svc.sync_legacy_columns_from_results(db, rec, saved_rows)
    db.add(rec)
    db.commit()
    db.refresh(rec)
    return rec


def _attach_production_reception(
    db: Session,
    item: Dict[str, Any],
    tenant_no: str,
    *,
    production_map: Optional[Dict[str, Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    rid = str(item.get("id") or "").strip()
    if not rid:
        item["productionReception"] = None
        return item
    if production_map is not None:
        item["productionReception"] = production_map.get(rid)
    else:
        item["productionReception"] = prod_reception_svc.get_by_reception_id(db, rid, tenant_no)
    return item


def _reception_detail_item(
    db: Session,
    rec: BizSampleReception,
    tenant_no: str,
    *,
    results_map: Optional[Dict[str, List[Dict[str, Any]]]] = None,
) -> Dict[str, Any]:
    cr = (results_map or {}).get(rec.id)
    if not cr:
        cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
    item = _reception_to_dict(rec, check_results=cr)
    item["receiveAttachmentFiles"] = urls_to_attachment_list(
        rec.receive_attachments, db, tenant_no
    )
    return item


def _reception_detail_item_with_production(
    db: Session,
    rec: BizSampleReception,
    tenant_no: str,
    *,
    results_map: Optional[Dict[str, List[Dict[str, Any]]]] = None,
    production_map: Optional[Dict[str, Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    item = _reception_detail_item(db, rec, tenant_no, results_map=results_map)
    return _attach_production_reception(db, item, tenant_no, production_map=production_map)


def _reception_sample_code(db: Session, rec: BizSampleReception) -> str:
    code = (rec.sample_code or "").strip()
    if code:
        return code
    sid = (rec.protocol_sample_id or "").strip()
    if not sid:
        return ""
    row = (
        db.query(BizProtocolSample.sample_code)
        .filter(BizProtocolSample.id == sid)
        .first()
    )
    return (row.sample_code or "").strip() if row else ""


def get_receptions_map_for_samples(
    db: Session,
    tenant_no: str,
    *,
    sample_code_by_id: Dict[str, str],
) -> Dict[str, Dict[str, Any]]:
    """按租户 + 样本编码关联接收记录（key 为当前协议样本 id），供协议单查看弹窗使用。"""
    if not tenant_no or not sample_code_by_id:
        return {}
    codes = list({(c or "").strip() for c in sample_code_by_id.values() if (c or "").strip()})
    if not codes:
        return {}
    rows = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
            BizSampleReception.reception_source == "cell_storage",
            BizSampleReception.sample_code.in_(codes),
        )
        .all()
    )
    if not rows:
        return {}
    code_to_rec: Dict[str, BizSampleReception] = {}
    for rec in rows:
        code = _reception_sample_code(db, rec)
        if code and code not in code_to_rec:
            code_to_rec[code] = rec
    reception_ids = [r.id for r in code_to_rec.values()]
    results_map = check_result_svc.list_results_by_reception_ids(
        db, reception_ids, tenant_no
    )
    production_map = prod_reception_svc.get_map_by_reception_ids(db, reception_ids, tenant_no)
    out: Dict[str, Dict[str, Any]] = {}
    for sid, raw_code in sample_code_by_id.items():
        code = (raw_code or "").strip()
        if not code:
            continue
        rec = code_to_rec.get(code)
        if not rec:
            continue
        out[sid] = _reception_detail_item_with_production(
            db, rec, tenant_no, results_map=results_map, production_map=production_map
        )
    return out


def get_receptions_map_for_donation_samples(
    db: Session,
    sample_ids: List[str],
    tenant_no: str,
) -> Dict[str, Dict[str, Any]]:
    """按捐赠样本 ID 批量查询接收记录，供捐赠协议单查看弹窗使用。"""
    if not sample_ids or not tenant_no:
        return {}
    rows = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.donation_sample_id.in_(sample_ids),
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
        )
        .all()
    )
    if not rows:
        return {}
    reception_ids = [r.id for r in rows]
    results_map = check_result_svc.list_results_by_reception_ids(db, reception_ids, tenant_no)
    production_map = prod_reception_svc.get_map_by_reception_ids(db, reception_ids, tenant_no)
    out: Dict[str, Dict[str, Any]] = {}
    for rec in rows:
        sid = rec.donation_sample_id
        if not sid:
            continue
        cr = results_map.get(rec.id)
        if not cr:
            cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no)
        item = _reception_to_dict(rec, check_results=cr)
        item["receiveAttachmentFiles"] = urls_to_attachment_list(
            rec.receive_attachments, db, tenant_no
        )
        _attach_production_reception(db, item, tenant_no, production_map=production_map)
        out[sid] = item
    return out


def get_lab_storage_code_map_by_sample_codes(
    db: Session,
    tenant_no: str,
    sample_codes: List[str],
    *,
    reception_source: Optional[str] = None,
) -> Dict[str, str]:
    """按样本编码查询已接收(js)记录的实验室入库编码；同码多条时取最新接收时间。"""
    codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()})
    if not codes or not tenant_no:
        return {}
    q = db.query(BizSampleReception.sample_code, BizSampleReception.lab_storage_code).filter(
        BizSampleReception.tenant_no == tenant_no,
        BizSampleReception.del_flag == 0,
        BizSampleReception.sample_code.in_(codes),
        BizSampleReception.accept_conclusion == "js",
        BizSampleReception.lab_storage_code.isnot(None),
        BizSampleReception.lab_storage_code != "",
    )
    if reception_source:
        q = q.filter(BizSampleReception.reception_source == reception_source)
    result: Dict[str, str] = {}
    for row in q.order_by(BizSampleReception.reception_time.desc()).all():
        sc = (row.sample_code or "").strip()
        lk = (row.lab_storage_code or "").strip()
        if sc and lk and sc not in result:
            result[sc] = lk

    missing = [c for c in codes if c not in result]
    if missing:
        join_q = (
            db.query(BizProtocolSample.sample_code, BizSampleReception.lab_storage_code)
            .join(
                BizSampleReception,
                and_(
                    BizSampleReception.protocol_sample_id == BizProtocolSample.id,
                    BizSampleReception.del_flag == 0,
                    BizSampleReception.tenant_no == tenant_no,
                    BizSampleReception.accept_conclusion == "js",
                    BizSampleReception.lab_storage_code.isnot(None),
                    BizSampleReception.lab_storage_code != "",
                ),
            )
            .filter(
                BizProtocolSample.sample_code.in_(missing),
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
            )
        )
        if reception_source:
            join_q = join_q.filter(BizSampleReception.reception_source == reception_source)
        for row in join_q.order_by(BizSampleReception.reception_time.desc()).all():
            sc = (row.sample_code or "").strip()
            lk = (row.lab_storage_code or "").strip()
            if sc and lk and sc not in result:
                result[sc] = lk
    return result


def format_lab_storage_codes_csv(
    sample_codes: List[str],
    code_map: Dict[str, str],
) -> str:
    """与 sampleCodes 顺序对齐，逗号拼接入库编码（无则空串）。"""
    parts: List[str] = []
    for raw in sample_codes:
        c = (raw or "").strip()
        if not c:
            continue
        parts.append((code_map.get(c) or "").strip())
    return ",".join(parts)


def _lab_storage_keyword_like(keyword: str) -> Optional[str]:
    kw = (keyword or "").strip()
    if not kw:
        return None
    return f"%{kw}%"


def sample_codes_matching_lab_storage_keyword(
    db: Session,
    tenant_no: str,
    keyword: str,
    *,
    reception_source: Optional[str] = None,
) -> List[str]:
    """关键词模糊匹配实验室入库编码，返回关联样本编码列表。"""
    like = _lab_storage_keyword_like(keyword)
    if not like or not tenant_no:
        return []
    q = db.query(BizSampleReception.sample_code).filter(
        BizSampleReception.tenant_no == tenant_no,
        BizSampleReception.del_flag == 0,
        BizSampleReception.lab_storage_code.isnot(None),
        BizSampleReception.lab_storage_code != "",
        BizSampleReception.lab_storage_code.like(like),
        BizSampleReception.sample_code.isnot(None),
        BizSampleReception.sample_code != "",
    )
    if reception_source:
        q = q.filter(BizSampleReception.reception_source == reception_source)
    return list({(row[0] or "").strip() for row in q.all() if (row[0] or "").strip()})


def reception_ids_matching_lab_storage_keyword(
    db: Session,
    tenant_no: str,
    keyword: str,
    *,
    reception_source: Optional[str] = None,
) -> List[str]:
    """关键词模糊匹配实验室入库编码，返回接收记录 id 列表。"""
    like = _lab_storage_keyword_like(keyword)
    if not like or not tenant_no:
        return []
    q = db.query(BizSampleReception.id).filter(
        BizSampleReception.tenant_no == tenant_no,
        BizSampleReception.del_flag == 0,
        BizSampleReception.lab_storage_code.isnot(None),
        BizSampleReception.lab_storage_code != "",
        BizSampleReception.lab_storage_code.like(like),
    )
    if reception_source:
        q = q.filter(BizSampleReception.reception_source == reception_source)
    return [row[0] for row in q.all() if row[0]]


def protocol_ids_matching_lab_storage_keyword(
    db: Session,
    tenant_no: str,
    keyword: str,
) -> List[str]:
    """入库编码关键词 → 含匹配样本的采集协议 id。"""
    codes = sample_codes_matching_lab_storage_keyword(
        db, tenant_no, keyword, reception_source="cell_storage"
    )
    if not codes:
        return []
    rows = (
        db.query(BizProtocolSample.collection_protocol_id)
        .filter(
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.del_flag == 0,
            BizProtocolSample.sample_code.in_(codes),
        )
        .distinct()
        .all()
    )
    return [r[0] for r in rows if r[0]]


def donation_ids_matching_lab_storage_keyword(
    db: Session,
    tenant_no: str,
    keyword: str,
) -> List[str]:
    """入库编码关键词 → 含匹配样本的捐赠协议 id。"""
    codes = sample_codes_matching_lab_storage_keyword(
        db, tenant_no, keyword, reception_source="research_donation"
    )
    if not codes:
        return []
    rows = (
        db.query(BizResearchDonationSample.research_donation_id)
        .filter(
            BizResearchDonationSample.tenant_no == tenant_no,
            BizResearchDonationSample.del_flag == 0,
            BizResearchDonationSample.sample_code.in_(codes),
        )
        .distinct()
        .all()
    )
    return [r[0] for r in rows if r[0]]


def get_reception_by_sample_code(
    db: Session,
    sample_code: str,
    source_type: Optional[str],
    tenant_no: str,
) -> Dict[str, Any]:
    """
    根据样本编码查询接收记录及样本信息。
    - 有记录 → 返回 {receptionId, sampleInfo}
    - 无记录 → 返回 {receptionId: null, sampleInfo}
    """
    code = (sample_code or "").strip()
    if not code:
        return {"receptionId": None, "sampleInfo": None}

    source = (source_type or "cell_storage").strip()
    # 先查询是否有已接收记录
    q_rec = db.query(BizSampleReception).filter(
        BizSampleReception.tenant_no == tenant_no,
        BizSampleReception.del_flag == 0,
        BizSampleReception.sample_code == code,
    )
    if source:
        q_rec = q_rec.filter(BizSampleReception.reception_source == source)

    reception = q_rec.order_by(BizSampleReception.reception_time.desc()).first()

    if source == "research_donation":
        # 捐赠样本查询
        rows = (
            _donation_sample_detail_rows_query(db)
            .filter(
                _donation_sample_id_col(BizResearchDonationSample.sample_code) == code,
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.tenant_no == tenant_no,
            )
            .all()
        )
        sample_info_list = _donation_rows_to_sample_list_items(db, rows, tenant_no)
        sample_info = sample_info_list[0] if sample_info_list else None
    else:
        # 细胞存储样本查询
        rows = (
            _sample_detail_rows_query(db)
            .filter(
                BizProtocolSample.sample_code == code,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.tenant_no == tenant_no,
            )
            .all()
        )
        if rows:
            if reception:
                sample_info_list = _rows_to_sample_list_items(
                    db, rows, tenant_no, reception_map={code: reception}
                )
            else:
                sample_info_list = _rows_to_sample_list_items(db, rows, tenant_no)
            sample_info = sample_info_list[0] if sample_info_list else None
        else:
            sample_info = None

    return {
        "receptionId": reception.id if reception else None,
        "sampleInfo": sample_info,
        "productionReception": (
            prod_reception_svc.get_by_reception_id(db, reception.id, tenant_no) if reception else None
        ),
    }