zkxw/backend/app/crud/production_work_order.py

"""入库生产工单 CRUD（含临时生产工单）"""
from __future__ import annotations

from collections import defaultdict
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

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

from app.core.db_collation import eq_unicode, like_unicode
from app.core.protocol_status_display import INACTIVE_PROTOCOL_STATUSES
from app.core.utils import generate_uuid
from app.crud.collection_protocol import urls_to_attachment_list
from app.crud import sample_reception as sample_reception_svc
from app.crud import sample_reception_check_result as check_result_svc
from app.crud import production_task as production_task_svc
from app.models.models import (
    BizCollectionProtocol,
    BizProductionTask,
    BizProductionTaskInput,
    BizProductionTaskOutput,
    BizProductionTaskStep,
    BizProductionTaskStepAssignee,
    BizProductionTaskStepConsumable,
    BizProductionTaskStepDataCollection,
    BizProductionTaskStepOutput,
    BizProductionTaskStepReagent,
    BizProductionWorkOrder,
    BizPilotInfusionSchedule,
    BizPilotInfusionScheduleProduct,
    BizProtocolSample,
    BizSampleReception,
)
from app.models.research_donation import BizResearchDonation, BizResearchDonationSample
from app.schemas.production_work_order import ProductionWorkOrderComplete, ProductionWorkOrderListQuery
from app.services.qc_work_order_fee_summary import enrich_work_order_fee_summaries

_WORK_ORDER_STATUS_LABELS = {
    "dpz": "待配置",
    "jxz": "进行中",
    "ywc": "已完成",
    "yzf": "已作废",
}
_RECEPTION_SOURCE_LABELS = {
    "cell_storage": "细胞存储",
    "research_donation": "科研捐赠",
    "outbound": "出库生产",
}
_ACCEPT_CONCLUSION_LABELS = {"js": "接收", "bj": "拒收"}


def apply_exclude_inactive_protocol_filter(query, *, exclude: bool = True):
    """排除关联采集协议/捐赠协议处于无效态（已作废、属地废弃）的入库生产工单。"""
    if not exclude:
        return query
    inactive_cell_storage = and_(
        BizProductionWorkOrder.work_order_type == "inbound",
        BizProductionWorkOrder.reception_source == "cell_storage",
        exists().where(
            and_(
                eq_unicode(
                    BizCollectionProtocol.id,
                    BizProductionWorkOrder.collection_protocol_id,
                ),
                BizCollectionProtocol.del_flag == 0,
                BizCollectionProtocol.protocol_status.in_(INACTIVE_PROTOCOL_STATUSES),
            )
        ),
    )
    inactive_research_donation = and_(
        BizProductionWorkOrder.work_order_type == "inbound",
        BizProductionWorkOrder.reception_source == "research_donation",
        exists().where(
            and_(
                eq_unicode(
                    BizResearchDonation.id,
                    BizProductionWorkOrder.collection_protocol_id,
                ),
                BizResearchDonation.del_flag == 0,
                BizResearchDonation.donation_status.in_(INACTIVE_PROTOCOL_STATUSES),
            )
        ),
    )
    return query.filter(~or_(inactive_cell_storage, inactive_research_donation))


def apply_planned_start_time_step_filter(
    query,
    *,
    tenant_no: str,
    work_order_id_column,
    planned_start_time_start: Optional[str] = None,
    planned_start_time_end: Optional[str] = None,
    mine_user_id: Optional[str] = None,
):
    """按活跃工单/任务下未完成工艺步骤的计划开始时间过滤。"""
    start = (planned_start_time_start or "").strip()
    end = (planned_start_time_end or "").strip()
    if not start and not end:
        return query

    step_conditions = [
        BizProductionTaskStep.del_flag == 0,
        BizProductionTaskStep.tenant_no == tenant_no,
        BizProductionTaskStep.planned_start_time.isnot(None),
        BizProductionTaskStep.step_status != "ywc",
        BizProductionTask.id == BizProductionTaskStep.task_id,
        BizProductionTask.del_flag == 0,
        BizProductionTask.tenant_no == tenant_no,
        BizProductionTask.task_status.in_(("dpz", "jxz")),
        BizProductionWorkOrder.id == BizProductionTask.work_order_id,
        BizProductionWorkOrder.del_flag == 0,
        BizProductionWorkOrder.tenant_no == tenant_no,
        BizProductionWorkOrder.work_order_status.in_(("dpz", "jxz")),
        work_order_id_column == BizProductionTask.work_order_id,
    ]
    if start:
        step_conditions.append(
            BizProductionTaskStep.planned_start_time >= _normalize_date_bound(start)
        )
    if end:
        step_conditions.append(
            BizProductionTaskStep.planned_start_time
            <= _normalize_date_bound(end, end_of_day=True)
        )

    uid = (mine_user_id or "").strip()
    if uid:
        assignee_match = exists().where(
            and_(
                BizProductionTaskStepAssignee.step_id == BizProductionTaskStep.id,
                BizProductionTaskStepAssignee.user_id == uid,
                BizProductionTaskStepAssignee.tenant_no == tenant_no,
                BizProductionTaskStepAssignee.del_flag == 0,
            )
        )
        step_conditions.append(
            or_(assignee_match, BizProductionTask.current_assignee_user_id == uid)
        )

    return query.filter(exists().where(and_(*step_conditions)))


def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
    val = (value or "").strip()
    if len(val) == 10:
        return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
    return val


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _soft_delete_sub(db: Session, model_class, parent_field: str, parent_id: str, now: datetime) -> None:
    rows = db.query(model_class).filter(
        and_(
            getattr(model_class, parent_field) == parent_id,
            model_class.del_flag == 0,
        )
    ).all()
    for r in rows:
        r.del_flag = 1
        r.update_time = now


def _generate_work_order_no(db: Session, tenant_no: str) -> str:
    today_str = datetime.now().strftime("%Y%m%d")
    prefix = f"SC{today_str}"
    last = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.work_order_no.like(f"{prefix}%"),
            BizProductionWorkOrder.del_flag == 0,
        )
        .order_by(BizProductionWorkOrder.work_order_no.desc())
        .first()
    )
    seq = 1
    if last:
        seq = int(last.work_order_no[-4:]) + 1
    return f"{prefix}{seq:04d}"


