zkxw/backend/app/crud/storage_release_review.py

from __future__ import annotations

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

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

from app.core.db_collation import eq_unicode
from app.core.protocol_status_display import INACTIVE_PROTOCOL_STATUSES
from app.core.utils import generate_uuid
from app.utils.sample_special import sample_special_to_dict
from app.models.models import (
    BizCollectionProtocol,
    BizProductionWorkOrder,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizProtocolSampleTransport,
    BizQcWorkOrder,
    BizSampleReception,
    BizStorageReleaseReview,
    SysUser,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationParty,
    BizResearchDonationSample,
    BizResearchDonationSampleTransport,
)

_RECEPTION_SOURCE_LABELS = {
    "cell_storage": "细胞存储",
    "research_donation": "科研捐赠",
}
_REVIEW_CONCLUSION_LABELS = {"tg": "通过", "btg": "不通过"}
_STORAGE_STATUS_LABELS = {"drk": "待入库", "yrk": "已入库"}
_PROTOCOL_STATUS_LABELS = {
    "dsx": "待生效",
    "sxz": "生效中",
    "ygq": "已过期",
    "yzf": "已作废",
    "sdfq": "属地废弃",
    "zky": "转科研",
}
_DONATION_STATUS_LABELS = {
    "dsx": "待生效",
    "sxz": "生效中",
    "ygq": "已过期",
    "yzf": "已作废",
    "sdfq": "属地废弃",
    "zckc": "转细胞存储",
}
def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))


def _row_to_dict(r: BizStorageReleaseReview) -> Dict[str, Any]:
    return {
        "id": r.id,
        "sampleCode": r.sample_code or "",
        "receptionSource": r.reception_source or "",
        "receptionSourceLabel": _RECEPTION_SOURCE_LABELS.get(r.reception_source or "", r.reception_source or ""),
        "protocolId": r.protocol_id or "",
        "receptionId": r.reception_id or "",
        "protocolNo": r.protocol_no or "",
        "sampleName": r.sample_name or "",
        "reviewConclusion": r.review_conclusion or "",
        "reviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(r.review_conclusion or "", r.review_conclusion or ""),
        "reviewRemark": r.review_remark or "",
        "reviewAttachments": r.review_attachments or "",
        "reviewBy": r.review_by or "",
        "reviewByName": r.review_by_name or "",
        "reviewTime": _fmt_dt(r.review_time),
        "qcReviewConclusion": r.qc_review_conclusion or "",
        "qcReviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(r.qc_review_conclusion or "", r.qc_review_conclusion or ""),
        "qcReviewRemark": r.qc_review_remark or "",
        "qcReviewAttachments": r.qc_review_attachments or "",
        "qcReviewBy": r.qc_review_by or "",
        "qcReviewByName": r.qc_review_by_name or "",
        "qcReviewTime": _fmt_dt(r.qc_review_time),
        "prodReviewConclusion": r.prod_review_conclusion or "",
        "prodReviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(r.prod_review_conclusion or "", r.prod_review_conclusion or ""),
        "prodReviewRemark": r.prod_review_remark or "",
        "prodReviewAttachments": r.prod_review_attachments or "",
        "prodReviewBy": r.prod_review_by or "",
        "prodReviewByName": r.prod_review_by_name or "",
        "prodReviewTime": _fmt_dt(r.prod_review_time),
        "storageStatus": r.storage_status or "drk",
        "storageStatusLabel": _STORAGE_STATUS_LABELS.get(r.storage_status or "drk", r.storage_status or "drk"),
        "storageBy": r.storage_by or "",
        "storageByName": r.storage_by_name or "",
        "storageTime": _fmt_dt(r.storage_time),
        "remark": r.remark or "",
        "createTime": _fmt_dt(r.create_time),
        "updateTime": _fmt_dt(r.update_time),
    }


