zkxw/backend/app/crud/pilot_application.py

"""中试申请单 CRUD（申请单、患者信息、输注方案、输注计划）"""
from __future__ import annotations

import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional

from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.models.models import (
    BizChannel,
    BizChannelServiceOrder,
    BizCustomer,
    BizPilotApplication,
    BizPilotDelivery,
    BizPilotInfusionConfirm,
    BizPilotInfusionPlan,
    BizPilotInfusionSchedule,
    BizPilotInfusionScheduleProduct,
    BizPilotPatient,
    BizProductionWorkOrder,
    BizQcWorkOrder,
)
from app.crud import pilot_checkup
from app.crud.collection_protocol import attachments_to_urls, urls_to_attachment_list


# ────────────────────────────────────────────
# Helpers: _row_to_dict
# ────────────────────────────────────────────


def _normalize_yyyymmdd(value: Any) -> str:
    if value is None:
        return ""
    s = str(value).strip().replace("-", "").replace("/", "")
    return s[:8] if len(s) >= 8 else s


def _format_yyyy_mm_dd(value: Any) -> str:
    s = _normalize_yyyymmdd(value)
    if len(s) == 8:
        return f"{s[:4]}-{s[4:6]}-{s[6:8]}"
    return str(value or "").strip()


def _parse_infusion_datetime(value: Any) -> Optional[datetime]:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        return value
    s = str(value).strip()
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.strptime(s[:19] if fmt.endswith("%S") else s[:10], fmt)
        except ValueError:
            continue
    compact = _normalize_yyyymmdd(s)
    if len(compact) == 8:
        try:
            return datetime.strptime(compact, "%Y%m%d")
        except ValueError:
            return None
    return None


def _release_application_no_for_soft_delete(application_no: str, row_id: str) -> str:
    """软删除时改写申请单号，避免 uk_application_no(application_no, del_flag) 与历史已删记录冲突。"""
    base = (application_no or "").strip()
    suffix = f"~{row_id.replace('-', '')[:20]}"
    max_base_len = max(0, 64 - len(suffix))
    return f"{base[:max_base_len]}{suffix}"