def _wo_to_dict(wo: BizProductionWorkOrder) -> Dict[str, Any]:
    return {
        "id": wo.id,
        "workOrderNo": wo.work_order_no,
        "workOrderType": getattr(wo, "work_order_type", None) or "inbound",
        "receptionId": wo.reception_id,
        "protocolSampleId": wo.protocol_sample_id,
        "collectionProtocolId": wo.collection_protocol_id,
        "protocolNo": wo.protocol_no or "",
        "sampleCode": wo.sample_code or "",
        "receptionSource": getattr(wo, "reception_source", "cell_storage") or "cell_storage",
        "receptionSourceLabel": _RECEPTION_SOURCE_LABELS.get(
            getattr(wo, "reception_source", "cell_storage") or "cell_storage",
            getattr(wo, "reception_source", "cell_storage") or "cell_storage",
        ),
        "infusionScheduleId": getattr(wo, "infusion_schedule_id", None) or None,
        "bizType": getattr(wo, "biz_type", None) or None,
        "bizTypeLabel": {
            "pilot_service": "中试服务",
            "clinical_application": "临床应用",
            "basic_experiment": "基础实验",
        }.get(getattr(wo, "biz_type", None) or "", ""),
        "workOrderStatus": wo.work_order_status,
        "workOrderStatusLabel": _WORK_ORDER_STATUS_LABELS.get(wo.work_order_status, wo.work_order_status),
        "qcConclusion": "",
        "qcConclusionLabel": "",
        "receptionTime": "",
        "handlerName": "",
        "sampleTypeName": "",
        "remark": wo.remark or "",
        "createTime": _fmt_dt(wo.create_time),
        "updateTime": _fmt_dt(wo.update_time),
    }