def create_review_from_reception(
    db: Session,
    *,
    reception: BizSampleReception,
    protocol_no: str,
    sample_name: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> BizStorageReleaseReview:
    source = getattr(reception, "reception_source", "cell_storage") or "cell_storage"
    if source == "research_donation":
        protocol_id = reception.research_donation_id or ""
    else:
        protocol_id = reception.collection_protocol_id or ""

    obj = BizStorageReleaseReview(
        id=generate_uuid(),
        sample_code=reception.sample_code or "",
        reception_source=source,
        protocol_id=protocol_id,
        reception_id=reception.id,
        protocol_no=protocol_no or None,
        sample_name=sample_name or None,
        storage_status="drk",
        tenant_no=tenant_no,
        create_by=user_id,
        create_dept_id=create_dept_id or "",
        update_by=user_id,
        create_time=datetime.now(),
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.flush()
    return obj


def normalize_review_phase(review_phase: Optional[str]) -> Optional[str]:
    """Return 'pending' | 'reviewed' or None when absent/invalid."""
    p = (review_phase or "").strip().lower()
    if p in ("pending", "reviewed"):
        return p
    return None


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


def aggregate_review_conclusion(
    prod: Optional[str], qc: Optional[str]
) -> Optional[str]:
    p = (prod or "").strip()
    q = (qc or "").strip()
    if p == "btg" or q == "btg":
        return "btg"
    if p == "tg" and q == "tg":
        return "tg"
    return None


def storage_transition(
    prod: Optional[str], qc: Optional[str], storage_status: Optional[str]
) -> str:
    """Return 'store' | 'unstore' | 'noop'."""
    agg = aggregate_review_conclusion(prod, qc)
    st = (storage_status or "drk").strip() or "drk"
    if agg == "tg" and st == "drk":
        return "store"
    if agg == "btg" and st == "yrk":
        return "unstore"
    return "noop"


def normalize_review_side(raw: Optional[str]) -> Optional[str]:
    s = (raw or "").strip().lower()
    if s in ("overview", "production", "qc"):
        return s
    return None


def review_side_required_menu(review_side: Optional[str]) -> Optional[str]:
    if review_side == "production":
        return "production:storageReleaseReview:list"
    if review_side == "qc":
        return "quality:storageReleaseReview:list"
    return None


def matches_dual_review_phase(
    review_side: Optional[str],
    phase: Optional[str],
    prod: Optional[str],
    qc: Optional[str],
) -> bool:
    """Pure predicate for Tab semantics; SQL filter must match this."""
    side = normalize_review_side(review_side) or "overview"
    ph = normalize_review_phase(phase)
    if not ph:
        return True
    prod_blank = conclusion_blank(prod)
    qc_blank = conclusion_blank(qc)
    if side == "production":
        return prod_blank if ph == "pending" else (not prod_blank)
    if side == "qc":
        return qc_blank if ph == "pending" else (not qc_blank)
    both_filled = (not prod_blank) and (not qc_blank)
    return (not both_filled) if ph == "pending" else both_filled


def should_copy_legacy_to_qc(
    qc_review_conclusion: Optional[str],
    legacy_review_conclusion: Optional[str],
) -> bool:
    return conclusion_blank(qc_review_conclusion) and not conclusion_blank(
        legacy_review_conclusion
    )


def normalize_qc_work_order_completed(raw) -> Optional[bool]:
    """Return True/False when recognizable; None means no filter."""
    if raw is None:
        return None
    if isinstance(raw, bool):
        return raw
    s = str(raw).strip().lower()
    if s in ("true", "1", "yes"):
        return True
    if s in ("false", "0", "no"):
        return False
    return None


def normalize_ready_for_review_only(raw) -> Optional[bool]:
    """Return True when ready-only filter should apply; None means no filter."""
    if raw is None:
        return None
    if isinstance(raw, bool):
        return True if raw else None
    s = str(raw).strip().lower()
    if s in ("true", "1", "yes"):
        return True
    return None


def is_inactive_protocol_status_for_hide(status) -> bool:
    """True when protocol/donation status is yzf or sdfq (list-hide rule)."""
    code = (status or "").strip().lower()
    return code in INACTIVE_PROTOCOL_STATUSES


def inbound_work_orders_all_void(statuses) -> bool:
    """True when there is ≥1 non-empty status and every one equals yzf."""
    cleaned = [(s or "").strip() for s in (statuses or [])]
    if not cleaned or not any(s for s in cleaned):
        return False
    return all(s.lower() == "yzf" for s in cleaned)


def _qc_work_order_exists_for_review_sample(
    tenant_no: str, *, unfinished_only: bool = False
):
    """Correlate biz_qc_work_order to BizStorageReleaseReview.sample_code."""
    conds = [
        BizQcWorkOrder.tenant_no == tenant_no,
        BizQcWorkOrder.del_flag == 0,
        BizQcWorkOrder.sample_code.isnot(None),
        BizQcWorkOrder.sample_code != "",
        # biz_qc_work_order 与 biz_storage_release_review 的 sample_code 排序规则不同
        eq_unicode(BizQcWorkOrder.sample_code, BizStorageReleaseReview.sample_code),
        BizStorageReleaseReview.sample_code.isnot(None),
        BizStorageReleaseReview.sample_code != "",
    ]
    if unfinished_only:
        conds.append(BizQcWorkOrder.work_order_status.in_(("dpz", "ypz")))
    return (
        exists()
        .where(and_(*conds))
        .correlate(BizStorageReleaseReview)
    )


def _production_work_order_unfinished_exists_for_review_sample(tenant_no: str):
    """True if sample has any unfinished inbound-side production WO (dpz/jxz)."""
    conds = [
        BizProductionWorkOrder.tenant_no == tenant_no,
        BizProductionWorkOrder.del_flag == 0,
        BizProductionWorkOrder.sample_code.isnot(None),
        BizProductionWorkOrder.sample_code != "",
        eq_unicode(
            BizProductionWorkOrder.sample_code, BizStorageReleaseReview.sample_code
        ),
        BizStorageReleaseReview.sample_code.isnot(None),
        BizStorageReleaseReview.sample_code != "",
        BizProductionWorkOrder.work_order_status.in_(("dpz", "jxz")),
    ]
    return (
        exists()
        .where(and_(*conds))
        .correlate(BizStorageReleaseReview)
    )


def _inactive_protocol_exists_for_review():
    """True when linked collection/donation protocol is yzf or sdfq."""
    inactive_cell = and_(
        or_(
            BizStorageReleaseReview.reception_source == "cell_storage",
            BizStorageReleaseReview.reception_source.is_(None),
            BizStorageReleaseReview.reception_source == "",
        ),
        BizStorageReleaseReview.protocol_id.isnot(None),
        BizStorageReleaseReview.protocol_id != "",
        exists()
        .where(
            and_(
                eq_unicode(
                    BizCollectionProtocol.id, BizStorageReleaseReview.protocol_id
                ),
                BizCollectionProtocol.del_flag == 0,
                BizCollectionProtocol.protocol_status.in_(INACTIVE_PROTOCOL_STATUSES),
            )
        )
        .correlate(BizStorageReleaseReview),
    )
    inactive_donation = and_(
        BizStorageReleaseReview.reception_source == "research_donation",
        BizStorageReleaseReview.protocol_id.isnot(None),
        BizStorageReleaseReview.protocol_id != "",
        exists()
        .where(
            and_(
                eq_unicode(
                    BizResearchDonation.id, BizStorageReleaseReview.protocol_id
                ),
                BizResearchDonation.del_flag == 0,
                BizResearchDonation.donation_status.in_(INACTIVE_PROTOCOL_STATUSES),
            )
        )
        .correlate(BizStorageReleaseReview),
    )
    return or_(inactive_cell, inactive_donation)


def _inbound_qc_all_void_for_review_sample(tenant_no: str):
    """True when sample has ≥1 inbound QC WO and every such WO is yzf."""
    base = [
        BizQcWorkOrder.tenant_no == tenant_no,
        BizQcWorkOrder.del_flag == 0,
        BizQcWorkOrder.qc_biz_type == "inbound",
        BizQcWorkOrder.sample_code.isnot(None),
        BizQcWorkOrder.sample_code != "",
        eq_unicode(BizQcWorkOrder.sample_code, BizStorageReleaseReview.sample_code),
        BizStorageReleaseReview.sample_code.isnot(None),
        BizStorageReleaseReview.sample_code != "",
    ]
    any_inbound = exists().where(and_(*base)).correlate(BizStorageReleaseReview)
    any_non_void = (
        exists()
        .where(
            and_(
                *base,
                or_(
                    BizQcWorkOrder.work_order_status.is_(None),
                    BizQcWorkOrder.work_order_status != "yzf",
                ),
            )
        )
        .correlate(BizStorageReleaseReview)
    )
    return and_(any_inbound, ~any_non_void)


def _inbound_production_all_void_for_review_sample(tenant_no: str):
    """True when sample has ≥1 inbound production WO and every such WO is yzf."""
    base = [
        BizProductionWorkOrder.tenant_no == tenant_no,
        BizProductionWorkOrder.del_flag == 0,
        BizProductionWorkOrder.work_order_type == "inbound",
        BizProductionWorkOrder.sample_code.isnot(None),
        BizProductionWorkOrder.sample_code != "",
        eq_unicode(
            BizProductionWorkOrder.sample_code, BizStorageReleaseReview.sample_code
        ),
        BizStorageReleaseReview.sample_code.isnot(None),
        BizStorageReleaseReview.sample_code != "",
    ]
    any_inbound = exists().where(and_(*base)).correlate(BizStorageReleaseReview)
    any_non_void = (
        exists()
        .where(
            and_(
                *base,
                or_(
                    BizProductionWorkOrder.work_order_status.is_(None),
                    BizProductionWorkOrder.work_order_status != "yzf",
                ),
            )
        )
        .correlate(BizStorageReleaseReview)
    )
    return and_(any_inbound, ~any_non_void)


def list_reviews(
    db: Session,
    *,
    tenant_no: str,
    page: int = 1,
    page_size: int = 10,
    reception_source: Optional[str] = None,
    review_conclusion: Optional[str] = None,
    storage_status: Optional[str] = None,
    keyword: Optional[str] = None,
    review_phase: Optional[str] = None,
    review_side: Optional[str] = None,
    prod_review_conclusion: Optional[str] = None,
    qc_review_conclusion: Optional[str] = None,
    qc_work_order_completed=None,
    ready_for_review_only=None,
    user_info=None,
) -> Dict[str, Any]:
    q0 = db.query(BizStorageReleaseReview).filter(
        BizStorageReleaseReview.tenant_no == tenant_no,
        BizStorageReleaseReview.del_flag == 0,
    )
    rs = (reception_source or "").strip()
    if rs:
        q0 = q0.filter(BizStorageReleaseReview.reception_source == rs)
    phase = normalize_review_phase(review_phase)
    side = normalize_review_side(review_side) or "overview"
    if phase:
        prod_blank = or_(
            BizStorageReleaseReview.prod_review_conclusion.is_(None),
            BizStorageReleaseReview.prod_review_conclusion == "",
        )
        qc_blank = or_(
            BizStorageReleaseReview.qc_review_conclusion.is_(None),
            BizStorageReleaseReview.qc_review_conclusion == "",
        )
        if side == "production":
            q0 = q0.filter(prod_blank if phase == "pending" else ~prod_blank)
        elif side == "qc":
            q0 = q0.filter(qc_blank if phase == "pending" else ~qc_blank)
        else:
            both_filled = and_(~prod_blank, ~qc_blank)
            q0 = q0.filter(~both_filled if phase == "pending" else both_filled)
    prc = (prod_review_conclusion or "").strip()
    if prc:
        q0 = q0.filter(BizStorageReleaseReview.prod_review_conclusion == prc)
    qrc = (qc_review_conclusion or "").strip()
    if qrc:
        q0 = q0.filter(BizStorageReleaseReview.qc_review_conclusion == qrc)
    rc = (review_conclusion or "").strip()
    if rc:
        q0 = q0.filter(BizStorageReleaseReview.review_conclusion == rc)
    ss = (storage_status or "").strip()
    if ss:
        q0 = q0.filter(BizStorageReleaseReview.storage_status == ss)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        kw_conds = [
            BizStorageReleaseReview.sample_code.like(like),
            BizStorageReleaseReview.sample_name.like(like),
        ]
        from app.crud import sample_reception as reception_svc

        rec_ids = reception_svc.reception_ids_matching_lab_storage_keyword(
            db, tenant_no, kw
        )
        if rec_ids:
            kw_conds.append(BizStorageReleaseReview.reception_id.in_(rec_ids))
        lab_codes = reception_svc.sample_codes_matching_lab_storage_keyword(db, tenant_no, kw)
        if lab_codes:
            kw_conds.append(BizStorageReleaseReview.sample_code.in_(lab_codes))
        q0 = q0.filter(or_(*kw_conds))
    qc_done = normalize_qc_work_order_completed(qc_work_order_completed)
    if qc_done is True:
        q0 = q0.filter(
            _qc_work_order_exists_for_review_sample(tenant_no, unfinished_only=False),
            ~_qc_work_order_exists_for_review_sample(tenant_no, unfinished_only=True),
        )
    elif qc_done is False:
        q0 = q0.filter(
            or_(
                ~_qc_work_order_exists_for_review_sample(
                    tenant_no, unfinished_only=False
                ),
                _qc_work_order_exists_for_review_sample(
                    tenant_no, unfinished_only=True
                ),
            )
        )
    if normalize_ready_for_review_only(ready_for_review_only) is True:
        # 无未完成质检/生产工单 ⇒ 该侧完成（含零工单）
        q0 = q0.filter(
            ~_qc_work_order_exists_for_review_sample(
                tenant_no, unfinished_only=True
            ),
            ~_production_work_order_unfinished_exists_for_review_sample(tenant_no),
        )
    # 协议无效 或 入库质检/生产「有工单且全部 yzf」→ 列表隐藏（两 Tab 常驻）
    q0 = q0.filter(
        ~or_(
            _inactive_protocol_exists_for_review(),
            _inbound_qc_all_void_for_review_sample(tenant_no),
            _inbound_production_all_void_for_review_sample(tenant_no),
        )
    )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q0 = apply_data_scope(q0, BizStorageReleaseReview, user_info, db, tenant_no)
    total = q0.order_by(None).with_entities(func.count(BizStorageReleaseReview.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q0.order_by(BizStorageReleaseReview.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = _enrich_review_list_items(db, [_row_to_dict(r) for r in rows])
    return {
        "list": items,
        "total": int(total),
        "page": page,
        "pageSize": page_size,
        "totalPages": total_pages(int(total), page_size),
    }


def _enrich_review_list_items(db: Session, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    if not items:
        return items
    from app.crud.qc_work_order import _enrich_work_order_list_items

    items = _enrich_work_order_list_items(db, items)

    cell_pids = [
        x["protocolId"]
        for x in items
        if (x.get("receptionSource") or "cell_storage") != "research_donation" and x.get("protocolId")
    ]
    donation_pids = [
        x["protocolId"]
        for x in items
        if (x.get("receptionSource") or "") == "research_donation" and x.get("protocolId")
    ]

    proto_map: Dict[str, BizCollectionProtocol] = {}
    if cell_pids:
        proto_map = {
            p.id: p
            for p in db.query(BizCollectionProtocol)
            .filter(BizCollectionProtocol.id.in_(cell_pids), BizCollectionProtocol.del_flag == 0)
            .all()
        }
    party_map: Dict[str, str] = {}
    if cell_pids:
        for party in (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id.in_(cell_pids),
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .all()
        ):
            party_map[party.collection_protocol_id] = party.mother_name or ""

    donation_map: Dict[str, BizResearchDonation] = {}
    if donation_pids:
        donation_map = {
            d.id: d
            for d in db.query(BizResearchDonation)
            .filter(BizResearchDonation.id.in_(donation_pids), BizResearchDonation.del_flag == 0)
            .all()
        }
    dparty_map: Dict[str, str] = {}
    if donation_pids:
        for party in (
            db.query(BizResearchDonationParty)
            .filter(
                BizResearchDonationParty.research_donation_id.in_(donation_pids),
                BizResearchDonationParty.del_flag == 0,
            )
            .all()
        ):
            dparty_map[party.research_donation_id] = party.mother_name or ""

    for item in items:
        source = item.get("receptionSource") or "cell_storage"
        pid = item.get("protocolId") or ""
        item.setdefault("motherName", "")
        item.setdefault("protocolStatus", "")
        item.setdefault("protocolStatusLabel", "")
        if source == "research_donation":
            donation = donation_map.get(pid)
            if donation:
                st = donation.donation_status or ""
                vt = donation.void_type or ""
                item["protocolStatus"] = st
                item["voidType"] = vt
                from app.core.protocol_status_display import format_protocol_status_label

                item["protocolStatusLabel"] = format_protocol_status_label(
                    st, vt, kind="donation", db=db, tenant_no=donation.tenant_no
                )
            item["motherName"] = dparty_map.get(pid, "")
        else:
            proto = proto_map.get(pid)
            if proto:
                st = getattr(proto, "protocol_status", "") or ""
                vt = getattr(proto, "void_type", "") or ""
                item["protocolStatus"] = st
                item["voidType"] = vt
                from app.core.protocol_status_display import format_protocol_status_label

                item["protocolStatusLabel"] = format_protocol_status_label(
                    st, vt, kind="protocol", db=db, tenant_no=proto.tenant_no
                )
            item["motherName"] = party_map.get(pid, "")
    return items


def _get_user_name(db: Session, user_id: str) -> str:
    u = db.query(SysUser).filter(SysUser.id == user_id).first()
    return (u.nick_name or "") if u else ""


def get_review_detail(db: Session, review_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
    rev = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.id == review_id,
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
        )
        .first()
    )
    if not rev:
        return None

    result: Dict[str, Any] = {"review": _row_to_dict(rev)}
    source = rev.reception_source or "cell_storage"

    if source == "cell_storage":
        protocol_info = _build_cell_storage_protocol_info(db, rev)
        result["protocolInfo"] = protocol_info
        transport_info = _build_cell_storage_transport_info(db, rev)
        result["transportInfo"] = transport_info
    else:
        protocol_info = _build_research_donation_protocol_info(db, rev)
        result["protocolInfo"] = protocol_info
        transport_info = _build_research_donation_transport_info(db, rev)
        result["transportInfo"] = transport_info

    from app.crud import sample_reception as sample_reception_svc

    reception_detail = sample_reception_svc.get_reception_by_id(
        db, rev.reception_id or "", tenant_no
    )
    if reception_detail:
        result["reception"] = reception_detail
        result["receptionInfo"] = {
            "receptionTime": reception_detail.get("receptionTime") or "",
            "handlerName": reception_detail.get("handlerName") or "",
            "receiveTemperature": reception_detail.get("receiveTemperature") or "",
            "volumeMl": reception_detail.get("volumeMl"),
            "netWeightG": reception_detail.get("netWeightG"),
            "acceptConclusion": reception_detail.get("acceptConclusion") or "",
            "acceptConclusionLabel": reception_detail.get("acceptConclusionLabel") or "",
            "labStorageCode": reception_detail.get("labStorageCode") or "",
            "remark": reception_detail.get("remark") or "",
            "checkResults": reception_detail.get("checkResults") or [],
            "receiveAttachments": reception_detail.get("receiveAttachments") or "",
            "receiveAttachmentFiles": reception_detail.get("receiveAttachmentFiles") or [],
            "transportType": reception_detail.get("transportType") or "",
            "expressCompanyName": reception_detail.get("expressCompanyName") or "",
            "waybillNo": reception_detail.get("waybillNo") or "",
            "carrierName": reception_detail.get("carrierName") or "",
            "carrierPhone": reception_detail.get("carrierPhone") or "",
            "etaTime": reception_detail.get("etaTime") or "",
        }
    else:
        result["receptionInfo"] = _build_reception_info(db, rev)

    result["qcInfo"] = _build_gate_qc_info(db, rev, tenant_no)
    result["processQcInfoList"] = _build_process_qc_info_list(db, rev, tenant_no)
    result["productionInfo"] = _build_production_info(db, rev, tenant_no)

    return result


def _build_cell_storage_protocol_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
    proto = (
        db.query(BizCollectionProtocol)
        .filter(
            BizCollectionProtocol.id == (rev.protocol_id or ""),
            BizCollectionProtocol.del_flag == 0,
        )
        .first()
    )
    if not proto:
        return {}
    sample = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.collection_protocol_id == proto.id,
            BizProtocolSample.sample_code == (rev.sample_code or ""),
            BizProtocolSample.tenant_no == rev.tenant_no,
            BizProtocolSample.del_flag == 0,
        )
        .first()
    )
    party = (
        db.query(BizProtocolPartyCustomer)
        .filter(
            BizProtocolPartyCustomer.collection_protocol_id == proto.id,
            BizProtocolPartyCustomer.del_flag == 0,
        )
        .first()
    )
    info: Dict[str, Any] = {
        "protocolNo": proto.protocol_no or "",
        "protocolStatus": getattr(proto, "protocol_status", "") or "",
        "motherName": "",
        "motherPhone": "",
        "fatherName": "",
        "fatherPhone": "",
        "newbornName": "",
        "sampleCode": rev.sample_code or "",
        "sampleTypeName": "",
        "collectAmount": None,
        "storageYears": None,
    }
    if sample:
        from app.crud.qc_work_order import _load_sample_type_name_map
        type_map = _load_sample_type_name_map(db, [sample.sample_type_id]) if sample.sample_type_id else {}
        info["sampleTypeName"] = type_map.get(sample.sample_type_id or "", "")
        info["collectAmount"] = float(sample.collect_amount) if sample.collect_amount else None
        info["storageYears"] = sample.storage_years
        info.update(sample_special_to_dict(sample))
    if party:
        info["motherName"] = party.mother_name or ""
        info["motherPhone"] = party.mother_phone or ""
        info["fatherName"] = party.father_name or ""
        info["fatherPhone"] = party.father_phone or ""
        info["newbornName"] = party.newborn_name or ""
    return info


def _build_research_donation_protocol_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
    donation = (
        db.query(BizResearchDonation)
        .filter(
            BizResearchDonation.id == (rev.protocol_id or ""),
            BizResearchDonation.del_flag == 0,
        )
        .first()
    )
    if not donation:
        return {}
    dsample = (
        db.query(BizResearchDonationSample)
        .filter(
            BizResearchDonationSample.research_donation_id == donation.id,
            BizResearchDonationSample.sample_code == (rev.sample_code or ""),
            BizResearchDonationSample.tenant_no == rev.tenant_no,
            BizResearchDonationSample.del_flag == 0,
        )
        .first()
    )
    party = (
        db.query(BizResearchDonationParty)
        .filter(
            BizResearchDonationParty.research_donation_id == donation.id,
            BizResearchDonationParty.del_flag == 0,
        )
        .first()
    )
    info: Dict[str, Any] = {
        "protocolNo": donation.donation_no or "",
        "protocolStatus": donation.donation_status or "",
        "motherName": "",
        "motherPhone": "",
        "fatherName": "",
        "fatherPhone": "",
        "newbornName": "",
        "sampleCode": rev.sample_code or "",
        "sampleTypeName": "",
        "collectAmount": None,
        "storageYears": None,
    }
    if dsample:
        from app.crud.qc_work_order import _load_sample_type_name_map
        type_map = _load_sample_type_name_map(db, [dsample.sample_type_id]) if dsample.sample_type_id else {}
        info["sampleTypeName"] = type_map.get(dsample.sample_type_id or "", "")
        info["collectAmount"] = float(dsample.collect_amount) if dsample.collect_amount else None
        info["storageYears"] = dsample.storage_years
        info.update(sample_special_to_dict(dsample))
    if party:
        info["motherName"] = party.mother_name or ""
        info["motherPhone"] = party.mother_phone or ""
        info["fatherName"] = party.father_name or ""
        info["fatherPhone"] = party.father_phone or ""
        info["newbornName"] = party.newborn_name or ""
    return info


def _build_cell_storage_transport_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
    sample = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.sample_code == (rev.sample_code or ""),
            BizProtocolSample.collection_protocol_id == (rev.protocol_id or ""),
            BizProtocolSample.tenant_no == rev.tenant_no,
            BizProtocolSample.del_flag == 0,
        )
        .first()
    )
    if not sample:
        return {}
    transport = (
        db.query(BizProtocolSampleTransport)
        .filter(
            BizProtocolSampleTransport.sample_id == sample.id,
            BizProtocolSampleTransport.collection_protocol_id == (rev.protocol_id or ""),
            BizProtocolSampleTransport.del_flag == 0,
        )
        .first()
    )
    if not transport:
        return {}
    return {
        "transportType": transport.transport_type or "",
        "waybillNo": transport.waybill_no or "",
        "expressFee": float(transport.express_fee) if transport.express_fee else None,
        "etaTime": _fmt_dt(transport.eta_time),
        "carrierName": transport.carrier_name or "",
        "carrierPhone": transport.carrier_phone or "",
    }


def _build_research_donation_transport_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
    dsample = (
        db.query(BizResearchDonationSample)
        .filter(
            BizResearchDonationSample.sample_code == (rev.sample_code or ""),
            BizResearchDonationSample.research_donation_id == (rev.protocol_id or ""),
            BizResearchDonationSample.tenant_no == rev.tenant_no,
            BizResearchDonationSample.del_flag == 0,
        )
        .first()
    )
    if not dsample:
        return {}
    transport = (
        db.query(BizResearchDonationSampleTransport)
        .filter(
            BizResearchDonationSampleTransport.sample_id == dsample.id,
            BizResearchDonationSampleTransport.research_donation_id == (rev.protocol_id or ""),
            BizResearchDonationSampleTransport.del_flag == 0,
        )
        .first()
    )
    if not transport:
        return {}
    return {
        "transportType": transport.transport_type or "",
        "waybillNo": transport.waybill_no or "",
        "expressFee": float(transport.express_fee) if transport.express_fee else None,
        "etaTime": _fmt_dt(transport.eta_time),
        "carrierName": transport.carrier_name or "",
        "carrierPhone": transport.carrier_phone or "",
    }


def _build_reception_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
    rec = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.id == (rev.reception_id or ""),
            BizSampleReception.tenant_no == rev.tenant_no,
            BizSampleReception.del_flag == 0,
        )
        .first()
    )
    if not rec:
        return {}
    return {
        "receptionTime": _fmt_dt(rec.reception_time),
        "handlerName": rec.handler_name or "",
        "receiveTemperature": rec.receive_temperature or "",
        "volumeMl": float(rec.volume_ml) if rec.volume_ml else None,
        "netWeightG": float(rec.net_weight_g) if rec.net_weight_g else None,
        "acceptConclusion": rec.accept_conclusion or "",
        "labStorageCode": rec.lab_storage_code or "",
    }


