"""科研捐赠管理 CRUD（捐赠单 / 审核 / 状态变更 / 转换）""" from __future__ import annotations import math from datetime import datetime from decimal import Decimal from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session from app.services.project_assignment import assignment_brief, filter_assignment, guard_project_edit from app.core.protocol_status_display import assert_protocol_status_transition from app.core.utils import generate_uuid, parse_datetime, format_date, optional_date from app.crud.child_row_sync import sync_child_rows from app.crud.base import camel_to_snake from app.crud.biz_no_generator import allocate_next_biz_no from app.crud.collection_protocol import ( append_protocol_status_change, attachments_to_urls, urls_to_attachment_list, validate_persisted_sample_codes_immutable, validate_sample_codes_not_used, _format_sample_collect_summary, _load_express_name_map, _load_sample_type_name_map, _load_site_name_map, _normalize_yyyymmdd_date, _owned_child_row, _retarget_protocol_sample_id, ) from app.crud.protocol_donation_convert import ( assert_biz_no_available, create_convert_link, get_convert_partner, has_active_convert_link, upsert_convert_link_for_to_storage, ) from app.utils.sample_special import ( apply_sample_special_fields, sample_special_to_dict, validate_sample_special_raw, ) from app.models.models import ( BizCollectionProtocol, BizCollectionRegion, BizCollectionHospital, BizItemUnit, BizProtocolCollectionInfo, BizProtocolPartyCustomer, BizProtocolSample, BizProtocolSampleTransport, BizProductionWorkOrder, BizQcSmsNotification, BizQcWorkOrder, BizSampleReception, SysUser, ) from app.models.research_donation import ( BizDonationStatusChange, BizResearchDonation, BizResearchDonationAudit, BizResearchDonationCollection, BizResearchDonationParty, BizResearchDonationSample, BizResearchDonationSampleTransport, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def total_pages(total: int, page_size: int) -> int: if total <= 0: return 0 return int(math.ceil(total / float(page_size))) _DONATION_STATUS_CHANGE_TYPE = { "sxz": "activate", "ygq": "expire", "yzf": "void", "zckc": "to_storage", } def _donation_status_change_to_dict( row: BizDonationStatusChange, *, operator_user_name: str = "", ) -> Dict[str, Any]: return { "id": row.id, "researchDonationId": row.research_donation_id, "changeType": row.change_type or "", "fromStatus": row.from_status or "", "toStatus": row.to_status or "", "changeTime": _fmt_dt(row.change_time), "changeDesc": row.change_desc or "", "voidType": row.void_type or "", "operatorUserId": row.operator_user_id or "", "operatorUserName": operator_user_name or "", "tenantNo": row.tenant_no, "createTime": _fmt_dt(row.create_time), "updateTime": _fmt_dt(row.update_time), } def append_donation_status_change( db: Session, *, research_donation_id: str, change_type: str, from_status: Optional[str], to_status: str, change_time: datetime, change_desc: Optional[str] = None, void_type: Optional[str] = None, tenant_no: str, user_id: Optional[str] = None, ) -> BizDonationStatusChange: """追加一条捐赠状态变更历史（不 commit）。""" now = datetime.now() obj = BizDonationStatusChange( id=generate_uuid(), research_donation_id=research_donation_id, change_type=(change_type or "").strip(), from_status=(from_status or "").strip() or None, to_status=(to_status or "").strip(), change_time=change_time, change_desc=(change_desc or "").strip() or None, void_type=(void_type or "").strip() or None, operator_user_id=user_id, tenant_no=tenant_no, order_num=0, remark=None, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(obj) return obj def list_donation_status_changes( db: Session, *, research_donation_id: str, tenant_no: str, ) -> List[Dict[str, Any]]: did = (research_donation_id or "").strip() if not did: return [] rows = ( db.query(BizDonationStatusChange) .filter( BizDonationStatusChange.research_donation_id == did, BizDonationStatusChange.tenant_no == tenant_no, BizDonationStatusChange.del_flag == 0, ) .order_by( BizDonationStatusChange.change_time.asc(), BizDonationStatusChange.create_time.asc(), ) .all() ) op_ids = list({r.operator_user_id for r in rows if r.operator_user_id}) name_map: Dict[str, str] = {} if op_ids: for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(op_ids)).all(): name_map[u.id] = u.nick_name or "" return [ _donation_status_change_to_dict( r, operator_user_name=name_map.get(r.operator_user_id or "", "") ) for r in rows ] _DONATION_SAMPLE_CODE_PREFIX = "KYJZ" _DONATION_SAMPLE_CODE_SEQ_LEN = 4 def _donation_sample_code_seq(code: str, *, date_str: str) -> Optional[int]: """解析 KYJZ+YYYYMM+序号 中的序号；不匹配则返回 None。""" prefix = f"{_DONATION_SAMPLE_CODE_PREFIX}{date_str}" c = (code or "").strip() if not c.startswith(prefix): return None suffix = c[len(prefix) :] if len(suffix) != _DONATION_SAMPLE_CODE_SEQ_LEN or not suffix.isdigit(): return None return int(suffix) def next_donation_sample_code( db: Session, *, tenant_no: str, extra_codes: Optional[List[str]] = None, ) -> str: """生成捐赠样本编码：KYJZ + YYYYMM + 4 位递增序号（0001 起）。""" date_str = datetime.now().strftime("%Y%m") prefix = f"{_DONATION_SAMPLE_CODE_PREFIX}{date_str}" max_seq = 0 def bump(code: Optional[str]) -> None: nonlocal max_seq seq = _donation_sample_code_seq(code or "", date_str=date_str) if seq is not None: max_seq = max(max_seq, seq) donation_rows = ( db.query(BizResearchDonationSample.sample_code) .filter( BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, BizResearchDonationSample.sample_code.like(f"{prefix}%"), ) .all() ) for row in donation_rows: bump(row[0]) protocol_rows = ( db.query(BizProtocolSample.sample_code) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_code.like(f"{prefix}%"), ) .all() ) for row in protocol_rows: bump(row[0]) from app.models.models import BizSampleCode pool_rows = ( db.query(BizSampleCode.sample_code) .filter( BizSampleCode.tenant_no == tenant_no, BizSampleCode.del_flag == 0, BizSampleCode.sample_code.like(f"{prefix}%"), ) .all() ) for row in pool_rows: bump(row[0]) for code in extra_codes or []: bump(code) return f"{prefix}{str(max_seq + 1).zfill(_DONATION_SAMPLE_CODE_SEQ_LEN)}" def allocate_monthly_verify_no( db: Session, *, tenant_no: str, create_time: datetime, ) -> str: """按租户 + create_time 年月分配 YYYYMM###；含软删占用，不回收。""" ym = create_time.strftime("%Y%m") prefix_len = len(ym) row = ( db.query(BizResearchDonation.monthly_verify_no) .filter( BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.monthly_verify_no.isnot(None), BizResearchDonation.monthly_verify_no.like(f"{ym}%"), ) .order_by(BizResearchDonation.monthly_verify_no.desc()) .first() ) max_no = (row[0] if row else None) or "" seq = 0 if max_no.startswith(ym) and len(max_no) >= prefix_len + 3: try: seq = int(max_no[prefix_len : prefix_len + 3]) except ValueError: seq = 0 nxt = seq + 1 if nxt > 999: raise ValueError(f"当月核量号已用尽（{ym}）") return f"{ym}{nxt:03d}" def _fmt_dt(t: Optional[datetime]) -> str: return t.strftime("%Y-%m-%d %H:%M:%S") if t else "" def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str: val = (value or "").strip() if len(val) == 10: return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00" return val def _dec(v: Any) -> Optional[float]: if v is None: return None if isinstance(v, Decimal): return float(v) return float(v) def _optional_decimal(v: Any) -> Optional[Decimal]: if v is None or v == "": return None return Decimal(str(v)) _SNAKE_ALIASES = { "apgar1min": "apgar_1min", "apgar5min": "apgar_5min", } def _normalize_fields(fields: Optional[dict]) -> dict: """将前端 camelCase 字段名转为数据库 snake_case。""" if not fields: return {} out: Dict[str, Any] = {} for k, v in fields.items(): sk = _SNAKE_ALIASES.get(k) or camel_to_snake(k) out[sk] = v return out def _merge_comma_separated_urls(*parts: Optional[str]) -> Optional[str]: """合并逗号分隔的附件 URL，去重并保持顺序。""" seen: set[str] = set() urls: List[str] = [] for part in parts: if not part: continue for url in str(part).split(","): u = url.strip() if not u or u in seen: continue seen.add(u) urls.append(u) return ",".join(urls) if urls else None def _resolve_donation_payload(raw: dict) -> None: """解析保存载荷：附件列表 → URL 字段；party/collection 别名。""" if raw.get("consent_attachments") is not None: raw["consent_form_urls"] = attachments_to_urls(raw["consent_attachments"]) or None elif (raw.get("consent_form_urls") or "").strip(): raw["consent_form_urls"] = (raw["consent_form_urls"] or "").strip() if raw.get("other_attachments") is not None: raw["attachment_urls"] = attachments_to_urls(raw["other_attachments"]) or None elif (raw.get("attachment_urls") or "").strip(): raw["attachment_urls"] = (raw["attachment_urls"] or "").strip() if raw.get("infectious_disease_report_attachments") is not None: raw["infectious_disease_report_urls"] = ( attachments_to_urls(raw["infectious_disease_report_attachments"]) or None ) elif (raw.get("infectious_disease_report_urls") or "").strip(): raw["infectious_disease_report_urls"] = ( (raw["infectious_disease_report_urls"] or "").strip() ) if not raw.get("party") and raw.get("party_customer"): raw["party"] = raw["party_customer"] if not raw.get("collection") and raw.get("collection_info"): raw["collection"] = raw["collection_info"] def _user_nick_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]: ids = list({x for x in user_ids if x}) if not ids: return {} rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all() return {u.id: u.nick_name or "" for u in rows} def _load_unit_name_map(db: Session, samples: List[BizResearchDonationSample]) -> Dict[str, str]: unit_ids = list({s.unit_id for s in samples if s.unit_id}) if not unit_ids: return {} unit_map: Dict[str, str] = {} for u in ( db.query(BizItemUnit) .filter(BizItemUnit.id.in_(unit_ids), BizItemUnit.del_flag == 0) .all() ): unit_map[u.id] = u.unit_name or "" return unit_map # --------------------------------------------------------------------------- # ORM → dict serializers # --------------------------------------------------------------------------- def _donation_to_dict( d: BizResearchDonation, db: Optional[Session] = None, tenant_no: Optional[str] = None, ) -> Dict[str, Any]: tn = tenant_no or d.tenant_no return { "id": d.id, **assignment_brief(db, tn, d.project_id), "donationNo": d.donation_no or "", "monthlyVerifyNo": d.monthly_verify_no or "", "donationStatus": d.donation_status or "", "donationDate": (d.donation_date or "").strip() if d.donation_date else "", "sourceType": d.source_type or "direct", "sourceProtocolId": d.source_protocol_id or "", "consentFormUrls": d.consent_form_urls or "", "attachmentUrls": d.attachment_urls or "", "infectiousDiseaseReportUrls": d.infectious_disease_report_urls or "", "consentAttachmentFiles": urls_to_attachment_list(d.consent_form_urls, db, tn) if db else [], "attachmentFiles": urls_to_attachment_list(d.attachment_urls, db, tn) if db else [], "infectiousDiseaseReportAttachmentFiles": ( urls_to_attachment_list(d.infectious_disease_report_urls, db, tn) if db else [] ), "collectorUserId": d.collector_user_id or "", "effectiveTime": _fmt_dt(d.effective_time), "effectiveDesc": d.effective_desc or "", "expireTime": _fmt_dt(d.expire_time), "expireDesc": d.expire_desc or "", "voidTime": _fmt_dt(d.void_time), "voidDesc": d.void_desc or "", "voidType": d.void_type or "", "toStorageTime": _fmt_dt(d.to_storage_time), "toStorageDesc": d.to_storage_desc or "", "tenantNo": d.tenant_no, "orderNum": d.order_num if d.order_num is not None else 0, "remark": d.remark or "", "createTime": _fmt_dt(d.create_time), "updateTime": _fmt_dt(d.update_time), } def _audit_to_dict( a: BizResearchDonationAudit, *, donation_no: str = "", auditor_user_name: str = "", ) -> Dict[str, Any]: return { "id": a.id, "researchDonationId": a.research_donation_id, "donationNo": donation_no, "auditorUserId": a.auditor_user_id or "", "auditorUserName": auditor_user_name or "", "auditRemark": a.audit_remark or "", "auditStatus": a.audit_status or "", "tenantNo": a.tenant_no, "orderNum": a.order_num if a.order_num is not None else 0, "remark": a.remark or "", "createTime": _fmt_dt(a.create_time), "updateTime": _fmt_dt(a.update_time), } def _party_to_dict(p: BizResearchDonationParty) -> Dict[str, Any]: return { "id": p.id, "researchDonationId": p.research_donation_id, "customerId": p.customer_id or "", "fatherCustomerId": p.father_customer_id or "", "newbornCustomerId": p.newborn_customer_id or "", "motherName": p.mother_name or "", "motherPhone": p.mother_phone or "", "motherIdCard": p.mother_id_card or "", "motherBirthYm": (p.mother_birth_ym or "").strip() if p.mother_birth_ym else "", "motherEthnic": p.mother_ethnic or "", "motherEmail": p.mother_email or "", "motherAddress": p.mother_address or "", "fatherName": p.father_name or "", "fatherPhone": p.father_phone or "", "fatherIdCard": p.father_id_card or "", "newbornName": p.newborn_name or "", "newbornIdCard": p.newborn_id_card or "", "expectedDeliveryDate": format_date(p.expected_delivery_date), "newbornDeliveryTime": _fmt_dt(p.newborn_delivery_time), "newbornSex": p.newborn_sex or "", "deliveryMode": p.delivery_mode or "", "birthOrder": p.birth_order, "newbornWeightGram": p.newborn_weight_gram, "apgar1min": _dec(p.apgar_1min), "apgar5min": _dec(p.apgar_5min), "collectorUserId": p.collector_user_id or "", "tenantNo": p.tenant_no, "orderNum": p.order_num if p.order_num is not None else 0, "remark": p.remark or "", "createTime": _fmt_dt(p.create_time), "updateTime": _fmt_dt(p.update_time), } def _collection_to_dict(c: BizResearchDonationCollection) -> Dict[str, Any]: return { "id": c.id, "researchDonationId": c.research_donation_id, "regionId": c.region_id or "", "hospitalId": c.hospital_id or "", "medicalStaff": c.medical_staff or "", "salesUserId": c.sales_user_id or "", "regionManagerUserId": c.region_manager_user_id or "", "sampleCollectTime": _fmt_dt(c.sample_collect_time), "maternalBloodCollectTime": _fmt_dt(c.maternal_blood_collect_time), "tenantNo": c.tenant_no, "orderNum": c.order_num if c.order_num is not None else 0, "remark": c.remark or "", "createTime": _fmt_dt(c.create_time), "updateTime": _fmt_dt(c.update_time), } def _sample_to_dict(s: BizResearchDonationSample) -> Dict[str, Any]: return { "id": s.id, "researchDonationId": s.research_donation_id, "sampleTypeId": s.sample_type_id or "", "sampleCodeId": s.sample_code_id or "", "sampleCode": s.sample_code or "", "collectAmount": _dec(s.collect_amount), "unitId": s.unit_id or "", "storageYears": s.storage_years, "sampleStatus": (s.sample_status or "").strip() if (s.sample_status or "").strip() else "", "storageInDate": (s.storage_in_date or "").strip() if s.storage_in_date else "", "detectStartDate": (s.detect_start_date or "").strip() if s.detect_start_date else "", "voidDate": (s.void_date or "").strip() if s.void_date else "", "storageExpireDate": (s.storage_expire_date or "").strip() if s.storage_expire_date else "", **sample_special_to_dict(s), "tenantNo": s.tenant_no, "orderNum": s.order_num if s.order_num is not None else 0, "remark": s.remark or "", "createTime": _fmt_dt(s.create_time), "updateTime": _fmt_dt(s.update_time), } def _transport_to_dict(t: BizResearchDonationSampleTransport) -> Dict[str, Any]: return { "id": t.id, "researchDonationId": t.research_donation_id, "sampleId": t.sample_id, "transportType": t.transport_type or "", "expressCompanyId": t.express_company_id or "", "waybillNo": t.waybill_no or "", "destSiteId": t.dest_site_id or "", "expressFee": _dec(t.express_fee), "etaTime": _fmt_dt(t.eta_time), "carrierName": t.carrier_name or "", "carrierPhone": t.carrier_phone or "", "remark": t.remark or "", "orderNum": t.order_num if t.order_num is not None else 0, "createTime": _fmt_dt(t.create_time), "updateTime": _fmt_dt(t.update_time), } # --------------------------------------------------------------------------- # Sub-resource queries # --------------------------------------------------------------------------- def get_party( db: Session, research_donation_id: str, tenant_no: str ) -> Optional[BizResearchDonationParty]: return ( db.query(BizResearchDonationParty) .filter( and_( BizResearchDonationParty.research_donation_id == research_donation_id, BizResearchDonationParty.tenant_no == tenant_no, BizResearchDonationParty.del_flag == 0, ) ) .first() ) def get_collection( db: Session, research_donation_id: str, tenant_no: str ) -> Optional[BizResearchDonationCollection]: return ( db.query(BizResearchDonationCollection) .filter( and_( BizResearchDonationCollection.research_donation_id == research_donation_id, BizResearchDonationCollection.tenant_no == tenant_no, BizResearchDonationCollection.del_flag == 0, ) ) .first() ) def list_samples( db: Session, research_donation_id: str, tenant_no: str ) -> List[BizResearchDonationSample]: return ( db.query(BizResearchDonationSample) .filter( and_( BizResearchDonationSample.research_donation_id == research_donation_id, BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, ) ) .order_by( BizResearchDonationSample.order_num.asc(), BizResearchDonationSample.create_time.asc(), ) .all() ) def list_sample_transports( db: Session, research_donation_id: str, tenant_no: str ) -> List[BizResearchDonationSampleTransport]: return ( db.query(BizResearchDonationSampleTransport) .filter( and_( BizResearchDonationSampleTransport.research_donation_id == research_donation_id, BizResearchDonationSampleTransport.tenant_no == tenant_no, BizResearchDonationSampleTransport.del_flag == 0, ) ) .order_by( BizResearchDonationSampleTransport.order_num.asc(), BizResearchDonationSampleTransport.create_time.asc(), ) .all() ) # --------------------------------------------------------------------------- # 1. list_donations # --------------------------------------------------------------------------- def list_donations( db: Session, *, tenant_no: str, project_id: Optional[str] = None, team_id: Optional[str] = None, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, donation_status: Optional[str] = None, create_time_start: Optional[str] = None, create_time_end: Optional[str] = None, sample_collect_time_start: Optional[str] = None, sample_collect_time_end: Optional[str] = None, user_info=None, ) -> Dict[str, Any]: def _base(): q = ( db.query(BizResearchDonation) .filter(BizResearchDonation.del_flag == 0) ) if tenant_no: q = q.filter(BizResearchDonation.tenant_no == tenant_no) q = filter_assignment(q, BizResearchDonation, db, tenant_no, project_id, team_id) if donation_status: q = q.filter(BizResearchDonation.donation_status == donation_status.strip()) if create_time_start: q = q.filter( BizResearchDonation.create_time >= _normalize_date_bound(create_time_start) ) if create_time_end: q = q.filter( BizResearchDonation.create_time <= _normalize_date_bound(create_time_end, end_of_day=True) ) if sample_collect_time_start or sample_collect_time_end: coll_q = db.query(BizResearchDonationCollection.research_donation_id).filter( BizResearchDonationCollection.del_flag == 0, ) if sample_collect_time_start: coll_q = coll_q.filter( BizResearchDonationCollection.sample_collect_time >= _normalize_date_bound(sample_collect_time_start) ) if sample_collect_time_end: coll_q = coll_q.filter( BizResearchDonationCollection.sample_collect_time <= _normalize_date_bound(sample_collect_time_end, end_of_day=True) ) q = q.filter(BizResearchDonation.id.in_(coll_q)) kw = (keyword or "").strip() if kw: like = f"%{kw}%" party_match = ( db.query(BizResearchDonationParty.research_donation_id) .filter( BizResearchDonationParty.del_flag == 0, BizResearchDonationParty.mother_name.like(like), ) .subquery() ) kw_conds = [ BizResearchDonation.id.in_(party_match), ] if tenant_no: from app.crud import sample_reception as reception_svc donation_ids = reception_svc.donation_ids_matching_lab_storage_keyword( db, tenant_no, kw ) if donation_ids: kw_conds.append(BizResearchDonation.id.in_(donation_ids)) q = q.filter(or_(*kw_conds)) if user_info: from app.core.data_scope import apply_data_scope q = apply_data_scope(q, BizResearchDonation, user_info, db, tenant_no) return q total = ( _base() .order_by(None) .with_entities(func.count(BizResearchDonation.id)) .scalar() ) or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _base() .order_by(BizResearchDonation.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) # 批量加载 party / collection 子表，避免 N+1 donation_ids = [r.id for r in rows] party_map: Dict[str, Dict[str, Any]] = {} collection_map: Dict[str, Dict[str, Any]] = {} if donation_ids: party_rows = ( db.query(BizResearchDonationParty) .filter( BizResearchDonationParty.research_donation_id.in_(donation_ids), BizResearchDonationParty.del_flag == 0, ) .all() ) for p in party_rows: party_map[p.research_donation_id] = _party_to_dict(p) collection_rows = ( db.query(BizResearchDonationCollection) .filter( BizResearchDonationCollection.research_donation_id.in_(donation_ids), BizResearchDonationCollection.del_flag == 0, ) .all() ) for c in collection_rows: collection_map[c.research_donation_id] = _collection_to_dict(c) # 批量解析地区 / 医院名称 region_ids = list( {collection_map.get(did, {}).get("regionId", "") for did in donation_ids} - {""} ) hospital_ids = list( {collection_map.get(did, {}).get("hospitalId", "") for did in donation_ids} - {""} ) region_name_map: Dict[str, str] = {} hospital_name_map: Dict[str, str] = {} if region_ids: for reg in ( db.query(BizCollectionRegion.id, BizCollectionRegion.region_name) .filter(BizCollectionRegion.id.in_(region_ids)) .all() ): region_name_map[reg.id] = reg.region_name or "" if hospital_ids: for hosp in ( db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name) .filter(BizCollectionHospital.id.in_(hospital_ids)) .all() ): hospital_name_map[hosp.id] = hosp.hospital_name or "" # 批量解析采集专员 / 市场专员 / 地区经理名称 sales_user_ids = list( {collection_map.get(did, {}).get("salesUserId", "") for did in donation_ids} - {""} ) region_manager_user_ids = list( { collection_map.get(did, {}).get("regionManagerUserId", "") for did in donation_ids } - {""} ) collector_ids = list({r.collector_user_id for r in rows if r.collector_user_id}) user_name_map = _user_nick_name_map( db, list(set(collector_ids + sales_user_ids + region_manager_user_ids)) ) # 批量加载捐赠样本（列表展示样本类型、样本编码、采集情况） sample_map: Dict[str, List[BizResearchDonationSample]] = {} sample_type_name_map: Dict[str, str] = {} if donation_ids: s_rows = ( db.query(BizResearchDonationSample) .filter( BizResearchDonationSample.research_donation_id.in_(donation_ids), BizResearchDonationSample.del_flag == 0, ) .order_by( BizResearchDonationSample.order_num.asc(), BizResearchDonationSample.create_time.asc(), ) .all() ) sample_type_name_map = _load_sample_type_name_map(db, s_rows) for s in s_rows: sample_map.setdefault(s.research_donation_id, []).append(s) all_sample_codes: List[str] = [] for samples in sample_map.values(): for s in samples: c = (s.sample_code or "").strip() if c: all_sample_codes.append(c) lab_code_map: Dict[str, str] = {} reception_summary_map: Dict[str, Dict[str, str]] = {} qc_status_map: Dict[str, Dict[str, str]] = {} from app.services.protocol_sample_list_summary import ( build_sample_item_reception_qc_fields, get_list_qc_status_map_by_sample_codes, get_list_reception_summary_map_by_sample_codes, ) if tenant_no and all_sample_codes: from app.crud import sample_reception as reception_svc lab_code_map = reception_svc.get_lab_storage_code_map_by_sample_codes( db, tenant_no, all_sample_codes, reception_source="research_donation" ) reception_summary_map = get_list_reception_summary_map_by_sample_codes( db, tenant_no, all_sample_codes, reception_source="research_donation" ) qc_status_map = get_list_qc_status_map_by_sample_codes( db, tenant_no, all_sample_codes, reception_source="research_donation" ) transport_by_sample_id: Dict[str, BizResearchDonationSampleTransport] = {} if donation_ids: for t in ( db.query(BizResearchDonationSampleTransport) .filter( BizResearchDonationSampleTransport.research_donation_id.in_(donation_ids), BizResearchDonationSampleTransport.tenant_no == tenant_no, BizResearchDonationSampleTransport.del_flag == 0, ) .all() ): transport_by_sample_id[t.sample_id] = t site_name_map = _load_site_name_map( db, [t.dest_site_id for t in transport_by_sample_id.values() if t.dest_site_id] ) express_name_map = _load_express_name_map( db, [ t.express_company_id for t in transport_by_sample_id.values() if t.express_company_id ], ) result = [] for r in rows: d = _donation_to_dict(r, db, tenant_no) d["collectorUserName"] = user_name_map.get(r.collector_user_id or "", "") pc = party_map.get(r.id) if pc: d["motherName"] = pc.get("motherName", "") d["motherPhone"] = pc.get("motherPhone", "") d["motherIdCard"] = pc.get("motherIdCard", "") d["fatherName"] = pc.get("fatherName", "") d["newbornName"] = pc.get("newbornName", "") d["expectedDeliveryDate"] = pc.get("expectedDeliveryDate", "") d["newbornDeliveryTime"] = pc.get("newbornDeliveryTime", "") d["newbornSex"] = pc.get("newbornSex", "") d["deliveryMode"] = pc.get("deliveryMode", "") d["newbornWeightGram"] = pc.get("newbornWeightGram") d["birthOrder"] = pc.get("birthOrder") else: d["motherName"] = "" d["motherPhone"] = "" d["motherIdCard"] = "" d["fatherName"] = "" d["newbornName"] = "" d["expectedDeliveryDate"] = "" d["newbornDeliveryTime"] = "" d["newbornSex"] = "" d["deliveryMode"] = "" d["newbornWeightGram"] = None d["birthOrder"] = None ci = collection_map.get(r.id) if ci: d["regionName"] = region_name_map.get(ci.get("regionId", ""), "") d["hospitalName"] = hospital_name_map.get(ci.get("hospitalId", ""), "") d["salesUserName"] = user_name_map.get(ci.get("salesUserId", ""), "") d["regionManagerUserName"] = user_name_map.get( ci.get("regionManagerUserId", ""), "" ) d["sampleCollectTime"] = ci.get("sampleCollectTime", "") d["maternalBloodCollectTime"] = ci.get("maternalBloodCollectTime", "") else: d["regionName"] = "" d["hospitalName"] = "" d["salesUserName"] = "" d["regionManagerUserName"] = "" d["sampleCollectTime"] = "" d["maternalBloodCollectTime"] = "" donation_samples = sample_map.get(r.id, []) type_names: List[str] = [] codes: List[str] = [] sample_items: List[Dict[str, Any]] = [] for s in donation_samples: type_name = sample_type_name_map.get(s.sample_type_id or "", "") if type_name: type_names.append(type_name) code = (s.sample_code or "").strip() if code: codes.append(code) transport_obj = transport_by_sample_id.get(s.id) transport_dict: Optional[Dict[str, Any]] = None if transport_obj and ( (transport_obj.transport_type or "").strip() or transport_obj.eta_time ): dest_site_id = transport_obj.dest_site_id or "" express_company_id = transport_obj.express_company_id or "" transport_dict = { "transportType": transport_obj.transport_type or "", "destSiteId": dest_site_id, "destSiteName": site_name_map.get(dest_site_id, ""), "expressCompanyId": express_company_id, "expressCompanyName": express_name_map.get(express_company_id, ""), "waybillNo": transport_obj.waybill_no or "", "carrierName": transport_obj.carrier_name or "", "carrierPhone": transport_obj.carrier_phone or "", "etaTime": _fmt_dt(transport_obj.eta_time), } sample_items.append( { "sampleTypeName": type_name, "sampleCode": code, "sampleStatus": (s.sample_status or "").strip(), "labStorageCode": lab_code_map.get(code, "") if code else "", "transport": transport_dict, **sample_special_to_dict(s), **build_sample_item_reception_qc_fields( reception_summary_map.get(code) if code else None, qc_status_map.get(code) if code else None, ), } ) d["sampleItems"] = sample_items d["sampleTypeNames"] = ",".join(type_names) d["sampleCodes"] = ",".join(codes) d["labStorageCodes"] = ",".join((lab_code_map.get(c) or "") for c in codes) d["sampleCollectSummary"] = _format_sample_collect_summary(donation_samples) result.append(d) return { "list": result, "total": int(total), "page": page, "pageSize": page_size, "totalPages": total_pages(int(total), page_size), } # --------------------------------------------------------------------------- # 2. get_donation_detail # --------------------------------------------------------------------------- def get_donation_detail( db: Session, donation_id: str, tenant_no: str, ) -> Optional[Dict[str, Any]]: """查询单条捐赠详情，含全部子资源嵌套。""" main = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == donation_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .first() ) if not main: return None out: Dict[str, Any] = {"donation": _donation_to_dict(main, db, tenant_no)} # 采集专员名称 user_ids: List[str] = [] if main.collector_user_id: user_ids.append(main.collector_user_id) # party party = get_party(db, donation_id, tenant_no) out["party"] = _party_to_dict(party) if party else None if party and party.collector_user_id: user_ids.append(party.collector_user_id) # collection coll = get_collection(db, donation_id, tenant_no) coll_dict = _collection_to_dict(coll) if coll else None if coll_dict and coll_dict.get("salesUserId"): user_ids.append(coll_dict["salesUserId"]) if coll_dict and coll_dict.get("regionManagerUserId"): user_ids.append(coll_dict["regionManagerUserId"]) if coll_dict: region_id = str(coll_dict.get("regionId") or "") if region_id: r = db.query(BizCollectionRegion.region_name).filter( BizCollectionRegion.id == region_id ).first() coll_dict["regionName"] = r.region_name if r else "" hospital_id = str(coll_dict.get("hospitalId") or "") if hospital_id: h = db.query(BizCollectionHospital.hospital_name).filter( BizCollectionHospital.id == hospital_id ).first() coll_dict["hospitalName"] = h.hospital_name if h else "" out["collection"] = coll_dict # samples + transport s_rows = list_samples(db, donation_id, tenant_no) transport_rows = list_sample_transports(db, donation_id, tenant_no) site_name_map = _load_site_name_map( db, [t.dest_site_id for t in transport_rows if t.dest_site_id] ) transport_by_sample: Dict[str, Dict[str, Any]] = {} for t in transport_rows: td = _transport_to_dict(t) td["destSiteName"] = site_name_map.get(t.dest_site_id or "", "") transport_by_sample[t.sample_id] = td from app.crud import sample_reception as sample_reception_svc from app.crud import qc_work_order as qc_work_order_svc from app.crud import production_work_order as production_work_order_svc from app.crud import storage_release_review as storage_release_review_svc sample_ids = [s.id for s in s_rows] sample_code_by_id = {s.id: s.sample_code or "" for s in s_rows} reception_by_sample = sample_reception_svc.get_receptions_map_for_donation_samples( db, sample_ids, tenant_no ) qc_by_sample = qc_work_order_svc.get_qc_info_map_for_samples( db, sample_ids, tenant_no, ) process_qc_by_sample = qc_work_order_svc.get_process_qc_info_map_for_samples( db, sample_ids, tenant_no, sample_code_by_id=sample_code_by_id, ) production_by_sample = production_work_order_svc.get_production_info_map_for_samples( db, sample_ids, tenant_no ) storage_release_by_sample = storage_release_review_svc.get_storage_release_map_for_samples( db, sample_ids, tenant_no, sample_code_by_id=sample_code_by_id, ) samples_out: List[Dict[str, Any]] = [] unit_name_map = _load_unit_name_map(db, s_rows) from app.crud import sample_freight as sample_freight_svc freight_by_sample = sample_freight_svc.freight_map_for_samples( db, tenant_no=tenant_no, sample_source=sample_freight_svc.SAMPLE_SOURCE_DONATION, sample_ids=sample_ids, ) for s in s_rows: item = _sample_to_dict(s) item["unitName"] = unit_name_map.get(s.unit_id or "", "") item["transport"] = transport_by_sample.get(s.id) item["freight"] = freight_by_sample.get(s.id) item["sampleReception"] = reception_by_sample.get(s.id) item["productionReception"] = (reception_by_sample.get(s.id) or {}).get( "productionReception" ) item["qcInfo"] = qc_by_sample.get(s.id) item["processQcInfoList"] = process_qc_by_sample.get(s.id, []) item["productionInfo"] = production_by_sample.get(s.id) item["storageReleaseInfo"] = storage_release_by_sample.get(s.id) samples_out.append(item) out["samples"] = samples_out # user name map user_name_map = _user_nick_name_map(db, user_ids) out["donation"]["collectorUserName"] = user_name_map.get( main.collector_user_id or "", "" ) if coll_dict and coll_dict.get("salesUserId"): coll_dict["salesUserName"] = user_name_map.get( coll_dict["salesUserId"], "" ) if coll_dict: coll_dict["regionManagerUserName"] = user_name_map.get( coll_dict.get("regionManagerUserId") or "", "" ) return out # --------------------------------------------------------------------------- # 3. create_donation # --------------------------------------------------------------------------- def create_donation( db: Session, *, data: dict, tenant_no: str, user_id: Optional[str] = None, create_dept_id: str = "", commit: bool = True, sync_customers: bool = True, ) -> BizResearchDonation: """创建科研捐赠单（含 party / collection / samples + transports）。""" raw = _normalize_fields(data) _resolve_donation_payload(raw) consent = (raw.get("consent_form_urls") or "").strip() donation_no = (raw.get("donation_no") or "").strip() if not donation_no: donation_no = allocate_next_biz_no( db, tenant_no=tenant_no, kind="donation", user_id=user_id ) assert_biz_no_available( db, tenant_no=tenant_no, biz_no=donation_no, for_donation=True ) now = datetime.now() monthly_verify_no = allocate_monthly_verify_no( db, tenant_no=tenant_no, create_time=now ) obj = BizResearchDonation( project_id=raw.get("project_id") or None, id=generate_uuid(), donation_no=donation_no, monthly_verify_no=monthly_verify_no, donation_status=(raw.get("donation_status") or "dsx").strip(), donation_date=_normalize_yyyymmdd_date(raw.get("donation_date")), source_type=(raw.get("source_type") or "direct").strip(), source_protocol_id=(raw.get("source_protocol_id") or "").strip() or None, consent_form_urls=consent or None, attachment_urls=(raw.get("attachment_urls") or "").strip() or None, infectious_disease_report_urls=( (raw.get("infectious_disease_report_urls") or "").strip() or None ), collector_user_id=(raw.get("collector_user_id") or "").strip() or None, tenant_no=tenant_no, order_num=0, remark=(raw.get("remark") or "").strip() or None, create_by=user_id, create_dept_id=create_dept_id, create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(obj) db.flush() donation_id = obj.id # party from app.services.party_customer_sync import sync_party_customers party_raw = _normalize_fields(raw.get("party") or {}) if sync_customers: party_raw = sync_party_customers( db, party_raw, tenant_no=tenant_no, user_id=user_id or "", commit=commit, create_dept_id=create_dept_id, ) _create_party( db, donation_id, tenant_no, user_id, party_raw, now, create_dept_id=create_dept_id, ) # collection coll_raw = _normalize_fields(raw.get("collection") or {}) _create_collection( db, donation_id, tenant_no, user_id, coll_raw, now, create_dept_id=create_dept_id, ) # samples（运费批次在 commit 后挂钩；停写旧运输依赖） samples_raw = raw.get("samples") or [] new_samples = _create_samples( db, donation_id, tenant_no, user_id, samples_raw, now, create_dept_id=create_dept_id, ) from app.crud import sample_freight as sample_freight_svc sample_freight_svc.apply_freight_on_protocol_save( db, tenant_no=tenant_no, user_id=user_id, sample_source=sample_freight_svc.SAMPLE_SOURCE_DONATION, protocol_id=donation_id, saved_samples=new_samples, input_samples=samples_raw, ) if commit: db.commit() db.refresh(obj) else: db.flush() return obj # --------------------------------------------------------------------------- # 4. update_donation # --------------------------------------------------------------------------- def update_donation( db: Session, donation_id: str, *, data: dict, tenant_no: str, user_id: Optional[str] = None, create_dept_id: str = "", commit: bool = True, sync_customers: bool = True, ) -> BizResearchDonation: """更新科研捐赠单：主表字段就地更新，子资源按行同步。""" obj = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == donation_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .first() ) if not obj: raise ValueError("捐赠单不存在或无权操作") guard_project_edit(obj, data) raw = _normalize_fields(data) for source_field in ("source_type", "source_protocol_id"): if source_field in raw and (raw[source_field] or None) != (getattr(obj, source_field) or None): raise ValueError("转换来源不可修改，请使用受控转换入口") _resolve_donation_payload(raw) now = datetime.now() # 主表字段 _MAIN_STR_FIELDS = [ "donation_status", "donation_date", "source_type", "source_protocol_id", "consent_form_urls", "attachment_urls", "infectious_disease_report_urls", "collector_user_id", "effective_desc", "expire_desc", "void_desc", "to_storage_desc", "remark", "donation_no", ] _MAIN_DT_FIELDS = [ "effective_time", "expire_time", "void_time", "to_storage_time", ] for k in _MAIN_STR_FIELDS: if k in raw: if k == "donation_no": pn = (raw[k] or "").strip() if pn and pn != (obj.donation_no or ""): raise ValueError("捐赠协议号不可修改") else: val = raw[k] if val is None: continue if k == "donation_date": setattr(obj, k, _normalize_yyyymmdd_date(val)) else: setattr(obj, k, (val or "").strip() or None) for k in _MAIN_DT_FIELDS: if k in raw: val = raw[k] if isinstance(val, str): val = parse_datetime(val) setattr(obj, k, val) obj.update_by = user_id obj.update_time = now db.add(obj) db.flush() if "party" in raw: from app.services.party_customer_sync import sync_party_customers party_raw = _normalize_fields(raw["party"] or {}) if sync_customers: party_raw = sync_party_customers( db, party_raw, tenant_no=tenant_no, user_id=user_id or "", commit=commit, create_dept_id=create_dept_id, ) _create_party( db, donation_id, tenant_no, user_id, party_raw, now, create_dept_id=create_dept_id, ) if "collection" in raw: coll_raw = _normalize_fields(raw["collection"] or {}) _create_collection( db, donation_id, tenant_no, user_id, coll_raw, now, create_dept_id=create_dept_id, ) # samples upsert if "samples" in raw: existing_samples = list_samples(db, donation_id, tenant_no) samples_raw = raw["samples"] or [] validate_persisted_sample_codes_immutable(existing_samples, samples_raw) old_id_to_code = { s.id: (s.sample_code or "").strip() for s in existing_samples if (s.sample_code or "").strip() } new_samples = _create_samples( db, donation_id, tenant_no, user_id, samples_raw, now, create_dept_id=create_dept_id, ) kept_ids = {s.id for s in new_samples} new_code_to_id = { (s.sample_code or "").strip(): s.id for s in new_samples if (s.sample_code or "").strip() } _migrate_donation_sample_child_refs_on_replace( db, tenant_no=tenant_no, old_id_to_code={ sid: code for sid, code in old_id_to_code.items() if sid not in kept_ids }, new_code_to_id=new_code_to_id, user_id=user_id, ) else: samples_raw = None new_samples = None if new_samples is not None and samples_raw is not None: from app.crud import sample_freight as sample_freight_svc sample_freight_svc.apply_freight_on_protocol_save( db, tenant_no=tenant_no, user_id=user_id, sample_source=sample_freight_svc.SAMPLE_SOURCE_DONATION, protocol_id=donation_id, saved_samples=new_samples, input_samples=samples_raw, ) if commit: db.commit() db.refresh(obj) else: db.flush() return obj # --------------------------------------------------------------------------- # 5. delete_donation / batch_delete_donations # --------------------------------------------------------------------------- def delete_donation( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str] = None, ) -> None: """软删除捐赠单 + 所有子资源。""" obj = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == donation_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .first() ) if not obj: raise ValueError("捐赠单不存在或无权操作") _soft_delete_all(db, donation_id, tenant_no, user_id) db.commit() def batch_delete_donations( db: Session, ids: List[str], tenant_no: str, user_id: Optional[str] = None, ) -> int: """批量软删除捐赠单。""" count = 0 for did in (ids or []): did = str(did).strip() if not did: continue obj = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == did, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .first() ) if not obj: continue _soft_delete_all(db, did, tenant_no, user_id) count += 1 db.commit() return count # --------------------------------------------------------------------------- # 6. list_audits # --------------------------------------------------------------------------- def list_audits( db: Session, *, tenant_no: str, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, audit_status: Optional[str] = None, research_donation_id: Optional[str] = None, user_info=None, ) -> Dict[str, Any]: q_base = ( db.query(BizResearchDonationAudit, BizResearchDonation.donation_no) .join( BizResearchDonation, and_( BizResearchDonationAudit.research_donation_id == BizResearchDonation.id, BizResearchDonation.del_flag == 0, ), ) .filter(BizResearchDonationAudit.del_flag == 0) ) if tenant_no: q_base = q_base.filter(BizResearchDonationAudit.tenant_no == tenant_no) if audit_status: q_base = q_base.filter(BizResearchDonationAudit.audit_status == audit_status.strip()) did = (research_donation_id or "").strip() if did: q_base = q_base.filter(BizResearchDonationAudit.research_donation_id == did) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q_base = q_base.filter( or_( BizResearchDonation.donation_no.like(like), BizResearchDonationAudit.audit_remark.like(like), ) ) if user_info: from app.core.data_scope import apply_data_scope q_base = apply_data_scope(q_base, BizResearchDonationAudit, user_info, db, tenant_no) total = ( q_base.order_by(None) .with_entities(func.count(BizResearchDonationAudit.id)) .scalar() ) or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q_base.order_by(BizResearchDonationAudit.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) auditor_ids = list({a.auditor_user_id for a, _ in rows if a.auditor_user_id}) auditor_name_map = _user_nick_name_map(db, auditor_ids) result = [ _audit_to_dict( a, donation_no=pn or "", auditor_user_name=auditor_name_map.get(a.auditor_user_id or "", ""), ) for a, pn in rows ] return { "list": result, "total": int(total), "page": page, "pageSize": page_size, "totalPages": total_pages(int(total), page_size), } # --------------------------------------------------------------------------- # 7. submit_audit # --------------------------------------------------------------------------- def submit_audit( db: Session, donation_id: str, *, audit_status: str, audit_remark: Optional[str] = None, user_id: Optional[str] = None, tenant_no: str = "", ) -> BizResearchDonationAudit: """提交审核。shtg → 捐赠状态变为 sxz；shth → 状态不变。""" # 校验捐赠单存在 main = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == donation_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .first() ) if not main: raise ValueError("捐赠单不存在") st = (audit_status or "").strip() if st not in ("shtg", "shth"): raise ValueError("审核状态无效") now = datetime.now() obj = BizResearchDonationAudit( id=generate_uuid(), research_donation_id=donation_id, auditor_user_id=user_id, audit_remark=(audit_remark or "").strip() or None, audit_status=st, tenant_no=tenant_no, order_num=0, remark=None, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(obj) # 审核通过 → 变更状态为 sxz if st == "shtg": main.donation_status = "sxz" main.effective_time = now main.update_by = user_id main.update_time = now db.add(main) db.commit() db.refresh(obj) return obj # --------------------------------------------------------------------------- # 8. change_status # --------------------------------------------------------------------------- def change_status( db: Session, donation_id: str, *, new_status: str, desc: Optional[str] = None, time_val: Optional[Any] = None, void_type: Optional[str] = None, user_id: Optional[str] = None, tenant_no: str = "", ) -> BizResearchDonation: """变更捐赠单状态（含合法性校验）。""" obj = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == donation_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .first() ) if not obj: raise ValueError("捐赠单不存在") current = obj.donation_status or "" ns = (new_status or "").strip() assert_protocol_status_transition(current, ns, kind="donation") now = datetime.now() if isinstance(time_val, str): time_val = parse_datetime(time_val) effective_time = time_val or now change_desc = (desc or "").strip() or None normalized_void_type: Optional[str] = None if current != ns: obj.donation_status = ns obj.update_by = user_id obj.update_time = now # 更新对应的时间/描述字段 if ns == "sxz": obj.effective_time = effective_time obj.effective_desc = change_desc elif ns == "yzf": obj.void_time = effective_time obj.void_desc = change_desc from app.crud.dict_data import normalize_void_type normalized_void_type = normalize_void_type(db, value=void_type or "", tenant_no=tenant_no) obj.void_type = normalized_void_type elif ns == "ygq": obj.expire_time = effective_time obj.expire_desc = change_desc elif ns == "zckc": obj.to_storage_time = effective_time obj.to_storage_desc = change_desc change_type = _DONATION_STATUS_CHANGE_TYPE.get(ns) if change_type: append_donation_status_change( db, research_donation_id=obj.id, change_type=change_type, from_status=current, to_status=ns, change_time=effective_time, change_desc=change_desc, void_type=normalized_void_type if ns == "yzf" else None, tenant_no=tenant_no, user_id=user_id, ) db.add(obj) if ns == "yzf" and (current or "").strip() != "yzf": from app.services.protocol_reception_discard_sync import after_protocol_voided after_protocol_voided( db, source="research_donation", protocol_id=obj.id, tenant_no=tenant_no, user_id=user_id, ) db.commit() db.refresh(obj) return obj # --------------------------------------------------------------------------- # 9. convert_from_storage # --------------------------------------------------------------------------- def convert_from_storage( db: Session, protocol_id: str, *, tenant_no: str, user_id: Optional[str] = None, ) -> BizResearchDonation: """从细胞存储协议单转换创建科研捐赠单。""" # 读取源协议 proto = ( db.query(BizCollectionProtocol) .filter( and_( BizCollectionProtocol.id == protocol_id, BizCollectionProtocol.tenant_no == tenant_no, BizCollectionProtocol.del_flag == 0, ) ) .populate_existing().with_for_update() .first() ) if not proto: raise ValueError("源协议单不存在") if proto.protocol_status != "sxz": raise ValueError("仅生效中的协议可转科研") if has_active_convert_link(db, tenant_no, protocol_id=protocol_id): raise ValueError("该协议已存在转移关联，无法重复转科研") pn_src = (proto.protocol_no or "").strip() if not pn_src: raise ValueError("源协议单协议号为空，无法转科研") # 读取源子资源 src_party = ( db.query(BizProtocolPartyCustomer) .filter( and_( BizProtocolPartyCustomer.collection_protocol_id == protocol_id, BizProtocolPartyCustomer.tenant_no == tenant_no, BizProtocolPartyCustomer.del_flag == 0, ) ) .first() ) src_coll = ( db.query(BizProtocolCollectionInfo) .filter( and_( BizProtocolCollectionInfo.collection_protocol_id == protocol_id, BizProtocolCollectionInfo.tenant_no == tenant_no, BizProtocolCollectionInfo.del_flag == 0, ) ) .first() ) src_samples = ( db.query(BizProtocolSample) .filter( and_( BizProtocolSample.collection_protocol_id == protocol_id, BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, ) ) .all() ) src_transports = ( db.query(BizProtocolSampleTransport) .filter( and_( BizProtocolSampleTransport.collection_protocol_id == protocol_id, BizProtocolSampleTransport.tenant_no == tenant_no, BizProtocolSampleTransport.del_flag == 0, ) ) .all() ) transport_by_sample = {t.sample_id: t for t in src_transports} now = datetime.now() donation_no = pn_src monthly_verify_no = allocate_monthly_verify_no( db, tenant_no=tenant_no, create_time=now ) # 创建主记录 donation = BizResearchDonation( project_id=proto.project_id, id=generate_uuid(), donation_no=donation_no, monthly_verify_no=monthly_verify_no, donation_status="dsx", donation_date=now.strftime("%Y%m%d"), source_type="from_storage", source_protocol_id=protocol_id, attachment_urls=(proto.protocol_attachments or "").strip() or None, collector_user_id=proto.collector_user_id, tenant_no=tenant_no, order_num=0, remark=f"由细胞存储协议单 {proto.protocol_no} 转入", create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(donation) db.flush() did = donation.id # 创建 party if src_party: party = BizResearchDonationParty( id=generate_uuid(), research_donation_id=did, customer_id=src_party.customer_id, father_customer_id=getattr(src_party, "father_customer_id", None), newborn_customer_id=getattr(src_party, "newborn_customer_id", None), mother_name=src_party.mother_name, mother_phone=src_party.mother_phone, mother_id_card=src_party.mother_id_card, mother_birth_ym=src_party.mother_birth_ym, mother_ethnic=src_party.mother_ethnic, mother_email=src_party.mother_email, mother_address=src_party.mother_address, father_name=src_party.father_name, father_phone=src_party.father_phone, father_id_card=src_party.father_id_card, newborn_name=src_party.newborn_name, newborn_id_card=src_party.newborn_id_card, expected_delivery_date=src_party.expected_delivery_date, newborn_delivery_time=src_party.newborn_delivery_time, newborn_sex=src_party.newborn_sex, delivery_mode=src_party.delivery_mode, birth_order=src_party.birth_order, newborn_weight_gram=src_party.newborn_weight_gram, apgar_1min=src_party.apgar_1min, apgar_5min=src_party.apgar_5min, collector_user_id=src_party.collector_user_id, tenant_no=tenant_no, order_num=0, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(party) # 创建 collection if src_coll: coll = BizResearchDonationCollection( id=generate_uuid(), research_donation_id=did, region_id=src_coll.region_id, hospital_id=src_coll.hospital_id, medical_staff=src_coll.medical_staff, sales_user_id=src_coll.sales_user_id, sample_collect_time=src_coll.sample_collect_time, maternal_blood_collect_time=src_coll.maternal_blood_collect_time, tenant_no=tenant_no, order_num=0, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(coll) # 创建 samples + transports for idx, ss in enumerate(src_samples): new_sample = BizResearchDonationSample( id=generate_uuid(), research_donation_id=did, sample_type_id=ss.sample_type_id, sample_code_id=ss.sample_code_id, sample_code=ss.sample_code, collect_amount=ss.collect_amount, unit_id=ss.unit_id, storage_years=ss.storage_years, sample_status=ss.sample_status, storage_in_date=ss.storage_in_date, detect_start_date=ss.detect_start_date, void_date=ss.void_date, storage_expire_date=ss.storage_expire_date, is_special=getattr(ss, "is_special", 0), special_remark=getattr(ss, "special_remark", None), tenant_no=tenant_no, order_num=idx, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(new_sample) db.flush() # 对应运输记录 src_t = transport_by_sample.get(ss.id) if src_t: new_t = BizResearchDonationSampleTransport( id=generate_uuid(), research_donation_id=did, sample_id=new_sample.id, sample_code=ss.sample_code or None, transport_type=src_t.transport_type, express_company_id=src_t.express_company_id, waybill_no=src_t.waybill_no, dest_site_id=src_t.dest_site_id, express_fee=src_t.express_fee, eta_time=src_t.eta_time, carrier_name=src_t.carrier_name, carrier_phone=src_t.carrier_phone, tenant_no=tenant_no, order_num=idx, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(new_t) # 更新源协议状态为 zky（转科研） from_status = (proto.protocol_status or "").strip() to_research_desc = f"已转入科研捐赠 {donation_no}" proto.protocol_status = "zky" proto.to_research_time = now proto.to_research_desc = to_research_desc proto.update_by = user_id proto.update_time = now db.add(proto) append_protocol_status_change( db, collection_protocol_id=protocol_id, change_type="to_research", from_status=from_status, to_status="zky", change_time=now, change_desc=to_research_desc, tenant_no=tenant_no, user_id=user_id, ) create_convert_link( db, convert_type="storage_to_research", protocol_id=protocol_id, research_donation_id=did, biz_no=donation_no, tenant_no=tenant_no, user_id=user_id, now=now, ) db.commit() db.refresh(donation) return donation # --------------------------------------------------------------------------- # 10. convert_to_storage # --------------------------------------------------------------------------- def convert_to_storage( db: Session, donation_id: str, *, tenant_no: str, user_id: Optional[str] = None, ) -> BizResearchDonation: """从科研捐赠单转换创建细胞存储协议单，返回更新后的捐赠单。""" # 读取源捐赠 donation = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == donation_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .populate_existing().with_for_update() .first() ) if not donation: raise ValueError("捐赠单不存在") if donation.donation_status != "sxz": raise ValueError("仅生效中的捐赠单可转细胞存储") partner = get_convert_partner(db, tenant_no, donation_id=donation_id) if partner: linked_proto = ( db.query(BizCollectionProtocol) .filter( and_( BizCollectionProtocol.id == partner["protocol_id"], BizCollectionProtocol.tenant_no == tenant_no, BizCollectionProtocol.del_flag == 0, ) ) .first() ) if linked_proto and linked_proto.protocol_status != "zky": raise ValueError("该捐赠单已转细胞存储，无法重复操作") dn_src = (donation.donation_no or "").strip() if not dn_src: raise ValueError("源捐赠单协议号为空，无法转细胞存储") src_party = get_party(db, donation_id, tenant_no) src_coll = get_collection(db, donation_id, tenant_no) src_samples = list_samples(db, donation_id, tenant_no) src_transports = list_sample_transports(db, donation_id, tenant_no) transport_by_sample = {t.sample_id: t for t in src_transports} now = datetime.now() protocol_no = dn_src protocol_no_id = None if (donation.source_protocol_id or "").strip(): src_proto = ( db.query(BizCollectionProtocol) .filter( and_( BizCollectionProtocol.id == donation.source_protocol_id.strip(), BizCollectionProtocol.tenant_no == tenant_no, BizCollectionProtocol.del_flag == 0, ) ) .first() ) if src_proto and src_proto.protocol_no_id: protocol_no_id = src_proto.protocol_no_id protocol_attachment_urls = _merge_comma_separated_urls( donation.consent_form_urls, donation.infectious_disease_report_urls, donation.attachment_urls, ) # 创建协议单 proto = BizCollectionProtocol( project_id=donation.project_id, id=generate_uuid(), protocol_no_id=protocol_no_id, protocol_no=protocol_no, protocol_status="dsx", register_date=now.strftime("%Y%m%d"), protocol_attachments=protocol_attachment_urls, collector_user_id=donation.collector_user_id, tenant_no=tenant_no, order_num=0, remark=f"由科研捐赠单 {donation.donation_no} 转入", create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(proto) db.flush() pid = proto.id # 创建 party if src_party: party = BizProtocolPartyCustomer( id=generate_uuid(), collection_protocol_id=pid, customer_id=src_party.customer_id, father_customer_id=getattr(src_party, "father_customer_id", None), newborn_customer_id=getattr(src_party, "newborn_customer_id", None), mother_name=src_party.mother_name, mother_phone=src_party.mother_phone, mother_id_card=src_party.mother_id_card, mother_birth_ym=src_party.mother_birth_ym, mother_ethnic=src_party.mother_ethnic, mother_email=src_party.mother_email, mother_address=src_party.mother_address, father_name=src_party.father_name, father_phone=src_party.father_phone, father_id_card=src_party.father_id_card, newborn_name=src_party.newborn_name, newborn_id_card=src_party.newborn_id_card, expected_delivery_date=src_party.expected_delivery_date, newborn_delivery_time=src_party.newborn_delivery_time, newborn_sex=src_party.newborn_sex, delivery_mode=src_party.delivery_mode, birth_order=src_party.birth_order, newborn_weight_gram=src_party.newborn_weight_gram, apgar_1min=src_party.apgar_1min, apgar_5min=src_party.apgar_5min, collector_user_id=src_party.collector_user_id, tenant_no=tenant_no, order_num=0, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(party) # 创建 collection if src_coll: coll = BizProtocolCollectionInfo( id=generate_uuid(), collection_protocol_id=pid, region_id=src_coll.region_id, hospital_id=src_coll.hospital_id, medical_staff=src_coll.medical_staff, sales_user_id=src_coll.sales_user_id, sample_collect_time=src_coll.sample_collect_time, maternal_blood_collect_time=src_coll.maternal_blood_collect_time, tenant_no=tenant_no, order_num=0, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(coll) # 创建 samples + transports（无 price 字段 → null） for idx, ss in enumerate(src_samples): new_sample = BizProtocolSample( id=generate_uuid(), collection_protocol_id=pid, sample_type_id=ss.sample_type_id, sample_code_id=ss.sample_code_id, sample_code=ss.sample_code, collect_amount=ss.collect_amount, unit_id=ss.unit_id, price=None, storage_years=ss.storage_years, sample_status=ss.sample_status, storage_in_date=ss.storage_in_date, detect_start_date=ss.detect_start_date, void_date=ss.void_date, storage_expire_date=ss.storage_expire_date, is_special=ss.is_special, special_remark=ss.special_remark, tenant_no=tenant_no, order_num=idx, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(new_sample) db.flush() src_t = transport_by_sample.get(ss.id) if src_t: new_t = BizProtocolSampleTransport( id=generate_uuid(), collection_protocol_id=pid, sample_id=new_sample.id, sample_code=ss.sample_code or None, transport_type=src_t.transport_type, express_company_id=src_t.express_company_id, waybill_no=src_t.waybill_no, dest_site_id=src_t.dest_site_id, express_fee=src_t.express_fee, eta_time=src_t.eta_time, carrier_name=src_t.carrier_name, carrier_phone=src_t.carrier_phone, tenant_no=tenant_no, order_num=idx, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(new_t) # 更新捐赠单状态为 zckc from_status = (donation.donation_status or "").strip() to_storage_desc = f"已转细胞存储 {protocol_no}" donation.donation_status = "zckc" donation.to_storage_time = now donation.to_storage_desc = to_storage_desc donation.update_by = user_id donation.update_time = now db.add(donation) append_donation_status_change( db, research_donation_id=donation_id, change_type="to_storage", from_status=from_status, to_status="zckc", change_time=now, change_desc=to_storage_desc, tenant_no=tenant_no, user_id=user_id, ) upsert_convert_link_for_to_storage( db, protocol_id=pid, research_donation_id=donation_id, biz_no=protocol_no, tenant_no=tenant_no, user_id=user_id, now=now, ) db.commit() db.refresh(donation) return donation # --------------------------------------------------------------------------- # Internal: sub-resource create helpers # --------------------------------------------------------------------------- _PARTY_STR_FIELDS = [ "customer_id", "father_customer_id", "newborn_customer_id", "mother_name", "mother_phone", "mother_id_card", "mother_birth_ym", "mother_ethnic", "mother_email", "mother_address", "father_name", "father_phone", "father_id_card", "newborn_name", "newborn_id_card", "newborn_sex", "delivery_mode", "collector_user_id", ] _PARTY_INT_FIELDS = ["birth_order", "newborn_weight_gram"] _PARTY_DEC_FIELDS = ["apgar_1min", "apgar_5min"] _PARTY_DATE_FIELDS = ["expected_delivery_date"] _PARTY_DT_FIELDS = ["newborn_delivery_time"] def _apply_single_child_fields(obj, raw, str_fields, int_fields=(), dec_fields=(), date_fields=(), dt_fields=()): obj.remark = (raw.get("remark") or "").strip() or None for key in str_fields: setattr(obj, key, ((raw.get(key) or "").strip() or None) if key in raw else None) for key in int_fields: setattr(obj, key, raw.get(key) if key in raw else None) for key in dec_fields: setattr(obj, key, _optional_decimal(raw.get(key)) if key in raw else None) for key in date_fields: setattr(obj, key, optional_date(raw.get(key)) if key in raw else None) for key in dt_fields: if key not in raw: setattr(obj, key, None) continue val = raw.get(key) if isinstance(val, str): val = parse_datetime(val) setattr(obj, key, val) def _create_party( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], raw: dict, now: datetime, create_dept_id: str = "", ) -> BizResearchDonationParty: def build_row(row_id): return BizResearchDonationParty( id=row_id, research_donation_id=donation_id, tenant_no=tenant_no, create_by=user_id, create_dept_id=create_dept_id, create_time=now, ) def apply_row(row, item, _idx, _now): row.order_num = 0 _apply_single_child_fields( row, item, _PARTY_STR_FIELDS, _PARTY_INT_FIELDS, _PARTY_DEC_FIELDS, _PARTY_DATE_FIELDS, _PARTY_DT_FIELDS, ) rows = sync_child_rows( db, BizResearchDonationParty, parent_attr="research_donation_id", parent_id=donation_id, tenant_no=tenant_no, items=[raw or {}], user_id=user_id, now=now, build_row=build_row, apply_fields=apply_row, natural_key=lambda _item: "party", row_natural_key=lambda _row: "party", duplicate_message="当事人信息重复", foreign_message="当事人信息不属于当前捐赠单", ) db.flush() return rows[0] _COLL_STR_FIELDS = [ "region_id", "hospital_id", "medical_staff", "sales_user_id", "region_manager_user_id", ] _COLL_DT_FIELDS = ["sample_collect_time", "maternal_blood_collect_time"] def _create_collection( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], raw: dict, now: datetime, create_dept_id: str = "", ) -> BizResearchDonationCollection: def build_row(row_id): return BizResearchDonationCollection( id=row_id, research_donation_id=donation_id, tenant_no=tenant_no, create_by=user_id, create_dept_id=create_dept_id, create_time=now, ) def apply_row(row, item, _idx, _now): row.order_num = 0 _apply_single_child_fields( row, item, _COLL_STR_FIELDS, dt_fields=_COLL_DT_FIELDS, ) rows = sync_child_rows( db, BizResearchDonationCollection, parent_attr="research_donation_id", parent_id=donation_id, tenant_no=tenant_no, items=[raw or {}], user_id=user_id, now=now, build_row=build_row, apply_fields=apply_row, natural_key=lambda _item: "collection", row_natural_key=lambda _row: "collection", duplicate_message="采集信息重复", foreign_message="采集信息不属于当前捐赠单", ) db.flush() return rows[0] _SAMPLE_STR_FIELDS = [ "sample_type_id", "sample_code_id", "sample_code", "unit_id", "sample_status", "storage_in_date", "detect_start_date", "void_date", "storage_expire_date", ] _SAMPLE_DEC_FIELDS = ["collect_amount"] _SAMPLE_INT_FIELDS = ["storage_years"] def _migrate_donation_sample_child_refs_on_replace( db: Session, *, tenant_no: str, old_id_to_code: Dict[str, str], new_code_to_id: Dict[str, str], user_id: Optional[str], ) -> None: """捐赠样本行重建后，将接收/质检工单的捐赠样本 ID 迁移到新样本 ID。""" if not old_id_to_code or not new_code_to_id: return old_ids = list(old_id_to_code.keys()) now = datetime.now() rec_rows = ( db.query(BizSampleReception) .filter( BizSampleReception.donation_sample_id.in_(old_ids), BizSampleReception.tenant_no == tenant_no, BizSampleReception.del_flag == 0, BizSampleReception.reception_source == "research_donation", ) .all() ) for row in rec_rows: old_sid = (row.donation_sample_id or "").strip() code = old_id_to_code.get(old_sid) or (row.sample_code or "").strip() new_sid = new_code_to_id.get(code) if code else None if not new_sid: continue row.donation_sample_id = new_sid row.update_by = user_id row.update_time = now db.add(row) wo_rows = ( db.query(BizQcWorkOrder) .filter( BizQcWorkOrder.protocol_sample_id.in_(old_ids), BizQcWorkOrder.tenant_no == tenant_no, BizQcWorkOrder.del_flag == 0, BizQcWorkOrder.reception_source == "research_donation", ) .all() ) for row in wo_rows: old_sid = (row.protocol_sample_id or "").strip() code = old_id_to_code.get(old_sid) or (row.sample_code or "").strip() new_sid = new_code_to_id.get(code) if code else None if not new_sid: continue row.protocol_sample_id = new_sid row.update_by = user_id row.update_time = now db.add(row) _retarget_protocol_sample_id( db, BizProductionWorkOrder, old_ids=old_ids, old_id_to_code=old_id_to_code, new_code_to_id=new_code_to_id, tenant_no=tenant_no, user_id=user_id, now=now, ) _retarget_protocol_sample_id( db, BizQcSmsNotification, old_ids=old_ids, old_id_to_code=old_id_to_code, new_code_to_id=new_code_to_id, tenant_no=tenant_no, user_id=user_id, now=now, del_flag_alive="0", ) def _apply_donation_sample_fields(obj, raw, *, order_num, user_id, now) -> None: obj.order_num = order_num obj.remark = (raw.get("remark") or "").strip() or None obj.update_by = user_id obj.update_time = now obj.del_flag = 0 for key in _SAMPLE_STR_FIELDS: if key == "sample_status": continue if key in raw: setattr(obj, key, (raw[key] or "").strip() or None) if "sample_status" in raw: status_val = raw.get("sample_status") obj.sample_status = ( str(status_val).strip() if status_val is not None and str(status_val).strip() else None ) for key in _SAMPLE_DEC_FIELDS: if key in raw and raw[key] is not None: setattr(obj, key, Decimal(str(raw[key]))) for key in _SAMPLE_INT_FIELDS: if key in raw: setattr(obj, key, raw[key]) apply_sample_special_fields(obj, raw) def _create_samples( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], samples: list, now: datetime, create_dept_id: str = "", ) -> List[BizResearchDonationSample]: existing = list_samples(db, donation_id, tenant_no) by_id = {sample.id: sample for sample in existing} by_code: Dict[str, List[BizResearchDonationSample]] = {} for sample in existing: code = (sample.sample_code or "").strip() if code: by_code.setdefault(code, []).append(sample) claimed = set() planned = [] for idx, raw_item in enumerate(samples or []): raw = _normalize_fields(raw_item) validate_sample_special_raw(raw, index=idx) sid = (raw.get("id") or "").strip() code = (raw.get("sample_code") or "").strip() row = by_id.get(sid) if sid else None if row is None and sid: row = _owned_child_row( db, BizResearchDonationSample, sid, parent_attr="research_donation_id", parent_id=donation_id, tenant_no=tenant_no, message="样本不属于当前捐赠单", ) if row is None and code: for candidate in by_code.get(code, []): if candidate.id not in claimed: row = candidate break if row is not None: if row.id in claimed: raise ValueError("样本编号重复") claimed.add(row.id) planned.append((idx, raw, row)) kept_codes = { (sample.sample_code or "").strip() for sample in existing if sample.id in claimed and (sample.sample_code or "").strip() } validate_sample_codes_not_used( db, tenant_no=tenant_no, samples=[ raw for _idx, raw, row in planned if row is None or (raw.get("sample_code") or "").strip() not in kept_codes ], exclude_donation_id=donation_id, ) result = [] assigned_codes: List[str] = [] kept_transport_ids = set() for idx, raw, obj in planned: code = (raw.get("sample_code") or "").strip() if obj is None and not code and (raw.get("sample_type_id") or "").strip(): code = next_donation_sample_code( db, tenant_no=tenant_no, extra_codes=assigned_codes, ) raw["sample_code"] = code if code: assigned_codes.append(code) if obj is None: obj = BizResearchDonationSample( id=(raw.get("id") or "").strip() or generate_uuid(), research_donation_id=donation_id, tenant_no=tenant_no, create_by=user_id, create_dept_id=create_dept_id, create_time=now, ) _apply_donation_sample_fields( obj, raw, order_num=idx, user_id=user_id, now=now ) obj.research_donation_id = donation_id obj.tenant_no = tenant_no db.add(obj) db.flush() result.append(obj) transport_raw = raw.get("transport") if isinstance(raw.get("transport"), dict) else {} if not transport_raw and raw.get("eta_time"): transport_raw = {"eta_time": raw.get("eta_time")} elif isinstance(transport_raw, dict) and raw.get("eta_time") and not ( transport_raw.get("eta_time") or transport_raw.get("etaTime") ): transport_raw = {**transport_raw, "eta_time": raw.get("eta_time")} if transport_raw: transport_row = _create_transport( db, donation_id, obj.id, tenant_no, user_id, transport_raw, idx, now, sample_code=obj.sample_code, ) if transport_row is not None: kept_transport_ids.add(transport_row.id) for sample in existing: if sample.id in claimed: continue sample.del_flag = 1 sample.update_by = user_id sample.update_time = now db.add(sample) for transport in ( db.query(BizResearchDonationSampleTransport) .filter( BizResearchDonationSampleTransport.research_donation_id == donation_id, BizResearchDonationSampleTransport.tenant_no == tenant_no, BizResearchDonationSampleTransport.del_flag == 0, ) .all() ): if transport.id in kept_transport_ids: continue transport.del_flag = 1 transport.update_by = user_id transport.update_time = now db.add(transport) return result def _create_transport( db: Session, donation_id: str, sample_id: str, tenant_no: str, user_id: Optional[str], raw: dict, order_num: int, now: datetime, sample_code: Optional[str] = None, ) -> Optional[BizResearchDonationSampleTransport]: """写回捐赠样本运输；可仅有预计到达时间（运费批次改造后）。""" tnorm = _normalize_fields(raw) transport_type = (tnorm.get("transport_type") or "").strip() eta_time = tnorm.get("eta_time") if isinstance(eta_time, str): eta_time = parse_datetime(eta_time) if not transport_type and not eta_time: return None express_fee = tnorm.get("express_fee") if express_fee is not None and express_fee != "": express_fee = Decimal(str(express_fee)) else: express_fee = None existing = ( db.query(BizResearchDonationSampleTransport) .filter( BizResearchDonationSampleTransport.research_donation_id == donation_id, BizResearchDonationSampleTransport.sample_id == sample_id, BizResearchDonationSampleTransport.tenant_no == tenant_no, BizResearchDonationSampleTransport.del_flag == 0, ) .order_by(BizResearchDonationSampleTransport.create_time.asc()) .all() ) obj = existing[0] if existing else BizResearchDonationSampleTransport( id=generate_uuid(), research_donation_id=donation_id, sample_id=sample_id, tenant_no=tenant_no, create_by=user_id, create_dept_id="", create_time=now, ) obj.sample_code = sample_code or None obj.transport_type = transport_type obj.express_company_id = (tnorm.get("express_company_id") or "").strip() or None obj.waybill_no = (tnorm.get("waybill_no") or "").strip() or None obj.dest_site_id = (tnorm.get("dest_site_id") or "").strip() or None obj.express_fee = express_fee obj.eta_time = eta_time obj.carrier_name = (tnorm.get("carrier_name") or "").strip() or None obj.carrier_phone = (tnorm.get("carrier_phone") or "").strip() or None obj.order_num = order_num obj.remark = (tnorm.get("remark") or "").strip() or None obj.del_flag = 0 obj.update_by = user_id obj.update_time = now db.add(obj) for extra in existing[1:]: extra.del_flag = 1 extra.update_by = user_id extra.update_time = now db.add(extra) db.flush() return obj # --------------------------------------------------------------------------- # Internal: soft delete helpers # --------------------------------------------------------------------------- def _soft_delete_party( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], ) -> None: rows = ( db.query(BizResearchDonationParty) .filter( and_( BizResearchDonationParty.research_donation_id == donation_id, BizResearchDonationParty.tenant_no == tenant_no, BizResearchDonationParty.del_flag == 0, ) ) .all() ) for r in rows: r.del_flag = 1 r.update_by = user_id r.update_time = datetime.now() db.add(r) if rows: db.flush() def _soft_delete_collection( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], ) -> None: rows = ( db.query(BizResearchDonationCollection) .filter( and_( BizResearchDonationCollection.research_donation_id == donation_id, BizResearchDonationCollection.tenant_no == tenant_no, BizResearchDonationCollection.del_flag == 0, ) ) .all() ) for r in rows: r.del_flag = 1 r.update_by = user_id r.update_time = datetime.now() db.add(r) if rows: db.flush() def _soft_delete_samples( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], ) -> None: rows = ( db.query(BizResearchDonationSample) .filter( and_( BizResearchDonationSample.research_donation_id == donation_id, BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, ) ) .all() ) for r in rows: r.del_flag = 1 r.update_by = user_id r.update_time = datetime.now() db.add(r) if rows: db.flush() def _soft_delete_transports( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], ) -> None: rows = ( db.query(BizResearchDonationSampleTransport) .filter( and_( BizResearchDonationSampleTransport.research_donation_id == donation_id, BizResearchDonationSampleTransport.tenant_no == tenant_no, BizResearchDonationSampleTransport.del_flag == 0, ) ) .all() ) for r in rows: r.del_flag = 1 r.update_by = user_id r.update_time = datetime.now() db.add(r) if rows: db.flush() def _soft_delete_all( db: Session, donation_id: str, tenant_no: str, user_id: Optional[str], ) -> None: """软删除主记录 + 所有子资源。""" now = datetime.now() # 主记录 main = ( db.query(BizResearchDonation) .filter( and_( BizResearchDonation.id == donation_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) ) .first() ) if main: main.del_flag = 1 main.update_by = user_id main.update_time = now db.add(main) _soft_delete_party(db, donation_id, tenant_no, user_id) _soft_delete_collection(db, donation_id, tenant_no, user_id) _soft_delete_samples(db, donation_id, tenant_no, user_id) _soft_delete_transports(db, donation_id, tenant_no, user_id)