def _enrich_wo_items(db: Session, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    from app.crud import qc_work_order as qc_wo_svc

    return qc_wo_svc._enrich_work_order_list_items(db, items)


def _enrich_inbound_qc_conclusions(
    db: Session, tenant_no: str, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """入库（细胞存储/科研捐赠）生产工单列表补充门禁质检结论。"""
    if not items or not (tenant_no or "").strip():
        return items
    from collections import defaultdict

    from app.services.protocol_sample_list_summary import (
        get_list_qc_conclusion_map_by_sample_codes,
    )

    codes_by_source: Dict[str, List[str]] = defaultdict(list)
    for item in items:
        wo_type = (item.get("workOrderType") or "inbound").strip() or "inbound"
        source = (item.get("receptionSource") or "").strip()
        code = (item.get("sampleCode") or "").strip()
        if wo_type != "inbound" or source not in ("cell_storage", "research_donation") or not code:
            item["qcConclusion"] = item.get("qcConclusion") or ""
            item["qcConclusionLabel"] = item.get("qcConclusionLabel") or ""
            continue
        codes_by_source[source].append(code)

    conclusion_maps: Dict[str, Dict[str, Dict[str, str]]] = {}
    for source, codes in codes_by_source.items():
        conclusion_maps[source] = get_list_qc_conclusion_map_by_sample_codes(
            db, tenant_no, codes, reception_source=source
        )

    for item in items:
        wo_type = (item.get("workOrderType") or "inbound").strip() or "inbound"
        source = (item.get("receptionSource") or "").strip()
        code = (item.get("sampleCode") or "").strip()
        if wo_type != "inbound" or source not in ("cell_storage", "research_donation") or not code:
            item["qcConclusion"] = ""
            item["qcConclusionLabel"] = ""
            continue
        fields = conclusion_maps.get(source, {}).get(code) or {}
        item["qcConclusion"] = fields.get("qcConclusion") or ""
        item["qcConclusionLabel"] = fields.get("qcConclusionLabel") or ""
    return items


_BIZ_TYPE_LABELS = {
    "pilot_service": "中试服务",
    "clinical_application": "临床应用",
    "basic_experiment": "基础实验",
}


def _dec_price(v: Any) -> Optional[float]:
    if v is None:
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def _enrich_outbound_items(
    db: Session, tenant_no: str, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """为 outbound 类型工单附加申请单号、用户姓名、疗程序号、细胞类型、规格、输注时间等关联数据。"""
    from app.crud.pilot_infusion_schedule_mgmt import (
        _get_plan_schedule_course_meta,
        _resolve_course_display,
    )
    from app.models.models import (
        BizPilotApplication,
        BizPilotInfusionPlan,
        BizPilotInfusionSchedule,
        BizPilotPatient,
    )

    outbound_ids = [
        it["infusionScheduleId"]
        for it in items
        if it.get("workOrderType") == "outbound" and it.get("infusionScheduleId")
    ]
    if not outbound_ids:
        return items

    products_map: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
    for product in (
        db.query(BizPilotInfusionScheduleProduct)
        .filter(
            BizPilotInfusionScheduleProduct.infusion_schedule_id.in_(outbound_ids),
            BizPilotInfusionScheduleProduct.tenant_no == tenant_no,
            BizPilotInfusionScheduleProduct.del_flag == 0,
        )
        .order_by(BizPilotInfusionScheduleProduct.order_num.asc())
        .all()
    ):
        products_map[product.infusion_schedule_id].append(
            {
                "projectCode": product.project_code or "",
                "projectName": product.project_name or "",
                "cellTypeName": product.cell_type_name or "",
                "specText": product.spec_text or "",
                "agreementPrice": _dec_price(product.agreement_price),
                "actualPrice": _dec_price(product.actual_price),
                "quantity": product.quantity,
                "productStatus": product.product_status or "",
            }
        )

    schedules = (
        db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.id.in_(outbound_ids),
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .all()
    )
    sched_map = {s.id: s for s in schedules}

    plan_ids = list({s.infusion_plan_id for s in schedules if s.infusion_plan_id})
    plan_map = {}
    if plan_ids:
        plans = (
            db.query(BizPilotInfusionPlan)
            .filter(
                BizPilotInfusionPlan.id.in_(plan_ids),
                BizPilotInfusionPlan.tenant_no == tenant_no,
                BizPilotInfusionPlan.del_flag == 0,
            )
            .all()
        )
        plan_map = {p.id: p for p in plans}

    app_ids = list({p.pilot_application_id for p in plan_map.values() if p.pilot_application_id})
    app_map = {}
    patient_map: Dict[str, BizPilotPatient] = {}
    if app_ids:
        apps = (
            db.query(BizPilotApplication)
            .filter(
                BizPilotApplication.id.in_(app_ids),
                BizPilotApplication.tenant_no == tenant_no,
                BizPilotApplication.del_flag == 0,
            )
            .all()
        )
        app_map = {a.id: a for a in apps}
        patients = (
            db.query(BizPilotPatient)
            .filter(
                BizPilotPatient.pilot_application_id.in_(app_ids),
                BizPilotPatient.tenant_no == tenant_no,
                BizPilotPatient.del_flag == 0,
            )
            .all()
        )
        patient_map = {p.pilot_application_id: p for p in patients}

    plan_course_meta = _get_plan_schedule_course_meta(db, plan_ids, tenant_no)

    for it in items:
        if it.get("workOrderType") != "outbound" or not it.get("infusionScheduleId"):
            continue
        sched = sched_map.get(it["infusionScheduleId"])
        if not sched:
            continue
        plan = plan_map.get(sched.infusion_plan_id) if sched.infusion_plan_id else None
        app = app_map.get(plan.pilot_application_id) if plan and plan.pilot_application_id else None
        patient = (
            patient_map.get(plan.pilot_application_id)
            if plan and plan.pilot_application_id
            else None
        )
        plan_meta = (
            plan_course_meta.get(sched.infusion_plan_id) if sched.infusion_plan_id else None
        )
        course = _resolve_course_display(sched, plan, plan_meta)

        it["applicationNo"] = app.application_no if app else ""
        it["applicationName"] = app.application_name if app else ""
        it["channelName"] = app.channel_name if app else ""
        if app:
            it["projectTitle"] = app.project_title or ""
            it["customerName"] = app.customer_name or ""
        it["patientName"] = patient.patient_name if patient else ""
        it["sortNo"] = course.get("sortNo")
        it["totalCourseCount"] = course.get("totalCourseCount")
        it["cellType"] = sched.cell_type or ""
        it["specText"] = sched.spec_text or ""
        it["infusionTime"] = _fmt_dt(sched.infusion_time) if sched.infusion_time else ""
        it["infusionLocation"] = sched.infusion_location or ""
        it["products"] = products_map.get(it["infusionScheduleId"], [])

    return items


def _outbound_work_order_ids_by_keyword(db: Session, tenant_no: str, keyword: str) -> List[str]:
    """出库工单关键词：按关联申请单号、申请单名称匹配。"""
    from app.models.models import (
        BizPilotApplication,
        BizPilotInfusionPlan,
        BizPilotInfusionSchedule,
    )

    kw = (keyword or "").strip()
    if not kw:
        return []
    like = f"%{kw}%"
    rows = (
        db.query(BizProductionWorkOrder.id)
        .join(
            BizPilotInfusionSchedule,
            eq_unicode(
                BizProductionWorkOrder.infusion_schedule_id,
                BizPilotInfusionSchedule.id,
            ),
        )
        .join(
            BizPilotInfusionPlan,
            eq_unicode(
                BizPilotInfusionSchedule.infusion_plan_id,
                BizPilotInfusionPlan.id,
            ),
        )
        .join(
            BizPilotApplication,
            eq_unicode(
                BizPilotInfusionPlan.pilot_application_id,
                BizPilotApplication.id,
            ),
        )
        .filter(
            BizProductionWorkOrder.del_flag == 0,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.work_order_type == "outbound",
            BizPilotInfusionSchedule.del_flag == 0,
            BizPilotInfusionPlan.del_flag == 0,
            BizPilotApplication.del_flag == 0,
            or_(
                like_unicode(BizPilotApplication.application_no, like),
                like_unicode(BizPilotApplication.application_name, like),
                like_unicode(BizPilotApplication.project_title, like),
                like_unicode(BizPilotApplication.customer_name, like),
            ),
        )
        .all()
    )
    return [r[0] for r in rows]


def _reception_source_label(source: Optional[str]) -> str:
    key = (source or "").strip() or "cell_storage"
    return _RECEPTION_SOURCE_LABELS.get(key, key)


def _accept_conclusion_label(v: Optional[str]) -> str:
    key = (v or "").strip()
    return _ACCEPT_CONCLUSION_LABELS.get(key, key)


def _attach_reception_detail(
    db: Session, wo: BizProductionWorkOrder, out: Dict[str, Any], tenant_no: str
) -> None:
    """详情响应附加运输与样本接收信息（与质检工单详情一致）。"""
    reception_detail = sample_reception_svc.get_reception_by_id(db, wo.reception_id, tenant_no)
    if reception_detail:
        out["reception"] = reception_detail
        out["productionReception"] = reception_detail.get("productionReception")
        return
    rec = (
        db.query(BizSampleReception)
        .filter(BizSampleReception.id == wo.reception_id, BizSampleReception.del_flag == 0)
        .first()
    )
    cr = check_result_svc.get_check_results_for_reception(db, rec, tenant_no) if rec else []
    out["reception"] = {
        "receptionTime": _fmt_dt(rec.reception_time) if rec else "",
        "handlerName": (rec.handler_name if rec else "") or "",
        "acceptConclusion": (rec.accept_conclusion if rec else "") or "",
        "acceptConclusionLabel": _accept_conclusion_label(rec.accept_conclusion if rec else ""),
        "sourceType": wo.reception_source or "cell_storage",
        "sourceTypeLabel": _reception_source_label(wo.reception_source),
        "sampleTypeName": out.get("sampleTypeName", ""),
        "checkResults": cr,
        "receiveAttachments": (rec.receive_attachments if rec else "") or "",
        "receiveAttachmentFiles": urls_to_attachment_list(
            rec.receive_attachments if rec else None, db, tenant_no
        ),
    }


def _attach_outbound_detail(
    db: Session, wo: BizProductionWorkOrder, out: Dict[str, Any]
) -> None:
    """outbound 工单详情附加申请单 + 输注计划信息。"""
    from app.crud.qc_work_order import build_outbound_pilot_context_by_schedule

    out["reception"] = {}
    schedule_id = getattr(wo, "infusion_schedule_id", None) or ""
    biz_type = getattr(wo, "biz_type", None) or ""

    if schedule_id and biz_type in ("pilot_service", "clinical_application", "basic_experiment"):
        pilot_ctx = build_outbound_pilot_context_by_schedule(
            db, schedule_id, wo.tenant_no or ""
        )
        out.update(pilot_ctx)
        return

    from app.models.models import (
        BizPilotApplication,
        BizPilotInfusionPlan,
        BizPilotInfusionSchedule,
    )

    _APP_STATUS_LABELS = {"dks": "待开始", "jxz": "进行中", "ywc": "已完成", "yzf": "已作废"}

    sched = (
        db.query(BizPilotInfusionSchedule)
        .filter(BizPilotInfusionSchedule.id == schedule_id, BizPilotInfusionSchedule.del_flag == 0)
        .first()
    )
    plan = None
    app = None
    if sched and sched.infusion_plan_id:
        plan = db.query(BizPilotInfusionPlan).filter(BizPilotInfusionPlan.id == sched.infusion_plan_id).first()
    if plan and plan.pilot_application_id:
        app = db.query(BizPilotApplication).filter(BizPilotApplication.id == plan.pilot_application_id).first()

    out["pilotApplication"] = {
        "applicationNo": app.application_no if app else "",
        "applicationName": app.application_name if app else "",
        "applicationStatus": app.application_status if app else "",
        "applicationStatusLabel": _APP_STATUS_LABELS.get(app.application_status, "") if app else "",
    } if app else {}
    out["infusionSchedule"] = {
        "cellType": sched.cell_type if sched else "",
        "specText": sched.spec_text if sched else "",
        "infusionTime": _fmt_dt(sched.infusion_time) if sched and sched.infusion_time else "",
        "infusionLocation": sched.infusion_location if sched else "",
        "infusionMethod": sched.infusion_method if sched else "",
        "scheduleStatus": sched.schedule_status if sched else "",
    } if sched else {}


def _wo_to_response_dict(db: Session, wo: BizProductionWorkOrder) -> Dict[str, Any]:
    """工单响应（含接收记录关联的样本类型、接收人、接收时间及 reception 详情）。"""
    out = _enrich_wo_items(db, [_wo_to_dict(wo)])[0]
    tenant_no = (wo.tenant_no or "").strip()
    if tenant_no:
        out = _enrich_inbound_qc_conclusions(db, tenant_no, [out])[0]
    if wo.work_order_type == "outbound" and wo.infusion_schedule_id:
        _attach_outbound_detail(db, wo, out)
    else:
        _attach_reception_detail(db, wo, out, wo.tenant_no or "")
    if tenant_no:
        out = enrich_work_order_fee_summaries(db, [out], tenant_no)[0]
    return out


def create_production_work_order_from_reception(
    db: Session,
    reception: BizSampleReception,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> Optional[BizProductionWorkOrder]:
    """接收结论为 js 时创建待配置入库生产工单（flush，不 commit）。"""
    if (reception.accept_conclusion or "").strip() != "js":
        return None
    exists = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.reception_id == reception.id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if exists:
        return exists

    source = getattr(reception, "reception_source", "cell_storage") or "cell_storage"

    if source == "research_donation":
        protocol_no = ""
        d = (
            db.query(BizResearchDonation)
            .filter(
                BizResearchDonation.id == (reception.research_donation_id or ""),
                BizResearchDonation.del_flag == 0,
            )
            .first()
        )
        if d:
            protocol_no = d.donation_no or ""
        sample_code = ""
        ds = (
            db.query(BizResearchDonationSample)
            .filter(
                BizResearchDonationSample.id == (reception.donation_sample_id or ""),
                BizResearchDonationSample.del_flag == 0,
            )
            .first()
        )
        if ds:
            sample_code = ds.sample_code or ""
        wo = BizProductionWorkOrder(
            id=generate_uuid(),
            work_order_no=_generate_work_order_no(db, tenant_no),
            reception_id=reception.id,
            protocol_sample_id=reception.donation_sample_id or "",
            collection_protocol_id=reception.research_donation_id or "",
            protocol_no=protocol_no or None,
            sample_code=sample_code or None,
            reception_source="research_donation",
            work_order_status="dpz",
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id=create_dept_id or "",
            update_by=user_id,
            create_time=datetime.now(),
            update_time=datetime.now(),
            del_flag=0,
        )
    else:
        protocol_no = ""
        p = (
            db.query(BizCollectionProtocol)
            .filter(
                BizCollectionProtocol.id == reception.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            )
            .first()
        )
        if p:
            protocol_no = p.protocol_no or ""
        sample_code = ""
        s = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.id == reception.protocol_sample_id,
                BizProtocolSample.del_flag == 0,
            )
            .first()
        )
        if s:
            sample_code = s.sample_code or ""
        wo = BizProductionWorkOrder(
            id=generate_uuid(),
            work_order_no=_generate_work_order_no(db, tenant_no),
            reception_id=reception.id,
            protocol_sample_id=reception.protocol_sample_id,
            collection_protocol_id=reception.collection_protocol_id,
            protocol_no=protocol_no or None,
            sample_code=sample_code or None,
            reception_source="cell_storage",
            work_order_status="dpz",
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id=create_dept_id or "",
            update_by=user_id,
            create_time=datetime.now(),
            update_time=datetime.now(),
            del_flag=0,
        )
    db.add(wo)
    db.flush()
    return wo


def list_production_work_orders(
    db: Session,
    *,
    tenant_no: str,
    query: Optional[ProductionWorkOrderListQuery] = None,
    page: int = 1,
    page_size: int = 20,
    user_info=None,
) -> Dict[str, Any]:
    def _filtered():
        q0 = db.query(BizProductionWorkOrder).filter(BizProductionWorkOrder.del_flag == 0)
        q0 = q0.filter(BizProductionWorkOrder.tenant_no == tenant_no)
        if not query:
            return q0
        wo_type = (query.work_order_type or "").strip()
        st_raw = (query.work_order_status or "").strip()
        if st_raw:
            statuses = [s.strip() for s in st_raw.split(",") if s.strip()]
            if statuses:
                q0 = q0.filter(BizProductionWorkOrder.work_order_status.in_(statuses))
        if wo_type:
            q0 = q0.filter(BizProductionWorkOrder.work_order_type == wo_type)
        rs = (query.reception_source or "").strip()
        if rs:
            q0 = q0.filter(BizProductionWorkOrder.reception_source == rs)
        bt = (query.biz_type or "").strip()
        if bt:
            q0 = q0.filter(BizProductionWorkOrder.biz_type == bt)
        kw = (query.keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            if wo_type == "temporary":
                q0 = q0.filter(BizProductionWorkOrder.remark.like(like))
            else:
                keyword_conditions = [
                    BizProductionWorkOrder.work_order_no.like(like),
                    BizProductionWorkOrder.sample_code.like(like),
                ]
                from app.crud import sample_reception as reception_svc

                rec_ids = reception_svc.reception_ids_matching_lab_storage_keyword(
                    db, tenant_no, kw
                )
                if rec_ids:
                    keyword_conditions.append(BizProductionWorkOrder.reception_id.in_(rec_ids))
                lab_codes = reception_svc.sample_codes_matching_lab_storage_keyword(
                    db, tenant_no, kw
                )
                if lab_codes:
                    keyword_conditions.append(BizProductionWorkOrder.sample_code.in_(lab_codes))
                if wo_type == "outbound":
                    outbound_ids = _outbound_work_order_ids_by_keyword(db, tenant_no, kw)
                    if outbound_ids:
                        keyword_conditions.append(BizProductionWorkOrder.id.in_(outbound_ids))
                q0 = q0.filter(or_(*keyword_conditions))
        elif query.work_order_no and query.work_order_no.strip():
            like = f"%{query.work_order_no.strip()}%"
            q0 = q0.filter(BizProductionWorkOrder.work_order_no.like(like))
        if query.sample_code and query.sample_code.strip():
            like = f"%{query.sample_code.strip()}%"
            q0 = q0.filter(BizProductionWorkOrder.sample_code.like(like))
        use_planned_start_time = (
            (query.planned_start_time_start and query.planned_start_time_start.strip())
            or (query.planned_start_time_end and query.planned_start_time_end.strip())
        )
        if use_planned_start_time:
            q0 = apply_planned_start_time_step_filter(
                q0,
                tenant_no=tenant_no,
                work_order_id_column=BizProductionWorkOrder.id,
                planned_start_time_start=query.planned_start_time_start,
                planned_start_time_end=query.planned_start_time_end,
            )
        else:
            use_infusion_time = wo_type == "outbound" and (
                (query.infusion_time_start and query.infusion_time_start.strip())
                or (query.infusion_time_end and query.infusion_time_end.strip())
            )
            if use_infusion_time:
                from app.models.models import BizPilotInfusionSchedule

                q0 = q0.join(
                    BizPilotInfusionSchedule,
                    eq_unicode(
                        BizProductionWorkOrder.infusion_schedule_id,
                        BizPilotInfusionSchedule.id,
                    ),
                ).filter(
                    BizPilotInfusionSchedule.del_flag == 0,
                    BizPilotInfusionSchedule.tenant_no == tenant_no,
                )
                if query.infusion_time_start and query.infusion_time_start.strip():
                    q0 = q0.filter(
                        BizPilotInfusionSchedule.infusion_time
                        >= _normalize_date_bound(query.infusion_time_start.strip())
                    )
                if query.infusion_time_end and query.infusion_time_end.strip():
                    q0 = q0.filter(
                        BizPilotInfusionSchedule.infusion_time
                        <= _normalize_date_bound(query.infusion_time_end.strip(), end_of_day=True)
                    )
            else:
                if query.create_time_start and query.create_time_start.strip():
                    q0 = q0.filter(
                        BizProductionWorkOrder.create_time
                        >= _normalize_date_bound(query.create_time_start.strip())
                    )
                if query.create_time_end and query.create_time_end.strip():
                    q0 = q0.filter(
                        BizProductionWorkOrder.create_time
                        <= _normalize_date_bound(query.create_time_end.strip(), end_of_day=True)
                    )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizProductionWorkOrder, user_info, db, tenant_no)
        exclude_inactive = True
        if query and query.exclude_void_protocol is not None:
            exclude_inactive = bool(query.exclude_void_protocol)
        q0 = apply_exclude_inactive_protocol_filter(q0, exclude=exclude_inactive)
        return q0

    effective_page = query.page if query and query.page else page
    effective_size = query.page_size if query and query.page_size else page_size

    total = _filtered().order_by(None).with_entities(func.count(BizProductionWorkOrder.id)).scalar() or 0
    effective_page = max(1, effective_page)
    effective_size = min(max(1, effective_size), 100)
    rows = (
        _filtered()
        .order_by(BizProductionWorkOrder.create_time.desc())
        .offset((effective_page - 1) * effective_size)
        .limit(effective_size)
        .all()
    )
    items = _enrich_wo_items(db, [_wo_to_dict(x) for x in rows])
    items = _enrich_inbound_qc_conclusions(db, tenant_no, items)
    items = _enrich_outbound_items(db, tenant_no, items)
    items = enrich_work_order_fee_summaries(db, items, tenant_no)
    return {"items": items, "total": int(total)}


def sync_production_work_order_status_from_tasks(
    db: Session,
    work_order_id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    """按关联生产任务回写工单：存在未删除任务则为 jxz，全部删除后回 dpz（已完成工单不改动）。"""
    if not (work_order_id or "").strip():
        return
    db.flush()
    wo = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == work_order_id.strip(),
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if not wo or (wo.work_order_status or "").strip() == "ywc":
        return
    task_count = (
        db.query(BizProductionTask.id)
        .filter(
            BizProductionTask.work_order_id == work_order_id.strip(),
            BizProductionTask.tenant_no == tenant_no,
            BizProductionTask.del_flag == 0,
        )
        .count()
    )
    new_status = "jxz" if task_count > 0 else "dpz"
    if (wo.work_order_status or "").strip() == new_status:
        return
    now = datetime.now()
    wo.work_order_status = new_status
    wo.update_by = user_id
    wo.update_time = now
    db.add(wo)
    from app.crud.pilot_outbound_status_sync import sync_schedule_from_production_work_order

    sync_schedule_from_production_work_order(db, wo, user_id=user_id)


def get_production_work_order(
    db: Session, row_id: str, tenant_no: str
) -> Optional[BizProductionWorkOrder]:
    return (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == row_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )


def update_production_work_order(
    db: Session,
    row_id: str,
    tenant_no: str,
    user_id: Optional[str],
    *,
    remark: Optional[str] = None,
) -> BizProductionWorkOrder:
    """更新生产工单备注（入库/临时均可，任意状态）"""
    wo = get_production_work_order(db, row_id, tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    if remark is not None:
        wo.remark = (remark or "").strip() or None
    wo.update_by = user_id
    wo.update_time = datetime.now()
    db.add(wo)
    db.flush()
    return wo


def complete_production_work_order(
    db: Session,
    row_id: str,
    tenant_no: str,
    user_id: str,
    data: Optional[ProductionWorkOrderComplete] = None,
) -> BizProductionWorkOrder:
    wo = get_production_work_order(db, row_id, tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    if wo.work_order_status == "ywc":
        raise ValueError("工单已完成")
    wo.work_order_status = "ywc"
    wo.update_by = user_id
    wo.update_time = datetime.now()
    if data and data.remark is not None:
        wo.remark = data.remark
    db.add(wo)
    from app.crud.pilot_outbound_status_sync import sync_schedule_from_production_work_order

    sync_schedule_from_production_work_order(db, wo, user_id=user_id)
    db.commit()
    db.refresh(wo)
    return wo


_TASK_STATUS_LABELS = {
    "dpz": "待配置",
    "dzx": "待执行",
    "jxz": "执行中",
    "ywc": "已完成",
}


def _production_task_info(db: Session, task: BizProductionTask) -> Dict[str, Any]:
    """将单条生产任务（含投入物、产出物）转为摘要字典。"""
    inputs = (
        db.query(BizProductionTaskInput)
        .filter(
            BizProductionTaskInput.task_id == task.id,
            BizProductionTaskInput.del_flag == 0,
        )
        .order_by(BizProductionTaskInput.order_num)
        .all()
    )
    outputs = (
        db.query(BizProductionTaskOutput)
        .filter(
            BizProductionTaskOutput.task_id == task.id,
            BizProductionTaskOutput.del_flag == 0,
        )
        .order_by(BizProductionTaskOutput.order_num)
        .all()
    )
    return {
        "id": task.id,
        "taskNo": task.task_no,
        "taskName": task.task_name or "",
        "taskStatus": task.task_status or "",
        "taskStatusLabel": _TASK_STATUS_LABELS.get(task.task_status, task.task_status or ""),
        "conclusion": (getattr(task, "conclusion", None) or "").strip(),
        "conclusionLabel": production_task_svc._production_conclusion_label(
            getattr(task, "conclusion", None)
        ),
        "startedTime": _fmt_dt(task.started_time),
        "completedTime": _fmt_dt(task.completed_time),
        "inputs": [
            {
                "inputType": (inp.input_type or ""),
                "inputCode": (inp.input_code or ""),
                "inputName": (inp.input_name or ""),
            }
            for inp in inputs
        ],
        "outputs": [
            {
                "outputType": "product",
                "outputCode": (out.output_barcode or ""),
                "outputName": (out.output_name or ""),
            }
            for out in outputs
        ],
    }


def _production_info_item(db: Session, wo: BizProductionWorkOrder) -> Dict[str, Any]:
    """将单条生产工单（含任务列表）转为摘要字典。"""
    tasks = (
        db.query(BizProductionTask)
        .filter(
            BizProductionTask.work_order_id == wo.id,
            BizProductionTask.tenant_no == wo.tenant_no,
            BizProductionTask.del_flag == 0,
        )
        .order_by(BizProductionTask.order_num, BizProductionTask.create_time)
        .all()
    )
    return {
        "workOrderNo": wo.work_order_no or "",
        "workOrderStatus": wo.work_order_status or "",
        "workOrderStatusLabel": _WORK_ORDER_STATUS_LABELS.get(wo.work_order_status, wo.work_order_status or ""),
        "tasks": [_production_task_info(db, t) for t in tasks],
    }


def get_production_info_map_for_samples(
    db: Session,
    sample_ids: List[str],
    tenant_no: str,
    *,
    sample_code_by_id: Optional[Dict[str, str]] = None,
) -> Dict[str, Dict[str, Any]]:
    """按样本批量返回生产工单摘要（含任务、投入物、产出物），供协议查看弹窗使用。

    细胞存储：传入 sample_code_by_id 时，按租户 + 样本编码关联。
    科研捐赠等：仅传 sample_ids 时，按 protocol_sample_id 关联。
    """
    if not tenant_no:
        return {}

    if sample_code_by_id:
        codes = list({(c or "").strip() for c in sample_code_by_id.values() if (c or "").strip()})
        if not codes:
            return {}
        wos = (
            db.query(BizProductionWorkOrder)
            .filter(
                BizProductionWorkOrder.tenant_no == tenant_no,
                BizProductionWorkOrder.del_flag == 0,
                BizProductionWorkOrder.sample_code.in_(codes),
            )
            .all()
        )
        code_to_wo: Dict[str, BizProductionWorkOrder] = {}
        for wo in wos:
            code = (wo.sample_code or "").strip()
            if code and code not in code_to_wo:
                code_to_wo[code] = wo
        out: Dict[str, Dict[str, Any]] = {}
        for sid, raw_code in sample_code_by_id.items():
            code = (raw_code or "").strip()
            if not code:
                continue
            wo = code_to_wo.get(code)
            if wo:
                out[sid] = _production_info_item(db, wo)
        return out

    if not sample_ids:
        return {}
    wos = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.protocol_sample_id.in_(sample_ids),
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .all()
    )
    out = {}
    for wo in wos:
        sid = (wo.protocol_sample_id or "").strip()
        if sid:
            out[sid] = _production_info_item(db, wo)
    return out


def create_production_work_order_manual(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    protocol_no: str = "",
    sample_code: str = "",
    reception_source: str = "cell_storage",
    remark: str = "",
) -> BizProductionWorkOrder:
    """手动创建生产工单（不关联接收记录）"""
    wo = BizProductionWorkOrder(
        id=generate_uuid(),
        work_order_no=_generate_work_order_no(db, tenant_no),
        work_order_type="inbound",
        reception_id=None,
        protocol_sample_id=None,
        collection_protocol_id=None,
        protocol_no=protocol_no or None,
        sample_code=sample_code or None,
        reception_source=reception_source,
        work_order_status="dpz",
        tenant_no=tenant_no,
        remark=remark,
        create_by=user_id,
        create_dept_id=create_dept_id,
    )
    db.add(wo)
    db.flush()
    return wo


def _generate_outbound_work_order_no(db: Session, tenant_no: str) -> str:
    today_str = datetime.now().strftime("%Y%m%d")
    prefix = f"CKSC{today_str}"
    last = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.work_order_no.like(f"{prefix}%"),
            BizProductionWorkOrder.del_flag == 0,
        )
        .order_by(BizProductionWorkOrder.work_order_no.desc())
        .first()
    )
    seq = 1
    if last:
        seq = int(last.work_order_no[-4:]) + 1
    return f"{prefix}{seq:04d}"


def create_outbound_production_work_order(
    db: Session,
    *,
    schedule_id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    remark: str = "",
    biz_type: str = "pilot_service",
) -> BizProductionWorkOrder:
    existing = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.infusion_schedule_id == schedule_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if existing:
        if (existing.biz_type or "") != biz_type:
            existing.biz_type = biz_type
            db.flush()
        return existing
    wo = BizProductionWorkOrder(
        id=generate_uuid(),
        work_order_no=_generate_outbound_work_order_no(db, tenant_no),
        work_order_type="outbound",
        reception_id=None,
        protocol_sample_id=None,
        collection_protocol_id=None,
        protocol_no=None,
        sample_code=None,
        reception_source="outbound",
        infusion_schedule_id=schedule_id,
        biz_type=biz_type,
        work_order_status="dpz",
        tenant_no=tenant_no,
        remark=remark,
        create_by=user_id,
        create_dept_id=create_dept_id,
    )
    db.add(wo)
    db.flush()
    return wo


def delete_production_work_order(db: Session, work_order_id: str, tenant_no: str, user_id: Optional[str]) -> None:
    wo = db.query(BizProductionWorkOrder).filter(
        and_(
            BizProductionWorkOrder.id == work_order_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
    ).first()
    if not wo:
        raise ValueError("工单不存在")
    if wo.work_order_status == "ywc":
        raise ValueError("已完成的工单不能删除")

    now = datetime.now()

    tasks = db.query(BizProductionTask).filter(
        and_(BizProductionTask.work_order_id == work_order_id, BizProductionTask.del_flag == 0)
    ).all()

    for task in tasks:
        steps = db.query(BizProductionTaskStep).filter(
            and_(BizProductionTaskStep.task_id == task.id, BizProductionTaskStep.del_flag == 0)
        ).all()
        for step in steps:
            _soft_delete_sub(db, BizProductionTaskStepReagent, "step_id", step.id, now)
            _soft_delete_sub(db, BizProductionTaskStepConsumable, "step_id", step.id, now)
            _soft_delete_sub(db, BizProductionTaskStepOutput, "step_id", step.id, now)
            _soft_delete_sub(db, BizProductionTaskStepDataCollection, "step_id", step.id, now)
            step.del_flag = 1
            step.update_by = user_id
            step.update_time = now

        _soft_delete_sub(db, BizProductionTaskInput, "task_id", task.id, now)
        _soft_delete_sub(db, BizProductionTaskOutput, "task_id", task.id, now)

        task.del_flag = 1
        task.update_by = user_id
        task.update_time = now

    wo.del_flag = 1
    wo.update_by = user_id
    wo.update_time = now

    db.commit()


# ---------------------------------------------------------------------------
# 临时生产工单（work_order_type='temporary'）
# ---------------------------------------------------------------------------


def _generate_temp_work_order_no(db: Session, tenant_no: str) -> str:
    today_str = datetime.now().strftime("%Y%m%d")
    prefix = f"LSSC{today_str}"
    last = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.work_order_no.like(f"{prefix}%"),
            BizProductionWorkOrder.del_flag == 0,
        )
        .order_by(BizProductionWorkOrder.work_order_no.desc())
        .first()
    )
    seq = 1
    if last:
        seq = int(last.work_order_no[-4:]) + 1
    return f"{prefix}{seq:04d}"


def _temp_wo_to_dict(wo: BizProductionWorkOrder) -> Dict[str, Any]:
    return {
        "id": wo.id,
        "workOrderNo": wo.work_order_no,
        "workOrderStatus": wo.work_order_status,
        "workOrderStatusLabel": _WORK_ORDER_STATUS_LABELS.get(wo.work_order_status, wo.work_order_status),
        "configuredTime": _fmt_dt(getattr(wo, "configured_time", None)),
        "configuredBy": getattr(wo, "configured_by", None) or "",
        "remark": wo.remark or "",
        "createTime": _fmt_dt(wo.create_time),
        "updateTime": _fmt_dt(wo.update_time),
    }


def create_temp_production_work_order(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    remark: str = "",
) -> BizProductionWorkOrder:
    wo = BizProductionWorkOrder(
        id=generate_uuid(),
        work_order_no=_generate_temp_work_order_no(db, tenant_no),
        work_order_type="temporary",
        work_order_status="dpz",
        tenant_no=tenant_no,
        remark=remark,
        create_by=user_id,
        create_dept_id=create_dept_id,
        update_by=user_id,
        create_time=datetime.now(),
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(wo)
    db.flush()
    return wo


def list_temp_production_work_orders(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    work_order_status: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = db.query(BizProductionWorkOrder).filter(
        BizProductionWorkOrder.del_flag == 0,
        BizProductionWorkOrder.work_order_type == "temporary",
    )
    if tenant_no:
        q = q.filter(BizProductionWorkOrder.tenant_no == tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(BizProductionWorkOrder.remark.like(like))
    st_raw = (work_order_status or "").strip()
    if st_raw:
        statuses = [s.strip() for s in st_raw.split(",") if s.strip()]
        if statuses:
            q = q.filter(BizProductionWorkOrder.work_order_status.in_(statuses))
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizProductionWorkOrder, user_info, db, tenant_no or "")
    total = q.order_by(None).with_entities(func.count(BizProductionWorkOrder.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizProductionWorkOrder.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = [_temp_wo_to_dict(x) for x in rows]
    return items, int(total)


def get_temp_production_work_order(
    db: Session, row_id: str, tenant_no: str
) -> Optional[BizProductionWorkOrder]:
    return (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == row_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
            BizProductionWorkOrder.work_order_type == "temporary",
        )
        .first()
    )


def get_temp_production_work_order_detail(
    db: Session, row_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    wo = get_temp_production_work_order(db, row_id, tenant_no)
    if not wo:
        return None
    result = _temp_wo_to_dict(wo)
    tasks = (
        db.query(BizProductionTask)
        .filter(
            BizProductionTask.work_order_id == row_id,
            BizProductionTask.order_source == "temporary",
            BizProductionTask.tenant_no == tenant_no,
            BizProductionTask.del_flag == 0,
        )
        .order_by(BizProductionTask.order_num, BizProductionTask.create_time)
        .all()
    )
    result["tasks"] = [production_task_svc._task_to_dict(t) for t in tasks]
    return result


def configure_temp_production_work_order(
    db: Session,
    work_order_id: str,
    *,
    task_name: str,
    task_type: str,
    route_id: Optional[str],
    description: str,
    remark: str,
    assignee_user_id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> BizProductionWorkOrder:
    wo = get_temp_production_work_order(db, work_order_id, tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    if wo.work_order_status != "dpz":
        raise ValueError("工单已配置，不能重复配置")

    production_task_svc.create_production_task(
        db,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=create_dept_id,
        data={
            "work_order_id": work_order_id,
            "task_name": task_name,
            "task_type": task_type,
            "route_id": route_id,
            "description": description,
            "remark": remark,
            "assignee_user_id": assignee_user_id,
        },
        order_source="temporary",
    )

    now = datetime.now()
    wo.work_order_status = "jxz"
    wo.configured_time = now
    wo.configured_by = user_id
    wo.update_by = user_id
    wo.update_time = now
    db.add(wo)
    db.flush()
    return wo


def sync_temp_production_work_order_status(
    db: Session,
    work_order_id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    if not (work_order_id or "").strip():
        return
    wo = get_temp_production_work_order(db, work_order_id.strip(), tenant_no)
    if not wo:
        return
    tasks = (
        db.query(BizProductionTask.task_status)
        .filter(
            BizProductionTask.work_order_id == work_order_id.strip(),
            BizProductionTask.order_source == "temporary",
            BizProductionTask.tenant_no == tenant_no,
            BizProductionTask.del_flag == 0,
        )
        .all()
    )
    if not tasks:
        new_status = "dpz"
    elif all(t.task_status == "ywc" for t in tasks):
        new_status = "ywc"
    else:
        new_status = "jxz"
    if wo.work_order_status == new_status:
        return
    now = datetime.now()
    wo.work_order_status = new_status
    wo.update_by = user_id
    wo.update_time = now
    db.add(wo)


def delete_temp_production_work_order(
    db: Session, work_order_id: str, tenant_no: str, user_id: Optional[str]
) -> None:
    wo = get_temp_production_work_order(db, work_order_id, tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    if wo.work_order_status == "ywc":
        raise ValueError("已完成的工单不能删除")

    now = datetime.now()

    tasks = db.query(BizProductionTask).filter(
        and_(
            BizProductionTask.work_order_id == work_order_id,
            BizProductionTask.order_source == "temporary",
            BizProductionTask.del_flag == 0,
        )
    ).all()

    for task in tasks:
        steps = db.query(BizProductionTaskStep).filter(
            and_(BizProductionTaskStep.task_id == task.id, BizProductionTaskStep.del_flag == 0)
        ).all()
        for step in steps:
            _soft_delete_sub(db, BizProductionTaskStepReagent, "step_id", step.id, now)
            _soft_delete_sub(db, BizProductionTaskStepConsumable, "step_id", step.id, now)
            _soft_delete_sub(db, BizProductionTaskStepOutput, "step_id", step.id, now)
            _soft_delete_sub(db, BizProductionTaskStepDataCollection, "step_id", step.id, now)
            step.del_flag = 1
            step.update_by = user_id
            step.update_time = now

        _soft_delete_sub(db, BizProductionTaskInput, "task_id", task.id, now)
        _soft_delete_sub(db, BizProductionTaskOutput, "task_id", task.id, now)

        task.del_flag = 1
        task.update_by = user_id
        task.update_time = now

    wo.del_flag = 1
    wo.update_by = user_id
    wo.update_time = now

    db.commit()