def _build_gate_qc_info(
    db: Session, rev: BizStorageReleaseReview, tenant_no: str
) -> Optional[Dict[str, Any]]:
    """入库放行门禁质检（仅 trigger_source=reception）。"""
    from app.crud import qc_work_order as qc_work_order_svc

    code = (rev.sample_code or "").strip()
    q = db.query(BizQcWorkOrder).filter(
        BizQcWorkOrder.tenant_no == tenant_no,
        BizQcWorkOrder.del_flag == 0,
        BizQcWorkOrder.trigger_source == "reception",
    )
    wo = q.filter(BizQcWorkOrder.sample_code == code).first() if code else None
    if not wo and rev.reception_id:
        wo = q.filter(BizQcWorkOrder.reception_id == rev.reception_id).first()
    if not wo:
        return None
    return qc_work_order_svc._qc_info_item(db, wo, tenant_no)


def _build_process_qc_info_list(
    db: Session, rev: BizStorageReleaseReview, tenant_no: str
) -> List[Dict[str, Any]]:
    """生产过程触发的入库质检（只读展示，不参与放行判定）。"""
    from app.crud import qc_work_order as qc_work_order_svc

    return qc_work_order_svc.list_process_qc_info_for_sample(
        db, sample_code=rev.sample_code or "", tenant_no=tenant_no
    )


