"""中试交付管理 CRUD(交付信息、交付物料、交付方式 + 费用扣减)""" from __future__ import annotations import uuid from datetime import datetime from typing import Any, Dict, List, Optional from sqlalchemy import and_, func from sqlalchemy.orm import Session from app.crud.child_row_sync import sync_child_rows from app.models.models import ( BizChannelAccount, BizChannelAgreementProductPrice, BizChannelProductQuota, BizChannelServiceOrder, BizChannelServiceOrderProduct, BizExpressCompany, BizItemUnit, BizPilotApplication, BizPilotDelivery, BizPilotDeliveryMaterial, BizPilotDeliveryMethod, BizPilotInfusionPlan, BizPilotInfusionSchedule, ) from app.crud.pilot_service import ( _get_or_create_channel_account, _reduce_channel_product_quota, ) # ──────────────────────────────────────────── # Helpers: _row_to_dict # ──────────────────────────────────────────── def _generate_delivery_no(db: Session, tenant_no: str) -> str: today_str = datetime.now().strftime("%Y%m%d") prefix = f"ZSJF{today_str}" existing = ( db.query(BizPilotDelivery.delivery_no) .filter( BizPilotDelivery.tenant_no == tenant_no, BizPilotDelivery.delivery_no.like(f"{prefix}%"), BizPilotDelivery.del_flag == 0, ) .order_by(BizPilotDelivery.delivery_no.desc()) .first() ) seq = 1 if existing and existing[0]: try: seq = int(existing[0][-4:]) + 1 except ValueError: seq = 1 return f"{prefix}{seq:04d}" def _row_to_dict_delivery(row, db=None, tenant_no=None) -> Dict[str, Any]: d: Dict[str, Any] = { "id": row.id, "infusionScheduleId": row.infusion_schedule_id or "", "deliveryNo": row.delivery_no or "", "tenantNo": row.tenant_no or "", "remark": row.remark or "", "createBy": row.create_by 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 db and tenant_no: # materialCount material_count = ( db.query(func.count(BizPilotDeliveryMaterial.id)) .filter( BizPilotDeliveryMaterial.delivery_id == row.id, BizPilotDeliveryMaterial.tenant_no == tenant_no, BizPilotDeliveryMaterial.del_flag == 0, ) .scalar() ) or 0 d["materialCount"] = material_count # hasMethod method = ( db.query(BizPilotDeliveryMethod.id) .filter( BizPilotDeliveryMethod.delivery_id == row.id, BizPilotDeliveryMethod.tenant_no == tenant_no, BizPilotDeliveryMethod.del_flag == 0, ) .first() ) d["hasMethod"] = method is not None return d def _row_to_dict_delivery_material(row, db=None) -> Dict[str, Any]: d: Dict[str, Any] = { "id": row.id, "deliveryId": row.delivery_id or "", "materialName": row.material_name or "", "qty": float(row.qty) if row.qty is not None else None, "unitId": row.unit_id or "", "tenantNo": row.tenant_no or "", "remark": row.remark or "", "createBy": row.create_by 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 "", } # unitName lookup if db and row.unit_id: unit = ( db.query(BizItemUnit.unit_name) .filter( BizItemUnit.id == row.unit_id, BizItemUnit.del_flag == 0, ) .first() ) d["unitName"] = unit.unit_name if unit else "" else: d["unitName"] = "" return d def _row_to_dict_delivery_method(row, db=None) -> Dict[str, Any]: d: Dict[str, Any] = { "id": row.id, "deliveryId": row.delivery_id or "", "deliveryMode": row.delivery_mode or "", "deliveryAttachments": row.delivery_attachments or "", "shipTime": row.ship_time.strftime("%Y-%m-%d %H:%M:%S") if row.ship_time else "", "handoverTemp": row.handover_temp or "", "pickupPerson": row.pickup_person or "", "pickupPhone": row.pickup_phone or "", "senderName": row.sender_name or "", "expressCompanyId": row.express_company_id or "", "waybillNo": row.waybill_no or "", "expressFee": float(row.express_fee) if row.express_fee is not None else None, "etaTime": row.eta_time.strftime("%Y-%m-%d %H:%M:%S") if row.eta_time else "", "tenantNo": row.tenant_no or "", "remark": row.remark or "", "createBy": row.create_by 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 "", } # expressCompanyName lookup if db and row.express_company_id: company = ( db.query(BizExpressCompany.express_name) .filter( BizExpressCompany.id == row.express_company_id, BizExpressCompany.del_flag == 0, ) .first() ) d["expressCompanyName"] = company.express_name if company else "" else: d["expressCompanyName"] = "" return d # ──────────────────────────────────────────── # Fee Deduction Logic # ──────────────────────────────────────────── def _deduct_fee_on_delivery(db: Session, schedule_id: str, tenant_no: str, user_id: str): """ 创建交付时扣减费用。 流程: 1. schedule → plan → application 2. application.channel_service_order_id → service order 3. service_order.order_type (yfk/gmcp) 4. schedule.cell_type, schedule.spec_text 5. yfk(预付款型): 查协议价 → 扣渠道账户余额 → 更新订单余额 6. gmcp(购买型): 扣份数 + 更新订单产品持仓 """ # 1. Find schedule schedule = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .first() ) if not schedule: raise ValueError("输注计划不存在") # 2. Find plan → application 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("中试申请单不存在") # 临床应用走 clinical_payment 模块缴费,不走服务订单扣费 if (application.biz_type or "") == "clinical_application": return # 3. Get service order 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 "" # 4. yfk: deduct from channel account balance if order_type == "yfk": # Find agreement price agreement_id = order.channel_agreement_id if not agreement_id: raise ValueError("服务订单未关联渠道协议,无法确定扣费金额") agreement_product = ( db.query(BizChannelAgreementProductPrice) .filter( BizChannelAgreementProductPrice.channel_agreement_id == agreement_id, BizChannelAgreementProductPrice.cell_type_code == (schedule.cell_type or ""), BizChannelAgreementProductPrice.spec_text == (schedule.spec_text or ""), BizChannelAgreementProductPrice.del_flag == 0, ) .first() ) deduction_amount = float(agreement_product.agreement_price) if agreement_product else 0 if deduction_amount <= 0: # No price found, skip deduction return # Get channel account with row lock 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 < deduction_amount: raise ValueError( f"渠道账户余额不足,当前余额 {balance_before} 元,需扣减 {deduction_amount} 元" ) # Deduct balance account.balance_amount = balance_before - deduction_amount balance_after = float(account.balance_amount) account.update_by = user_id account.update_time = datetime.now() db.add(account) db.flush() # Update order current_balance order.current_balance = float(order.current_balance or 0) - deduction_amount order.balance_before = balance_before order.balance_after = balance_after order.update_by = user_id order.update_time = datetime.now() db.add(order) db.flush() # 5. gmcp: deduct quota and update order product elif order_type == "gmcp": # Find matching service order product for project_code order_product = ( db.query(BizChannelServiceOrderProduct) .filter( BizChannelServiceOrderProduct.service_order_id == order.id, BizChannelServiceOrderProduct.cell_type == (schedule.cell_type or ""), BizChannelServiceOrderProduct.spec_text == (schedule.spec_text or ""), BizChannelServiceOrderProduct.del_flag == 0, ) .first() ) project_code = order_product.project_code if order_product else "" # Reduce channel product quota _reduce_channel_product_quota( db, channel_id=order.channel_id, project_code=project_code, cell_type_code=schedule.cell_type or "", spec_text=schedule.spec_text or "", reduce_qty=1, user_id=user_id, ) # Update BizChannelServiceOrderProduct.current_qty if order_product: current_qty = float(order_product.current_qty or 0) if current_qty < 1: raise ValueError("订单产品持仓数量不足") order_product.current_qty = current_qty - 1 order_product.update_by = user_id order_product.update_time = datetime.now() db.add(order_product) db.flush() # ──────────────────────────────────────────── # Delivery CRUD # ──────────────────────────────────────────── def get_delivery_by_schedule( db: Session, schedule_id: str, tenant_no: str ) -> Optional[Dict[str, Any]]: row = ( db.query(BizPilotDelivery) .filter( BizPilotDelivery.infusion_schedule_id == schedule_id, BizPilotDelivery.tenant_no == tenant_no, BizPilotDelivery.del_flag == 0, ) .first() ) if not row: return None return _row_to_dict_delivery(row, db, tenant_no) def list_deliveries_by_application( db: Session, application_id: str, tenant_no: str ) -> List[Dict[str, Any]]: """List all deliveries for an application (through plan → schedule → delivery).""" # Find all plan IDs for this application plan_ids = [ r[0] for r in db.query(BizPilotInfusionPlan.id) .filter( BizPilotInfusionPlan.pilot_application_id == application_id, BizPilotInfusionPlan.tenant_no == tenant_no, BizPilotInfusionPlan.del_flag == 0, ) .all() ] if not plan_ids: return [] # Find all schedule IDs for these plans schedule_ids = [ r[0] for r in db.query(BizPilotInfusionSchedule.id) .filter( BizPilotInfusionSchedule.infusion_plan_id.in_(plan_ids), BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .all() ] if not schedule_ids: return [] # Find all deliveries for these schedules rows = ( db.query(BizPilotDelivery) .filter( BizPilotDelivery.infusion_schedule_id.in_(schedule_ids), BizPilotDelivery.tenant_no == tenant_no, BizPilotDelivery.del_flag == 0, ) .order_by(BizPilotDelivery.create_time.desc()) .all() ) return [_row_to_dict_delivery(r, db, tenant_no) for r in rows] def create_delivery( db: Session, schedule_id: str, data: dict, tenant_no: str, user_id: str, dept_id: str, ) -> Dict[str, Any]: # Validate schedule exists and belongs to tenant schedule = ( db.query(BizPilotInfusionSchedule) .filter( BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.tenant_no == tenant_no, BizPilotInfusionSchedule.del_flag == 0, ) .first() ) if not schedule: raise ValueError("输注计划不存在") # Check if delivery already exists for this schedule existing = ( db.query(BizPilotDelivery) .filter( BizPilotDelivery.infusion_schedule_id == schedule_id, BizPilotDelivery.tenant_no == tenant_no, BizPilotDelivery.del_flag == 0, ) .first() ) if existing: raise ValueError("该输注计划已创建交付信息") now = datetime.now() delivery_no = _generate_delivery_no(db, tenant_no) row = BizPilotDelivery( id=str(uuid.uuid4()), infusion_schedule_id=schedule_id, delivery_no=delivery_no, tenant_no=tenant_no, order_num=0, remark=data.get("remark") or "", create_by=user_id, create_dept_id=dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(row) db.flush() # Trigger fee deduction try: _deduct_fee_on_delivery(db, schedule_id, tenant_no, user_id) except ValueError: db.rollback() raise # Update schedule status to yjf (已交付) schedule.schedule_status = "yjf" schedule.update_by = user_id schedule.update_time = now from app.crud.pilot_outbound_status_sync import sync_pilot_outbound_from_schedule sync_pilot_outbound_from_schedule(db, schedule, user_id=user_id) db.commit() db.refresh(row) return _row_to_dict_delivery(row, db, tenant_no) def soft_delete_delivery(db: Session, delivery_id: str, tenant_no: str) -> None: now = datetime.now() row = ( db.query(BizPilotDelivery) .filter( BizPilotDelivery.id == delivery_id, BizPilotDelivery.tenant_no == tenant_no, BizPilotDelivery.del_flag == 0, ) .first() ) if not row: raise ValueError("交付信息不存在") row.del_flag = 1 row.update_time = now # Cascade soft-delete to materials db.query(BizPilotDeliveryMaterial).filter( BizPilotDeliveryMaterial.delivery_id == delivery_id, BizPilotDeliveryMaterial.tenant_no == tenant_no, BizPilotDeliveryMaterial.del_flag == 0, ).update({"del_flag": 1, "update_time": now}, synchronize_session=False) # Cascade soft-delete to method db.query(BizPilotDeliveryMethod).filter( BizPilotDeliveryMethod.delivery_id == delivery_id, BizPilotDeliveryMethod.tenant_no == tenant_no, BizPilotDeliveryMethod.del_flag == 0, ).update({"del_flag": 1, "update_time": now}, synchronize_session=False) db.commit() # ──────────────────────────────────────────── # Delivery Material CRUD # ──────────────────────────────────────────── def list_delivery_materials( db: Session, delivery_id: str, tenant_no: str ) -> List[Dict[str, Any]]: rows = ( db.query(BizPilotDeliveryMaterial) .filter( BizPilotDeliveryMaterial.delivery_id == delivery_id, BizPilotDeliveryMaterial.tenant_no == tenant_no, BizPilotDeliveryMaterial.del_flag == 0, ) .order_by( BizPilotDeliveryMaterial.order_num.asc(), BizPilotDeliveryMaterial.create_time.asc(), ) .all() ) return [_row_to_dict_delivery_material(r, db) for r in rows] def save_delivery_materials( db: Session, delivery_id: str, items_json: list, tenant_no: str, user_id: str, ) -> List[Dict[str, Any]]: """按物料名称更新已有行,只插入新物料,只软删去掉的物料。""" now = datetime.now() # Validate delivery exists delivery = ( db.query(BizPilotDelivery) .filter( BizPilotDelivery.id == delivery_id, BizPilotDelivery.tenant_no == tenant_no, BizPilotDelivery.del_flag == 0, ) .first() ) if not delivery: raise ValueError("交付信息不存在") def build_row(row_id): return BizPilotDeliveryMaterial( id=row_id, delivery_id=delivery_id, material_name="", tenant_no=tenant_no, create_by=user_id, create_dept_id="", create_time=now, ) def apply_row(row, item, idx, _now): row.material_name = item.get("material_name") or "" row.qty = item.get("qty") row.unit_id = item.get("unit_id") or "" row.order_num = idx + 1 row.remark = item.get("remark") or "" created = sync_child_rows( db, BizPilotDeliveryMaterial, parent_attr="delivery_id", parent_id=delivery_id, tenant_no=tenant_no, items=list(items_json or []), user_id=user_id, now=now, build_row=build_row, apply_fields=apply_row, natural_key=lambda item: (item.get("material_name") or "").strip(), row_natural_key=lambda row: (row.material_name or "").strip(), duplicate_message="交付物料重复", foreign_message="交付物料不属于当前交付单", ) db.commit() for m in created: db.refresh(m) return [_row_to_dict_delivery_material(m, db) for m in created] def delete_delivery_material( db: Session, material_id: str, tenant_no: str ) -> None: row = ( db.query(BizPilotDeliveryMaterial) .filter( BizPilotDeliveryMaterial.id == material_id, BizPilotDeliveryMaterial.tenant_no == tenant_no, BizPilotDeliveryMaterial.del_flag == 0, ) .first() ) if not row: raise ValueError("交付物料不存在") row.del_flag = 1 row.update_time = datetime.now() db.commit() # ──────────────────────────────────────────── # Delivery Method CRUD # ──────────────────────────────────────────── def get_delivery_method( db: Session, delivery_id: str, tenant_no: str ) -> Optional[Dict[str, Any]]: row = ( db.query(BizPilotDeliveryMethod) .filter( BizPilotDeliveryMethod.delivery_id == delivery_id, BizPilotDeliveryMethod.tenant_no == tenant_no, BizPilotDeliveryMethod.del_flag == 0, ) .first() ) if not row: return None return _row_to_dict_delivery_method(row, db) def save_delivery_method( db: Session, delivery_id: str, data: dict, tenant_no: str, user_id: str, ) -> Dict[str, Any]: """Create or update delivery method (upsert pattern).""" now = datetime.now() # Validate delivery exists delivery = ( db.query(BizPilotDelivery) .filter( BizPilotDelivery.id == delivery_id, BizPilotDelivery.tenant_no == tenant_no, BizPilotDelivery.del_flag == 0, ) .first() ) if not delivery: raise ValueError("交付信息不存在") existing = ( db.query(BizPilotDeliveryMethod) .filter( BizPilotDeliveryMethod.delivery_id == delivery_id, BizPilotDeliveryMethod.tenant_no == tenant_no, BizPilotDeliveryMethod.del_flag == 0, ) .first() ) # Parse datetime fields ship_time = data.get("ship_time") eta_time = data.get("eta_time") if existing: # Update existing.delivery_mode = data.get("delivery_mode") or existing.delivery_mode existing.delivery_attachments = data.get("delivery_attachments") if data.get("delivery_attachments") is not None else existing.delivery_attachments existing.ship_time = ship_time if ship_time is not None else existing.ship_time existing.handover_temp = data.get("handover_temp") if data.get("handover_temp") is not None else existing.handover_temp existing.pickup_person = data.get("pickup_person") if data.get("pickup_person") is not None else existing.pickup_person existing.pickup_phone = data.get("pickup_phone") if data.get("pickup_phone") is not None else existing.pickup_phone existing.sender_name = data.get("sender_name") if data.get("sender_name") is not None else existing.sender_name existing.express_company_id = data.get("express_company_id") if data.get("express_company_id") is not None else existing.express_company_id existing.waybill_no = data.get("waybill_no") if data.get("waybill_no") is not None else existing.waybill_no existing.express_fee = data.get("express_fee") if data.get("express_fee") is not None else existing.express_fee existing.eta_time = eta_time if eta_time is not None else existing.eta_time existing.remark = data.get("remark") if data.get("remark") is not None else existing.remark existing.update_by = user_id existing.update_time = now db.commit() db.refresh(existing) return _row_to_dict_delivery_method(existing, db) # Create row = BizPilotDeliveryMethod( id=str(uuid.uuid4()), delivery_id=delivery_id, delivery_mode=data.get("delivery_mode") or "", delivery_attachments=data.get("delivery_attachments") or "", ship_time=ship_time, handover_temp=data.get("handover_temp") or "", pickup_person=data.get("pickup_person") or "", pickup_phone=data.get("pickup_phone") or "", sender_name=data.get("sender_name") or "", express_company_id=data.get("express_company_id") or "", waybill_no=data.get("waybill_no") or "", express_fee=data.get("express_fee"), eta_time=eta_time, tenant_no=tenant_no, order_num=0, remark=data.get("remark") or "", create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(row) db.commit() db.refresh(row) return _row_to_dict_delivery_method(row, db)