167 lines
6.7 KiB
Python
167 lines
6.7 KiB
Python
"""样本信息列表：批量补进度标识（接收/质检/入库/证保险）与客户服务宽表字段。"""
|
|
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
|
|
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),
|
|
"_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] = {}
|
|
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 ""
|
|
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["insuranceHandleTime"] = _fmt_dt(getattr(ins_row, "handle_time", None)) 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"))
|