def _build_production_info(
    db: Session, rev: BizStorageReleaseReview, tenant_no: str
) -> Optional[Dict[str, Any]]:
    from app.crud import production_work_order as production_work_order_svc

    code = (rev.sample_code or "").strip()
    q = db.query(BizProductionWorkOrder).filter(
        BizProductionWorkOrder.tenant_no == tenant_no,
        BizProductionWorkOrder.del_flag == 0,
    )
    pwo = q.filter(BizProductionWorkOrder.sample_code == code).first() if code else None
    if not pwo and rev.reception_id:
        pwo = q.filter(BizProductionWorkOrder.reception_id == rev.reception_id).first()
    if not pwo:
        return None
    return production_work_order_svc._production_info_item(db, pwo)


def submit_review(
    db: Session,
    review_id: str,
    *,
    review_side: str,
    review_conclusion: str,
    review_remark: Optional[str] = None,
    review_attachments: Optional[str] = None,
    tenant_no: str,
    user_id: str,
) -> BizStorageReleaseReview:
    rev = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.id == review_id,
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
        )
        .first()
    )
    if not rev:
        raise ValueError("审核记录不存在")
    side = normalize_review_side(review_side)
    if side not in ("production", "qc"):
        raise ValueError("审核侧无效")
    conclusion = (review_conclusion or "").strip()
    if conclusion not in ("tg", "btg"):
        raise ValueError("审核结论无效")
    user_name = _get_user_name(db, user_id)
    now = datetime.now()
    remark = (review_remark or "").strip() or None
    attachments = (review_attachments or "").strip() or None
    if side == "production":
        rev.prod_review_conclusion = conclusion
        rev.prod_review_remark = remark
        rev.prod_review_attachments = attachments
        rev.prod_review_by = user_id
        rev.prod_review_by_name = user_name
        rev.prod_review_time = now
    else:
        rev.qc_review_conclusion = conclusion
        rev.qc_review_remark = remark
        rev.qc_review_attachments = attachments
        rev.qc_review_by = user_id
        rev.qc_review_by_name = user_name
        rev.qc_review_time = now
    agg = aggregate_review_conclusion(rev.prod_review_conclusion, rev.qc_review_conclusion)
    rev.review_conclusion = agg
    rev.review_remark = remark
    rev.review_attachments = attachments
    rev.review_by = user_id
    rev.review_by_name = user_name
    rev.review_time = now
    rev.update_by = user_id
    rev.update_time = now
    db.add(rev)
    db.flush()
    action = storage_transition(
        rev.prod_review_conclusion, rev.qc_review_conclusion, rev.storage_status
    )
    if action == "store":
        store_sample(db, review_id, tenant_no=tenant_no, user_id=user_id)
        db.refresh(rev)
    elif action == "unstore":
        unstore_sample(db, review_id, tenant_no=tenant_no, user_id=user_id)
        db.refresh(rev)
    return rev


