"""采集协议单 / 审核 / 缴费明细 CRUD（细胞存储 2.2.6～2.2.10）""" from __future__ import annotations import math import calendar from datetime import datetime, timedelta 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.core.utils import generate_uuid, parse_datetime from app.crud.base import camel_to_snake from app.crud.protocol_donation_convert import ( assert_biz_no_available, get_partner_sample_codes, ) from app.models.research_donation import BizResearchDonation, BizResearchDonationSample from app.models.models import ( BizAttachment, BizCollectionProtocol, BizProtocolAudit, BizProtocolCollectionInfo, BizProtocolFullmoonTodo, BizProtocolPartyCustomer, BizProtocolPayment, BizProtocolPaymentDetail, BizProtocolSample, BizProtocolSampleTransport, BizProtocolStorageCert, BizCollectionRegion, BizCollectionHospital, SysUser, ) def get_protocol_nos_used_in_protocols( db: Session, *, tenant_no: str, protocol_nos: List[str], ) -> set: """返回已在采集协议单或科研捐赠单中占用的协议编号集合（同租户、未删除）。""" nos = list({(n or "").strip() for n in protocol_nos if (n or "").strip()}) if not nos: return set() used: set = set() proto_rows = ( db.query(BizCollectionProtocol.protocol_no) .filter( BizCollectionProtocol.tenant_no == tenant_no, BizCollectionProtocol.del_flag == 0, BizCollectionProtocol.protocol_no.in_(nos), ) .all() ) used.update(r[0] for r in proto_rows if r[0]) donation_rows = ( db.query(BizResearchDonation.donation_no) .filter( BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, BizResearchDonation.donation_no.in_(nos), ) .all() ) used.update(r[0] for r in donation_rows if r[0]) return used def get_sample_codes_used_in_protocol_samples( db: Session, *, tenant_no: str, sample_codes: List[str], ) -> set: """返回已在采集协议单或科研捐赠单样本表中占用的样本编码集合（同租户、未删除）。""" codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()}) if not codes: return set() used: set = set() proto_rows = ( db.query(BizProtocolSample.sample_code) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_code.in_(codes), ) .all() ) used.update(r[0] for r in proto_rows if r[0]) donation_rows = ( db.query(BizResearchDonationSample.sample_code) .filter( BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, BizResearchDonationSample.sample_code.in_(codes), ) .all() ) used.update(r[0] for r in donation_rows if r[0]) return used def validate_sample_codes_not_used( db: Session, *, tenant_no: str, samples: list, exclude_protocol_id: Optional[str] = None, exclude_donation_id: Optional[str] = None, ) -> None: """创建/更新样本行前校验：样本编码不得已被任一业务单占用（转移配对单豁免）。""" codes: List[str] = [] for raw_item in samples or []: raw = _normalize_fields(raw_item) c = (raw.get("sample_code") or "").strip() if c: codes.append(c) if not codes: return used = get_sample_codes_used_in_protocol_samples( db, tenant_no=tenant_no, sample_codes=codes ) exempt = get_partner_sample_codes( db, tenant_no, protocol_id=exclude_protocol_id, donation_id=exclude_donation_id, ) used -= exempt conflicts = set(codes) & used if conflicts: first = sorted(conflicts)[0] raise ValueError(f"样本编码「{first}」已被占用") def total_pages(total: int, page_size: int) -> int: if total <= 0: return 0 return int(math.ceil(total / float(page_size))) def _fmt_dt(t: Optional[datetime]) -> str: return t.strftime("%Y-%m-%d %H:%M:%S") if t else "" 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)) def _today_yyyymmdd() -> str: return datetime.now().strftime("%Y%m%d") def _normalize_yyyymmdd_date(val: Any) -> Optional[str]: """将 YYYYMMDD / YYYY-MM-DD 等格式规范为 8 位日期字符串。""" if val is None: return None s = str(val).strip() if not s: return None compact = s.replace("-", "").replace("/", "")[:8] if len(compact) == 8 and compact.isdigit(): return compact return None _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 _pop_attachment_urls(raw: Dict[str, Any], *keys: str) -> Optional[str]: """从保存载荷中取出附件字段（支持 snake/camel、URL 串或附件对象列表）。""" for key in keys: if key not in raw: continue urls = attachments_to_urls(raw.pop(key)) if urls: return urls return None def attachments_to_urls(attachments: Any) -> Optional[str]: """附件列表或字符串 → 逗号分隔 URL（MinIO）。""" if attachments is None: return None if isinstance(attachments, str): return attachments.strip() or None if isinstance(attachments, list): urls: List[str] = [] for item in attachments: if isinstance(item, dict): u = (item.get("fileUrl") or item.get("file_url") or "").strip() if u: urls.append(u) elif isinstance(item, str) and item.strip(): urls.append(item.strip()) return ",".join(urls) if urls else None return None def _normalize_attachment_url_key(url: str) -> str: u = (url or "").strip() if not u: return "" if "?" in u: u = u.split("?", 1)[0] return u.lstrip("/") def _format_attachment_file_size(size_bytes: int) -> str: if size_bytes < 1024: return f"{size_bytes} B" if size_bytes < 1024 * 1024: return f"{size_bytes / 1024:.1f} KB" return f"{size_bytes / (1024 * 1024):.1f} MB" def _lookup_attachment_by_url( db: Session, url: str, tenant_no: Optional[str], ) -> Optional[BizAttachment]: key = _normalize_attachment_url_key(url) if not key: return None q = db.query(BizAttachment).filter(BizAttachment.del_flag == 0) if tenant_no: q = q.filter(BizAttachment.tenant_no == tenant_no) row = q.filter(BizAttachment.file_url == key).first() if row: return row tail = key.rsplit("/", 1)[-1] if tail: row = ( q.filter(BizAttachment.file_url.like(f"%/{tail}")) .order_by(BizAttachment.create_time.desc()) .first() ) return row def urls_to_attachment_list( urls_str: Optional[str], db: Optional[Session] = None, tenant_no: Optional[str] = None, ) -> List[Dict[str, Any]]: """逗号分隔 URL → 前端附件卡片结构；有 db 时从 biz_attachment 取原始文件名。""" if not urls_str: return [] result: List[Dict[str, Any]] = [] for url in urls_str.split(","): u = url.strip() if not u: continue att = _lookup_attachment_by_url(db, u, tenant_no) if db else None if att: size = att.file_size or 0 result.append( { "id": att.id, "fileUrl": u, "fileName": att.original_name or u.rsplit("/", 1)[-1], "fileSize": size, "fileSizeDisplay": _format_attachment_file_size(size), "fileType": att.file_type or "", } ) else: name = u.rsplit("/", 1)[-1] if "/" in u else u result.append({"fileUrl": u, "fileName": name, "fileSize": 0}) return result 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 _protocol_to_dict(p: BizCollectionProtocol) -> Dict[str, Any]: return { "id": p.id, "protocolNoId": p.protocol_no_id or "", "protocolNo": p.protocol_no or "", "protocolStatus": p.protocol_status or "", "protocolAttachments": p.protocol_attachments or "", "registerDate": (p.register_date or "").strip() if p.register_date else "", "effectiveTime": _fmt_dt(p.effective_time), "effectiveDesc": p.effective_desc or "", "expireTime": _fmt_dt(p.expire_time), "expireDesc": p.expire_desc or "", "voidTime": _fmt_dt(p.void_time), "voidDesc": p.void_desc or "", "toResearchTime": _fmt_dt(p.to_research_time), "toResearchDesc": p.to_research_desc or "", "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 _payment_to_dict(p: BizProtocolPayment) -> Dict[str, Any]: return { "id": p.id, "collectionProtocolId": p.collection_protocol_id, "payMode": p.pay_mode or "", "contractAmount": _dec(p.contract_amount), "installmentCount": p.installment_count, "installmentIntervalMonths": p.installment_interval_months, "lumpSumDeadline": (p.lump_sum_deadline or "").strip() if p.lump_sum_deadline else "", "installmentStartDate": (p.installment_start_date or "").strip() if p.installment_start_date else "", "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 _detail_to_dict( d: BizProtocolPaymentDetail, *, protocol_no: str = "", collection_protocol_id: str = "", protocol_status: str = "", pay_mode: str = "", mother_name: str = "", mother_id_card: str = "", newborn_name: str = "", newborn_id_card: str = "", ) -> Dict[str, Any]: return { "id": d.id, "paymentId": d.payment_id, "collectionProtocolId": collection_protocol_id, "protocolNo": protocol_no, "protocolStatus": protocol_status, "payMode": pay_mode, "motherName": mother_name, "motherIdCard": mother_id_card, "newbornName": newborn_name, "newbornIdCard": newborn_id_card, "transferType": d.transfer_type or "", "dueAmount": _dec(d.due_amount), "transferAmount": _dec(d.transfer_amount), "transferTime": _fmt_dt(d.transfer_time), "transferAttachments": d.transfer_attachments or "", "financeConfirmAttachments": d.finance_confirm_attachments or "", "financeRemark": d.finance_remark or "", "refundFinanceConfirmAttachments": d.refund_finance_confirm_attachments or "", "refundFinanceRemark": d.refund_finance_remark or "", "payStatus": d.pay_status or "", "installmentNo": d.installment_no, "installmentDueDate": (d.installment_due_date or "").strip() if d.installment_due_date else "", "refundAmount": _dec(d.refund_amount), "refundRemark": d.refund_remark or "", "refundAttachments": d.refund_attachments 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: BizProtocolAudit, *, protocol_no: str = "", auditor_user_name: str = "", ) -> Dict[str, Any]: return { "id": a.id, "collectionProtocolId": a.collection_protocol_id, "protocolNo": protocol_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_customer_to_dict(p: BizProtocolPartyCustomer) -> Dict[str, Any]: return { "id": p.id, "collectionProtocolId": p.collection_protocol_id, "customerId": p.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 "", "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 "", "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_info_to_dict(c: BizProtocolCollectionInfo) -> Dict[str, Any]: return { "id": c.id, "collectionProtocolId": c.collection_protocol_id, "regionId": c.region_id or "", "hospitalId": c.hospital_id or "", "medicalStaff": c.medical_staff or "", "salesUserId": c.sales_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: BizProtocolSample) -> Dict[str, Any]: return { "id": s.id, "collectionProtocolId": s.collection_protocol_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 "", "price": _dec(s.price), "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 "", "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: BizProtocolSampleTransport) -> Dict[str, Any]: return { "id": t.id, "collectionProtocolId": t.collection_protocol_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), } def _storage_cert_to_dict(s: BizProtocolStorageCert) -> Dict[str, Any]: return { "id": s.id, "collectionProtocolId": s.collection_protocol_id, "makeStatus": s.make_status or "0", "certImageUrls": s.cert_image_urls or "", "makeDate": (s.make_date or "").strip() if s.make_date else "", "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 list_collection_protocols( db: Session, *, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, protocol_status: Optional[str] = None, tenant_no: Optional[str] = None, scope_collector_user_id: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.del_flag == 0) if tenant_no: q0 = q0.filter(BizCollectionProtocol.tenant_no == tenant_no) if scope_collector_user_id: q0 = q0.filter( BizCollectionProtocol.collector_user_id == scope_collector_user_id ) if protocol_status: q0 = q0.filter(BizCollectionProtocol.protocol_status == protocol_status.strip()) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter( or_( BizCollectionProtocol.protocol_no.like(like), BizCollectionProtocol.remark.like(like), ) ) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCollectionProtocol.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizCollectionProtocol.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) # 批量加载子表数据，避免 N+1 查询 protocol_ids = [r.id for r in rows] pc_map: Dict[str, Dict[str, Any]] = {} ci_map: Dict[str, Dict[str, Any]] = {} if protocol_ids: pc_rows = ( db.query(BizProtocolPartyCustomer) .filter( BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids), BizProtocolPartyCustomer.del_flag == 0, ) .all() ) for pc in pc_rows: pc_map[pc.collection_protocol_id] = _party_customer_to_dict(pc) ci_rows = ( db.query(BizProtocolCollectionInfo) .filter( BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids), BizProtocolCollectionInfo.del_flag == 0, ) .all() ) for ci in ci_rows: ci_map[ci.collection_protocol_id] = _collection_info_to_dict(ci) # 批量解析地区/医院名称 region_ids = list({ci_map.get(pid, {}).get("regionId", "") for pid in protocol_ids} - {""}) hospital_ids = list({ci_map.get(pid, {}).get("hospitalId", "") for pid in protocol_ids} - {""}) region_name_map: Dict[str, str] = {} hospital_name_map: Dict[str, str] = {} if region_ids: for r in db.query(BizCollectionRegion.id, BizCollectionRegion.region_name).filter(BizCollectionRegion.id.in_(region_ids)).all(): region_name_map[r.id] = r.region_name or "" if hospital_ids: for h in db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name).filter(BizCollectionHospital.id.in_(hospital_ids)).all(): hospital_name_map[h.id] = h.hospital_name or "" # 批量解析销售人员名称 sales_user_ids = list({ci_map.get(pid, {}).get("salesUserId", "") for pid in protocol_ids} - {""}) collector_user_ids = list({_protocol_to_dict(r).get("collectorUserId", "") for r in rows} - {""}) all_user_ids = list(set(sales_user_ids + collector_user_ids)) user_name_map: Dict[str, str] = {} if all_user_ids: for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(all_user_ids)).all(): user_name_map[u.id] = u.nick_name or "" # 合并主表 + 子表数据 result = [] for r in rows: d = _protocol_to_dict(r) pc = pc_map.get(r.id) ci = ci_map.get(r.id) d["collectorUserName"] = user_name_map.get(d.get("collectorUserId", ""), "") if pc: d["motherName"] = pc.get("motherName", "") d["motherPhone"] = pc.get("motherPhone", "") d["motherIdCard"] = pc.get("motherIdCard", "") d["motherBirthYm"] = pc.get("motherBirthYm", "") d["motherEthnic"] = pc.get("motherEthnic", "") d["fatherName"] = pc.get("fatherName", "") d["fatherPhone"] = pc.get("fatherPhone", "") d["newbornDeliveryTime"] = pc.get("newbornDeliveryTime", "") d["newbornSex"] = pc.get("newbornSex", "") d["deliveryMode"] = pc.get("deliveryMode", "") d["birthOrder"] = pc.get("birthOrder") d["newbornWeightGram"] = pc.get("newbornWeightGram") else: d["motherName"] = "" d["motherPhone"] = "" d["motherIdCard"] = "" d["motherBirthYm"] = "" d["motherEthnic"] = "" d["fatherName"] = "" d["fatherPhone"] = "" d["newbornDeliveryTime"] = "" d["newbornSex"] = "" d["deliveryMode"] = "" d["birthOrder"] = None d["newbornWeightGram"] = None 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["sampleCollectTime"] = ci.get("sampleCollectTime", "") else: d["regionName"] = "" d["hospitalName"] = "" d["salesUserName"] = "" d["sampleCollectTime"] = "" result.append(d) return result, int(total) def get_collection_protocol( db: Session, protocol_id: str, tenant_no: str, *, scope_collector_user_id: Optional[str] = None, ) -> Optional[BizCollectionProtocol]: filters = [ BizCollectionProtocol.id == protocol_id, BizCollectionProtocol.del_flag == 0, BizCollectionProtocol.tenant_no == tenant_no, ] if scope_collector_user_id: filters.append( BizCollectionProtocol.collector_user_id == scope_collector_user_id ) return db.query(BizCollectionProtocol).filter(and_(*filters)).first() def get_payment_by_protocol_id(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolPayment]: return ( db.query(BizProtocolPayment) .filter( and_( BizProtocolPayment.collection_protocol_id == collection_protocol_id, BizProtocolPayment.tenant_no == tenant_no, BizProtocolPayment.del_flag == 0, ) ) .first() ) def get_protocol_detail_bundle( db: Session, protocol_id: str, tenant_no: str, *, scope_collector_user_id: Optional[str] = None, ) -> Optional[Dict[str, Any]]: p = get_collection_protocol( db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id ) if not p: return None pay = get_payment_by_protocol_id(db, protocol_id, tenant_no) pc = get_party_customer(db, protocol_id, tenant_no) ci = get_collection_info(db, protocol_id, tenant_no) s_rows = list_samples(db, protocol_id, tenant_no) sc = get_storage_cert(db, protocol_id, tenant_no) protocol_dict = _protocol_to_dict(p) ci_dict = _collection_info_to_dict(ci) if ci else None user_ids: List[str] = [] if protocol_dict.get("collectorUserId"): user_ids.append(str(protocol_dict["collectorUserId"])) if ci_dict and ci_dict.get("salesUserId"): user_ids.append(str(ci_dict["salesUserId"])) user_name_map = _user_nick_name_map(db, user_ids) protocol_dict["collectorUserName"] = user_name_map.get( str(protocol_dict.get("collectorUserId") or ""), "" ) if ci_dict: ci_dict["salesUserName"] = user_name_map.get(str(ci_dict.get("salesUserId") or ""), "") region_id = str(ci_dict.get("regionId") or "") if region_id: r = ( db.query(BizCollectionRegion.region_name) .filter(BizCollectionRegion.id == region_id) .first() ) ci_dict["regionName"] = r.region_name if r else "" hospital_id = str(ci_dict.get("hospitalId") or "") if hospital_id: h = ( db.query(BizCollectionHospital.hospital_name) .filter(BizCollectionHospital.id == hospital_id) .first() ) ci_dict["hospitalName"] = h.hospital_name if h else "" out: Dict[str, Any] = {"protocol": protocol_dict} out["payment"] = _payment_to_dict(pay) if pay else None out["partyCustomer"] = _party_customer_to_dict(pc) if pc else None out["collectionInfo"] = ci_dict transport_rows = list_sample_transports(db, protocol_id, tenant_no) transport_by_sample = {t.sample_id: _transport_to_dict(t) for t in transport_rows} from app.crud import sample_reception as sample_reception_svc sample_ids = [s.id for s in s_rows] reception_by_sample = sample_reception_svc.get_receptions_map_for_samples( db, sample_ids, tenant_no ) from app.crud import qc_work_order as qc_work_order_svc qc_by_sample = qc_work_order_svc.get_qc_info_map_for_samples( db, sample_ids, tenant_no ) samples_out: List[Dict[str, Any]] = [] for s in s_rows: item = _sample_to_dict(s) item["transport"] = transport_by_sample.get(s.id) item["sampleReception"] = reception_by_sample.get(s.id) item["qcInfo"] = qc_by_sample.get(s.id) samples_out.append(item) out["samples"] = samples_out out["storageCert"] = _storage_cert_to_dict(sc) if sc else None out["attachments"] = urls_to_attachment_list(p.protocol_attachments, db, tenant_no) if pay: detail_rows = list_payment_details(db, pay.id, tenant_no) details_out: List[Dict[str, Any]] = [] for d in detail_rows: item = _detail_to_dict( d, protocol_no=p.protocol_no or "", collection_protocol_id=protocol_id, ) item["attachments"] = urls_to_attachment_list(d.transfer_attachments, db, tenant_no) item["financeConfirmAttachmentFiles"] = urls_to_attachment_list( d.finance_confirm_attachments, db, tenant_no ) item["refundFinanceConfirmAttachmentFiles"] = urls_to_attachment_list( d.refund_finance_confirm_attachments, db, tenant_no ) item["refundAttachmentFiles"] = urls_to_attachment_list( d.refund_attachments, db, tenant_no ) details_out.append(item) out["paymentDetails"] = details_out else: out["paymentDetails"] = [] return out def list_audits( db: Session, *, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, audit_status: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: q_base = ( db.query(BizProtocolAudit, BizCollectionProtocol.protocol_no) .join( BizCollectionProtocol, and_( BizProtocolAudit.collection_protocol_id == BizCollectionProtocol.id, BizCollectionProtocol.del_flag == 0, ), ) .filter(BizProtocolAudit.del_flag == 0) ) if tenant_no: q_base = q_base.filter(BizProtocolAudit.tenant_no == tenant_no) if audit_status: q_base = q_base.filter(BizProtocolAudit.audit_status == audit_status.strip()) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q_base = q_base.filter( or_(BizCollectionProtocol.protocol_no.like(like), BizProtocolAudit.audit_remark.like(like)) ) total = q_base.order_by(None).with_entities(func.count(BizProtocolAudit.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q_base.order_by(BizProtocolAudit.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: Dict[str, str] = {} if auditor_ids: for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(auditor_ids)).all(): auditor_name_map[u.id] = u.nick_name or "" out = [ _audit_to_dict( a, protocol_no=pn or "", auditor_user_name=auditor_name_map.get(a.auditor_user_id or "", ""), ) for a, pn in rows ] return out, int(total) def create_audit( db: Session, *, collection_protocol_id: str, audit_status: str, audit_remark: Optional[str], tenant_no: str, user_id: Optional[str], ) -> BizProtocolAudit: p = get_collection_protocol(db, collection_protocol_id, tenant_no) if not p: raise ValueError("协议单不存在") st = (audit_status or "").strip() if st not in ("shtg", "shth"): raise ValueError("审核状态无效") obj = BizProtocolAudit( id=generate_uuid(), collection_protocol_id=collection_protocol_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=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def _detail_query_with_protocol( db: Session, tenant_no: str, pay_status_filter: str, ): return ( db.query(BizProtocolPaymentDetail, BizProtocolPayment, BizCollectionProtocol) .join(BizProtocolPayment, BizProtocolPaymentDetail.payment_id == BizProtocolPayment.id) .join(BizCollectionProtocol, BizProtocolPayment.collection_protocol_id == BizCollectionProtocol.id) .filter( BizProtocolPaymentDetail.del_flag == 0, BizProtocolPayment.del_flag == 0, BizCollectionProtocol.del_flag == 0, BizProtocolPaymentDetail.tenant_no == tenant_no, BizProtocolPaymentDetail.pay_status == pay_status_filter, ) ) def _rows_to_payment_detail_dicts( db: Session, rows: List[Any], tenant_no: str, *, post_process: Optional[Any] = None, ) -> List[Dict[str, Any]]: """缴费明细行 → 前端字典（含协议、客户、转账附件）。""" protocol_ids = [ pay.collection_protocol_id for _, pay, _ in rows if pay and pay.collection_protocol_id ] pc_map = _party_customer_map_by_protocol_ids(db, protocol_ids) out: List[Dict[str, Any]] = [] for d, pay, prot in rows: cpid = pay.collection_protocol_id if pay else "" pc = pc_map.get(cpid) item = _detail_to_dict( d, protocol_no=(prot.protocol_no if prot else "") or "", collection_protocol_id=cpid, protocol_status=(prot.protocol_status if prot else "") or "", pay_mode=(pay.pay_mode if pay else "") or "", mother_name=(pc.mother_name if pc else "") or "", mother_id_card=(pc.mother_id_card if pc else "") or "", newborn_name=(pc.newborn_name if pc else "") or "", newborn_id_card=(pc.newborn_id_card if pc else "") or "", ) item["attachments"] = urls_to_attachment_list(d.transfer_attachments, db, tenant_no) if post_process: post_process(item, d) out.append(item) return out def list_receipt_confirm_details( db: Session, *, page: int = 1, page_size: int = 10, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: if not tenant_no: return [], 0 q_base = _detail_query_with_protocol(db, tenant_no, "jfqrz") total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q_base.order_by(BizProtocolPaymentDetail.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return _rows_to_payment_detail_dicts(db, rows, tenant_no), int(total) def list_receipt_confirmed_details( db: Session, *, page: int = 1, page_size: int = 10, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: """财务已确认收款（yjf），按确认时间倒序。""" if not tenant_no: return [], 0 q_base = _detail_query_with_protocol(db, tenant_no, "yjf") total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q_base.order_by(BizProtocolPaymentDetail.update_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) def _add_finance_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None: item["financeConfirmAttachmentFiles"] = urls_to_attachment_list( d.finance_confirm_attachments, db, tenant_no ) return ( _rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_finance_files), int(total), ) def confirm_receipt( db: Session, *, detail_id: str, tenant_no: str, user_id: Optional[str], finance_confirm_attachments: Optional[str] = None, finance_remark: Optional[str] = None, ) -> BizProtocolPaymentDetail: d = ( db.query(BizProtocolPaymentDetail) .filter( and_( BizProtocolPaymentDetail.id == detail_id, BizProtocolPaymentDetail.tenant_no == tenant_no, BizProtocolPaymentDetail.del_flag == 0, ) ) .first() ) if not d: raise ValueError("缴费明细不存在") if d.pay_status != "jfqrz": if d.pay_status == "yjf": raise ValueError("该明细已确认收款") raise ValueError("当前状态不可确认收款") d.pay_status = "yjf" if finance_confirm_attachments is not None: d.finance_confirm_attachments = finance_confirm_attachments.strip() or None if finance_remark is not None: d.finance_remark = finance_remark.strip() or None d.update_by = user_id d.update_time = datetime.now() db.add(d) db.commit() db.refresh(d) return d def list_refund_confirm_details( db: Session, *, page: int = 1, page_size: int = 10, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: if not tenant_no: return [], 0 q_base = _detail_query_with_protocol(db, tenant_no, "tfqrz") total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q_base.order_by(BizProtocolPaymentDetail.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) def _add_refund_apply_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None: item["refundAttachmentFiles"] = urls_to_attachment_list( d.refund_attachments, db, tenant_no ) return ( _rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_refund_apply_files), int(total), ) def list_refund_confirmed_details( db: Session, *, page: int = 1, page_size: int = 10, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: """财务已确认退费（ytf），按确认时间倒序。""" if not tenant_no: return [], 0 q_base = _detail_query_with_protocol(db, tenant_no, "ytf") total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q_base.order_by(BizProtocolPaymentDetail.update_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) def _add_refund_confirmed_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None: item["refundAttachmentFiles"] = urls_to_attachment_list( d.refund_attachments, db, tenant_no ) item["refundFinanceConfirmAttachmentFiles"] = urls_to_attachment_list( d.refund_finance_confirm_attachments, db, tenant_no ) return ( _rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_refund_confirmed_files), int(total), ) def confirm_refund( db: Session, *, detail_id: str, tenant_no: str, user_id: Optional[str], refund_finance_confirm_attachments: Optional[str] = None, refund_finance_remark: Optional[str] = None, ) -> BizProtocolPaymentDetail: d = ( db.query(BizProtocolPaymentDetail) .filter( and_( BizProtocolPaymentDetail.id == detail_id, BizProtocolPaymentDetail.tenant_no == tenant_no, BizProtocolPaymentDetail.del_flag == 0, ) ) .first() ) if not d: raise ValueError("缴费明细不存在") if d.pay_status != "tfqrz": if d.pay_status == "ytf": raise ValueError("该明细已确认退费") raise ValueError("当前状态不可确认退费") d.pay_status = "ytf" if refund_finance_confirm_attachments is not None: d.refund_finance_confirm_attachments = refund_finance_confirm_attachments.strip() or None if refund_finance_remark is not None: d.refund_finance_remark = refund_finance_remark.strip() or None d.update_by = user_id d.update_time = datetime.now() db.add(d) db.commit() db.refresh(d) return d def _get_payment_detail_or_raise( db: Session, *, detail_id: str, tenant_no: str, ) -> BizProtocolPaymentDetail: d = ( db.query(BizProtocolPaymentDetail) .filter( and_( BizProtocolPaymentDetail.id == detail_id, BizProtocolPaymentDetail.tenant_no == tenant_no, BizProtocolPaymentDetail.del_flag == 0, ) ) .first() ) if not d: raise ValueError("缴费明细不存在") return d def _apply_submit_receipt_finance( d: BizProtocolPaymentDetail, *, user_id: Optional[str], ) -> None: if d.pay_status != "dfj": raise ValueError("仅待缴费状态可提交财务确认") d.pay_status = "jfqrz" d.update_by = user_id d.update_time = datetime.now() def submit_receipt_finance( db: Session, *, detail_id: str, tenant_no: str, user_id: Optional[str], ) -> BizProtocolPaymentDetail: d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no) _apply_submit_receipt_finance(d, user_id=user_id) db.add(d) db.commit() db.refresh(d) return d def batch_submit_receipt_finance( db: Session, *, detail_ids: List[str], tenant_no: str, user_id: Optional[str], ) -> int: ids = [str(i).strip() for i in (detail_ids or []) if str(i).strip()] if not ids: raise ValueError("请选择缴费明细") for detail_id in ids: d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no) _apply_submit_receipt_finance(d, user_id=user_id) db.add(d) db.commit() return len(ids) def _apply_submit_refund_finance( d: BizProtocolPaymentDetail, *, user_id: Optional[str], refund_amount: Any, refund_remark: Optional[str] = None, refund_attachments: Optional[str] = None, ) -> None: if d.pay_status != "yjf": raise ValueError("仅已缴费状态可提交退费财务确认") if refund_amount is None or Decimal(str(refund_amount)) <= 0: raise ValueError("请填写有效的退费金额") d.pay_status = "tfqrz" d.refund_amount = Decimal(str(refund_amount)) d.refund_remark = (refund_remark or "").strip() or None d.refund_attachments = (refund_attachments or "").strip() or None d.update_by = user_id d.update_time = datetime.now() def submit_refund_finance( db: Session, *, detail_id: str, tenant_no: str, user_id: Optional[str], refund_amount: Any, refund_remark: Optional[str] = None, refund_attachments: Optional[str] = None, ) -> BizProtocolPaymentDetail: d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no) _apply_submit_refund_finance( d, user_id=user_id, refund_amount=refund_amount, refund_remark=refund_remark, refund_attachments=refund_attachments, ) db.add(d) db.commit() db.refresh(d) return d def batch_submit_refund_finance( db: Session, *, detail_ids: List[str], tenant_no: str, user_id: Optional[str], refund_amount: Any, refund_remark: Optional[str] = None, refund_attachments: Optional[str] = None, ) -> int: ids = [str(i).strip() for i in (detail_ids or []) if str(i).strip()] if not ids: raise ValueError("请选择缴费明细") for detail_id in ids: d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no) _apply_submit_refund_finance( d, user_id=user_id, refund_amount=refund_amount, refund_remark=refund_remark, refund_attachments=refund_attachments, ) db.add(d) db.commit() return len(ids) def _party_customer_map_by_protocol_ids( db: Session, protocol_ids: List[str], ) -> Dict[str, BizProtocolPartyCustomer]: if not protocol_ids: return {} rows = ( db.query(BizProtocolPartyCustomer) .filter( BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids), BizProtocolPartyCustomer.del_flag == 0, ) .all() ) return {r.collection_protocol_id: r for r in rows} def list_payment_details_page( db: Session, *, page: int = 1, page_size: int = 10, pay_status: Optional[str] = None, keyword: Optional[str] = None, transfer_type: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: if not tenant_no: return [], 0 q = ( db.query(BizProtocolPaymentDetail, BizProtocolPayment, BizCollectionProtocol) .join(BizProtocolPayment, BizProtocolPaymentDetail.payment_id == BizProtocolPayment.id) .join(BizCollectionProtocol, BizProtocolPayment.collection_protocol_id == BizCollectionProtocol.id) .filter( BizProtocolPaymentDetail.del_flag == 0, BizProtocolPayment.del_flag == 0, BizCollectionProtocol.del_flag == 0, BizProtocolPaymentDetail.tenant_no == tenant_no, ) ) if pay_status: q = q.filter(BizProtocolPaymentDetail.pay_status == pay_status) if keyword and keyword.strip(): like = f"%{keyword.strip()}%" q = q.filter(BizCollectionProtocol.protocol_no.like(like)) if transfer_type and transfer_type.strip(): q = q.filter(BizProtocolPaymentDetail.transfer_type == transfer_type.strip()) total = q.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( q.order_by(BizProtocolPaymentDetail.installment_due_date.asc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) protocol_ids = [ pay.collection_protocol_id for _, pay, _ in rows if pay and pay.collection_protocol_id ] pc_map = _party_customer_map_by_protocol_ids(db, protocol_ids) out = [] for d, pay, prot in rows: cpid = pay.collection_protocol_id if pay else "" pc = pc_map.get(cpid) out.append( _detail_to_dict( d, protocol_no=(prot.protocol_no if prot else "") or "", collection_protocol_id=cpid, protocol_status=(prot.protocol_status if prot else "") or "", pay_mode=(pay.pay_mode if pay else "") or "", mother_name=(pc.mother_name if pc else "") or "", mother_id_card=(pc.mother_id_card if pc else "") or "", newborn_name=(pc.newborn_name if pc else "") or "", newborn_id_card=(pc.newborn_id_card if pc else "") or "", ) ) return out, int(total) def _classify_installment_warning(due: str, today: str) -> str: """overdue | warn | upcoming — due 为 YYYYMMDD""" if not due or len(due) != 8 or not due.isdigit(): return "unknown" if due < today: return "overdue" last_warn = (datetime.strptime(today, "%Y%m%d") + timedelta(days=15)).strftime("%Y%m%d") if today <= due <= last_warn: return "warn" return "upcoming" def list_installment_warning( db: Session, *, page: int = 1, page_size: int = 10, warning_type: str = "all", tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: if not tenant_no: return [], 0 today = _today_yyyymmdd() q0 = _detail_query_with_protocol(db, tenant_no, "dfj").filter( BizProtocolPaymentDetail.installment_no.isnot(None), ) # 拉取当前租户下候选（分期号非空）；日期过滤在内存中完成以保证与 classify 一致 rows_all = q0.order_by(BizProtocolPaymentDetail.installment_due_date.asc()).all() flattened: List[Dict[str, Any]] = [] for d, pay, prot in rows_all: due_raw = (d.installment_due_date or "").strip() if d.installment_due_date else "" wt = _classify_installment_warning(due_raw, today) cpid = pay.collection_protocol_id if pay else "" item = _detail_to_dict( d, protocol_no=(prot.protocol_no if prot else "") or "", collection_protocol_id=cpid, protocol_status=(prot.protocol_status if prot else "") or "", pay_mode=(pay.pay_mode if pay else "") or "", ) item["warningType"] = wt wt_filter = (warning_type or "all").strip().lower() if wt_filter == "overdue" and wt != "overdue": continue if wt_filter == "warn" and wt != "warn": continue if wt_filter == "all": pass elif wt_filter not in ("all", "warn", "overdue"): pass flattened.append(item) total = len(flattened) page = max(1, page) page_size = min(max(1, page_size), 100) start = (page - 1) * page_size page_rows = flattened[start : start + page_size] return page_rows, total def create_collection_protocol( db: Session, *, protocol_no_id: Optional[str] = None, protocol_no: str = "", protocol_status: str = "dsx", protocol_attachments: Optional[str] = None, register_date: Optional[str] = None, collector_user_id: Optional[str] = None, remark: Optional[str] = None, tenant_no: str = "zkxw", user_id: Optional[str] = None, ) -> BizCollectionProtocol: pn = (protocol_no or "").strip() assert_biz_no_available(db, tenant_no=tenant_no, biz_no=pn) obj = BizCollectionProtocol( id=generate_uuid(), protocol_no_id=(protocol_no_id or "").strip() or None, protocol_no=pn, protocol_status=protocol_status.strip() or "dsx", protocol_attachments=(protocol_attachments or "").strip() or None, register_date=(register_date or "").strip() or None, collector_user_id=(collector_user_id or "").strip() or None, tenant_no=tenant_no, order_num=0, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_collection_protocol( db: Session, protocol_id: str, *, tenant_no: str, user_id: Optional[str] = None, scope_collector_user_id: Optional[str] = None, protocol_no: Optional[str] = None, protocol_status: Optional[str] = None, protocol_attachments: Optional[str] = None, register_date: Optional[str] = None, effective_time: Optional[datetime] = None, effective_desc: Optional[str] = None, expire_time: Optional[datetime] = None, expire_desc: Optional[str] = None, void_time: Optional[datetime] = None, void_desc: Optional[str] = None, to_research_time: Optional[datetime] = None, to_research_desc: Optional[str] = None, collector_user_id: Optional[str] = None, remark: Optional[str] = None, ) -> BizCollectionProtocol: obj = get_collection_protocol( db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id ) if not obj: raise ValueError("协议单不存在或无权操作") if protocol_no is not None: pn = protocol_no.strip() if not pn: raise ValueError("协议编号不能为空") if pn != obj.protocol_no: assert_biz_no_available( db, tenant_no=tenant_no, biz_no=pn, exclude_protocol_id=protocol_id, ) obj.protocol_no = pn if protocol_status is not None: obj.protocol_status = protocol_status.strip() if protocol_attachments is not None: obj.protocol_attachments = protocol_attachments.strip() or None if register_date is not None: obj.register_date = register_date.strip() or None if effective_time is not None: obj.effective_time = effective_time if effective_desc is not None: obj.effective_desc = effective_desc.strip() or None if expire_time is not None: obj.expire_time = expire_time if expire_desc is not None: obj.expire_desc = expire_desc.strip() or None if void_time is not None: obj.void_time = void_time if void_desc is not None: obj.void_desc = void_desc.strip() or None if to_research_time is not None: obj.to_research_time = to_research_time if to_research_desc is not None: obj.to_research_desc = to_research_desc.strip() or None if collector_user_id is not None: obj.collector_user_id = collector_user_id.strip() or None if remark is not None: obj.remark = remark.strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_collection_protocol( db: Session, protocol_id: str, tenant_no: str, user_id: Optional[str] = None, scope_collector_user_id: Optional[str] = None, ) -> None: obj = get_collection_protocol( db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id ) if not obj: raise ValueError("协议单不存在或无权操作") obj.del_flag = 1 obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() # --------------------------------------------------------------------------- # 子表：当事人客户 / 采集信息 / 样本 / 储存证书 # --------------------------------------------------------------------------- def get_party_customer(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolPartyCustomer]: return ( db.query(BizProtocolPartyCustomer) .filter( and_( BizProtocolPartyCustomer.collection_protocol_id == collection_protocol_id, BizProtocolPartyCustomer.tenant_no == tenant_no, BizProtocolPartyCustomer.del_flag == 0, ) ) .first() ) def upsert_party_customer( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], **fields, ) -> BizProtocolPartyCustomer: fields = _normalize_fields(fields) obj = get_party_customer(db, collection_protocol_id, tenant_no) _STR_FIELDS = [ "customer_id", "mother_name", "mother_phone", "mother_id_card", "mother_birth_ym", "mother_ethnic", "mother_address", "father_name", "father_phone", "father_id_card", "newborn_name", "newborn_id_card", "newborn_sex", "delivery_mode", "collector_user_id", ] _INT_FIELDS = ["birth_order", "newborn_weight_gram"] _DEC_FIELDS = ["apgar_1min", "apgar_5min"] _DT_FIELDS = ["newborn_delivery_time"] if obj: for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) for k in _INT_FIELDS: if k in fields: setattr(obj, k, fields[k]) for k in _DEC_FIELDS: if k in fields: setattr(obj, k, _optional_decimal(fields[k])) for k in _DT_FIELDS: if k in fields: setattr(obj, k, fields[k]) if "remark" in fields: obj.remark = (fields["remark"] or "").strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj obj = BizProtocolPartyCustomer( id=generate_uuid(), collection_protocol_id=collection_protocol_id, tenant_no=tenant_no, order_num=0, remark=(fields.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) for k in _INT_FIELDS: if k in fields: setattr(obj, k, fields[k]) for k in _DEC_FIELDS: if k in fields: setattr(obj, k, _optional_decimal(fields[k])) for k in _DT_FIELDS: if k in fields: setattr(obj, k, fields[k]) db.add(obj) db.commit() db.refresh(obj) return obj def get_collection_info(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolCollectionInfo]: return ( db.query(BizProtocolCollectionInfo) .filter( and_( BizProtocolCollectionInfo.collection_protocol_id == collection_protocol_id, BizProtocolCollectionInfo.tenant_no == tenant_no, BizProtocolCollectionInfo.del_flag == 0, ) ) .first() ) def upsert_collection_info( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], **fields, ) -> BizProtocolCollectionInfo: fields = _normalize_fields(fields) obj = get_collection_info(db, collection_protocol_id, tenant_no) _STR_FIELDS = ["region_id", "hospital_id", "medical_staff", "sales_user_id"] _DT_FIELDS = ["sample_collect_time", "maternal_blood_collect_time"] if obj: for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) for k in _DT_FIELDS: if k in fields: setattr(obj, k, fields[k]) if "remark" in fields: obj.remark = (fields["remark"] or "").strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj obj = BizProtocolCollectionInfo( id=generate_uuid(), collection_protocol_id=collection_protocol_id, tenant_no=tenant_no, order_num=0, remark=(fields.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) for k in _DT_FIELDS: if k in fields: setattr(obj, k, fields[k]) db.add(obj) db.commit() db.refresh(obj) return obj def list_samples(db: Session, collection_protocol_id: str, tenant_no: str) -> List[BizProtocolSample]: return ( db.query(BizProtocolSample) .filter( and_( BizProtocolSample.collection_protocol_id == collection_protocol_id, BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, ) ) .order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc()) .all() ) def save_samples( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], samples: list, ) -> List[BizProtocolSample]: existing = list_samples(db, collection_protocol_id, tenant_no) for s in existing: s.del_flag = 1 s.update_by = user_id s.update_time = datetime.now() db.add(s) if existing: db.flush() validate_sample_codes_not_used( db, tenant_no=tenant_no, samples=samples, exclude_protocol_id=collection_protocol_id, ) _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", ] _DEC_FIELDS = ["collect_amount", "price"] _INT_FIELDS = ["storage_years"] result = [] for idx, raw in enumerate(samples): raw = _normalize_fields(raw) obj = BizProtocolSample( id=generate_uuid(), collection_protocol_id=collection_protocol_id, tenant_no=tenant_no, order_num=idx, remark=(raw.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) for k in _STR_FIELDS: if k == "sample_status": continue if k in raw: setattr(obj, k, (raw[k] or "").strip() or None) # 样本状态：显式支持 null/空字符串，避免 ORM/库默认值 ycj 回填 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 k in _DEC_FIELDS: if k in raw and raw[k] is not None: setattr(obj, k, Decimal(str(raw[k]))) for k in _INT_FIELDS: if k in raw: setattr(obj, k, raw[k]) db.add(obj) result.append(obj) db.commit() for o in result: db.refresh(o) save_sample_transports(db, collection_protocol_id, tenant_no, user_id, result, samples) return result def list_sample_transports( db: Session, collection_protocol_id: str, tenant_no: str, ) -> List[BizProtocolSampleTransport]: return ( db.query(BizProtocolSampleTransport) .filter( and_( BizProtocolSampleTransport.collection_protocol_id == collection_protocol_id, BizProtocolSampleTransport.tenant_no == tenant_no, BizProtocolSampleTransport.del_flag == 0, ) ) .order_by(BizProtocolSampleTransport.order_num.asc(), BizProtocolSampleTransport.create_time.asc()) .all() ) def _soft_delete_transports_for_protocol( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], ) -> None: rows = list_sample_transports(db, collection_protocol_id, tenant_no) 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 save_sample_transports( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], saved_samples: List[BizProtocolSample], input_samples: list, ) -> None: _soft_delete_transports_for_protocol(db, collection_protocol_id, tenant_no, user_id) for idx, sample_obj in enumerate(saved_samples): if idx >= len(input_samples): break raw = _normalize_fields(input_samples[idx] if isinstance(input_samples[idx], dict) else {}) transport_raw = raw.get("transport") if not isinstance(transport_raw, dict): continue tnorm = _normalize_fields(transport_raw) transport_type = (tnorm.get("transport_type") or "").strip() if not transport_type: continue eta_time = tnorm.get("eta_time") if isinstance(eta_time, str): eta_time = parse_datetime(eta_time) 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 obj = BizProtocolSampleTransport( id=generate_uuid(), collection_protocol_id=collection_protocol_id, sample_id=sample_obj.id, sample_code=sample_obj.sample_code or None, transport_type=transport_type, express_company_id=(tnorm.get("express_company_id") or "").strip() or None, waybill_no=(tnorm.get("waybill_no") or "").strip() or None, dest_site_id=(tnorm.get("dest_site_id") or "").strip() or None, express_fee=express_fee, eta_time=eta_time, carrier_name=(tnorm.get("carrier_name") or "").strip() or None, carrier_phone=(tnorm.get("carrier_phone") or "").strip() or None, tenant_no=tenant_no, order_num=idx, remark=(tnorm.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() def get_storage_cert(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolStorageCert]: return ( db.query(BizProtocolStorageCert) .filter( and_( BizProtocolStorageCert.collection_protocol_id == collection_protocol_id, BizProtocolStorageCert.tenant_no == tenant_no, BizProtocolStorageCert.del_flag == 0, ) ) .first() ) def _add_months_yyyymmdd(date_str: str, months: int) -> str: raw = (date_str or "").strip().replace("-", "") if len(raw) != 8: return "" y = int(raw[0:4]) m = int(raw[4:6]) d = int(raw[6:8]) month_index = m - 1 + months ny = y + month_index // 12 nm = month_index % 12 + 1 last_day = calendar.monthrange(ny, nm)[1] nd = min(d, last_day) return f"{ny:04d}{nm:02d}{nd:02d}" def _calc_installment_due_amounts( contract_amount: Optional[Decimal], count: int, ) -> List[Decimal]: if contract_amount is None or count < 1: return [] total_cents = int(round(float(contract_amount) * 100)) base_cents = total_cents // count amounts: List[Decimal] = [] sum_cents = 0 for i in range(count): if i == count - 1: cents = total_cents - sum_cents else: cents = base_cents sum_cents += base_cents amounts.append(Decimal(cents) / Decimal(100)) return amounts def align_fq_payment_details_payload( pay: BizProtocolPayment, details: Optional[list], ) -> list: if pay.pay_mode != "fq" or not pay.installment_count or int(pay.installment_count) < 1: return list(details or []) count = int(pay.installment_count) rows = [_normalize_fields(dict(d)) for d in (details or [])] if len(rows) >= count: needed = set(range(1, count + 1)) have = { int(r["installment_no"]) for r in rows if r.get("installment_no") is not None } if needed - have == set(): return rows[:count] by_no: Dict[int, Dict[str, Any]] = {} unnumbered: List[Dict[str, Any]] = [] for r in rows: no = r.get("installment_no") if no is not None: by_no[int(no)] = r else: unnumbered.append(r) amounts = _calc_installment_due_amounts(pay.contract_amount, count) start = (pay.installment_start_date or "").strip() interval = int(pay.installment_interval_months or 0) result: List[Dict[str, Any]] = [] for i in range(count): no = i + 1 if no in by_no: result.append(by_no[no]) continue if i < len(unnumbered): raw = dict(unnumbered[i]) raw["installment_no"] = no if not raw.get("installment_due_date") and start and interval: raw["installment_due_date"] = _add_months_yyyymmdd(start, i * interval) result.append(raw) continue due_date = _add_months_yyyymmdd(start, i * interval) if start and interval else "" due = amounts[i] if i < len(amounts) else None result.append( { "installment_no": no, "installment_due_date": due_date or None, "due_amount": float(due) if due is not None else None, "pay_status": "dfj", "transfer_type": "", "transfer_amount": None, "attachments": [], } ) return result def list_payment_details( db: Session, payment_id: str, tenant_no: str, ) -> List[BizProtocolPaymentDetail]: return ( db.query(BizProtocolPaymentDetail) .filter( and_( BizProtocolPaymentDetail.payment_id == payment_id, BizProtocolPaymentDetail.tenant_no == tenant_no, BizProtocolPaymentDetail.del_flag == 0, ) ) .order_by( BizProtocolPaymentDetail.installment_no.asc(), BizProtocolPaymentDetail.order_num.asc(), BizProtocolPaymentDetail.create_time.asc(), ) .all() ) def upsert_payment( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], **fields, ) -> BizProtocolPayment: fields = _normalize_fields(fields) obj = get_payment_by_protocol_id(db, collection_protocol_id, tenant_no) _STR_FIELDS = ["pay_mode"] _STR_DATE_FIELDS = ["lump_sum_deadline", "installment_start_date"] _DEC_FIELDS = ["contract_amount"] _INT_FIELDS = ["installment_count", "installment_interval_months"] def _apply(target: BizProtocolPayment) -> None: for k in _STR_FIELDS: if k in fields: setattr(target, k, (fields[k] or "").strip() or None) for k in _STR_DATE_FIELDS: if k in fields: setattr(target, k, _normalize_yyyymmdd_date(fields[k])) for k in _DEC_FIELDS: if k in fields and fields[k] is not None: setattr(target, k, Decimal(str(fields[k]))) for k in _INT_FIELDS: if k in fields: setattr(target, k, fields[k]) if "remark" in fields: target.remark = (fields["remark"] or "").strip() or None if obj: _apply(obj) obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj pay_mode = (fields.get("pay_mode") or "").strip() or "dj" obj = BizProtocolPayment( id=generate_uuid(), collection_protocol_id=collection_protocol_id, pay_mode=pay_mode, tenant_no=tenant_no, order_num=0, remark=(fields.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) _apply(obj) db.add(obj) db.commit() db.refresh(obj) return obj def save_payment_details( db: Session, payment_id: str, tenant_no: str, user_id: Optional[str], details: list, ) -> List[BizProtocolPaymentDetail]: existing = list_payment_details(db, payment_id, tenant_no) for d in existing: d.del_flag = 1 d.update_by = user_id d.update_time = datetime.now() db.add(d) if existing: db.flush() _STR_FIELDS = [ "transfer_type", "pay_status", "finance_remark", "refund_finance_remark", "refund_remark", ] _STR_DATE_FIELDS = ["installment_due_date"] _DEC_FIELDS = ["due_amount", "transfer_amount", "refund_amount"] _INT_FIELDS = ["installment_no"] result: List[BizProtocolPaymentDetail] = [] for idx, raw in enumerate(details or []): raw = _normalize_fields(dict(raw)) transfer_att = _pop_attachment_urls( raw, "attachments", "transfer_attachments", "transferAttachments" ) finance_confirm_att = _pop_attachment_urls( raw, "finance_confirm_attachments", "financeConfirmAttachments", "finance_confirm_attachment_files", "financeConfirmAttachmentFiles", ) refund_att = _pop_attachment_urls( raw, "refund_attachments", "refundAttachments", "refund_attachment_files", "refundAttachmentFiles", ) refund_finance_att = _pop_attachment_urls( raw, "refund_finance_confirm_attachments", "refundFinanceConfirmAttachments", "refund_finance_confirm_attachment_files", "refundFinanceConfirmAttachmentFiles", ) transfer_time = raw.get("transfer_time") if isinstance(transfer_time, str): transfer_time = parse_datetime(transfer_time) installment_no = raw.get("installment_no") if installment_no is None and raw.get("installment_due_date"): installment_no = idx + 1 obj = BizProtocolPaymentDetail( id=generate_uuid(), payment_id=payment_id, tenant_no=tenant_no, order_num=idx, remark=(raw.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, transfer_attachments=transfer_att, finance_confirm_attachments=finance_confirm_att, refund_attachments=refund_att, refund_finance_confirm_attachments=refund_finance_att, transfer_time=transfer_time, installment_no=installment_no, ) for k in _STR_FIELDS: if k in raw: setattr(obj, k, (raw[k] or "").strip() or None) for k in _STR_DATE_FIELDS: if k in raw: setattr(obj, k, _normalize_yyyymmdd_date(raw[k])) for k in _DEC_FIELDS: if k in raw and raw[k] is not None: setattr(obj, k, Decimal(str(raw[k]))) for k in _INT_FIELDS: if k in raw and raw[k] is not None: setattr(obj, k, raw[k]) if "pay_status" not in raw or not (raw.get("pay_status") or "").strip(): obj.pay_status = "dfj" db.add(obj) result.append(obj) db.commit() for o in result: db.refresh(o) return result def upsert_storage_cert( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], **fields, ) -> BizProtocolStorageCert: fields = _normalize_fields(fields) obj = get_storage_cert(db, collection_protocol_id, tenant_no) _STR_FIELDS = ["make_status", "cert_image_urls", "make_date"] if obj: for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) if "remark" in fields: obj.remark = (fields["remark"] or "").strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj obj = BizProtocolStorageCert( id=generate_uuid(), collection_protocol_id=collection_protocol_id, tenant_no=tenant_no, order_num=0, remark=(fields.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) db.add(obj) db.commit() db.refresh(obj) return obj # --------------------------------------------------------------------------- # 子表：满月待办 # --------------------------------------------------------------------------- def _fullmoon_todo_to_dict(ft: BizProtocolFullmoonTodo) -> Dict[str, Any]: return { "id": ft.id, "collectionProtocolId": ft.collection_protocol_id, "todoStatus": ft.todo_status or "0", "handleDate": (ft.handle_date or "").strip() if ft.handle_date else "", "handleRemark": ft.handle_remark or "", "insuranceAttachments": ft.insurance_attachments or "", "insuranceCompanyId": ft.insurance_company_id or "", "insuranceCompanyName": ft.insurance_company_name or "", "insuranceType": ft.insurance_type or "", "insurancePolicyNo": ft.insurance_policy_no or "", "insuranceEffectiveDate": (ft.insurance_effective_date or "").strip() if ft.insurance_effective_date else "", "insuranceExpiryDate": (ft.insurance_expiry_date or "").strip() if ft.insurance_expiry_date else "", "insuranceSubscribed": ft.insurance_subscribed or "0", "insuranceFee": _dec(ft.insurance_fee), "tenantNo": ft.tenant_no, "orderNum": ft.order_num if ft.order_num is not None else 0, "remark": ft.remark or "", "createTime": _fmt_dt(ft.create_time), "updateTime": _fmt_dt(ft.update_time), } def get_fullmoon_todo( db: Session, collection_protocol_id: str, tenant_no: str, ) -> Optional[BizProtocolFullmoonTodo]: return ( db.query(BizProtocolFullmoonTodo) .filter( and_( BizProtocolFullmoonTodo.collection_protocol_id == collection_protocol_id, BizProtocolFullmoonTodo.tenant_no == tenant_no, BizProtocolFullmoonTodo.del_flag == 0, ) ) .first() ) def upsert_fullmoon_todo( db: Session, collection_protocol_id: str, tenant_no: str, user_id: Optional[str], **fields, ) -> BizProtocolFullmoonTodo: fields = _normalize_fields(fields) obj = get_fullmoon_todo(db, collection_protocol_id, tenant_no) _STR_FIELDS = [ "todo_status", "handle_date", "handle_remark", "insurance_attachments", "insurance_company_id", "insurance_company_name", "insurance_type", "insurance_policy_no", "insurance_effective_date", "insurance_expiry_date", "insurance_subscribed", ] _DEC_FIELDS = ["insurance_fee"] if obj: for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) for k in _DEC_FIELDS: if k in fields and fields[k] is not None: setattr(obj, k, Decimal(str(fields[k]))) if "remark" in fields: obj.remark = (fields["remark"] or "").strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj obj = BizProtocolFullmoonTodo( id=generate_uuid(), collection_protocol_id=collection_protocol_id, tenant_no=tenant_no, order_num=0, remark=(fields.get("remark") or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) for k in _STR_FIELDS: if k in fields: setattr(obj, k, (fields[k] or "").strip() or None) for k in _DEC_FIELDS: if k in fields and fields[k] is not None: setattr(obj, k, Decimal(str(fields[k]))) db.add(obj) db.commit() db.refresh(obj) return obj