zkxw/backend/app/services/sample_info_list_enrich.py

"""样本信息列表：批量补进度标识（接收/质检/入库/证保险）。"""
from __future__ import annotations

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

from sqlalchemy.orm import Session

from app.models.models import (
    BizProtocolFullmoonTodo,
    BizProtocolStorageCert,
    BizStorageReleaseReview,
)
from app.services.protocol_sample_list_summary import (
    get_list_qc_conclusion_map_by_sample_codes,
    get_list_reception_summary_map_by_sample_codes,
)
from app.services.sample_info_progress import (
    collection_progress,
    insurance_handled,
    qc_progress,
    reception_progress,
    storage_cert_handled,
    storage_progress,
)


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


def get_storage_review_map_by_sample_codes(
    db: Session,
    tenant_no: str,
    sample_codes: List[str],
    *,
    reception_source: str,
) -> Dict[str, Dict[str, str]]:
    codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()})
    if not tenant_no or not codes or not (reception_source or "").strip():
        return {}
    rows = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
            BizStorageReleaseReview.reception_source == reception_source,
            BizStorageReleaseReview.sample_code.in_(codes),
        )
        .all()
    )
    out: Dict[str, Dict[str, Any]] = {}
    for rev in rows:
        code = (rev.sample_code or "").strip()
        if not code:
            continue
        prev = out.get(code)
        prev_t = prev.get("_t") if prev else None
        cur_t = rev.storage_time or rev.update_time or rev.create_time
        if prev and cur_t and prev_t and cur_t <= prev_t:
            continue
        if prev and cur_t is None and prev_t is not None:
            continue
        out[code] = {
            "storageStatus": (rev.storage_status or "").strip(),
            "reviewConclusion": (rev.review_conclusion or "").strip(),
            "storageTime": _fmt_dt(rev.storage_time),
            "_t": cur_t,
        }
    for v in out.values():
        v.pop("_t", None)
    return out


def enrich_sample_info_items(
    db: Session,
    tenant_no: str,
    items: List[Dict[str, Any]],
    *,
    reception_source: str,
    include_handle_flags: bool,
) -> None:
    if not items or not tenant_no:
        return
    codes = [str(i.get("sampleCode") or "").strip() for i in items]
    codes = [c for c in codes if c]
    rec_map = get_list_reception_summary_map_by_sample_codes(
        db, tenant_no, codes, reception_source=reception_source
    )
    qc_map = get_list_qc_conclusion_map_by_sample_codes(
        db, tenant_no, codes, reception_source=reception_source
    )
    st_map = get_storage_review_map_by_sample_codes(
        db, tenant_no, codes, reception_source=reception_source
    )
    cert_map: Dict[str, Any] = {}
    ins_map: Dict[str, Any] = {}
    if include_handle_flags:
        pids = list({str(i.get("collectionProtocolId") or "").strip() for i in items if i.get("collectionProtocolId")})
        if pids:
            for sc in (
                db.query(BizProtocolStorageCert)
                .filter(
                    BizProtocolStorageCert.collection_protocol_id.in_(pids),
                    BizProtocolStorageCert.tenant_no == tenant_no,
                    BizProtocolStorageCert.del_flag == 0,
                )
                .all()
            ):
                cert_map[sc.collection_protocol_id] = sc
            for ft in (
                db.query(BizProtocolFullmoonTodo)
                .filter(
                    BizProtocolFullmoonTodo.collection_protocol_id.in_(pids),
                    BizProtocolFullmoonTodo.tenant_no == tenant_no,
                    BizProtocolFullmoonTodo.del_flag == 0,
                )
                .all()
            ):
                ins_map[ft.collection_protocol_id] = ft

    for item in items:
        code = str(item.get("sampleCode") or "").strip()
        cflag, clabel = collection_progress(str(item.get("sampleStatus") or ""))
        item["collectionFlag"] = cflag
        item["collectionFlagLabel"] = clabel
        rec = rec_map.get(code)
        rflag, rlabel = reception_progress(rec)
        item["receptionFlag"] = rflag
        item["receptionFlagLabel"] = rlabel
        item["receptionTime"] = (rec or {}).get("receptionTime") or ""
        qc = qc_map.get(code)
        qflag, qlabel = qc_progress(qc)
        item["qcFlag"] = qflag
        item["qcFlagLabel"] = qlabel
        item["qcCompleteTime"] = ""
        st = st_map.get(code)
        sflag, slabel = storage_progress(st)
        item["storageFlag"] = sflag
        item["storageFlagLabel"] = slabel
        item["storageTime"] = (st or {}).get("storageTime") or ""
        if include_handle_flags:
            pid = str(item.get("collectionProtocolId") or "")
            item["storageCertHandled"] = storage_cert_handled(cert_map.get(pid))
            item["insuranceHandled"] = insurance_handled(ins_map.get(pid))