def batch_submit_review(
    db: Session,
    ids: List[str],
    *,
    review_side: str,
    review_conclusion: str,
    review_remark: Optional[str] = None,
    review_attachments: Optional[str] = None,
    tenant_no: str,
    user_id: str,
) -> Dict[str, Any]:
    success = 0
    fail = 0
    for rid in ids:
        try:
            submit_review(
                db,
                rid,
                review_side=review_side,
                review_conclusion=review_conclusion,
                review_remark=review_remark,
                review_attachments=review_attachments,
                tenant_no=tenant_no,
                user_id=user_id,
            )
            success += 1
        except Exception:
            fail += 1
    return {"success": success, "fail": fail}


def store_sample(
    db: Session,
    review_id: str,
    *,
    tenant_no: str,
    user_id: str,
) -> BizStorageReleaseReview:
    rev = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.id == review_id,
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
        )
        .first()
    )
    if not rev:
        raise ValueError("审核记录不存在")
    if aggregate_review_conclusion(rev.prod_review_conclusion, rev.qc_review_conclusion) != "tg":
        raise ValueError("生产与质检均通过后才能入库")
    if rev.storage_status != "drk":
        raise ValueError("当前状态不允许入库操作")
    user_name = _get_user_name(db, user_id)
    rev.storage_status = "yrk"
    rev.storage_by = user_id
    rev.storage_by_name = user_name
    rev.storage_time = datetime.now()
    rev.update_by = user_id
    rev.update_time = datetime.now()
    db.add(rev)
    storage_in_yyyymmdd = rev.storage_time.strftime("%Y%m%d")

    if (rev.reception_source or "cell_storage") == "cell_storage":
        sample_code = (rev.sample_code or "").strip()
        if sample_code:
            ps = (
                db.query(BizProtocolSample)
                .filter(
                    BizProtocolSample.sample_code == sample_code,
                    BizProtocolSample.tenant_no == tenant_no,
                    BizProtocolSample.del_flag == 0,
                )
                .first()
            )
            if ps:
                ps.sample_status = "yrk"
                ps.storage_in_date = storage_in_yyyymmdd
                ps.update_time = datetime.now()
                db.add(ps)

    db.flush()
    return rev


