"""实验室样本处理台账列表与补录 CRUD。""" from __future__ import annotations from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import and_ from sqlalchemy.orm import Session from app.core.data_scope import apply_data_scope from app.core.utils import generate_uuid from app.crud import sample_production_reception as prod_svc from app.models.models import ( BizCellSampleType, BizCollectionHospital, BizCollectionProtocol, BizLabSampleProcessSupplement, BizProtocolCollectionInfo, BizProtocolPartyCustomer, BizProtocolSample, BizSampleProductionReception, BizSampleReception, BizSampleReceptionCheckResult, BizStorageReleaseReview, ) from app.models.research_donation import ( BizResearchDonation, BizResearchDonationCollection, BizResearchDonationParty, BizResearchDonationSample, ) from app.services import lab_sample_process as ledger def _fmt_raw(dt: Optional[datetime]) -> str: return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else "" def _normalize_bound(value: Optional[str], *, end_of_day: bool = False) -> Optional[datetime]: text = (value or "").strip() if not text: return None for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): try: dt = datetime.strptime(text[:19] if " " in text else text[:10], fmt) if end_of_day and fmt == "%Y-%m-%d": return dt.replace(hour=23, minute=59, second=59) return dt except ValueError: continue return None def _type_ids_by_name(db: Session, tenant_no: str, type_name: str) -> List[str]: rows = ( db.query(BizCellSampleType.id) .filter( BizCellSampleType.tenant_no == tenant_no, BizCellSampleType.del_flag == 0, BizCellSampleType.type_name == type_name, ) .all() ) return [r.id for r in rows if r.id] def _get_supplement_map( db: Session, tenant_no: str, codes: List[str] ) -> Dict[str, Dict[str, Any]]: codes = [c for c in {(x or "").strip() for x in codes} if c] if not codes: return {} rows = ( db.query(BizLabSampleProcessSupplement) .filter( BizLabSampleProcessSupplement.tenant_no == tenant_no, BizLabSampleProcessSupplement.del_flag == 0, BizLabSampleProcessSupplement.lab_storage_code.in_(codes), ) .all() ) out: Dict[str, Dict[str, Any]] = {} for row in rows: out[row.lab_storage_code] = { "id": row.id, "payload": ledger.parse_payload(row.payload), "remark": row.remark or "", "updateTime": _fmt_raw(row.update_time), } return out def get_supplement( db: Session, *, tenant_no: str, lab_storage_code: str ) -> Optional[Dict[str, Any]]: code = (lab_storage_code or "").strip() if not code: return None row = ( db.query(BizLabSampleProcessSupplement) .filter( BizLabSampleProcessSupplement.tenant_no == tenant_no, BizLabSampleProcessSupplement.lab_storage_code == code, BizLabSampleProcessSupplement.del_flag == 0, ) .first() ) if not row: return None return { "id": row.id, "labStorageCode": row.lab_storage_code, "receptionSource": row.reception_source, "sampleTypeName": row.sample_type_name or "", "payload": ledger.parse_payload(row.payload), "remark": row.remark or "", "createTime": _fmt_raw(row.create_time), "updateTime": _fmt_raw(row.update_time), } def upsert_supplement( db: Session, *, tenant_no: str, user_id: str, dept_id: str, lab_storage_code: str, reception_source: str, sample_type_name: str, payload: Dict[str, Any], remark: Optional[str] = None, ) -> Dict[str, Any]: code = (lab_storage_code or "").strip() if not code: raise ValueError("实验室入库编码不能为空") src = (reception_source or "").strip() or "cell_storage" if src not in ("cell_storage", "research_donation"): raise ValueError("来源类型无效") type_name = (sample_type_name or "").strip() clean_payload = { str(k): ("" if v is None else str(v).strip()) for k, v in (payload or {}).items() if str(k).strip() } row = ( db.query(BizLabSampleProcessSupplement) .filter( BizLabSampleProcessSupplement.tenant_no == tenant_no, BizLabSampleProcessSupplement.lab_storage_code == code, BizLabSampleProcessSupplement.del_flag == 0, ) .first() ) now = datetime.now() if row: row.reception_source = src row.sample_type_name = type_name or row.sample_type_name row.payload = ledger.dumps_payload(clean_payload) if remark is not None: row.remark = (remark or "").strip() or None row.update_by = user_id row.update_time = now else: row = BizLabSampleProcessSupplement( id=generate_uuid(), lab_storage_code=code, reception_source=src, sample_type_name=type_name or None, payload=ledger.dumps_payload(clean_payload), tenant_no=tenant_no, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id=dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(row) db.commit() db.refresh(row) return get_supplement(db, tenant_no=tenant_no, lab_storage_code=code) or {} def _check_flag_map(db: Session, tenant_no: str, reception_ids: List[str]) -> Dict[str, Dict[str, bool]]: ids = [i for i in reception_ids if i] if not ids: return {} rows = ( db.query(BizSampleReceptionCheckResult) .filter( BizSampleReceptionCheckResult.tenant_no == tenant_no, BizSampleReceptionCheckResult.del_flag == 0, BizSampleReceptionCheckResult.reception_id.in_(ids), ) .all() ) out: Dict[str, Dict[str, bool]] = {} for r in rows: bucket = out.setdefault(r.reception_id, {}) name = (r.item_name or "").strip() if name: bucket[name] = bool(r.checked) return out def _storage_status_map(db: Session, tenant_no: str, sample_codes: List[str]) -> Dict[str, str]: codes = [c for c in sample_codes if c] if not codes: return {} rows = ( db.query(BizStorageReleaseReview.sample_code, BizStorageReleaseReview.storage_status) .filter( BizStorageReleaseReview.tenant_no == tenant_no, BizStorageReleaseReview.del_flag == 0, BizStorageReleaseReview.sample_code.in_(codes), ) .all() ) labels = {"drk": "待入库", "yrk": "已入库"} return {r.sample_code: labels.get(r.storage_status or "", r.storage_status or "") for r in rows} def _hospital_name_map(db: Session, hospital_ids: List[str]) -> Dict[str, str]: ids = [i for i in hospital_ids if i] if not ids: return {} rows = ( db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name) .filter(BizCollectionHospital.id.in_(ids), BizCollectionHospital.del_flag == 0) .all() ) return {r.id: r.hospital_name or "" for r in rows} def _build_cell_system( *, tab_key: str, rec: BizSampleReception, sample: BizProtocolSample, party: Optional[BizProtocolPartyCustomer], coll: Optional[BizProtocolCollectionInfo], protocol: Optional[BizCollectionProtocol], prod: Optional[Dict[str, Any]], hospital_name: str, check_flags: Dict[str, bool], storage_status: str, qc_mother: Dict[str, str], qc_process: Dict[str, str], ) -> Dict[str, Any]: prod = prod or {} maternal_time = coll.maternal_blood_collect_time if coll else None clear_flag = None for name, checked in check_flags.items(): if "标识清晰" in name or "样本信息正确" in name: clear_flag = checked break base: Dict[str, Any] = { "receptionId": rec.id, "sampleCode": (rec.sample_code or sample.sample_code or "").strip(), "labStorageCode": (rec.lab_storage_code or "").strip(), "motherName": (party.mother_name if party else "") or "", "collectionTime": ledger.format_cn_datetime(coll.sample_collect_time if coll else None), "collectionTimeRaw": _fmt_raw(coll.sample_collect_time if coll else None), "hospitalName": hospital_name, "receptionTime": ledger.format_cn_datetime(rec.reception_time), "receptionTimeRaw": _fmt_raw(rec.reception_time), "receiveTemperature": rec.receive_temperature or "", "handlerName": rec.handler_name or "", "remarkReception": rec.remark or "", "createTime": ledger.format_cn_datetime(rec.create_time), "createTimeRaw": _fmt_raw(rec.create_time), "operatorName": prod.get("handlerName") or "", "remarkProcess": prod.get("remark") or "", "storageStatus": storage_status, } if tab_key == "cell_hsc": base.update( { "cellCode": (rec.lab_storage_code or "").strip(), "hasMaternalBlood": "是" if maternal_time else "否", "remeasureMaternalBlood": "", "consentForm": ledger.yes_no_from_urls( protocol.consent_form_urls if protocol else None ), "transportPackingList": "", "cordBloodReceiveConfirm": "", "healthSurvey": ledger.yes_no_from_urls( protocol.health_survey_form_urls if protocol else None ), "labelClear": ledger.yes_no_from_bool(clear_flag) if clear_flag is not None else "", "receiveCheckText": "", "originalBloodWeightG": prod.get("originalBloodWeightG"), "hesVolumeMl": prod.get("hesVolumeMl"), "bloodType": prod.get("bloodType") or "", "infectiousFourTestLabel": prod.get("infectiousFourTestResultLabel") or ledger.infectious_label(prod.get("infectiousFourTestResult")), "infectiousFourExternal": "", "preEnrichNucleatedCount": "", "enrichedBloodWeightG": prod.get("enrichedBloodWeightG"), "postEnrichNucleatedCount": "", "cryopreservationVolumeMl": prod.get("cryopreservationVolumeMl"), "cd34Count1e4": "", } ) else: # cell_msc base.update( { "labFlowCode": (rec.lab_storage_code or "").strip(), "cordCollectTime": ledger.format_cn_datetime( coll.sample_collect_time if coll else None ), "transportTempC": rec.receive_temperature or "", "cordReceiveTime": ledger.format_cn_datetime(rec.reception_time), "inoculationAmount": prod.get("inoculationAmount") or "", "cryopreservationTime": ledger.format_cn_datetime(prod.get("cryopreservationTime")), "storageTotalTubes": prod.get("storageTotalTubes"), "availableCellCount1e7": prod.get("availableCellCount1e7"), "remark": rec.remark or prod.get("remark") or "", "storageDateP2": "", } ) base.update(qc_process) base.update(qc_mother) return base def _build_donation_system( *, tab_key: str, rec: BizSampleReception, sample: BizResearchDonationSample, party: Optional[BizResearchDonationParty], coll: Optional[BizResearchDonationCollection], prod: Optional[Dict[str, Any]], storage_status: str, qc_mother: Dict[str, str], qc_process: Dict[str, str], ) -> Dict[str, Any]: prod = prod or {} base: Dict[str, Any] = { "receptionId": rec.id, "sampleCode": (rec.sample_code or sample.sample_code or "").strip(), "labStorageCode": (rec.lab_storage_code or "").strip(), "motherName": (party.mother_name if party else "") or "", "receptionTime": ledger.format_cn_datetime(rec.reception_time), "receptionTimeRaw": _fmt_raw(rec.reception_time), "receiveTemperature": rec.receive_temperature or "", "handlerName": rec.handler_name or "", "remarkReception": rec.remark or "", "createTime": ledger.format_cn_datetime(rec.create_time), "createTimeRaw": _fmt_raw(rec.create_time), "operatorName": prod.get("handlerName") or "", "remarkProcess": prod.get("remark") or "", "storageStatus": storage_status, } if tab_key == "donation_hsc": base.update( { "cellCode": (rec.lab_storage_code or "").strip(), "labMaterialCode": "", "inboundCode": (rec.lab_storage_code or "").strip(), "discardReason": "", "receiveCheckText": "", "originalBloodVolumeMl": prod.get("originalBloodVolumeMl"), "nucleatedCellCount1e8": prod.get("nucleatedCellCount1e8"), "preOpNucleatedCount": "", "infectiousFourTestLabel": prod.get("infectiousFourTestResultLabel") or ledger.infectious_label(prod.get("infectiousFourTestResult")), } ) # 科研捐赠母血 5 项(不含核酸) for k in ( "motherHbsag", "motherAntiHcv", "motherAntiTp", "motherAntiHiv", "motherCmvIgm", ): if qc_mother.get(k): base[k] = qc_mother[k] else: base.update( { "labFlowCode": (rec.lab_storage_code or "").strip(), "cordCollectTime": ledger.format_cn_datetime( coll.sample_collect_time if coll else None ), "transportTempC": rec.receive_temperature or "", "cordReceiveTime": ledger.format_cn_datetime(rec.reception_time), "inoculationAmount": prod.get("inoculationAmount") or "", "cryopreservationTime": ledger.format_cn_datetime(prod.get("cryopreservationTime")), "storageTotalTubes": prod.get("storageTotalTubes"), "availableCellCount1e7": prod.get("availableCellCount1e7"), "remark": rec.remark or prod.get("remark") or "", "storageDateP2": "", } ) base.update(qc_process) base.update(qc_mother) return base def list_ledger( db: Session, *, tab_key: str, page: int = 1, page_size: int = 10, tenant_no: str, lab_storage_code: Optional[str] = None, mother_name: Optional[str] = None, create_time_start: Optional[str] = None, create_time_end: Optional[str] = None, user_info=None, ) -> Tuple[List[Dict[str, Any]], int]: meta = ledger.TAB_META.get(tab_key) if not meta: raise ValueError("无效的 Tab") source = meta["source"] type_name = meta["sampleTypeName"] type_ids = _type_ids_by_name(db, tenant_no, type_name) if not type_ids: return [], 0 q = db.query(BizSampleReception).filter( BizSampleReception.tenant_no == tenant_no, BizSampleReception.del_flag == 0, BizSampleReception.reception_source == source, BizSampleReception.lab_storage_code.isnot(None), BizSampleReception.lab_storage_code != "", BizSampleReception.accept_conclusion == "js", ) code_kw = (lab_storage_code or "").strip() if code_kw: q = q.filter(BizSampleReception.lab_storage_code.like(f"%{code_kw}%")) start = _normalize_bound(create_time_start) end = _normalize_bound(create_time_end, end_of_day=True) if start: q = q.filter(BizSampleReception.create_time >= start) if end: q = q.filter(BizSampleReception.create_time <= end) if user_info: q = apply_data_scope(q, BizSampleReception, user_info, db, tenant_no) mother_kw = (mother_name or "").strip() # 按来源限制样本类型,并可选母亲姓名 if source == "cell_storage": sample_q = ( db.query(BizProtocolSample.id) .join( BizCollectionProtocol, and_( BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id, BizCollectionProtocol.del_flag == 0, BizCollectionProtocol.tenant_no == tenant_no, ), ) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_type_id.in_(type_ids), ) ) if mother_kw: sample_q = sample_q.join( BizProtocolPartyCustomer, and_( BizProtocolPartyCustomer.collection_protocol_id == BizProtocolSample.collection_protocol_id, BizProtocolPartyCustomer.del_flag == 0, BizProtocolPartyCustomer.tenant_no == tenant_no, ), ).filter(BizProtocolPartyCustomer.mother_name.like(f"%{mother_kw}%")) sample_ids = [r.id for r in sample_q.all()] if not sample_ids: return [], 0 q = q.filter(BizSampleReception.protocol_sample_id.in_(sample_ids)) else: sample_q = ( db.query(BizResearchDonationSample.id) .join( BizResearchDonation, and_( BizResearchDonation.id == BizResearchDonationSample.research_donation_id, BizResearchDonation.del_flag == 0, BizResearchDonation.tenant_no == tenant_no, ), ) .filter( BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, BizResearchDonationSample.sample_type_id.in_(type_ids), ) ) if mother_kw: sample_q = sample_q.join( BizResearchDonationParty, and_( BizResearchDonationParty.research_donation_id == BizResearchDonationSample.research_donation_id, BizResearchDonationParty.del_flag == 0, BizResearchDonationParty.tenant_no == tenant_no, ), ).filter(BizResearchDonationParty.mother_name.like(f"%{mother_kw}%")) sample_ids = [r.id for r in sample_q.all()] if not sample_ids: return [], 0 q = q.filter(BizSampleReception.donation_sample_id.in_(sample_ids)) total = q.count() page = max(1, int(page or 1)) page_size = min(max(1, int(page_size or 10)), 200) receptions = ( q.order_by(BizSampleReception.create_time.desc(), BizSampleReception.id.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) if not receptions: return [], int(total) reception_ids = [r.id for r in receptions] codes = [(r.lab_storage_code or "").strip() for r in receptions] sample_codes = [(r.sample_code or "").strip() for r in receptions] prod_map = _prod_map(db, tenant_no, reception_ids) check_map = _check_flag_map(db, tenant_no, reception_ids) storage_map = _storage_status_map(db, tenant_no, [c for c in sample_codes if c]) sup_map = _get_supplement_map(db, tenant_no, codes) items: List[Dict[str, Any]] = [] if source == "cell_storage": items = _assemble_cell_rows( db, tenant_no=tenant_no, tab_key=tab_key, receptions=receptions, prod_map=prod_map, check_map=check_map, storage_map=storage_map, sup_map=sup_map, ) else: items = _assemble_donation_rows( db, tenant_no=tenant_no, tab_key=tab_key, receptions=receptions, prod_map=prod_map, storage_map=storage_map, sup_map=sup_map, ) return items, int(total) def _prod_map(db: Session, tenant_no: str, reception_ids: List[str]) -> Dict[str, Dict[str, Any]]: rows = ( db.query(BizSampleProductionReception) .filter( BizSampleProductionReception.tenant_no == tenant_no, BizSampleProductionReception.del_flag == 0, BizSampleProductionReception.reception_id.in_(reception_ids), ) .all() ) return {r.reception_id: prod_svc._row_to_dict(r) for r in rows} def _assemble_cell_rows( db: Session, *, tenant_no: str, tab_key: str, receptions: List[BizSampleReception], prod_map: Dict[str, Dict[str, Any]], check_map: Dict[str, Dict[str, bool]], storage_map: Dict[str, str], sup_map: Dict[str, Dict[str, Any]], ) -> List[Dict[str, Any]]: sample_ids = [r.protocol_sample_id for r in receptions if r.protocol_sample_id] samples = ( db.query(BizProtocolSample) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.id.in_(sample_ids), ) .all() if sample_ids else [] ) sample_map = {s.id: s for s in samples} protocol_ids = list({s.collection_protocol_id for s in samples if s.collection_protocol_id}) parties = ( db.query(BizProtocolPartyCustomer) .filter( BizProtocolPartyCustomer.tenant_no == tenant_no, BizProtocolPartyCustomer.del_flag == 0, BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids), ) .all() if protocol_ids else [] ) party_map = {p.collection_protocol_id: p for p in parties} colls = ( db.query(BizProtocolCollectionInfo) .filter( BizProtocolCollectionInfo.tenant_no == tenant_no, BizProtocolCollectionInfo.del_flag == 0, BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids), ) .all() if protocol_ids else [] ) coll_map = {c.collection_protocol_id: c for c in colls} protocols = ( db.query(BizCollectionProtocol) .filter( BizCollectionProtocol.tenant_no == tenant_no, BizCollectionProtocol.del_flag == 0, BizCollectionProtocol.id.in_(protocol_ids), ) .all() if protocol_ids else [] ) protocol_map = {p.id: p for p in protocols} hospital_ids = [c.hospital_id for c in colls if c.hospital_id] hospital_map = _hospital_name_map(db, hospital_ids) out: List[Dict[str, Any]] = [] for rec in receptions: sample = sample_map.get(rec.protocol_sample_id or "") if not sample: continue pid = sample.collection_protocol_id sample_code = (rec.sample_code or sample.sample_code or "").strip() tasks = ledger.load_qc_tasks_for_sample(db, tenant_no, sample_code) qc_mother = ledger.extract_mother_blood(tasks, include_nucleic=True) qc_process = ledger.extract_process_tests(tasks) system = _build_cell_system( tab_key=tab_key, rec=rec, sample=sample, party=party_map.get(pid), coll=coll_map.get(pid), protocol=protocol_map.get(pid), prod=prod_map.get(rec.id), hospital_name=hospital_map.get((coll_map.get(pid).hospital_id if coll_map.get(pid) else "") or "", ""), check_flags=check_map.get(rec.id, {}), storage_status=storage_map.get(sample_code, ""), qc_mother=qc_mother, qc_process=qc_process, ) lab_code = system.get("labStorageCode") or "" merged = ledger.merge_row( tab_key=tab_key, system_values=system, supplement=(sup_map.get(lab_code) or {}).get("payload"), ) merged["supplementId"] = (sup_map.get(lab_code) or {}).get("id") or "" merged["hasSupplement"] = bool((sup_map.get(lab_code) or {}).get("payload")) out.append(merged) return out def _assemble_donation_rows( db: Session, *, tenant_no: str, tab_key: str, receptions: List[BizSampleReception], prod_map: Dict[str, Dict[str, Any]], storage_map: Dict[str, str], sup_map: Dict[str, Dict[str, Any]], ) -> List[Dict[str, Any]]: sample_ids = [r.donation_sample_id for r in receptions if r.donation_sample_id] samples = ( db.query(BizResearchDonationSample) .filter( BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, BizResearchDonationSample.id.in_(sample_ids), ) .all() if sample_ids else [] ) sample_map = {s.id: s for s in samples} donation_ids = list({s.research_donation_id for s in samples if s.research_donation_id}) parties = ( db.query(BizResearchDonationParty) .filter( BizResearchDonationParty.tenant_no == tenant_no, BizResearchDonationParty.del_flag == 0, BizResearchDonationParty.research_donation_id.in_(donation_ids), ) .all() if donation_ids else [] ) party_map = {p.research_donation_id: p for p in parties} colls = ( db.query(BizResearchDonationCollection) .filter( BizResearchDonationCollection.tenant_no == tenant_no, BizResearchDonationCollection.del_flag == 0, BizResearchDonationCollection.research_donation_id.in_(donation_ids), ) .all() if donation_ids else [] ) coll_map = {c.research_donation_id: c for c in colls} out: List[Dict[str, Any]] = [] for rec in receptions: sample = sample_map.get(rec.donation_sample_id or "") if not sample: continue did = sample.research_donation_id sample_code = (rec.sample_code or sample.sample_code or "").strip() tasks = ledger.load_qc_tasks_for_sample(db, tenant_no, sample_code) qc_mother = ledger.extract_mother_blood(tasks, include_nucleic=True) qc_process = ledger.extract_process_tests(tasks) system = _build_donation_system( tab_key=tab_key, rec=rec, sample=sample, party=party_map.get(did), coll=coll_map.get(did), prod=prod_map.get(rec.id), storage_status=storage_map.get(sample_code, ""), qc_mother=qc_mother, qc_process=qc_process, ) lab_code = system.get("labStorageCode") or "" merged = ledger.merge_row( tab_key=tab_key, system_values=system, supplement=(sup_map.get(lab_code) or {}).get("payload"), ) merged["supplementId"] = (sup_map.get(lab_code) or {}).get("id") or "" merged["hasSupplement"] = bool((sup_map.get(lab_code) or {}).get("payload")) out.append(merged) return out