def _generate_application_no(db: Session, tenant_no: str, biz_type: str = "pilot_service") -> str:
    today_str = datetime.now().strftime("%Y%m%d")
    if biz_type == "clinical_application":
        prefix = f"LCSQ{today_str}"
    elif biz_type == "basic_experiment":
        prefix = f"JCSY{today_str}"
    else:
        prefix = f"ZSSQ{today_str}"
    existing = (
        db.query(BizPilotApplication.application_no)
        .filter(
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.application_no.like(f"{prefix}%"),
            BizPilotApplication.del_flag == 0,
        )
        .order_by(BizPilotApplication.application_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 _resolve_application_attachment_payload(raw: dict) -> None:
    """解析保存载荷：附件列表 → URL 字段。"""
    if raw.get("consentAttachments") is not None:
        raw["consentFormUrls"] = attachments_to_urls(raw["consentAttachments"]) or None
    elif (raw.get("consentFormUrls") or "").strip():
        raw["consentFormUrls"] = (raw["consentFormUrls"] or "").strip()

    if raw.get("otherAttachments") is not None:
        raw["attachmentUrls"] = attachments_to_urls(raw["otherAttachments"]) or None
    elif (raw.get("attachmentUrls") or "").strip():
        raw["attachmentUrls"] = (raw["attachmentUrls"] or "").strip()


def _row_to_dict_application(row, db=None, tenant_no=None) -> Dict[str, Any]:
    tn = tenant_no or row.tenant_no
    d: Dict[str, Any] = {
        "id": row.id,
        "applicationNo": row.application_no or "",
        "applicationName": row.application_name or "",
        "projectTitle": row.project_title or "",
        "customerName": row.customer_name or "",
        "customerContact": row.customer_contact or "",
        "customerPhone": row.customer_phone or "",
        "applicationStatus": row.application_status or "",
        "bizType": row.biz_type or "pilot_service",
        "channelServiceOrderId": row.channel_service_order_id or "",
        "channelId": row.channel_id or "",
        "channelName": row.channel_name or "",
        "tenantNo": row.tenant_no or "",
        "remark": row.remark or "",
        "consentFormUrls": row.consent_form_urls or "",
        "attachmentUrls": row.attachment_urls or "",
        "consentAttachmentFiles": urls_to_attachment_list(row.consent_form_urls, db, tn) if db else [],
        "attachmentFiles": urls_to_attachment_list(row.attachment_urls, db, tn) if db else [],
        "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:
        # patientName
        patient = (
            db.query(BizPilotPatient.patient_name)
            .filter(
                BizPilotPatient.pilot_application_id == row.id,
                BizPilotPatient.tenant_no == tenant_no,
                BizPilotPatient.del_flag == 0,
            )
            .first()
        )
        d["patientName"] = patient.patient_name if patient else ""

        # infusionScheduleCount：该申请单下全部输注计划条数（跨输注方案汇总）
        schedule_count = (
            db.query(func.count(BizPilotInfusionSchedule.id))
            .join(
                BizPilotInfusionPlan,
                and_(
                    BizPilotInfusionSchedule.infusion_plan_id == BizPilotInfusionPlan.id,
                    BizPilotInfusionPlan.del_flag == 0,
                ),
            )
            .filter(
                BizPilotInfusionPlan.pilot_application_id == row.id,
                BizPilotInfusionPlan.tenant_no == tenant_no,
                BizPilotInfusionSchedule.tenant_no == tenant_no,
                BizPilotInfusionSchedule.del_flag == 0,
            )
            .scalar()
        ) or 0
        d["infusionScheduleCount"] = schedule_count

        # serviceOrderNo + orderType
        svc_order = (
            db.query(BizChannelServiceOrder)
            .filter(BizChannelServiceOrder.id == row.channel_service_order_id)
            .first()
        )
        if svc_order:
            name = (svc_order.order_name or "").strip()
            if name:
                d["serviceOrderNo"] = name
            else:
                oid = svc_order.id or ""
                d["serviceOrderNo"] = oid[-8:] if len(oid) >= 8 else oid
            d["orderType"] = svc_order.order_type or ""
        else:
            d["serviceOrderNo"] = ""
            d["orderType"] = ""

    return d


def _row_to_dict_patient(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "pilotApplicationId": row.pilot_application_id or "",
        "customerId": row.customer_id or "",
        "patientName": row.patient_name or "",
        "sex": row.sex or "",
        "age": row.age if row.age is not None else None,
        "idCardNo": row.id_card_no or "",
        "phoneNo": row.phone_no or "",
        "emergencyContact": row.emergency_contact or "",
        "emergencyPhone": row.emergency_phone or "",
        "clinicalDiagnosis": row.clinical_diagnosis or "",
        "diagnosisAttachments": row.diagnosis_attachments or "",
        "needCheckup": row.need_checkup if row.need_checkup is not None else 1,
        "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 "",
    }


def _row_to_dict_infusion_plan(row, db=None, tenant_no=None) -> Dict[str, Any]:
    d: Dict[str, Any] = {
        "id": row.id,
        "pilotApplicationId": row.pilot_application_id or "",
        "cellType": row.cell_type or "",
        "specText": row.spec_text or "",
        "totalCourseCount": row.total_course_count if row.total_course_count is not None else 0,
        "courseInterval": row.course_interval or "",
        "infusionRoute": row.infusion_route or "",
        "infusionStartDate": _format_yyyy_mm_dd(row.infusion_start_date) if row.infusion_start_date else "",
        "channelAgreementId": row.channel_agreement_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 "",
    }

    if db and tenant_no:
        # scheduleCount (all schedules)
        schedule_count = (
            db.query(func.count(BizPilotInfusionSchedule.id))
            .filter(
                BizPilotInfusionSchedule.infusion_plan_id == row.id,
                BizPilotInfusionSchedule.tenant_no == tenant_no,
                BizPilotInfusionSchedule.del_flag == 0,
            )
            .scalar()
        ) or 0
        d["scheduleCount"] = schedule_count

        # completedCount (schedule_status='ywc')
        completed_count = (
            db.query(func.count(BizPilotInfusionSchedule.id))
            .filter(
                BizPilotInfusionSchedule.infusion_plan_id == row.id,
                BizPilotInfusionSchedule.tenant_no == tenant_no,
                BizPilotInfusionSchedule.del_flag == 0,
                BizPilotInfusionSchedule.schedule_status == "ywc",
            )
            .scalar()
        ) or 0
        d["completedCount"] = completed_count

    return d


def _row_to_dict_schedule_product(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "infusionScheduleId": row.infusion_schedule_id or "",
        "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 "",
        "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,
        "status": row.product_status or "",
        "remark": row.remark or "",
    }


def _row_to_dict_infusion_schedule(row, cell_products: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
    infusion_time = ""
    if row.infusion_time:
        infusion_time = row.infusion_time.strftime("%Y-%m-%d")
    return {
        "id": row.id,
        "infusionPlanId": row.infusion_plan_id or "",
        "cellType": row.cell_type or "",
        "specText": row.spec_text or "",
        "infusionTime": infusion_time,
        "infusionLocation": row.infusion_location or "",
        "infusionMethod": row.infusion_method or "",
        "scheduleStatus": row.schedule_status or "",
        "infusionStatus": row.infusion_status or "dtz",
        "cancelReason": row.cancel_reason or "",
        "qcStatus": row.qc_status or "dzc",
        "cellProducts": cell_products if cell_products is not None else [],
        "attachments": row.attachments or "",
        "amount": float(row.amount) if row.amount else None,
        "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 "",
    }


# ────────────────────────────────────────────
# Application CRUD
# ────────────────────────────────────────────


def list_applications(
    db: Session,
    tenant_no: str,
    page: int = 1,
    page_size: int = 10,
    channel_id: Optional[str] = None,
    application_status: Optional[str] = None,
    channel_service_order_id: Optional[str] = None,
    keyword: Optional[str] = None,
    biz_type: Optional[str] = None,
):
    q = db.query(BizPilotApplication).filter(
        BizPilotApplication.tenant_no == tenant_no,
        BizPilotApplication.del_flag == 0,
    )
    if channel_id:
        q = q.filter(BizPilotApplication.channel_id == channel_id)
    if application_status:
        statuses = [s.strip() for s in application_status.split(",") if s.strip()]
        if statuses:
            q = q.filter(BizPilotApplication.application_status.in_(statuses))
    if channel_service_order_id:
        q = q.filter(BizPilotApplication.channel_service_order_id == channel_service_order_id)
    if biz_type:
        q = q.filter(BizPilotApplication.biz_type == biz_type)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        patient_app_ids = (
            db.query(BizPilotPatient.pilot_application_id)
            .filter(
                BizPilotPatient.tenant_no == tenant_no,
                BizPilotPatient.del_flag == 0,
                BizPilotPatient.patient_name.like(like),
            )
            .distinct()
        )
        order_ids = (
            db.query(BizChannelServiceOrder.id)
            .filter(
                BizChannelServiceOrder.tenant_no == tenant_no,
                BizChannelServiceOrder.del_flag == 0,
                BizChannelServiceOrder.order_name.like(like),
            )
            .distinct()
        )
        q = q.filter(
            or_(
                BizPilotApplication.application_no.like(like),
                BizPilotApplication.application_name.like(like),
                BizPilotApplication.channel_name.like(like),
                BizPilotApplication.id.in_(patient_app_ids),
                BizPilotApplication.channel_service_order_id.in_(order_ids),
            )
        )

    total = q.order_by(None).with_entities(func.count(BizPilotApplication.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizPilotApplication.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = [_row_to_dict_application(r, db, tenant_no) for r in rows]
    return {
        "list": items,
        "pagination": {"page": page, "pageSize": page_size, "total": total},
    }


def get_application(db: Session, app_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
    row = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == app_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    if not row:
        return None
    result = _row_to_dict_application(row, db, tenant_no)
    plans = list_infusion_plans(db, app_id, tenant_no)
    if plans:
        plan = plans[0]
        result["infusionPlan"] = {
            "id": plan.get("id"),
            "totalCourseCount": plan.get("totalCourseCount"),
            "courseInterval": plan.get("courseInterval"),
            "infusionStartDate": plan.get("infusionStartDate"),
            "channelAgreementId": plan.get("channelAgreementId"),
        }
        plan_id = plan.get("id")
        if plan_id:
            result["infusionSchedules"] = list_schedules_by_plan(db, plan_id, tenant_no)
            if row.biz_type in ("clinical_application", "basic_experiment") and result["infusionSchedules"]:
                from app.crud import clinical_payment

                payment_map = clinical_payment.map_payments_by_schedule_id(
                    db, application_id=app_id, tenant_no=tenant_no
                )
                for schedule in result["infusionSchedules"]:
                    schedule_id = schedule.get("id")
                    if schedule_id and schedule_id in payment_map:
                        schedule["payment"] = payment_map[schedule_id]
    else:
        result["infusionPlan"] = None
        result["infusionSchedules"] = []
    return result


def create_application(
    db: Session,
    data: dict,
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> Dict[str, Any]:
    channel_service_order_id = data.get("channelServiceOrderId") or ""
    biz_type = data.get("bizType") or "pilot_service"
    channel_id = (data.get("channelId") or "").strip()
    channel_name = (data.get("channelName") or "").strip()

    project_title = ""
    customer_name = ""
    customer_contact = ""
    customer_phone = ""
    consent_form_urls = data.get("consentFormUrls") or None

    if biz_type == "basic_experiment":
        if not (data.get("projectTitle") or "").strip():
            raise ValueError("请填写课题名称")
        if not (data.get("customerName") or "").strip():
            raise ValueError("请填写客户名称")
        if not (data.get("customerContact") or "").strip():
            raise ValueError("请填写客户联系人")
        if not (data.get("customerPhone") or "").strip():
            raise ValueError("请填写客户联系方式")
        project_title = (data.get("projectTitle") or "").strip()
        customer_name = (data.get("customerName") or "").strip()
        customer_contact = (data.get("customerContact") or "").strip()
        customer_phone = (data.get("customerPhone") or "").strip()
        channel_service_order_id = ""
        channel_id = ""
        channel_name = ""
        consent_form_urls = None
    elif not channel_service_order_id and biz_type != "clinical_application":
        raise ValueError("请选择渠道服务订单")

    # Validate service order exists and is active
    svc_order = None
    if channel_service_order_id:
        svc_order = (
            db.query(BizChannelServiceOrder)
            .filter(
                BizChannelServiceOrder.id == channel_service_order_id,
                BizChannelServiceOrder.del_flag == 0,
            )
            .first()
        )
        if not svc_order:
            raise ValueError("渠道服务订单不存在")
        if svc_order.order_status != "sxz":
            raise ValueError("渠道服务订单未生效，无法创建中试申请")

    if biz_type == "clinical_application":
        if svc_order:
            channel_id = svc_order.channel_id or channel_id
            channel_name = svc_order.channel_name or channel_name
        if not channel_id:
            raise ValueError("请选择渠道")
        if not channel_name:
            ch = (
                db.query(BizChannel)
                .filter(
                    BizChannel.id == channel_id,
                    BizChannel.tenant_no == tenant_no,
                    BizChannel.del_flag == 0,
                )
                .first()
            )
            if not ch:
                raise ValueError("渠道不存在")
            channel_name = ch.channel_name or ""

    _resolve_application_attachment_payload(data)

    now = datetime.now()
    application_no = _generate_application_no(db, tenant_no, biz_type=biz_type)

    row = BizPilotApplication(
        id=str(uuid.uuid4()),
        application_no=application_no,
        application_name=data.get("applicationName") or "",
        project_title=project_title or None,
        customer_name=customer_name or None,
        customer_contact=customer_contact or None,
        customer_phone=customer_phone or None,
        application_status="dks",
        channel_service_order_id=channel_service_order_id or "",
        channel_id=(
            (svc_order.channel_id or "" if svc_order else "")
            or channel_id
        ),
        channel_name=(
            (svc_order.channel_name or "" if svc_order else "")
            or channel_name
        ),
        tenant_no=tenant_no,
        order_num=0,
        remark=data.get("remark") or "",
        consent_form_urls=consent_form_urls,
        attachment_urls=data.get("attachmentUrls") or None,
        create_by=user_id,
        create_dept_id=dept_id or "",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
        biz_type=biz_type,
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return _row_to_dict_application(row, db, tenant_no)


def update_application(db: Session, app_id: str, data: dict, tenant_no: str) -> Dict[str, Any]:
    row = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == app_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("申请单不存在")
    if row.application_status not in ("dks", "jxz"):
        raise ValueError("当前状态不允许编辑")

    _resolve_application_attachment_payload(data)

    if "remark" in data and data["remark"] is not None:
        row.remark = data["remark"]
    if "applicationName" in data and data["applicationName"] is not None:
        row.application_name = data["applicationName"]
    if "projectTitle" in data and data["projectTitle"] is not None:
        row.project_title = (data["projectTitle"] or "").strip() or None
    if "customerName" in data and data["customerName"] is not None:
        row.customer_name = (data["customerName"] or "").strip() or None
    if "customerContact" in data and data["customerContact"] is not None:
        row.customer_contact = (data["customerContact"] or "").strip() or None
    if "customerPhone" in data and data["customerPhone"] is not None:
        row.customer_phone = (data["customerPhone"] or "").strip() or None
    if "consentFormUrls" in data:
        row.consent_form_urls = data.get("consentFormUrls") or None
    if "attachmentUrls" in data:
        row.attachment_urls = data.get("attachmentUrls") or None

    db.commit()
    db.refresh(row)
    return _row_to_dict_application(row, db, tenant_no)


def soft_delete_application(db: Session, app_id: str, tenant_no: str) -> None:
    row = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == app_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("申请单不存在")
    if row.application_status != "dks":
        raise ValueError("当前状态不允许删除")

    row.application_no = _release_application_no_for_soft_delete(row.application_no or "", row.id)
    row.del_flag = 1
    row.update_time = datetime.now()
    db.commit()


def start_application(db: Session, app_id: str, tenant_no: str, user_id: str) -> Dict[str, Any]:
    row = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == app_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("申请单不存在")
    if row.application_status != "dks":
        raise ValueError("当前状态不允许开始")

    row.application_status = "jxz"
    row.update_by = user_id
    row.update_time = datetime.now()
    db.commit()
    db.refresh(row)
    return _row_to_dict_application(row, db, tenant_no)


def complete_application(db: Session, app_id: str, tenant_no: str, user_id: str) -> Dict[str, Any]:
    row = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == app_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("申请单不存在")
    if row.application_status != "jxz":
        raise ValueError("当前状态不允许完成")

    row.application_status = "ywc"
    row.update_by = user_id
    row.update_time = datetime.now()
    db.commit()
    db.refresh(row)
    return _row_to_dict_application(row, db, tenant_no)


def void_application(db: Session, app_id: str, tenant_no: str, user_id: str) -> Dict[str, Any]:
    row = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == app_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("申请单不存在")
    if row.application_status not in ("dks", "jxz"):
        raise ValueError("当前状态不允许作废")

    row.application_status = "yzf"
    row.update_by = user_id
    row.update_time = datetime.now()
    db.commit()
    db.refresh(row)
    return _row_to_dict_application(row, db, tenant_no)


# ────────────────────────────────────────────
# Patient CRUD
# ────────────────────────────────────────────


def get_patient_by_application(
    db: Session, app_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    row = (
        db.query(BizPilotPatient)
        .filter(
            BizPilotPatient.pilot_application_id == app_id,
            BizPilotPatient.tenant_no == tenant_no,
            BizPilotPatient.del_flag == 0,
        )
        .first()
    )
    if not row:
        return None
    return _row_to_dict_patient(row)


def save_patient(
    db: Session,
    app_id: str,
    data: dict,
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> Dict[str, Any]:
    now = datetime.now()
    customer_id = data.get("customerId") or ""

    if customer_id:
        customer = (
            db.query(BizCustomer)
            .filter(BizCustomer.id == customer_id, BizCustomer.del_flag == 0)
            .first()
        )
        if customer:
            if data.get("patientName"):
                customer.customer_name = data["patientName"]
            if data.get("sex"):
                customer.sex = data["sex"]
            if data.get("age") is not None:
                customer.age = data["age"]
            if data.get("idCardNo"):
                customer.id_card_no = data["idCardNo"]
            if data.get("phoneNo"):
                customer.phone_no = data["phoneNo"]
            customer.update_by = user_id
            customer.update_time = now
    else:
        customer_id = str(uuid.uuid4())
        customer = BizCustomer(
            id=customer_id,
            customer_name=data.get("patientName") or "",
            sex=data.get("sex") or "2",
            id_card_no=data.get("idCardNo") or "",
            age=data.get("age"),
            phone_no=data.get("phoneNo") or "",
            tenant_no=tenant_no,
            order_num=0,
            remark="",
            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(customer)
        db.flush()

    existing = (
        db.query(BizPilotPatient)
        .filter(
            BizPilotPatient.pilot_application_id == app_id,
            BizPilotPatient.tenant_no == tenant_no,
            BizPilotPatient.del_flag == 0,
        )
        .first()
    )

    if existing:
        existing.customer_id = customer_id
        existing.patient_name = data.get("patientName") or ""
        existing.sex = data.get("sex") or ""
        existing.age = data.get("age")
        existing.id_card_no = data.get("idCardNo") or ""
        existing.phone_no = data.get("phoneNo") or ""
        existing.emergency_contact = data.get("emergencyContact") or ""
        existing.emergency_phone = data.get("emergencyPhone") or ""
        existing.clinical_diagnosis = data.get("clinicalDiagnosis") or ""
        existing.diagnosis_attachments = data.get("diagnosisAttachments") or ""
        existing.need_checkup = data.get("needCheckup") if data.get("needCheckup") is not None else 1
        existing.remark = data.get("remark") or ""
        existing.update_by = user_id
        existing.update_time = now
    else:
        existing = BizPilotPatient(
            id=str(uuid.uuid4()),
            pilot_application_id=app_id,
            customer_id=customer_id,
            patient_name=data.get("patientName") or "",
            sex=data.get("sex") or "",
            age=data.get("age"),
            id_card_no=data.get("idCardNo") or "",
            phone_no=data.get("phoneNo") or "",
            emergency_contact=data.get("emergencyContact") or "",
            emergency_phone=data.get("emergencyPhone") or "",
            clinical_diagnosis=data.get("clinicalDiagnosis") or "",
            diagnosis_attachments=data.get("diagnosisAttachments") or "",
            need_checkup=data.get("needCheckup") if data.get("needCheckup") is not None else 1,
            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(existing)

    db.commit()
    db.refresh(existing)

    need_checkup = data.get("needCheckup") if data.get("needCheckup") is not None else 1
    if need_checkup == 1:
        pilot_checkup.ensure_checkup_for_patient(
            db, patient_id=existing.id,
            application_id=app_id, tenant_no=tenant_no, user_id=user_id,
        )
    else:
        pilot_checkup.remove_checkup_for_patient(
            db, patient_id=existing.id, tenant_no=tenant_no,
        )
    db.commit()

    return _row_to_dict_patient(existing)


# ────────────────────────────────────────────
# InfusionPlan CRUD
# ────────────────────────────────────────────


def list_infusion_plans(
    db: Session, app_id: str, tenant_no: str
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.pilot_application_id == app_id,
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .order_by(BizPilotInfusionPlan.order_num.asc(), BizPilotInfusionPlan.create_time.asc())
        .all()
    )
    return [_row_to_dict_infusion_plan(r, db, tenant_no) for r in rows]


def create_infusion_plan(
    db: Session,
    app_id: str,
    data: dict,
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> Dict[str, Any]:
    now = datetime.now()
    row = BizPilotInfusionPlan(
        id=str(uuid.uuid4()),
        pilot_application_id=app_id,
        cell_type=data.get("cellType") or "",
        spec_text=data.get("specText") or "",
        total_course_count=data.get("totalCourseCount") or 0,
        course_interval=data.get("courseInterval") or "",
        infusion_route=data.get("infusionRoute") or "",
        infusion_start_date=_normalize_yyyymmdd(data.get("infusionStartDate")),
        channel_agreement_id=data.get("channelAgreementId") or "",
        tenant_no=tenant_no,
        order_num=data.get("orderNum") or 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.commit()
    db.refresh(row)
    return _row_to_dict_infusion_plan(row, db, tenant_no)


_INFUSION_STATUS_CN_TO_CODE = {
    "待输注": "dtz",
    "已用户确认": "yrqr",
    "已取消": "yqx",
    "已完成": "ywc",
}


def _normalize_infusion_status_code(raw: Any) -> str:
    s = (str(raw).strip() if raw is not None else "") or ""
    if not s:
        return ""
    return _INFUSION_STATUS_CN_TO_CODE.get(s, s)


def _schedule_infusion_editable(row: BizPilotInfusionSchedule) -> bool:
    return (row.infusion_status or "") in ("dtz", "yrqr")


def _schedule_locked(row: BizPilotInfusionSchedule) -> bool:
    return (row.infusion_status or "") in ("ywc", "yqx")


def _schedule_has_blocking_links(
    db: Session, schedule_id: str, tenant_no: str
) -> bool:
    delivery = (
        db.query(BizPilotDelivery.id)
        .filter(
            BizPilotDelivery.infusion_schedule_id == schedule_id,
            BizPilotDelivery.tenant_no == tenant_no,
            BizPilotDelivery.del_flag == 0,
        )
        .first()
    )
    if delivery:
        return True
    confirm = (
        db.query(BizPilotInfusionConfirm.id)
        .filter(
            BizPilotInfusionConfirm.infusion_schedule_id == schedule_id,
            BizPilotInfusionConfirm.tenant_no == tenant_no,
            BizPilotInfusionConfirm.del_flag == 0,
        )
        .first()
    )
    return confirm is not None


def _schedule_deletable(
    db: Session, row: BizPilotInfusionSchedule, tenant_no: str
) -> tuple[bool, str]:
    label = f"第{row.order_num or '?'}疗程"
    st = row.infusion_status or ""
    if st in ("ywc", "yqx"):
        return False, f"{label}已完成或已取消，不可删除"
    if st != "dtz":
        return False, f"{label}仅待输注状态可删除"
    if (row.schedule_status or "") != "dsc":
        return False, f"{label}生产已开始，不可删除"
    wo = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.infusion_schedule_id == row.id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
            BizProductionWorkOrder.work_order_type == "outbound",
        )
        .first()
    )
    if wo and (wo.work_order_status or "") not in ("dpz", "yzf"):
        return False, f"{label}关联出库工单进行中，不可删除"
    if _schedule_has_blocking_links(db, row.id, tenant_no):
        return False, f"{label}已有关联交付或确认记录，不可删除"
    return True, ""


def _get_schedule_for_plan(
    db: Session,
    schedule_id: str,
    plan_id: str,
    tenant_no: str,
) -> BizPilotInfusionSchedule:
    row = (
        db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.id == schedule_id,
            BizPilotInfusionSchedule.infusion_plan_id == plan_id,
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("输注计划不存在或不属于当前申请单")
    return row


def _prepare_infusion_plan_payload(
    plan_data: Optional[dict],
    schedules_data: Optional[List[dict]],
) -> dict:
    plan_payload = dict(plan_data or {})
    cell_type = plan_payload.get("cellType") or ""
    spec_text = plan_payload.get("specText") or ""
    if not cell_type and schedules_data:
        for sched in schedules_data:
            for cp in sched.get("cellProducts") or []:
                if cp.get("cellTypeName"):
                    cell_type = cp.get("cellTypeName")
                    spec_text = cp.get("specText") or spec_text
                    break
            if cell_type:
                break
    plan_payload.setdefault("cellType", cell_type or "-")
    plan_payload.setdefault("specText", spec_text)
    plan_payload.setdefault(
        "totalCourseCount",
        plan_payload.get("totalCourseCount") or len(schedules_data or []),
    )
    return plan_payload


def update_infusion_plan(
    db: Session, plan_id: str, data: dict, tenant_no: str, *, commit: bool = True
) -> Dict[str, Any]:
    row = (
        db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.id == plan_id,
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("输注方案不存在")

    fields = {
        "cellType": "cell_type",
        "specText": "spec_text",
        "totalCourseCount": "total_course_count",
        "courseInterval": "course_interval",
        "infusionRoute": "infusion_route",
        "infusionStartDate": "infusion_start_date",
        "channelAgreementId": "channel_agreement_id",
        "remark": "remark",
    }
    for camel, snake in fields.items():
        if camel == "infusionStartDate" and data[camel] is not None:
            setattr(row, snake, _normalize_yyyymmdd(data[camel]))
            continue
        if camel in data and data[camel] is not None:
            setattr(row, snake, data[camel])

    row.update_time = datetime.now()
    if commit:
        db.commit()
        db.refresh(row)
    else:
        db.flush()
    return _row_to_dict_infusion_plan(row, db, tenant_no)


def soft_delete_infusion_plan(db: Session, plan_id: str, tenant_no: str) -> None:
    now = datetime.now()
    row = (
        db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.id == plan_id,
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("输注方案不存在")

    row.del_flag = 1
    row.update_time = now

    schedule_ids = [
        r[0]
        for r in db.query(BizPilotInfusionSchedule.id)
        .filter(
            BizPilotInfusionSchedule.infusion_plan_id == plan_id,
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .all()
    ]
    if schedule_ids:
        db.query(BizPilotInfusionSchedule).filter(
            BizPilotInfusionSchedule.id.in_(schedule_ids),
        ).update({"del_flag": 1, "update_time": now}, synchronize_session=False)
        db.query(BizPilotInfusionScheduleProduct).filter(
            BizPilotInfusionScheduleProduct.infusion_schedule_id.in_(schedule_ids),
            BizPilotInfusionScheduleProduct.tenant_no == tenant_no,
            BizPilotInfusionScheduleProduct.del_flag == 0,
        ).update({"del_flag": 1, "update_time": now}, synchronize_session=False)

    db.commit()


# ────────────────────────────────────────────
# InfusionSchedule CRUD
# ────────────────────────────────────────────


def list_schedules_by_plan(
    db: Session, plan_id: str, tenant_no: str
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.infusion_plan_id == plan_id,
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .order_by(BizPilotInfusionSchedule.order_num.asc())
        .all()
    )
    if not rows:
        return []
    schedule_ids = [r.id for r in rows]
    product_rows = (
        db.query(BizPilotInfusionScheduleProduct)
        .filter(
            BizPilotInfusionScheduleProduct.infusion_schedule_id.in_(schedule_ids),
            BizPilotInfusionScheduleProduct.tenant_no == tenant_no,
            BizPilotInfusionScheduleProduct.del_flag == 0,
        )
        .order_by(BizPilotInfusionScheduleProduct.order_num.asc())
        .all()
    )
    products_by_schedule: Dict[str, List[Dict[str, Any]]] = {}
    for p in product_rows:
        products_by_schedule.setdefault(p.infusion_schedule_id, []).append(
            _row_to_dict_schedule_product(p)
        )
    return [
        _row_to_dict_infusion_schedule(r, products_by_schedule.get(r.id, []))
        for r in rows
    ]


def auto_generate_schedules(
    db: Session,
    plan_id: str,
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> Dict[str, Any]:
    plan = (
        db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.id == plan_id,
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .first()
    )
    if not plan:
        raise ValueError("输注方案不存在")

    # 追溯申请单获取正确的 biz_type
    _app = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == plan.pilot_application_id,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    schedule_biz_type = (_app.biz_type if _app else None) or "pilot_service"

    total = plan.total_course_count or 0
    if total <= 0:
        raise ValueError("总疗程数必须大于0")

    now = datetime.now()
    for i in range(1, total + 1):
        sched = BizPilotInfusionSchedule(
            id=str(uuid.uuid4()),
            infusion_plan_id=plan_id,
            cell_type=plan.cell_type or "",
            spec_text=plan.spec_text or "",
            schedule_status="dsc",
            infusion_status="dtz",
            qc_status="dzc",
            tenant_no=tenant_no,
            order_num=i,
            remark="",
            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(sched)
        from app.crud import production_work_order as pwo_svc

        pwo_svc.create_outbound_production_work_order(
            db,
            schedule_id=sched.id,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=dept_id,
            biz_type=schedule_biz_type,
        )
        from app.crud import qc_work_order as qc_svc

        qc_svc.create_outbound_qc_work_order(
            db,
            schedule_id=sched.id,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=dept_id,
            biz_type=schedule_biz_type,
        )

    db.commit()
    return {"generatedCount": total}


def _create_schedule_products(
    db: Session,
    schedule_id: str,
    products: List[dict],
    tenant_no: str,
    user_id: str,
    dept_id: str,
    now: datetime,
) -> List[Dict[str, Any]]:
    result: List[Dict[str, Any]] = []
    for idx, item in enumerate(products or []):
        if not item:
            continue
        row = BizPilotInfusionScheduleProduct(
            id=str(uuid.uuid4()),
            infusion_schedule_id=schedule_id,
            price_id=item.get("priceId") or "",
            project_code=item.get("projectCode") or "",
            project_name=item.get("projectName") or "",
            cell_type_name=item.get("cellTypeName") or "",
            spec_text=item.get("specText") or "",
            agreement_price=item.get("agreementPrice"),
            actual_price=item.get("actualPrice"),
            quantity=item.get("quantity") if item.get("quantity") is not None else 1,
            product_status=item.get("status") or item.get("productStatus") or "dtz",
            tenant_no=tenant_no,
            order_num=item.get("sortNo") or (idx + 1),
            remark=item.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)
        result.append(_row_to_dict_schedule_product(row))
    return result


def create_schedule(
    db: Session,
    plan_id: str,
    data: dict,
    tenant_no: str,
    user_id: str,
    dept_id: str,
    *,
    commit: bool = True,
) -> Dict[str, Any]:
    now = datetime.now()
    schedule_id = str(uuid.uuid4())
    cell_products = data.get("cellProducts") or []
    first_product = cell_products[0] if cell_products else {}

    # 追溯申请单获取正确的 biz_type
    _plan = (
        db.query(BizPilotInfusionPlan)
        .filter(BizPilotInfusionPlan.id == plan_id, BizPilotInfusionPlan.del_flag == 0)
        .first()
    )
    _app = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == (_plan.pilot_application_id if _plan else None),
            BizPilotApplication.del_flag == 0,
        )
        .first()
    ) if _plan else None
    schedule_biz_type = (_app.biz_type if _app else None) or "pilot_service"

    row = BizPilotInfusionSchedule(
        id=schedule_id,
        infusion_plan_id=plan_id,
        cell_type=data.get("cellType") or first_product.get("cellTypeName") or "",
        spec_text=data.get("specText") or first_product.get("specText") or "",
        infusion_time=_parse_infusion_datetime(data.get("infusionTime")),
        infusion_location=data.get("infusionLocation") or "",
        infusion_method=data.get("infusionMethod") or "",
        schedule_status=data.get("scheduleStatus") or "dsc",
        infusion_status=data.get("infusionStatus") or "dtz",
        qc_status=data.get("qcStatus") or "dzc",
        attachments=data.get("attachments") or "",
        amount=data.get("amount") if data.get("amount") else None,
        tenant_no=tenant_no,
        order_num=data.get("sortNo") or data.get("orderNum") or 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)
    products = _create_schedule_products(
        db, schedule_id, cell_products, tenant_no, user_id, dept_id, now
    )
    # 自动创建出库生产工单
    from app.crud import production_work_order as pwo_svc

    pwo_svc.create_outbound_production_work_order(
        db,
        schedule_id=schedule_id,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=dept_id,
        biz_type=schedule_biz_type,
    )
    # 自动创建出库质检工单
    from app.crud import qc_work_order as qc_svc

    qc_svc.create_outbound_qc_work_order(
        db,
        schedule_id=schedule_id,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=dept_id,
        biz_type=schedule_biz_type,
    )
    # 临床应用自动创建缴费记录
    plan_row = _plan
    if plan_row:
        application = _app
        if (
            application
            and application.biz_type in ("clinical_application", "basic_experiment")
            and data.get("amount")
        ):
            from app.crud.clinical_payment import auto_create_payment

            auto_create_payment(
                db,
                application_id=application.id,
                schedule_id=schedule_id,
                amount=data["amount"],
                tenant_no=tenant_no,
                user_id=user_id,
                dept_id=dept_id,
                biz_type=application.biz_type,
            )
    if commit:
        db.commit()
        db.refresh(row)
    return _row_to_dict_infusion_schedule(row, products)


def _update_schedule_editable(
    db: Session,
    row: BizPilotInfusionSchedule,
    data: dict,
    tenant_no: str,
    user_id: str,
) -> None:
    cell_products = data.get("cellProducts") or []
    first_product = cell_products[0] if cell_products else {}
    if data.get("infusionTime") is not None:
        row.infusion_time = _parse_infusion_datetime(data.get("infusionTime"))
    if data.get("infusionLocation") is not None:
        row.infusion_location = data.get("infusionLocation") or ""
    if data.get("infusionMethod") is not None:
        row.infusion_method = data.get("infusionMethod") or ""
    if data.get("sortNo") is not None or data.get("orderNum") is not None:
        row.order_num = data.get("sortNo") or data.get("orderNum") or row.order_num
    if cell_products:
        row.cell_type = data.get("cellType") or first_product.get("cellTypeName") or row.cell_type
        row.spec_text = data.get("specText") or first_product.get("specText") or row.spec_text
    if data.get("remark") is not None:
        row.remark = data.get("remark") or ""
    row.update_by = user_id
    row.update_time = datetime.now()


def _upsert_schedule_products(
    db: Session,
    schedule_row: BizPilotInfusionSchedule,
    products: List[dict],
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> None:
    now = datetime.now()
    schedule_id = schedule_row.id
    existing_rows = (
        db.query(BizPilotInfusionScheduleProduct)
        .filter(
            BizPilotInfusionScheduleProduct.infusion_schedule_id == schedule_id,
            BizPilotInfusionScheduleProduct.tenant_no == tenant_no,
            BizPilotInfusionScheduleProduct.del_flag == 0,
        )
        .all()
    )
    existing_by_id = {r.id: r for r in existing_rows}
    keep_ids: set[str] = set()

    default_status = schedule_row.infusion_status or "dtz"

    for idx, raw in enumerate(products or []):
        if not raw:
            continue
        item = dict(raw)
        pid = (item.get("id") or "").strip()
        if pid and pid in existing_by_id:
            prow = existing_by_id[pid]
            keep_ids.add(pid)
            prow.price_id = item.get("priceId") or ""
            prow.project_code = item.get("projectCode") or ""
            prow.project_name = item.get("projectName") or ""
            prow.cell_type_name = item.get("cellTypeName") or ""
            prow.spec_text = item.get("specText") or ""
            if item.get("agreementPrice") is not None:
                prow.agreement_price = item.get("agreementPrice")
            if item.get("actualPrice") is not None:
                prow.actual_price = item.get("actualPrice")
            if item.get("quantity") is not None:
                prow.quantity = item.get("quantity")
            prow.order_num = item.get("sortNo") or (idx + 1)
            prow.remark = item.get("remark") or ""
            prow.update_by = user_id
            prow.update_time = now
            continue

        status_raw = item.get("status") or item.get("productStatus")
        product_status = _normalize_infusion_status_code(status_raw) or default_status
        row = BizPilotInfusionScheduleProduct(
            id=str(uuid.uuid4()),
            infusion_schedule_id=schedule_id,
            price_id=item.get("priceId") or "",
            project_code=item.get("projectCode") or "",
            project_name=item.get("projectName") or "",
            cell_type_name=item.get("cellTypeName") or "",
            spec_text=item.get("specText") or "",
            agreement_price=item.get("agreementPrice"),
            actual_price=item.get("actualPrice"),
            quantity=item.get("quantity") if item.get("quantity") is not None else 1,
            product_status=product_status,
            tenant_no=tenant_no,
            order_num=item.get("sortNo") or (idx + 1),
            remark=item.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)

    for eid, erow in existing_by_id.items():
        if eid not in keep_ids:
            erow.del_flag = 1
            erow.update_by = user_id
            erow.update_time = now


def _update_schedule_order_only(
    db: Session,
    row: BizPilotInfusionSchedule,
    data: dict,
    user_id: str,
) -> None:
    if data.get("sortNo") is not None or data.get("orderNum") is not None:
        row.order_num = data.get("sortNo") or data.get("orderNum") or row.order_num
        row.update_by = user_id
        row.update_time = datetime.now()


def upsert_application_infusion(
    db: Session,
    app_id: str,
    plan_data: Optional[dict],
    schedules_data: Optional[List[dict]],
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> None:
    """编辑申请单：增量更新输注方案/计划，保留状态与 schedule id。"""
    if not plan_data and not schedules_data:
        return

    plan_payload = _prepare_infusion_plan_payload(plan_data, schedules_data)
    plans = list_infusion_plans(db, app_id, tenant_no)
    plan_id = (plans[0].get("id") if plans else None) or ""

    if plan_id:
        update_infusion_plan(
            db, plan_id, plan_payload, tenant_no, commit=False
        )
    else:
        plan_result = create_infusion_plan(
            db,
            app_id=app_id,
            data=plan_payload,
            tenant_no=tenant_no,
            user_id=user_id,
            dept_id=dept_id,
        )
        plan_id = plan_result.get("id") or ""
        if not plan_id:
            raise ValueError("创建输注方案失败")

    existing_schedules = list_schedules_by_plan(db, plan_id, tenant_no)
    existing_ids = {s["id"] for s in existing_schedules if s.get("id")}
    payload_ids: set[str] = set()

    for idx, sched in enumerate(schedules_data or []):
        sched_data = dict(sched)
        sched_data["sortNo"] = sched_data.get("sortNo") or (idx + 1)
        sid = (sched_data.get("id") or "").strip()
        label = f"第{idx + 1}疗程"

        if sid:
            if sid not in existing_ids:
                raise ValueError(f"{label}输注计划不存在或不属于当前申请单")
            row = _get_schedule_for_plan(db, sid, plan_id, tenant_no)
            payload_ids.add(sid)
            if _schedule_locked(row):
                _update_schedule_order_only(db, row, sched_data, user_id)
                continue
            if not _schedule_infusion_editable(row):
                raise ValueError(f"{label}当前状态不允许编辑")
            _update_schedule_editable(db, row, sched_data, tenant_no, user_id)
            _upsert_schedule_products(
                db, row, sched_data.get("cellProducts") or [], tenant_no, user_id, dept_id
            )
            continue

        create_schedule(
            db,
            plan_id=plan_id,
            data=sched_data,
            tenant_no=tenant_no,
            user_id=user_id,
            dept_id=dept_id,
            commit=False,
        )

    for sid in existing_ids - payload_ids:
        row = _get_schedule_for_plan(db, sid, plan_id, tenant_no)
        ok, msg = _schedule_deletable(db, row, tenant_no)
        if not ok:
            raise ValueError(msg)
        soft_delete_schedule(db, sid, tenant_no, commit=False)

    plan_row = (
        db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.id == plan_id,
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .first()
    )
    if plan_row:
        active_count = (
            db.query(func.count(BizPilotInfusionSchedule.id))
            .filter(
                BizPilotInfusionSchedule.infusion_plan_id == plan_id,
                BizPilotInfusionSchedule.tenant_no == tenant_no,
                BizPilotInfusionSchedule.del_flag == 0,
            )
            .scalar()
        ) or 0
        plan_row.total_course_count = int(active_count)
        plan_row.update_by = user_id
        plan_row.update_time = datetime.now()

    db.commit()


def _get_or_create_infusion_plan(
    db: Session,
    application_id: str,
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> BizPilotInfusionPlan:
    """获取或创建申请单下的输注/售卖方案（基础实验批量创建售卖计划用）。"""
    plans = list_infusion_plans(db, application_id, tenant_no)
    if plans:
        plan_row = (
            db.query(BizPilotInfusionPlan)
            .filter(
                BizPilotInfusionPlan.id == plans[0].get("id"),
                BizPilotInfusionPlan.tenant_no == tenant_no,
                BizPilotInfusionPlan.del_flag == 0,
            )
            .first()
        )
        if plan_row:
            return plan_row

    plan_result = create_infusion_plan(
        db,
        app_id=application_id,
        data={"cellType": "-", "totalCourseCount": 0},
        tenant_no=tenant_no,
        user_id=user_id,
        dept_id=dept_id,
    )
    plan_id = plan_result.get("id") or ""
    plan_row = (
        db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.id == plan_id,
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .first()
    )
    if not plan_row:
        raise ValueError("创建售卖方案失败")
    return plan_row


def create_basic_experiment_sale_schedules(
    db: Session,
    application_id: str,
    product_lines: list,
    tenant_no: str,
    user_id: str,
    dept_id: str,
) -> List[Dict[str, Any]]:
    """基础实验：每个产品行创建一条售卖计划（一产品一计划）。"""
    app = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == application_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    if not app:
        raise ValueError("申请单不存在")
    if (app.biz_type or "") != "basic_experiment":
        raise ValueError("仅基础实验申请单可批量创建售卖计划")
    if not product_lines:
        raise ValueError("请至少添加一行售卖产品")

    plan = _get_or_create_infusion_plan(
        db, application_id, tenant_no, user_id, dept_id
    )
    results: List[Dict[str, Any]] = []
    for idx, line in enumerate(product_lines):
        schedule_data = {
            "cellType": line.get("cellTypeName") or "",
            "specText": line.get("specText") or "",
            "infusionTime": line.get("pickupTime"),
            "amount": line.get("amount"),
            "sortNo": idx + 1,
            "cellProducts": [
                {
                    "priceId": line.get("priceId"),
                    "projectCode": line.get("projectCode"),
                    "projectName": line.get("projectName"),
                    "cellTypeName": line.get("cellTypeName"),
                    "specText": line.get("specText"),
                    "agreementPrice": line.get("agreementPrice"),
                    "actualPrice": line.get("actualPrice"),
                    "quantity": line.get("quantity") or 1,
                }
            ],
        }
        results.append(
            create_schedule(
                db,
                plan.id,
                schedule_data,
                tenant_no,
                user_id,
                dept_id,
                commit=False,
            )
        )

    active_count = (
        db.query(func.count(BizPilotInfusionSchedule.id))
        .filter(
            BizPilotInfusionSchedule.infusion_plan_id == plan.id,
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .scalar()
    ) or 0
    plan.total_course_count = int(active_count)
    plan.update_by = user_id
    plan.update_time = datetime.now()
    db.commit()
    return results


def save_application_infusion(
    db: Session,
    app_id: str,
    plan_data: Optional[dict],
    schedules_data: Optional[List[dict]],
    tenant_no: str,
    user_id: str,
    dept_id: str,
    *,
    replace_existing: bool = False,
) -> None:
    if replace_existing:
        for plan in list_infusion_plans(db, app_id, tenant_no):
            plan_id = plan.get("id")
            if plan_id:
                soft_delete_infusion_plan(db, plan_id, tenant_no)

    if not plan_data and not schedules_data:
        return

    plan_payload = _prepare_infusion_plan_payload(plan_data, schedules_data)

    plan_result = create_infusion_plan(
        db,
        app_id=app_id,
        data=plan_payload,
        tenant_no=tenant_no,
        user_id=user_id,
        dept_id=dept_id,
    )
    plan_id = plan_result.get("id")
    if not plan_id:
        raise ValueError("创建输注方案失败")

    for idx, sched in enumerate(schedules_data or []):
        sched_data = dict(sched)
        sched_data["sortNo"] = sched_data.get("sortNo") or (idx + 1)
        create_schedule(
            db,
            plan_id=plan_id,
            data=sched_data,
            tenant_no=tenant_no,
            user_id=user_id,
            dept_id=dept_id,
            commit=False,
        )
    db.commit()


def update_schedule(
    db: Session, schedule_id: str, data: dict, tenant_no: 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("输注计划不存在")

    if "infusionTime" in data and data["infusionTime"] is not None:
        row.infusion_time = data["infusionTime"]
    if "remark" in data and data["remark"] is not None:
        row.remark = data["remark"]
    if "attachments" in data and data["attachments"] is not None:
        row.attachments = data["attachments"]

    row.update_time = datetime.now()
    db.commit()
    db.refresh(row)
    return _row_to_dict_infusion_schedule(row)


def soft_delete_schedule(
    db: Session, schedule_id: str, tenant_no: str, *, commit: bool = True
) -> None:
    row = (
        db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.id == schedule_id,
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .first()
    )
    if not row:
        raise ValueError("输注计划不存在")

    now = datetime.now()
    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",
        )
        .first()
    )
    row.schedule_status = "yqx"
    row.qc_status = "yqx"

    if outbound_wo:
        outbound_wo.work_order_status = "yzf"
        outbound_wo.update_time = now
        from app.crud.pilot_outbound_status_sync import sync_schedule_from_production_work_order

        sync_schedule_from_production_work_order(db, outbound_wo)

    for qc_wo in (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.infusion_schedule_id == schedule_id,
            BizQcWorkOrder.work_order_status.notin_(["ywc", "yzf"]),
            BizQcWorkOrder.del_flag == 0,
        )
        .all()
    ):
        qc_wo.work_order_status = "yzf"
        qc_wo.update_time = now
        from app.crud.pilot_outbound_status_sync import sync_schedule_from_qc_work_order

        sync_schedule_from_qc_work_order(db, qc_wo)

    db.query(BizPilotInfusionScheduleProduct).filter(
        BizPilotInfusionScheduleProduct.infusion_schedule_id == schedule_id,
        BizPilotInfusionScheduleProduct.tenant_no == tenant_no,
        BizPilotInfusionScheduleProduct.del_flag == 0,
    ).update({"del_flag": 1, "update_time": now}, synchronize_session=False)

    row.del_flag = 1
    row.update_time = now
    if commit:
        db.commit()
    else:
        db.flush()


def mark_schedule_complete(
    db: Session, schedule_id: 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("输注计划不存在")

    row.schedule_status = "ysc"
    row.update_by = user_id
    row.update_time = datetime.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 _row_to_dict_infusion_schedule(row)