def unstore_sample(
    db: Session,
    review_id: str,
    *,
    tenant_no: str,
    user_id: str,
) -> BizStorageReleaseReview:
    rev = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.id == review_id,
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
        )
        .first()
    )
    if not rev:
        raise ValueError("审核记录不存在")
    if (rev.storage_status or "") != "yrk":
        return rev
    rev.storage_status = "drk"
    rev.storage_by = None
    rev.storage_by_name = None
    rev.storage_time = None
    rev.update_by = user_id
    rev.update_time = datetime.now()
    db.add(rev)
    if (rev.reception_source or "cell_storage") == "cell_storage":
        sample_code = (rev.sample_code or "").strip()
        if sample_code:
            ps = (
                db.query(BizProtocolSample)
                .filter(
                    BizProtocolSample.sample_code == sample_code,
                    BizProtocolSample.tenant_no == tenant_no,
                    BizProtocolSample.del_flag == 0,
                )
                .first()
            )
            if ps and (ps.sample_status or "") == "yrk":
                ps.sample_status = "jcz"
                ps.storage_in_date = None
                ps.update_time = datetime.now()
                db.add(ps)
    db.flush()
    return rev


def batch_store_samples(
    db: Session,
    ids: List[str],
    *,
    tenant_no: str,
    user_id: str,
) -> Dict[str, Any]:
    success = 0
    fail = 0
    for rid in ids:
        try:
            store_sample(db, rid, tenant_no=tenant_no, user_id=user_id)
            success += 1
        except Exception:
            fail += 1
    return {"success": success, "fail": fail}


