"""中试输注计划管理 CRUD（输注计划列表、详情、状态变更、编辑）""" from __future__ import annotations import json from datetime import datetime from typing import Any, Dict, List, Optional from sqlalchemy import and_, desc, func, or_ from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.crud.child_row_sync import sync_child_rows from app.models.models import ( BizChannelAgreementProductPrice, BizChannelServiceOrder, BizChannelServiceOrderProduct, BizChannelServiceOrderTxn, BizPilotApplication, BizPilotInfusionPlan, BizPilotInfusionSchedule, BizPilotInfusionScheduleProduct, BizPilotPatient, BizProductionWorkOrder, BizQcWorkOrder, ) from app.crud.pilot_service import ( _get_or_create_channel_account, _reduce_channel_product_quota, _resolve_agreement_price_cell_type, ) from app.crud import outbound_handover as handover_svc from app.crud import outbound_release_review as orr_svc from app.crud import pilot_infusion_confirm as confirm_svc # ──────────────────────────────────────────── # Status transition map # ──────────────────────────────────────────── VALID_TRANSITIONS: Dict[str, set] = { "dtz": {"yrqr", "yqx"}, "yrqr": {"ywc", "yqx"}, } QC_VALID_TRANSITIONS: Dict[str, set] = { "dzc": {"yzj", "yqx"}, } # ──────────────────────────────────────────── # Helpers: _row_to_dict # ──────────────────────────────────────────── def _row_to_dict_schedule( row, application=None, plan=None, patient=None, products=None, service_order=None, plan_meta: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: course = _resolve_course_display(row, plan, plan_meta) d: Dict[str, Any] = { "id": row.id, "infusionPlanId": row.infusion_plan_id or "", "sortNo": course.get("sortNo"), "totalCourseCount": course.get("totalCourseCount"), "cellType": row.cell_type or "", "specText": row.spec_text or "", "packageId": getattr(row, "package_id", None) or "", "packageText": getattr(row, "package_text", None) or "", "infusionTime": row.infusion_time.strftime("%Y-%m-%d %H:%M:%S") if row.infusion_time else "", "infusionLocation": row.infusion_location or "", "infusionMethod": row.infusion_method or "", "scheduleStatus": row.schedule_status or "", "infusionStatus": row.infusion_status or "", "cancelReason": row.cancel_reason or "", "qcStatus": row.qc_status or "", "attachments": row.attachments or "", "remark": row.remark or "", "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "", "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "", } if application: d["applicationNo"] = application.application_no or "" d["applicationName"] = application.application_name or "" d["applicationStatus"] = application.application_status or "" d["channelName"] = application.channel_name or "" d["channelId"] = application.channel_id or "" d["pilotApplicationId"] = application.id or "" d["channelServiceOrderId"] = application.channel_service_order_id or "" d["projectTitle"] = application.project_title or "" d["customerName"] = application.customer_name or "" d["bizType"] = application.biz_type or "pilot_service" else: d["applicationNo"] = "" d["applicationName"] = "" d["applicationStatus"] = "" d["channelName"] = "" d["channelId"] = "" d["pilotApplicationId"] = "" d["channelServiceOrderId"] = "" d["projectTitle"] = "" d["customerName"] = "" d["bizType"] = "" if plan: d["channelAgreementId"] = plan.channel_agreement_id or "" else: d["channelAgreementId"] = "" if patient: d["patientName"] = patient.patient_name or "" d["patientSex"] = patient.sex or "" d["patientAge"] = patient.age else: d["patientName"] = "" d["patientSex"] = "" d["patientAge"] = None if service_order: d["orderType"] = service_order.order_type or "" d["channelServiceOrderName"] = service_order.order_name or "" else: d["orderType"] = "" d["channelServiceOrderName"] = "" if products is not None: d["products"] = products else: d["products"] = [] return d def _row_to_dict_product(row) -> Dict[str, Any]: return { "id": row.id, "priceId": row.price_id or "", "projectCode": row.project_code or "", "projectName": row.project_name or "", "cellTypeName": row.cell_type_name or "", "specText": row.spec_text or "", "packageId": getattr(row, "package_id", None) or "", "packageText": getattr(row, "package_text", None) or "", "agreementPrice": float(row.agreement_price) if row.agreement_price is not None else None, "actualPrice": float(row.actual_price) if row.actual_price is not None else None, "quantity": row.quantity if row.quantity is not None else 1, "remark": row.remark or "", "productStatus": row.product_status or "", "status": row.product_status or "", } def _get_plan_schedule_course_meta( db: Session, plan_ids: List[str], tenant_no: str ) -> Dict[str, Dict[str, Any]]: """按输注方案汇总：计划条数、各计划内每条 schedule 的疗程序号（1 起）。""" unique_ids = list({pid for pid in plan_ids if pid}) if not unique_ids: return {} rows = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.infusion_plan_id.in_(unique_ids), BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .order_by( BizPilotInfusionSchedule.infusion_plan_id.asc(), BizPilotInfusionSchedule.order_num.asc(), BizPilotInfusionSchedule.create_time.asc(), ) .all() ) meta: Dict[str, Dict[str, Any]] = {} by_plan: Dict[str, List[BizPilotInfusionSchedule]] = {} for row in rows: by_plan.setdefault(row.infusion_plan_id, []).append(row) for plan_id, schedules in by_plan.items(): sort_no_by_id: Dict[str, int] = {} for idx, sch in enumerate(schedules): order_num = sch.order_num if sch.order_num is not None else 0 sort_no_by_id[sch.id] = order_num if order_num > 0 else idx + 1 meta[plan_id] = { "scheduleCount": len(schedules), "sortNoByScheduleId": sort_no_by_id, } return meta def _resolve_course_display( row: BizPilotInfusionSchedule, plan: Optional[BizPilotInfusionPlan], plan_meta: Optional[Dict[str, Any]] = None, ) -> Dict[str, Optional[int]]: sort_no: Optional[int] = None total: Optional[int] = None if plan_meta: sort_no = plan_meta.get("sortNoByScheduleId", {}).get(row.id) plan_total = plan.total_course_count if plan else None if plan_total is not None and plan_total > 0: total = int(plan_total) else: total = int(plan_meta.get("scheduleCount") or 0) or None elif plan and row.infusion_plan_id: order_num = row.order_num if row.order_num is not None else 0 if order_num > 0: sort_no = int(order_num) plan_total = plan.total_course_count if plan_total is not None and plan_total > 0: total = int(plan_total) if sort_no is None or sort_no <= 0: order_num = row.order_num if row.order_num is not None else 0 sort_no = int(order_num) if order_num > 0 else None return {"sortNo": sort_no, "totalCourseCount": total if total and total > 0 else None} def _get_products_for_schedule(db: Session, schedule_id: str, tenant_no: str) -> List[Dict[str, Any]]: rows = ( db.query(BizPilotInfusionScheduleProduct) .filter( BizPilotInfusionScheduleProduct.infusion_schedule_id == schedule_id, BizPilotInfusionScheduleProduct.tenant_no == tenant_no, BizPilotInfusionScheduleProduct.del_flag == 0, ) .order_by( BizPilotInfusionScheduleProduct.order_num.asc(), BizPilotInfusionScheduleProduct.create_time.asc(), ) .all() ) return [_row_to_dict_product(r) for r in rows] # ──────────────────────────────────────────── # List (Paginated, Multi-table JOIN) # ──────────────────────────────────────────── def list_infusion_schedules_mgmt( db: Session, tenant_no: str, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, infusion_status: Optional[str] = None, qc_status: Optional[str] = None, infusion_time_start: Optional[str] = None, infusion_time_end: Optional[str] = None, biz_type: Optional[str] = None, ) -> Dict[str, Any]: sch = BizPilotInfusionSchedule plan = BizPilotInfusionPlan app = BizPilotApplication pat = BizPilotPatient prod = BizPilotInfusionScheduleProduct q = ( db.query(sch) .outerjoin(plan, and_(sch.infusion_plan_id == plan.id, plan.del_flag == 0)) .outerjoin(app, and_(plan.pilot_application_id == app.id, app.del_flag == 0)) .outerjoin(pat, and_(app.id == pat.pilot_application_id, pat.del_flag == 0)) .filter( sch.tenant_no == tenant_no, sch.del_flag == 0, ) ) bt = (biz_type or "").strip() if bt: q = q.filter(app.biz_type == bt) # keyword: application_no, channel_name, patient_name; basic_experiment also project_title/customer_name if keyword: kw = f"%{keyword}%" q = q.filter( or_( app.application_no.like(kw), app.channel_name.like(kw), app.project_title.like(kw), app.customer_name.like(kw), pat.patient_name.like(kw), ) ) if infusion_status: statuses = [s.strip() for s in infusion_status.split(",") if s.strip()] if statuses: q = q.filter(sch.infusion_status.in_(statuses)) if qc_status: qc_statuses = [s.strip() for s in qc_status.split(",") if s.strip()] if qc_statuses: q = q.filter(sch.qc_status.in_(qc_statuses)) time_bounds = [] if infusion_time_start: try: start_dt = datetime.strptime(infusion_time_start[:10], "%Y-%m-%d") time_bounds.append(sch.infusion_time >= start_dt) except ValueError: pass if infusion_time_end: try: end_dt = datetime.strptime(infusion_time_end[:10], "%Y-%m-%d") end_dt = end_dt.replace(hour=23, minute=59, second=59) time_bounds.append(sch.infusion_time <= end_dt) except ValueError: pass # 未填写输注时间的计划（申请单新建时常见）在限定时间范围时仍应可见 if time_bounds: q = q.filter(or_(and_(*time_bounds), sch.infusion_time.is_(None))) total = q.order_by(None).with_entities(func.count(sch.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) # Use distinct to avoid duplicates from JOINs # MySQL 5.7 不支持 NULLS LAST：先排「有输注时间」再按时间倒序，未填时间的排最后 schedule_rows = ( q.order_by(sch.infusion_time.is_(None), desc(sch.infusion_time), desc(sch.create_time)) .distinct() .offset((page - 1) * page_size) .limit(page_size) .all() ) plan_ids = [r.infusion_plan_id for r in schedule_rows if r.infusion_plan_id] plan_course_meta = _get_plan_schedule_course_meta(db, plan_ids, tenant_no) items = [] for row in schedule_rows: # Fetch related data for each schedule plan_row = ( db.query(BizPilotInfusionPlan) .filter( BizPilotInfusionPlan.id == row.infusion_plan_id, BizPilotInfusionPlan.del_flag == 0, ) .first() ) application = None patient = None if plan_row: application = ( db.query(BizPilotApplication) .filter( BizPilotApplication.id == plan_row.pilot_application_id, BizPilotApplication.del_flag == 0, ) .first() ) if application: patient = ( db.query(BizPilotPatient) .filter( BizPilotPatient.pilot_application_id == application.id, BizPilotPatient.del_flag == 0, ) .first() ) service_order = None if application and application.channel_service_order_id: service_order = ( db.query(BizChannelServiceOrder) .filter( BizChannelServiceOrder.id == application.channel_service_order_id, BizChannelServiceOrder.del_flag == 0, ) .first() ) products = _get_products_for_schedule(db, row.id, tenant_no) plan_meta = plan_course_meta.get(row.infusion_plan_id) if row.infusion_plan_id else None items.append( _row_to_dict_schedule( row, application, plan_row, patient, products, service_order, plan_meta=plan_meta, ) ) schedule_ids = [item["id"] for item in items if item.get("id")] if schedule_ids: from app.crud import clinical_payment payment_map = clinical_payment.map_payments_by_schedule_ids( db, schedule_ids=schedule_ids, tenant_no=tenant_no ) for item in items: payment = payment_map.get(item["id"]) if not payment: continue item["payment"] = payment item["amount"] = payment.get("amount") item["payStatus"] = payment.get("payStatus") return { "list": items, "pagination": {"page": page, "pageSize": page_size, "total": total}, } # ──────────────────────────────────────────── # Detail # ──────────────────────────────────────────── def get_infusion_schedule_detail( db: Session, schedule_id: str, tenant_no: str ) -> Optional[Dict[str, Any]]: row = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .first() ) if not row: return None plan_row = ( db.query(BizPilotInfusionPlan) .filter( BizPilotInfusionPlan.id == row.infusion_plan_id, BizPilotInfusionPlan.del_flag == 0, ) .first() ) application = None patient = None if plan_row: application = ( db.query(BizPilotApplication) .filter( BizPilotApplication.id == plan_row.pilot_application_id, BizPilotApplication.del_flag == 0, ) .first() ) if application: patient = ( db.query(BizPilotPatient) .filter( BizPilotPatient.pilot_application_id == application.id, BizPilotPatient.del_flag == 0, ) .first() ) service_order = None if application and application.channel_service_order_id: service_order = ( db.query(BizChannelServiceOrder) .filter( BizChannelServiceOrder.id == application.channel_service_order_id, BizChannelServiceOrder.del_flag == 0, ) .first() ) products = _get_products_for_schedule(db, row.id, tenant_no) plan_meta = None if row.infusion_plan_id: batch_meta = _get_plan_schedule_course_meta(db, [row.infusion_plan_id], tenant_no) plan_meta = batch_meta.get(row.infusion_plan_id) flat = _row_to_dict_schedule( row, application, plan_row, patient, products, service_order, plan_meta=plan_meta ) aggregated = orr_svc.get_schedule_detail(db, schedule_id, tenant_no) or {} app = dict(aggregated.get("application") or {}) app["applicationNo"] = app.get("applicationNo") or flat.get("applicationNo") or "" app["applicationName"] = app.get("applicationName") or flat.get("applicationName") or "" app["applicationStatus"] = app.get("applicationStatus") or flat.get("applicationStatus") or "" app["channelName"] = app.get("channelName") or flat.get("channelName") or "" app["orderType"] = flat.get("orderType") or "" app["channelServiceOrderName"] = flat.get("channelServiceOrderName") or "" app["channelServiceOrderId"] = flat.get("channelServiceOrderId") or "" app["projectTitle"] = app.get("projectTitle") or flat.get("projectTitle") or "" app["customerName"] = app.get("customerName") or flat.get("customerName") or "" if application: app["bizType"] = application.biz_type or "pilot_service" if not app.get("projectTitle"): app["projectTitle"] = application.project_title or "" if not app.get("customerName"): app["customerName"] = application.customer_name or "" plan = dict(aggregated.get("infusionPlan") or {}) if plan_row: plan.setdefault("cellType", plan_row.cell_type or "") plan.setdefault("specText", plan_row.spec_text or "") plan.setdefault("totalCourseCount", plan_row.total_course_count) plan.setdefault("courseInterval", plan_row.course_interval or "") plan.setdefault("infusionStartDate", plan_row.infusion_start_date or "") sched = dict(aggregated.get("infusionSchedule") or {}) sched["sortNo"] = flat.get("sortNo") sched["totalCourseCount"] = flat.get("totalCourseCount") sched["cellType"] = flat.get("cellType") or sched.get("cellType") or "" sched["specText"] = flat.get("specText") or sched.get("specText") or "" sched["packageId"] = flat.get("packageId") or sched.get("packageId") or "" sched["packageText"] = flat.get("packageText") or sched.get("packageText") or "" sched["infusionTime"] = flat.get("infusionTime") or sched.get("infusionTime") or "" sched["infusionLocation"] = flat.get("infusionLocation") or sched.get("infusionLocation") or "" sched["infusionMethod"] = flat.get("infusionMethod") or sched.get("infusionMethod") or "" sched["scheduleStatus"] = flat.get("scheduleStatus") or sched.get("scheduleStatus") or "" sched["infusionStatus"] = flat.get("infusionStatus") or sched.get("infusionStatus") or "" sched["qcStatus"] = flat.get("qcStatus") or sched.get("qcStatus") or "" sched["cancelReason"] = flat.get("cancelReason") or sched.get("cancelReason") or "" sched["remark"] = flat.get("remark") or "" patient_data = dict(aggregated.get("patient") or {}) if patient: patient_data.setdefault("patientName", patient.patient_name or "") patient_data.setdefault("sex", patient.sex or "") patient_data.setdefault("age", str(patient.age) if patient.age is not None else "") patient_data.setdefault("phoneNo", patient.phone_no or "") patient_data.setdefault("idCardNo", patient.id_card_no or "") patient_data.setdefault("emergencyContact", patient.emergency_contact or "") patient_data.setdefault("emergencyPhone", patient.emergency_phone or "") patient_data.setdefault("clinicalDiagnosis", patient.clinical_diagnosis or "") patient_data.setdefault("diagnosisAttachments", patient.diagnosis_attachments or "") patient_data.setdefault("needCheckup", patient.need_checkup) result: Dict[str, Any] = { **aggregated, **flat, "application": app, "patient": patient_data, "infusionPlan": plan, "infusionSchedule": sched, "products": flat.get("products") or aggregated.get("products") or [], "productionInfo": aggregated.get("productionInfo") or {}, "qcInfo": aggregated.get("qcInfo") or {}, "processQcInfoList": aggregated.get("processQcInfoList") or [], "reviewInfo": aggregated.get("reviewInfo") or {}, "outboundMaterial": aggregated.get("outboundMaterial") or {}, "outboundHandover": handover_svc.get_outbound_handover_view(db, schedule_id, tenant_no), "infusionConfirm": confirm_svc.get_confirm_by_schedule(db, schedule_id, tenant_no) or {}, } if application and (application.biz_type or "") in ("clinical_application", "basic_experiment"): from app.crud import clinical_payment payment = clinical_payment.get_payment_by_schedule( db, schedule_id=schedule_id, tenant_no=tenant_no ) if payment: result["payment"] = payment return result # ──────────────────────────────────────────── # Status update # ──────────────────────────────────────────── def update_infusion_status( db: Session, schedule_id: str, infusion_status: str, cancel_reason: Optional[str], tenant_no: str, user_id: str, ) -> Dict[str, Any]: row = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .first() ) if not row: raise ValueError("输注计划不存在") current = row.infusion_status or "" if current not in VALID_TRANSITIONS: raise ValueError(f"当前状态 {current} 不允许变更") if infusion_status not in VALID_TRANSITIONS.get(current, set()): raise ValueError(f"不允许从 {current} 变更为 {infusion_status}") # Cancel requires reason if infusion_status == "yqx" and not cancel_reason: raise ValueError("取消时需填写取消原因") now = datetime.now() row.infusion_status = infusion_status row.update_by = user_id row.update_time = now if infusion_status == "yqx": row.cancel_reason = cancel_reason or "" row.schedule_status = "yqx" row.qc_status = "yqx" # 自动作废关联的未完成出库生产工单 outbound_wo = ( db.query(BizProductionWorkOrder) .filter( BizProductionWorkOrder.infusion_schedule_id == schedule_id, BizProductionWorkOrder.tenant_no == tenant_no, BizProductionWorkOrder.del_flag == 0, BizProductionWorkOrder.work_order_type == "outbound", BizProductionWorkOrder.work_order_status.notin_(["ywc", "yzf"]), ) .first() ) if outbound_wo: outbound_wo.work_order_status = "yzf" outbound_wo.update_by = user_id outbound_wo.update_time = now # 自动作废关联的未完成出库质检工单 pending_qc_orders = ( db.query(BizQcWorkOrder) .filter( BizQcWorkOrder.infusion_schedule_id == schedule_id, BizQcWorkOrder.work_order_status.notin_(["ywc", "yzf"]), BizQcWorkOrder.del_flag == 0, ) .all() ) for order in pending_qc_orders: order.work_order_status = "yzf" order.update_by = user_id order.update_time = now from app.crud.pilot_outbound_status_sync import sync_pilot_outbound_from_schedule sync_pilot_outbound_from_schedule(db, row, user_id=user_id) # Sync all related products' product_status product_rows = ( db.query(BizPilotInfusionScheduleProduct) .filter( BizPilotInfusionScheduleProduct.infusion_schedule_id == schedule_id, BizPilotInfusionScheduleProduct.tenant_no == tenant_no, BizPilotInfusionScheduleProduct.del_flag == 0, ) .all() ) for p_row in product_rows: p_row.product_status = infusion_status p_row.update_by = user_id p_row.update_time = now db.flush() # If completing, trigger fee deduction if infusion_status == "ywc": _deduct_fee_on_complete(db, schedule_id, tenant_no, user_id) db.commit() db.refresh(row) return get_infusion_schedule_detail(db, row.id, tenant_no) def update_qc_status( db: Session, schedule_id: str, qc_status: str, tenant_no: str, user_id: str, ) -> Dict[str, Any]: row = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .first() ) if not row: raise ValueError("输注计划不存在") current = row.qc_status or "" if current not in QC_VALID_TRANSITIONS: raise ValueError(f"当前质检状态 {current} 不允许变更") if qc_status not in QC_VALID_TRANSITIONS.get(current, set()): raise ValueError(f"不允许从质检状态 {current} 变更为 {qc_status}") now = datetime.now() row.qc_status = qc_status row.update_by = user_id row.update_time = now from app.crud.pilot_outbound_status_sync import sync_pilot_outbound_from_schedule sync_pilot_outbound_from_schedule(db, row, user_id=user_id) db.commit() db.refresh(row) return get_infusion_schedule_detail(db, row.id, tenant_no) def _settlement_remark_marker(schedule_id: str) -> str: """幂等检索用标记（中文）；兼容历史英文标记 infusionScheduleId=""" return f"输注计划ID={schedule_id}" def _settlement_remark_markers_for_lookup(schedule_id: str) -> List[str]: return [_settlement_remark_marker(schedule_id), f"infusionScheduleId={schedule_id}"] def _find_existing_settlement_txn( db: Session, order_id: str, schedule_id: str ) -> Optional[BizChannelServiceOrderTxn]: marker_filters = [ BizChannelServiceOrderTxn.remark.like(f"%{m}%") for m in _settlement_remark_markers_for_lookup(schedule_id) ] return ( db.query(BizChannelServiceOrderTxn) .filter( BizChannelServiceOrderTxn.service_order_id == order_id, BizChannelServiceOrderTxn.txn_type == "settlement", or_(*marker_filters), BizChannelServiceOrderTxn.txn_status != "yzf", BizChannelServiceOrderTxn.del_flag == 0, ) .first() ) def _resolve_yfk_unit_price( db: Session, agreement_id: str, product: BizPilotInfusionScheduleProduct ) -> float: spec = (product.spec_text or "").strip() if agreement_id and (product.project_code or "").strip(): row = ( db.query(BizChannelAgreementProductPrice) .filter( BizChannelAgreementProductPrice.channel_agreement_id == agreement_id, BizChannelAgreementProductPrice.project_code == (product.project_code or ""), BizChannelAgreementProductPrice.spec_text == spec, BizChannelAgreementProductPrice.del_flag == 0, ) .first() ) if row and row.agreement_price is not None: return float(row.agreement_price) cell_name = (product.cell_type_name or "").strip() if agreement_id and cell_name: row = ( db.query(BizChannelAgreementProductPrice) .filter( BizChannelAgreementProductPrice.channel_agreement_id == agreement_id, BizChannelAgreementProductPrice.spec_text == spec, BizChannelAgreementProductPrice.del_flag == 0, or_( BizChannelAgreementProductPrice.cell_type_code == cell_name, BizChannelAgreementProductPrice.cell_type_name == cell_name, ), ) .first() ) if row and row.agreement_price is not None: return float(row.agreement_price) if product.actual_price is not None: return float(product.actual_price) if product.agreement_price is not None: return float(product.agreement_price) return 0.0 def _find_service_order_product( db: Session, order_id: str, product: BizPilotInfusionScheduleProduct ) -> Optional[BizChannelServiceOrderProduct]: project_code = (product.project_code or "").strip() spec = (product.spec_text or "").strip() cell_type = (product.cell_type_name or "").strip() if project_code: row = ( db.query(BizChannelServiceOrderProduct) .filter( BizChannelServiceOrderProduct.service_order_id == order_id, BizChannelServiceOrderProduct.project_code == project_code, BizChannelServiceOrderProduct.spec_text == spec, BizChannelServiceOrderProduct.del_flag == 0, ) .first() ) if row: return row if cell_type: return ( db.query(BizChannelServiceOrderProduct) .filter( BizChannelServiceOrderProduct.service_order_id == order_id, BizChannelServiceOrderProduct.cell_type == cell_type, BizChannelServiceOrderProduct.spec_text == spec, BizChannelServiceOrderProduct.del_flag == 0, ) .first() ) return None def _schedule_product_to_txn_item( product: BizPilotInfusionScheduleProduct, *, qty: int, amount: float ) -> Dict[str, Any]: return { "projectCode": product.project_code or "", "projectName": product.project_name or "", "cellType": product.cell_type_name or "", "specText": product.spec_text or "", "qty": qty, "amount": round(amount, 2), } def _create_settlement_txn( db: Session, *, order: BizChannelServiceOrder, schedule_id: str, application_no: str, amount: float, products_json: List[Dict[str, Any]], balance_before: Optional[float], balance_after: Optional[float], tenant_no: str, user_id: str, ) -> BizChannelServiceOrderTxn: now = datetime.now() remark_parts = [ "输注计划完成结算", _settlement_remark_marker(schedule_id), ] if application_no: remark_parts.append(f"申请单号={application_no}") txn = BizChannelServiceOrderTxn( id=generate_uuid(), service_order_id=order.id, txn_type="settlement", amount=round(amount, 2), txn_status="yqr", products_json=json.dumps(products_json, ensure_ascii=False) if products_json else None, remark=" ".join(remark_parts), balance_before=balance_before, balance_after=balance_after, finance_confirm_time=now, finance_confirm_by=user_id, finance_remark="输注计划完成自动结算", tenant_no=tenant_no or order.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(txn) db.flush() return txn def _deduct_fee_on_complete(db: Session, schedule_id: str, tenant_no: str, user_id: str): schedule = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .first() ) if not schedule: raise ValueError("输注计划不存在") plan = ( db.query(BizPilotInfusionPlan) .filter( BizPilotInfusionPlan.id == schedule.infusion_plan_id, BizPilotInfusionPlan.del_flag == 0, ) .first() ) if not plan: raise ValueError("输注方案不存在") application = ( db.query(BizPilotApplication) .filter( BizPilotApplication.id == plan.pilot_application_id, BizPilotApplication.tenant_no == tenant_no, BizPilotApplication.del_flag == 0, ) .first() ) if not application: raise ValueError("中试申请单不存在") # 临床应用/基础实验走缴费模块，不走服务订单扣费 if (application.biz_type or "") in ("clinical_application", "basic_experiment"): return if not application.channel_service_order_id: raise ValueError("申请单未关联渠道服务订单，无法扣费") order = ( db.query(BizChannelServiceOrder) .filter( BizChannelServiceOrder.id == application.channel_service_order_id, BizChannelServiceOrder.del_flag == 0, ) .first() ) if not order: raise ValueError("渠道服务订单不存在") order_type = order.order_type or "" application_no = application.application_no or "" if _find_existing_settlement_txn(db, order.id, schedule_id): return products = ( db.query(BizPilotInfusionScheduleProduct) .filter( BizPilotInfusionScheduleProduct.infusion_schedule_id == schedule_id, BizPilotInfusionScheduleProduct.tenant_no == tenant_no, BizPilotInfusionScheduleProduct.del_flag == 0, ) .order_by( BizPilotInfusionScheduleProduct.order_num.asc(), BizPilotInfusionScheduleProduct.create_time.asc(), ) .all() ) if not products: raise ValueError("输注计划未配置细胞产品，无法完成结算，请先在编辑中维护产品明细") now = datetime.now() if order_type == "yfk": agreement_id = order.channel_agreement_id if not agreement_id: raise ValueError("服务订单未关联渠道协议，无法确定扣费金额") total_deduction = 0.0 txn_products: List[Dict[str, Any]] = [] for p in products: qty = max(1, int(p.quantity or 1)) unit_price = _resolve_yfk_unit_price(db, agreement_id, p) if unit_price <= 0: label = p.project_name or p.cell_type_name or p.project_code or "未知产品" raise ValueError(f"无法确定产品「{label}」的协议单价，请检查渠道协议价格或输注计划实际单价") line_amount = unit_price * qty total_deduction += line_amount txn_products.append( _schedule_product_to_txn_item(p, qty=qty, amount=line_amount) ) account = _get_or_create_channel_account(db, order.channel_id, tenant_no, user_id) balance_before = float(account.balance_amount) if account.balance_amount else 0 if balance_before < total_deduction: raise ValueError( f"渠道账户余额不足，当前余额 {balance_before} 元，需扣减 {total_deduction} 元" ) account.balance_amount = balance_before - total_deduction balance_after = float(account.balance_amount) account.update_by = user_id account.update_time = now db.add(account) db.flush() order.current_balance = float(order.current_balance or 0) - total_deduction order.balance_before = balance_before order.balance_after = balance_after order.update_by = user_id order.update_time = now db.add(order) db.flush() _create_settlement_txn( db, order=order, schedule_id=schedule_id, application_no=application_no, amount=total_deduction, products_json=txn_products, balance_before=balance_before, balance_after=balance_after, tenant_no=tenant_no, user_id=user_id, ) elif order_type == "gmcp": txn_products = [] total_amount = 0.0 for p in products: order_product = _find_service_order_product(db, order.id, p) if not order_product: label = p.project_name or p.cell_type_name or p.project_code or "未知产品" raise ValueError(f"服务订单中未找到匹配的产品持仓：{label}") project_code = order_product.project_code or "" spec_text = order_product.spec_text or "" qty = max(1, int(p.quantity or 1)) cell_type_code, cell_type_name = _resolve_agreement_price_cell_type( db, order.channel_agreement_id, project_code, spec_text, order_product.cell_type or p.cell_type_name or "", ) if not cell_type_code: cell_type_code = cell_type_name or (p.cell_type_name or "") _reduce_channel_product_quota( db, channel_id=order.channel_id, project_code=project_code, cell_type_code=cell_type_code, spec_text=spec_text, reduce_qty=qty, user_id=user_id, ) current_qty = float(order_product.current_qty or 0) if current_qty < qty: raise ValueError( f"订单产品「{order_product.project_name or project_code}」持仓不足，" f"当前 {current_qty} 份，需扣减 {qty} 份" ) order_product.current_qty = current_qty - qty order_product.update_by = user_id order_product.update_time = now db.add(order_product) db.flush() unit_price = float(order_product.agreement_unit_price or 0) if unit_price <= 0 and p.actual_price is not None: unit_price = float(p.actual_price) line_amount = unit_price * qty total_amount += line_amount txn_products.append( _schedule_product_to_txn_item(p, qty=qty, amount=line_amount) ) _create_settlement_txn( db, order=order, schedule_id=schedule_id, application_no=application_no, amount=total_amount, products_json=txn_products, balance_before=None, balance_after=None, tenant_no=tenant_no, user_id=user_id, ) else: raise ValueError(f"不支持的渠道服务订单类型：{order_type or '未知'}") # ──────────────────────────────────────────── # Update schedule + products # ──────────────────────────────────────────── def update_infusion_schedule_mgmt( db: Session, schedule_id: str, data: dict, tenant_no: str, user_id: str, ) -> Dict[str, Any]: row = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .first() ) if not row: raise ValueError("输注计划不存在") status = row.infusion_status or "" if status not in ("dtz", "yrqr"): raise ValueError("当前状态不允许编辑") now = datetime.now() fields = { "cellType": "cell_type", "specText": "spec_text", "packageId": "package_id", "packageText": "package_text", "infusionLocation": "infusion_location", "infusionMethod": "infusion_method", } for camel, snake in fields.items(): if camel in data and data[camel] is not None: setattr(row, snake, data[camel]) if "infusionTime" in data and data["infusionTime"] is not None: val = data["infusionTime"] if isinstance(val, str) and val: for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"): try: row.infusion_time = datetime.strptime(val, fmt) break except ValueError: continue row.update_by = user_id row.update_time = now products_data = data.get("products") if products_data is not None: def _price_id(item): return (item.get("priceId") or item.get("price_id") or "").strip() def _product_key(item): price_id = _price_id(item) if price_id: return "price:" + price_id project_code = (item.get("projectCode") or item.get("project_code") or "").strip() spec_text = (item.get("specText") or item.get("spec_text") or "").strip() return f"proj:{project_code}|{spec_text}" def build_product(row_id): return BizPilotInfusionScheduleProduct( id=row_id, infusion_schedule_id=schedule_id, tenant_no=tenant_no, create_by=user_id, create_dept_id="", create_time=now, ) def apply_product(row, item, idx, _now): row.price_id = _price_id(item) row.project_code = item.get("projectCode") or item.get("project_code") or "" row.project_name = item.get("projectName") or item.get("project_name") or "" row.cell_type_name = item.get("cellTypeName") or item.get("cell_type_name") or "" row.spec_text = item.get("specText") or item.get("spec_text") or "" row.package_id = item.get("packageId") or item.get("package_id") or "" row.package_text = item.get("packageText") or item.get("package_text") or "" row.agreement_price = item.get("agreementPrice") if item.get("agreementPrice") is not None else ( item.get("agreement_price") if item.get("agreement_price") is not None else None ) row.actual_price = item.get("actualPrice") if item.get("actualPrice") is not None else ( item.get("actual_price") if item.get("actual_price") is not None else None ) row.quantity = item.get("quantity") if item.get("quantity") is not None else 1 row.product_status = status row.order_num = idx + 1 row.remark = item.get("remark") or "" sync_child_rows( db, BizPilotInfusionScheduleProduct, parent_attr="infusion_schedule_id", parent_id=schedule_id, tenant_no=tenant_no, items=list(products_data or []), user_id=user_id, now=now, build_row=build_product, apply_fields=apply_product, natural_key=_product_key, row_natural_key=lambda row: ( "price:" + (row.price_id or "").strip() if (row.price_id or "").strip() else f"proj:{(row.project_code or '').strip()}|{(row.spec_text or '').strip()}" ), duplicate_message="输注产品重复", foreign_message="输注产品不属于当前计划", ) db.commit() db.refresh(row) return get_infusion_schedule_detail(db, row.id, tenant_no)