"""实验室样本处理数据 — 聚合市场/生产/质检已发生业务数据与部门字段回写。 Excel 仅作为字段目录示例；列表以协议样本为主表，关联样本接收与生产接收， 补充表只承载源系统没有的列，且不得覆盖源表字段。 """ from __future__ import annotations import json import uuid from datetime import datetime from decimal import Decimal from typing import Any, Dict, List, Optional, Tuple from types import SimpleNamespace from sqlalchemy import or_, text from sqlalchemy.orm import Session from app.models.models import ( BizCellSampleType, BizCollectionHospital, BizProtocolCollectionInfo, BizProtocolPartyCustomer, BizProtocolSample, BizSampleProductionReception, BizSampleReception, ) from app.services.lab_sample_processing_fields import ( DATASET_HSC, DATASET_MSC, DATASET_UBMC, DEPT_EDIT_PERMS, PERM_LIST, PERM_MANAGER, SOURCE_EXTRA, SOURCE_PRODUCTION, SOURCE_PROTOCOL, SOURCE_RECEPTION, editable_depts_for_user, field_map, fields_for, resolve_dataset_by_type_name, ) def _new_id() -> str: return uuid.uuid4().hex.upper() def _fmt_dt(v: Any) -> Optional[str]: if v is None: return None if isinstance(v, datetime): return v.strftime("%Y-%m-%d %H:%M:%S") return str(v) def _fmt_num(v: Any) -> Any: if v is None: return None if isinstance(v, Decimal): return float(v) return v def _parse_extra(raw: Optional[str]) -> Dict[str, Any]: if not raw: return {} try: data = json.loads(raw) except Exception: return {} return data if isinstance(data, dict) else {} def user_acl(db: Session, user: dict) -> Dict[str, Any]: from app.core.menu_guard import user_has_menu_perm is_manager = user_has_menu_perm(db, user, PERM_MANAGER) dept_flags = { dept: user_has_menu_perm(db, user, code) for dept, code in DEPT_EDIT_PERMS.items() } editable = editable_depts_for_user(is_manager=is_manager, dept_flags=dept_flags) return { "canList": user_has_menu_perm(db, user, PERM_LIST) or is_manager or any(dept_flags.values()), "isManager": is_manager, "editableDepts": sorted(editable), "deptFlags": dept_flags, } def _type_name_filters(dataset: str): if dataset == DATASET_HSC: return "%造血%" if dataset == DATASET_MSC: return "%间充质%" if dataset == DATASET_UBMC: return "%单个核%" return None def _code_prefix_filters(dataset: str) -> List[str]: if dataset == DATASET_HSC: return ["B%"] if dataset == DATASET_MSC: return ["BC%", "UC%"] if dataset == DATASET_UBMC: return ["UBMC%"] return [] def _match_keyword(row: Dict[str, Any], kw: str) -> bool: if not kw: return True text = f"{row.get('sampleCode') or ''} {row.get('motherName') or ''}" return kw.lower() in text.lower() def _match_date_range(row: Dict[str, Any], date_from: Optional[str], date_to: Optional[str]) -> bool: if not date_from and not date_to: return True for key in ("receptionTime", "collectTime", "cryoTime", "storageDateP2"): val = row.get(key) if not val: continue day = str(val)[:10].replace(".", "-").replace("/", "-") # normalize compact dates like 2025-9-8 parts = day.split("-") if len(parts) == 3: try: day = f"{int(parts[0]):04d}-{int(parts[1]):02d}-{int(parts[2]):02d}" except Exception: pass if date_from and day < date_from: continue if date_to and day > date_to: continue return True return False def list_rows( db: Session, *, tenant_no: str, dataset: str, keyword: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, page: int = 1, page_size: int = 20, ) -> Tuple[List[Dict[str, Any]], int]: from app.models.models import BizLabSampleProcessingExtra type_like = _type_name_filters(dataset) code_likes = _code_prefix_filters(dataset) q = ( db.query( BizProtocolSample, BizCellSampleType, BizProtocolPartyCustomer, BizProtocolCollectionInfo, BizCollectionHospital, ) .outerjoin( BizCellSampleType, (BizCellSampleType.id == BizProtocolSample.sample_type_id) & (BizCellSampleType.del_flag == 0), ) .outerjoin( BizProtocolPartyCustomer, (BizProtocolPartyCustomer.collection_protocol_id == BizProtocolSample.collection_protocol_id) & (BizProtocolPartyCustomer.tenant_no == tenant_no) & (BizProtocolPartyCustomer.del_flag == 0), ) .outerjoin( BizProtocolCollectionInfo, (BizProtocolCollectionInfo.collection_protocol_id == BizProtocolSample.collection_protocol_id) & (BizProtocolCollectionInfo.tenant_no == tenant_no) & (BizProtocolCollectionInfo.del_flag == 0), ) .outerjoin( BizCollectionHospital, (BizCollectionHospital.id == BizProtocolCollectionInfo.hospital_id) & (BizCollectionHospital.del_flag == 0), ) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_code.isnot(None), BizProtocolSample.sample_code != "", ) ) type_or_code = [] if type_like: type_or_code.append(BizCellSampleType.type_name.like(type_like)) for pref in code_likes: type_or_code.append(BizProtocolSample.sample_code.like(pref)) if type_or_code: q = q.filter(or_(*type_or_code)) rows = q.order_by(BizProtocolSample.create_time.desc()).all() filtered = [] for sample, stype, party, coll, hospital in rows: type_name = (stype.type_name if stype else "") or "" code = sample.sample_code or "" if resolve_dataset_by_type_name(type_name, code) != dataset: continue filtered.append((sample, stype, party, coll, hospital)) extras = { e.sample_code: e for e in db.query(BizLabSampleProcessingExtra) .filter( BizLabSampleProcessingExtra.tenant_no == tenant_no, BizLabSampleProcessingExtra.del_flag == 0, BizLabSampleProcessingExtra.dataset == dataset, ) .all() } # 仅聚合真实协议样本；补充表只挂到已有样本上，不再单独展示 Excel 演示行 codes = [s.sample_code for s, *_ in filtered if s.sample_code] sample_ids = [s.id for s, *_ in filtered if s.id] receptions_by_sample_id: Dict[str, BizSampleReception] = {} receptions_by_code: Dict[str, BizSampleReception] = {} productions: Dict[str, BizSampleProductionReception] = {} if sample_ids or codes: recv_filters = [] if sample_ids: recv_filters.append(BizSampleReception.protocol_sample_id.in_(sample_ids)) if codes: recv_filters.append(BizSampleReception.sample_code.in_(codes)) for r in ( db.query(BizSampleReception) .filter( BizSampleReception.tenant_no == tenant_no, BizSampleReception.del_flag == 0, or_(*recv_filters) if recv_filters else False, ) .order_by(BizSampleReception.reception_time.desc()) .all() ): if r.protocol_sample_id: receptions_by_sample_id.setdefault(r.protocol_sample_id, r) if r.sample_code: receptions_by_code.setdefault(r.sample_code, r) if receptions_by_sample_id or receptions_by_code: rids = list( { *(r.id for r in receptions_by_sample_id.values()), *(r.id for r in receptions_by_code.values()), } ) for p in ( db.query(BizSampleProductionReception) .filter( BizSampleProductionReception.tenant_no == tenant_no, BizSampleProductionReception.del_flag == 0, BizSampleProductionReception.reception_id.in_(rids), ) .order_by(BizSampleProductionReception.process_time.desc()) .all() ): productions.setdefault(p.reception_id, p) built_map: Dict[str, Dict[str, Any]] = {} for sample, stype, party, coll, hospital in filtered: code = sample.sample_code reception = receptions_by_sample_id.get(sample.id) or receptions_by_code.get(code) production = productions.get(reception.id) if reception else None extra = _parse_extra(extras[code].payload_json if code in extras else None) built_map[code] = _build_row( dataset=dataset, sample=sample, type_name=(stype.type_name if stype else None), party=party, coll=coll, hospital=hospital, reception=reception, production=production, extra=extra, ) # 单个核：业务已发生数据主要在科研捐献接收（lab_storage_code=UBMC*），无细胞储存协议样本时仍需展示 if dataset == DATASET_UBMC: ubmc_recs = ( db.query(BizSampleReception) .filter( BizSampleReception.tenant_no == tenant_no, BizSampleReception.del_flag == 0, BizSampleReception.lab_storage_code.like("UBMC%"), ) .order_by(BizSampleReception.reception_time.desc()) .all() ) mother_by_donation: Dict[str, str] = {} donation_ids = sorted( {r.research_donation_id for r in ubmc_recs if r.research_donation_id} ) if donation_ids: rows = db.execute( text( "SELECT research_donation_id, mother_name FROM biz_research_donation_party " "WHERE del_flag=0 AND tenant_no=:tenant_no AND research_donation_id IN (" + ",".join(f":d{i}" for i in range(len(donation_ids))) + ")" ), { "tenant_no": tenant_no, **{f"d{i}": did for i, did in enumerate(donation_ids)}, }, ).fetchall() for did, mother in rows: if did and mother and did not in mother_by_donation: mother_by_donation[str(did)] = mother missing_rids = [ r.id for r in ubmc_recs if r.id and r.id not in productions ] if missing_rids: for p in ( db.query(BizSampleProductionReception) .filter( BizSampleProductionReception.tenant_no == tenant_no, BizSampleProductionReception.del_flag == 0, BizSampleProductionReception.reception_id.in_(missing_rids), ) .order_by(BizSampleProductionReception.process_time.desc()) .all() ): productions.setdefault(p.reception_id, p) for reception in ubmc_recs: code = (reception.lab_storage_code or reception.sample_code or "").strip() if not code or code in built_map: continue production = productions.get(reception.id) extra = _parse_extra(extras[code].payload_json if code in extras else None) mother = mother_by_donation.get(str(reception.research_donation_id or "")) fake_sample = SimpleNamespace( id=reception.donation_sample_id or reception.id, sample_code=code, collection_protocol_id=None, ) fake_party = SimpleNamespace(mother_name=mother) if mother else None built_map[code] = _build_row( dataset=dataset, sample=fake_sample, type_name="脐血单个核细胞", party=fake_party, coll=None, hospital=None, reception=reception, production=production, extra=extra, ) kw = (keyword or "").strip() built = [] for row in built_map.values(): if not _match_keyword(row, kw): continue if not _match_date_range(row, date_from, date_to): continue built.append(row) # 稳定排序：优先按样本编码 built.sort(key=lambda r: str(r.get("sampleCode") or "")) total = len(built) page = max(1, int(page or 1)) page_size = max(1, min(5000, int(page_size or 20))) start = (page - 1) * page_size return built[start : start + page_size], total def _build_row( *, dataset: str, sample: Optional[BizProtocolSample], type_name: Optional[str], party: Optional[BizProtocolPartyCustomer], coll: Optional[BizProtocolCollectionInfo], hospital: Optional[BizCollectionHospital], reception: Optional[BizSampleReception], production: Optional[BizSampleProductionReception], extra: Dict[str, Any], ) -> Dict[str, Any]: protocol_vals = { "sample_code": sample.sample_code if sample else extra.get("sampleCode"), "mother_name": party.mother_name if party else None, "sample_collect_time": coll.sample_collect_time if coll else None, "hospital_name": hospital.hospital_name if hospital else None, } reception_vals = {} if reception: reception_vals = { "reception_time": reception.reception_time, "receive_temperature": reception.receive_temperature, "handler_name": reception.handler_name, "remark": reception.remark, "lab_storage_code": reception.lab_storage_code, } production_vals = {} if production: production_vals = { "original_blood_weight_g": production.original_blood_weight_g, "hes_volume_ml": production.hes_volume_ml, "blood_type": production.blood_type, "infectious_four_test_result": production.infectious_four_test_result, "enriched_blood_weight_g": production.enriched_blood_weight_g, "cryopreservation_volume_ml": production.cryopreservation_volume_ml, "handler_name": production.handler_name, "remark": production.remark, "original_blood_volume_ml": production.original_blood_volume_ml, "nucleated_cell_count_1e8": production.nucleated_cell_count_1e8, "inoculation_amount": production.inoculation_amount, "cryopreservation_time": production.cryopreservation_time, "storage_total_tubes": production.storage_total_tubes, "available_cell_count_1e7": production.available_cell_count_1e7, } sample_code = ( (sample.sample_code if sample else None) or extra.get("sampleCode") or "" ) data: Dict[str, Any] = { "id": (sample.id if sample else f"extra:{dataset}:{sample_code}"), "dataset": dataset, "sampleCode": sample_code, "sampleTypeName": type_name, "protocolSampleId": sample.id if sample else None, "collectionProtocolId": sample.collection_protocol_id if sample else None, "receptionId": reception.id if reception else None, "productionReceptionId": production.id if production else None, "fromExtraOnly": sample is None, } for f in fields_for(dataset): val = None if f.source == SOURCE_PROTOCOL: val = protocol_vals.get(f.attr) elif f.source == SOURCE_RECEPTION: val = reception_vals.get(f.attr) elif f.source == SOURCE_PRODUCTION: val = production_vals.get(f.attr) elif f.source == SOURCE_EXTRA: # 仅源系统没有的列走补充表；业务源表字段不被补充表覆盖 val = extra.get(f.key) if isinstance(val, datetime): val = _fmt_dt(val) elif isinstance(val, Decimal): val = _fmt_num(val) data[f.key] = val if not data.get("sampleCode"): data["sampleCode"] = sample_code return data def upsert_extra_payload( db: Session, *, tenant_no: str, dataset: str, sample_code: str, payload: Dict[str, Any], operator_id: str = "import", replace: bool = False, ) -> None: from app.models.models import BizLabSampleProcessingExtra code = (sample_code or "").strip() if not code: raise ValueError("样本编码不能为空") now = datetime.now() row = ( db.query(BizLabSampleProcessingExtra) .filter( BizLabSampleProcessingExtra.tenant_no == tenant_no, BizLabSampleProcessingExtra.dataset == dataset, BizLabSampleProcessingExtra.sample_code == code, BizLabSampleProcessingExtra.del_flag == 0, ) .first() ) current = {} if replace else _parse_extra(row.payload_json if row else None) for k, v in (payload or {}).items(): if v in (None, ""): continue current[k] = v current["sampleCode"] = code if row is None: db.add( BizLabSampleProcessingExtra( id=_new_id(), dataset=dataset, sample_code=code, payload_json=json.dumps(current, ensure_ascii=False), tenant_no=tenant_no, create_by=operator_id, create_dept_id="", create_time=now, update_by=operator_id, update_time=now, del_flag=0, ) ) else: row.payload_json = json.dumps(current, ensure_ascii=False) row.update_by = operator_id row.update_time = now def update_row( db: Session, *, tenant_no: str, dataset: str, sample_code: str, payload: Dict[str, Any], editable_depts: set, operator_id: str, ) -> Dict[str, Any]: code = (sample_code or "").strip() if not code: raise ValueError("样本编码不能为空") fmap = field_map(dataset) if not fmap: raise ValueError("未知数据集") denied = [] accepted: Dict[str, Any] = {} for key, value in (payload or {}).items(): f = fmap.get(key) if not f: continue if f.dept not in editable_depts: denied.append(key) continue accepted[key] = value if denied: raise PermissionError(f"无权编辑字段: {', '.join(denied)}") if not accepted: raise ValueError("没有可保存的字段") sample = ( db.query(BizProtocolSample) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.sample_code == code, BizProtocolSample.del_flag == 0, ) .first() ) reception = ( db.query(BizSampleReception) .filter( BizSampleReception.tenant_no == tenant_no, BizSampleReception.sample_code == code, BizSampleReception.del_flag == 0, ) .order_by(BizSampleReception.reception_time.desc()) .first() ) production = None if reception: production = ( db.query(BizSampleProductionReception) .filter( BizSampleProductionReception.tenant_no == tenant_no, BizSampleProductionReception.reception_id == reception.id, BizSampleProductionReception.del_flag == 0, ) .order_by(BizSampleProductionReception.process_time.desc()) .first() ) party = None coll = None if sample: party = ( db.query(BizProtocolPartyCustomer) .filter( BizProtocolPartyCustomer.collection_protocol_id == sample.collection_protocol_id, BizProtocolPartyCustomer.tenant_no == tenant_no, BizProtocolPartyCustomer.del_flag == 0, ) .first() ) coll = ( db.query(BizProtocolCollectionInfo) .filter( BizProtocolCollectionInfo.collection_protocol_id == sample.collection_protocol_id, BizProtocolCollectionInfo.tenant_no == tenant_no, BizProtocolCollectionInfo.del_flag == 0, ) .first() ) now = datetime.now() extra_payload: Dict[str, Any] = {} def _set_dt(obj, attr, value): if value in (None, ""): setattr(obj, attr, None) return if isinstance(value, datetime): setattr(obj, attr, value) return text = str(value).strip().replace("/", "-").replace(".", "-") for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"): try: setattr(obj, attr, datetime.strptime(text, fmt)) return except ValueError: continue setattr(obj, attr, value) for key, value in accepted.items(): f = fmap[key] wrote = False if f.source == SOURCE_PROTOCOL and sample is not None: if f.attr == "mother_name" and party is not None: party.mother_name = None if value in (None, "") else str(value) party.update_by = operator_id party.update_time = now wrote = True elif f.attr == "sample_collect_time" and coll is not None: _set_dt(coll, "sample_collect_time", value) coll.update_by = operator_id coll.update_time = now wrote = True elif f.source == SOURCE_RECEPTION and reception is not None: if f.attr == "reception_time": _set_dt(reception, f.attr, value) elif f.attr in {"receive_temperature", "handler_name", "remark", "lab_storage_code"}: setattr(reception, f.attr, None if value in (None, "") else value) reception.update_by = operator_id reception.update_time = now wrote = True elif f.source == SOURCE_PRODUCTION and production is not None: numeric_attrs = { "original_blood_weight_g", "hes_volume_ml", "enriched_blood_weight_g", "cryopreservation_volume_ml", "original_blood_volume_ml", "nucleated_cell_count_1e8", "available_cell_count_1e7", } if f.attr == "cryopreservation_time": _set_dt(production, f.attr, value) elif f.attr == "storage_total_tubes": production.storage_total_tubes = None if value in (None, "") else int(value) elif f.attr in numeric_attrs: production_val = None if value in (None, "") else Decimal(str(value)) setattr(production, f.attr, production_val) elif f.attr in { "blood_type", "infectious_four_test_result", "handler_name", "remark", "inoculation_amount", }: setattr(production, f.attr, None if value in (None, "") else value) production.update_by = operator_id production.update_time = now wrote = True # 补充表始终保留一份，保证导入台账与无权源表场景可编辑 extra_payload[key] = value _ = wrote upsert_extra_payload( db, tenant_no=tenant_no, dataset=dataset, sample_code=code, payload=extra_payload, operator_id=operator_id, replace=False, ) db.commit() rows, _ = list_rows(db, tenant_no=tenant_no, dataset=dataset, keyword=code, page=1, page_size=20) for row in rows: if row.get("sampleCode") == code: return row raise LookupError("保存成功但读取失败") def export_rows( db: Session, *, tenant_no: str, dataset: str, keyword: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, ) -> Tuple[List[str], List[List[Any]]]: rows, _ = list_rows( db, tenant_no=tenant_no, dataset=dataset, keyword=keyword, date_from=date_from, date_to=date_to, page=1, page_size=5000, ) headers = [f.label for f in fields_for(dataset)] keys = [f.key for f in fields_for(dataset)] body = [[row.get(k) for k in keys] for row in rows] return headers, body