def get_storage_release_map_for_samples(
    db: Session,
    sample_ids: List[str],
    tenant_no: str,
    *,
    sample_code_by_id: Optional[Dict[str, str]] = None,
) -> Dict[str, Dict[str, Any]]:
    """按样本批量返回存储放行审核摘要。

    当传入 sample_code_by_id 时，按租户 + 样本编码关联存储放行审核记录，
    返回以 sample_id 为 key 的审核信息映射。
    """
    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(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
            BizStorageReleaseReview.sample_code.in_(codes),
        )
        .all()
    )
    code_to_review: Dict[str, BizStorageReleaseReview] = {}
    for r in rows:
        code = (r.sample_code or "").strip()
        if code and code not in code_to_review:
            code_to_review[code] = r

    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
        rev = code_to_review.get(code)
        if not rev:
            continue
        rc = rev.review_conclusion or ""
        out[sid] = {
            "reviewConclusion": rc,
            "reviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(rc, rc),
            "reviewByName": rev.review_by_name or "",
            "reviewTime": _fmt_dt(rev.review_time),
            "reviewRemark": rev.review_remark or "",
            "reviewAttachments": rev.review_attachments or "",
            "qcReviewConclusion": rev.qc_review_conclusion or "",
            "qcReviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(rev.qc_review_conclusion or "", rev.qc_review_conclusion or ""),
            "qcReviewByName": rev.qc_review_by_name or "",
            "qcReviewTime": _fmt_dt(rev.qc_review_time),
            "qcReviewRemark": rev.qc_review_remark or "",
            "qcReviewAttachments": rev.qc_review_attachments or "",
            "prodReviewConclusion": rev.prod_review_conclusion or "",
            "prodReviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(rev.prod_review_conclusion or "", rev.prod_review_conclusion or ""),
            "prodReviewByName": rev.prod_review_by_name or "",
            "prodReviewTime": _fmt_dt(rev.prod_review_time),
            "prodReviewRemark": rev.prod_review_remark or "",
            "prodReviewAttachments": rev.prod_review_attachments or "",
            "storageStatus": rev.storage_status or "drk",
            "storageStatusLabel": _STORAGE_STATUS_LABELS.get(rev.storage_status or "drk", rev.storage_status or "drk"),
        }
    return out