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.crud import sample_customer_service as sample_cs_crud
from app.crud import sample_insurance as sample_insurance_crud
from app.crud import sample_storage_cert as sample_storage_cert_crud
from app.models.models import BizStorageReleaseReview, SysUser
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 _has_urls(raw: Optional[str]) -> bool:
    return bool((raw or "").strip())


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),
            "storageBy": (rev.storage_by or "").strip(),
            "storageByName": (rev.storage_by_name or "").strip(),
            "storageAttachments": (rev.review_attachments or "").strip(),
            "storageRemark": (rev.remark or "").strip(),
            "_t": cur_t,
        }
    for v in out.values():
        v.pop("_t", None)
    return out


def _user_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
    ids = list({(uid or "").strip() for uid in user_ids if (uid or "").strip()})
    if not ids:
        return {}
    rows = (
        db.query(SysUser.id, SysUser.nick_name)
        .filter(SysUser.id.in_(ids), SysUser.del_flag == 0)
        .all()
    )
    return {str(row[0]): (row[1] or "") for row in rows}


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] = {}
    cs_map: Dict[str, Any] = {}
    if include_handle_flags and reception_source == "cell_storage":
        cert_map = sample_storage_cert_crud.get_map_by_sample_codes(db, tenant_no, codes)
        ins_map = sample_insurance_crud.get_map_by_sample_codes(db, tenant_no, codes)
        cs_map = sample_cs_crud.get_map_by_sample_codes(db, tenant_no, codes)

    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 ""
        item["storageBy"] = (st or {}).get("storageBy") or ""
        item["storageByName"] = (st or {}).get("storageByName") or ""
        item["storageAttachments"] = (st or {}).get("storageAttachments") or ""
        item["storageRemark"] = (st or {}).get("storageRemark") or ""
        if include_handle_flags:
            cert_row = cert_map.get(code)
            ins_row = ins_map.get(code)
            cert_done = storage_cert_handled(cert_row)
            ins_done = insurance_handled(ins_row)
            item["storageCertHandled"] = cert_done
            item["insuranceHandled"] = ins_done
            item["mailCertTime"] = _fmt_dt(getattr(cert_row, "mail_cert_time", None)) if cert_row else ""
            item["storageCertHandleTime"] = (
                _fmt_dt(getattr(cert_row, "handle_time", None)) if cert_row else ""
            )
            item["storageCertHandlerUserId"] = (
                getattr(cert_row, "handler_user_id", None) or ""
            ) if cert_row else ""
            item["storageCertAttachments"] = (
                getattr(cert_row, "cert_image_urls", None) or ""
            ) if cert_row else ""
            item["storageCertRemark"] = (getattr(cert_row, "remark", None) or "") if cert_row else ""
            item["insuranceHandleTime"] = _fmt_dt(getattr(ins_row, "handle_time", None)) if ins_row else ""
            item["insuranceHandlerUserId"] = (
                getattr(ins_row, "handler_user_id", None) or ""
            ) if ins_row else ""
            item["insuranceAttachments"] = (
                getattr(ins_row, "insurance_attachments", None) or ""
            ) if ins_row else ""
            item["insuranceRemark"] = (getattr(ins_row, "handle_remark", None) or "") if ins_row else ""
            item["policySendTime"] = _fmt_dt(getattr(ins_row, "policy_send_time", None)) if ins_row else ""
            # 购买保险时间 = 保险办理时间
            item["purchaseInsuranceTime"] = item["insuranceHandleTime"]

            cs_row = cs_map.get(code)
            if cs_row:
                cs = sample_cs_crud.row_to_dict(cs_row)
                item["smsSendDate"] = cs.get("smsSendDate") or ""
                item["smsSendStatus"] = cs.get("smsSendStatus") or ""
                item["microbeFollowupDate"] = cs.get("microbeFollowupDate") or ""
                item["microbeFollowupStatus"] = cs.get("microbeFollowupStatus") or ""
                item["fullmoonFollowupDate"] = cs.get("fullmoonFollowupDate") or ""
                item["fullmoonFollowupStatus"] = cs.get("fullmoonFollowupStatus") or ""
                item["householdRegistered"] = cs.get("householdRegistered") or ""
                item["einvoiceSendTime"] = cs.get("einvoiceSendTime") or ""
                item["effectiveAddress"] = cs.get("effectiveAddress") or ""
            else:
                item["smsSendDate"] = ""
                item["smsSendStatus"] = ""
                item["microbeFollowupDate"] = ""
                item["microbeFollowupStatus"] = ""
                item["fullmoonFollowupDate"] = ""
                item["fullmoonFollowupStatus"] = ""
                item["householdRegistered"] = ""
                item["einvoiceSendTime"] = ""
                item["effectiveAddress"] = ""

            item["protocolAttachmentPresent"] = _has_urls(item.get("protocolAttachments"))
            item["insuranceDocPresent"] = bool(ins_done)
            item["informedConsentPresent"] = _has_urls(item.get("consentFormUrls"))
            item["healthSurveyPresent"] = _has_urls(item.get("healthSurveyFormUrls"))

    name_ids = [
        str(item.get("storageBy") or "")
        for item in items
        if str(item.get("storageBy") or "").strip()
        and not str(item.get("storageByName") or "").strip()
    ]
    if include_handle_flags:
        name_ids.extend(
            str(item.get("storageCertHandlerUserId") or "")
            for item in items
            if str(item.get("storageCertHandlerUserId") or "").strip()
        )
        name_ids.extend(
            str(item.get("insuranceHandlerUserId") or "")
            for item in items
            if str(item.get("insuranceHandlerUserId") or "").strip()
        )
    name_map = _user_name_map(db, name_ids)
    for item in items:
        if not str(item.get("storageByName") or "").strip():
            item["storageByName"] = name_map.get(str(item.get("storageBy") or ""), "")
        if include_handle_flags:
            cert_uid = str(item.get("storageCertHandlerUserId") or "")
            ins_uid = str(item.get("insuranceHandlerUserId") or "")
            item["storageCertHandlerName"] = name_map.get(cert_uid, "") if cert_uid else ""
            item["insuranceHandlerName"] = name_map.get(ins_uid, "") if ins_uid else ""