zkxw/backend/app/crud/qc_work_order.py

"""质检工单 CRUD"""
from __future__ import annotations

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

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

from app.core.db_collation import eq_unicode, like_unicode, unicode_ci
from app.core.utils import generate_uuid
from app.crud import qc_lab_test_item as lab_item_svc
from app.crud import qc_task as task_svc
from app.crud import qc_task_assignee as assignee_svc
from app.crud import qc_third_party_item as third_item_svc
from app.crud.qc_third_party_company import get_company
from app.crud.qc_inbound_lab_storage_sort import (
    inbound_lab_storage_order_by_clauses,
    uses_inbound_lab_storage_list_sort,
)
from app.crud.collection_protocol import urls_to_attachment_list
from app.utils.sample_special import apply_special_to_item_from_samples
from app.crud import sample_reception as sample_reception_svc
from app.crud.temp_qc_target import resolve_temp_qc_target
from app.crud.sample_reception import _donation_sample_id_col
from app.crud import sample_reception_check_result as check_result_svc
from app.crud.cell_product import _enrich_product_dicts as _cp_enrich, _row_to_dict as _cp_row_to_dict, get_cell_product
from app.services.qc_work_order_fee_summary import enrich_work_order_fee_summaries
from app.services.qc_work_order_conclusion import (
    build_qc_conclusion_fields,
    work_order_qc_conclusion_label,
)
from app.models.models import (
    BizCellProduct,
    BizCellSampleType,
    BizCollectionProtocol,
    BizPilotApplication,
    BizPilotInfusionPlan,
    BizPilotInfusionSchedule,
    BizPilotInfusionScheduleProduct,
    BizPilotPatient,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizProductionTask,
    BizProductionTaskStep,
    BizProductionTaskStepOutput,
    BizProductionWorkOrder,
    BizOutputType,
    BizQcLabTestItem,
    BizQcTask,
    BizQcTaskAssignee,
    BizQcThirdPartyItem,
    BizQcWorkOrder,
    BizQcWorkOrderLabItem,
    BizQcWorkOrderThirdItem,
    BizSampleReception,
    SysUser,
)
from app.models.qc_param import BizQcTaskResult
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationParty,
    BizResearchDonationSample,
)

_RECEPTION_SOURCE_LABELS = {
    "cell_storage": "细胞存储",
    "research_donation": "科研捐赠",
    "outbound": "出库",
}
_QC_BIZ_TYPE_LABELS = {
    "inbound": "入库质检",
    "outbound": "出库质检",
    "temporary": "临时质检",
}
_TRIGGER_SOURCE_LABELS = {
    "reception": "样本接收",
    "schedule": "计划质检",
    "production": "生产过程",
    "manual": "手动创建",
}
_ACCEPT_CONCLUSION_LABELS = {"js": "接收", "bj": "拒收"}
_WORK_ORDER_STATUS_LABELS = {"dpz": "待配置", "ypz": "已配置", "ywc": "已完成", "yzf": "已作废"}
_VALID_QC_WORK_ORDER_STATUSES = frozenset(_WORK_ORDER_STATUS_LABELS.keys())
_VALID_QC_WORK_ORDER_CONCLUSIONS = frozenset({"tg", "btg"})
_VALID_QC_DISPATCH_OUTPUT_TYPE_NAME = "有效物"
_TASK_TYPE_LABELS = {"sysz": "实验室自检", "dsfjc": "第三方检测"}
_TASK_STATUS_LABELS = {"dzx": "待执行", "jxz": "进行中", "ywc": "已完成"}
_PROTOCOL_STATUS_LABELS = {
    "dsx": "待生效", "sxz": "生效中", "ygq": "已过期", "yzf": "已作废", "sdfq": "属地废弃", "zky": "转科研"
}
_DONATION_STATUS_LABELS = {
    "dsx": "待生效", "sxz": "生效中", "ygq": "已过期", "yzf": "已作废", "sdfq": "属地废弃", "zckc": "转细胞存储"
}
_CONCLUSION_LABELS = {"tg": "通过", "btg": "不通过"}
_OUTBOUND_BIZ_TYPE_LABELS = {
    "pilot_service": "中试服务",
    "clinical_application": "临床应用",
    "basic_experiment": "基础实验",
}
_OUTBOUND_BIZ_TYPES = tuple(_OUTBOUND_BIZ_TYPE_LABELS.keys())


def resolve_outbound_biz_type_for_schedule(
    db: Session, schedule_id: str, tenant_no: str
) -> str:
    """按输注计划追溯申请单，得到出库生产/质检工单应使用的 biz_type。"""
    if not schedule_id:
        return "pilot_service"
    sched = (
        db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.id == schedule_id,
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .first()
    )
    if not sched or not sched.infusion_plan_id:
        return "pilot_service"
    plan = (
        db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.id == sched.infusion_plan_id,
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .first()
    )
    if not plan or not plan.pilot_application_id:
        return "pilot_service"
    app = (
        db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id == plan.pilot_application_id,
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .first()
    )
    bt = ((app.biz_type if app else None) or "").strip()
    return bt if bt in _OUTBOUND_BIZ_TYPE_LABELS else "pilot_service"


def _dec(v: Any) -> Optional[float]:
    if v is None:
        return None
    return float(v)


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


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 _reception_source_label(source: Optional[str]) -> str:
    key = (source or "").strip() or "cell_storage"
    return _RECEPTION_SOURCE_LABELS.get(key, key)


def _qc_biz_type_label(biz_type: Optional[str]) -> str:
    key = (biz_type or "").strip() or "inbound"
    return _QC_BIZ_TYPE_LABELS.get(key, key)


def _work_order_qc_biz_type(wo: BizQcWorkOrder) -> str:
    wo_type = getattr(wo, "work_order_type", None) or "inbound"
    if wo_type == "temporary":
        return "temporary"
    return getattr(wo, "qc_biz_type", None) or "inbound"


def _validate_cell_product_id(
    db: Session, cell_product_id: Optional[str], tenant_no: str
) -> str:
    cp_id = (cell_product_id or "").strip()
    if not cp_id:
        raise ValueError("请选择质检对象（细胞产品）")
    cp = get_cell_product(db, cp_id, tenant_no)
    if not cp:
        raise ValueError("细胞产品不存在")
    return cp_id


def _enrich_work_order_cell_products(
    db: Session, tenant_no: str, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    if not items:
        return items
    cp_ids = list({x.get("cellProductId") for x in items if x.get("cellProductId")})
    if not cp_ids:
        return items
    products = (
        db.query(BizCellProduct)
        .filter(
            BizCellProduct.id.in_(cp_ids),
            BizCellProduct.tenant_no == tenant_no,
            BizCellProduct.del_flag == 0,
        )
        .all()
    )
    product_dicts = _cp_enrich(db, tenant_no, [_cp_row_to_dict(p) for p in products])
    product_map = {d["id"]: d for d in product_dicts}
    for item in items:
        cp = product_map.get(item.get("cellProductId") or "")
        item["cellProductCode"] = cp.get("productCode", "") if cp else ""
        item["cellProductName"] = cp.get("productName", "") if cp else ""
        item["cellProductTypeName"] = cp.get("productTypeName", "") if cp else ""
        item["cellProductPassage"] = cp.get("passage", "") if cp else ""
        item["cellProductSpecName"] = cp.get("specName", "") if cp else ""
    return items


def _enrich_work_order_step_outputs(
    db: Session, tenant_no: str, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """步骤产出物触发的质检：无细胞产品时用工单条码/名称展示质检对象。"""
    if not items:
        return items
    step_output_ids = list(
        {
            x.get("productionStepOutputId")
            for x in items
            if (x.get("productionStepOutputId") or "").strip()
            and not (x.get("cellProductId") or "").strip()
        }
    )
    if not step_output_ids:
        return items
    rows = (
        db.query(BizProductionTaskStepOutput)
        .filter(
            BizProductionTaskStepOutput.id.in_(step_output_ids),
            BizProductionTaskStepOutput.tenant_no == tenant_no,
            BizProductionTaskStepOutput.del_flag == 0,
        )
        .all()
    )
    row_map = {r.id: r for r in rows}
    for item in items:
        if (item.get("cellProductId") or "").strip():
            continue
        so = row_map.get(item.get("productionStepOutputId") or "")
        if not so:
            continue
        if not (item.get("cellProductCode") or "").strip():
            item["cellProductCode"] = (so.output_barcode or "").strip()
        if not (item.get("cellProductName") or "").strip():
            item["cellProductName"] = (so.output_name or "").strip()
    return items


def _enrich_work_order_qc_targets(
    db: Session, tenant_no: str, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    items = _enrich_work_order_cell_products(db, tenant_no, items)
    return _enrich_work_order_step_outputs(db, tenant_no, items)


def _enrich_outbound_work_orders(
    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,
    )

    outbound = [x for x in items if x.get("workOrderType") == "outbound"]
    if not outbound:
        return items
    schedule_ids = list({x.get("infusionScheduleId") for x in outbound if x.get("infusionScheduleId")})
    if not schedule_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_(schedule_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(product.agreement_price),
                "actualPrice": _dec(product.actual_price),
                "quantity": product.quantity,
                "productStatus": product.product_status or "",
            }
        )

    # 查输注计划 → 输注方案 → 申请单 → 患者
    schedules = {
        s.id: s
        for s in db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.id.in_(schedule_ids),
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .all()
    }
    plan_ids = list({s.infusion_plan_id for s in schedules.values() if s.infusion_plan_id})
    plans = {
        p.id: p
        for p in db.query(BizPilotInfusionPlan)
        .filter(
            BizPilotInfusionPlan.id.in_(plan_ids),
            BizPilotInfusionPlan.tenant_no == tenant_no,
            BizPilotInfusionPlan.del_flag == 0,
        )
        .all()
    } if plan_ids else {}
    app_ids = list({p.pilot_application_id for p in plans.values() if p.pilot_application_id})
    apps = {
        a.id: a
        for a in db.query(BizPilotApplication)
        .filter(
            BizPilotApplication.id.in_(app_ids),
            BizPilotApplication.tenant_no == tenant_no,
            BizPilotApplication.del_flag == 0,
        )
        .all()
    } if app_ids else {}
    patient_map: Dict[str, BizPilotPatient] = {}
    if app_ids:
        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 item in outbound:
        sid = item.get("infusionScheduleId") or ""
        sched = schedules.get(sid)
        if not sched:
            continue
        plan = plans.get(sched.infusion_plan_id or "")
        app = apps.get(plan.pilot_application_id or "") if plan 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)

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


def _outbound_qc_work_order_ids_by_keyword(
    db: Session, tenant_no: str, keyword: str
) -> List[str]:
    """出库质检工单关键词：按关联申请单号、申请单名称匹配。"""
    kw = (keyword or "").strip()
    if not kw:
        return []
    like = f"%{kw}%"
    rows = (
        db.query(BizQcWorkOrder.id)
        .join(
            BizPilotInfusionSchedule,
            eq_unicode(BizQcWorkOrder.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(
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.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 _inbound_qc_work_order_ids_by_mother_name(
    db: Session,
    tenant_no: str,
    mother_name: str,
    *,
    reception_source: Optional[str] = None,
) -> List[str]:
    """入库质检工单：按母亲姓名匹配（细胞存储协议 / 科研捐赠单）。"""
    mn = (mother_name or "").strip()
    if not mn or not tenant_no:
        return []
    like = f"%{mn}%"
    ids: set = set()
    rs = (reception_source or "").strip()

    if not rs or rs == "cell_storage":
        for (wid,) in (
            db.query(BizQcWorkOrder.id)
            .join(
                BizProtocolPartyCustomer,
                eq_unicode(
                    BizProtocolPartyCustomer.collection_protocol_id,
                    BizQcWorkOrder.collection_protocol_id,
                ),
            )
            .filter(
                BizQcWorkOrder.del_flag == 0,
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.work_order_type == "inbound",
                or_(
                    BizQcWorkOrder.reception_source == "cell_storage",
                    BizQcWorkOrder.reception_source.is_(None),
                    BizQcWorkOrder.reception_source == "",
                ),
                BizProtocolPartyCustomer.del_flag == 0,
                like_unicode(BizProtocolPartyCustomer.mother_name, like),
            )
            .all()
        ):
            ids.add(wid)

    if not rs or rs == "research_donation":
        for (wid,) in (
            db.query(BizQcWorkOrder.id)
            .join(
                BizResearchDonationParty,
                eq_unicode(
                    BizResearchDonationParty.research_donation_id,
                    BizQcWorkOrder.collection_protocol_id,
                ),
            )
            .filter(
                BizQcWorkOrder.del_flag == 0,
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.work_order_type == "inbound",
                BizQcWorkOrder.reception_source == "research_donation",
                BizResearchDonationParty.del_flag == 0,
                like_unicode(BizResearchDonationParty.mother_name, like),
            )
            .all()
        ):
            ids.add(wid)

    return list(ids)


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


def _load_sample_type_name_map(db: Session, type_ids: List[str]) -> Dict[str, str]:
    ids = [x for x in type_ids if x]
    if not ids:
        return {}
    return {
        t.id: t.type_name or ""
        for t in db.query(BizCellSampleType)
        .filter(BizCellSampleType.id.in_(ids), BizCellSampleType.del_flag == 0)
        .all()
    }


def _effective_reception_source(item: Dict[str, Any], rec: Optional[BizSampleReception]) -> str:
    """列表展示以工单 receptionSource 为准，接收记录作兜底。"""
    return (
        (item.get("receptionSource") or "").strip()
        or (getattr(rec, "reception_source", None) if rec else None)
        or "cell_storage"
    )


def _effective_donation_sample_id(item: Dict[str, Any], rec: Optional[BizSampleReception]) -> str:
    """科研捐赠：接收记录 donation_sample_id；工单 protocol_sample_id 存捐赠样本 id。"""
    if _effective_reception_source(item, rec) != "research_donation":
        return ""
    return (
        (getattr(rec, "donation_sample_id", None) if rec else None) or ""
    ).strip() or (item.get("protocolSampleId") or "").strip()


def _effective_protocol_sample_id(item: Dict[str, Any], rec: Optional[BizSampleReception]) -> str:
    if _effective_reception_source(item, rec) == "research_donation":
        return ""
    return (
        (getattr(rec, "protocol_sample_id", None) if rec else None) or ""
    ).strip() or (item.get("protocolSampleId") or "").strip()


def _sample_type_name_for_item(
    *,
    source: str,
    donation_sample_id: str,
    protocol_sample_id: str,
    type_map: Dict[str, str],
    protocol_samples: Dict[str, BizProtocolSample],
    donation_samples: Dict[str, BizResearchDonationSample],
) -> str:
    if source == "research_donation":
        sample = donation_samples.get(donation_sample_id or "")
        tid = sample.sample_type_id if sample else ""
    else:
        sample = protocol_samples.get(protocol_sample_id or "")
        tid = sample.sample_type_id if sample else ""
    return type_map.get(tid or "", "")


def _sample_type_name_for_reception(
    db: Session,
    rec: BizSampleReception,
    type_map: Dict[str, str],
    *,
    protocol_samples: Dict[str, BizProtocolSample],
    donation_samples: Dict[str, BizResearchDonationSample],
    item: Optional[Dict[str, Any]] = None,
) -> str:
    """按接收记录解析样本类型；传入 item 时与工单列表 enrich 规则一致。"""
    _ = db
    ctx = item if item is not None else {}
    source = _effective_reception_source(ctx, rec)
    return _sample_type_name_for_item(
        source=source,
        donation_sample_id=_effective_donation_sample_id(ctx, rec),
        protocol_sample_id=_effective_protocol_sample_id(ctx, rec),
        type_map=type_map,
        protocol_samples=protocol_samples,
        donation_samples=donation_samples,
    )


def _enrich_work_order_list_items(
    db: Session, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    if not items:
        return items
    rec_ids = [x["receptionId"] for x in items if x.get("receptionId")]
    sample_codes_no_rec = list(
        {
            (x.get("sampleCode") or "").strip()
            for x in items
            if not (x.get("receptionId") or "").strip() and (x.get("sampleCode") or "").strip()
        }
    )
    rec_map = {}
    if rec_ids:
        rec_map = {
            r.id: r
            for r in db.query(BizSampleReception)
            .filter(BizSampleReception.id.in_(rec_ids), BizSampleReception.del_flag == 0)
            .all()
        }
    sample_code_rec_map: Dict[str, BizSampleReception] = {}
    if sample_codes_no_rec:
        for r in (
            db.query(BizSampleReception)
            .filter(
                BizSampleReception.sample_code.in_(sample_codes_no_rec),
                BizSampleReception.del_flag == 0,
            )
            .order_by(BizSampleReception.create_time.desc())
            .all()
        ):
            code = (r.sample_code or "").strip()
            if code and code not in sample_code_rec_map:
                sample_code_rec_map[code] = r
    if not rec_map and not sample_code_rec_map:
        return items

    def _resolve_reception(item: Dict[str, Any]) -> Optional[BizSampleReception]:
        rec = rec_map.get(item.get("receptionId") or "")
        if rec:
            return rec
        code = (item.get("sampleCode") or "").strip()
        return sample_code_rec_map.get(code) if code else None

    cell_sample_ids: List[str] = []
    donation_sample_ids: List[str] = []
    cell_proto_ids: List[str] = []
    donation_ids: List[str] = []
    for item in items:
        rec = _resolve_reception(item)
        source = _effective_reception_source(item, rec)
        if source == "research_donation":
            dsid = _effective_donation_sample_id(item, rec)
            if dsid:
                donation_sample_ids.append(dsid)
            did = (
                (getattr(rec, "research_donation_id", None) if rec else None) or ""
            ).strip() or (item.get("collectionProtocolId") or "").strip()
            if did:
                donation_ids.append(did)
        else:
            psid = _effective_protocol_sample_id(item, rec)
            if psid:
                cell_sample_ids.append(psid)
            pid = (
                (getattr(rec, "collection_protocol_id", None) if rec else None) or ""
            ).strip() or (item.get("collectionProtocolId") or "").strip()
            if pid:
                cell_proto_ids.append(pid)
    donation_sample_ids = list(dict.fromkeys(donation_sample_ids))
    cell_sample_ids = list(dict.fromkeys(cell_sample_ids))
    donation_ids = list(dict.fromkeys(donation_ids))
    cell_proto_ids = list(dict.fromkeys(cell_proto_ids))
    protocol_samples: Dict[str, BizProtocolSample] = {}
    if cell_sample_ids:
        protocol_samples = {
            s.id: s
            for s in db.query(BizProtocolSample)
            .filter(BizProtocolSample.id.in_(cell_sample_ids), BizProtocolSample.del_flag == 0)
            .all()
        }
    donation_samples: Dict[str, BizResearchDonationSample] = {}
    if donation_sample_ids:
        donation_samples = {}
        for s in (
            db.query(BizResearchDonationSample)
            .filter(
                _donation_sample_id_col(BizResearchDonationSample.id).in_(donation_sample_ids),
                BizResearchDonationSample.del_flag == 0,
            )
            .all()
        ):
            donation_samples[s.id] = s
            donation_samples[(s.id or "").strip()] = s
    type_ids = list(
        {
            s.sample_type_id
            for s in list(protocol_samples.values()) + list(donation_samples.values())
            if s.sample_type_id
        }
    )
    type_map = _load_sample_type_name_map(db, type_ids)
    proto_status_map: Dict[str, str] = {}
    proto_void_type_map: Dict[str, str] = {}
    tenant_no_for_status = ""
    if cell_proto_ids:
        for p in (
            db.query(
                BizCollectionProtocol.id,
                BizCollectionProtocol.protocol_status,
                BizCollectionProtocol.void_type,
                BizCollectionProtocol.tenant_no,
            )
            .filter(BizCollectionProtocol.id.in_(cell_proto_ids), BizCollectionProtocol.del_flag == 0)
            .all()
        ):
            proto_status_map[p.id] = p.protocol_status or ""
            proto_void_type_map[p.id] = p.void_type or ""
            if not tenant_no_for_status and (p.tenant_no or "").strip():
                tenant_no_for_status = (p.tenant_no or "").strip()
    donation_status_map: Dict[str, str] = {}
    donation_void_type_map: Dict[str, str] = {}
    if donation_ids:
        for d in (
            db.query(
                BizResearchDonation.id,
                BizResearchDonation.donation_status,
                BizResearchDonation.void_type,
                BizResearchDonation.tenant_no,
            )
            .filter(BizResearchDonation.id.in_(donation_ids), BizResearchDonation.del_flag == 0)
            .all()
        ):
            donation_status_map[d.id] = d.donation_status or ""
            donation_void_type_map[d.id] = d.void_type or ""
            if not tenant_no_for_status and (d.tenant_no or "").strip():
                tenant_no_for_status = (d.tenant_no or "").strip()
    cell_party_map: Dict[str, Tuple[str, str]] = {}
    if cell_proto_ids:
        for party in (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id.in_(cell_proto_ids),
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .all()
        ):
            cell_party_map[party.collection_protocol_id] = (
                party.mother_name or "",
                party.newborn_name or "",
            )
    donation_party_map: Dict[str, Tuple[str, str]] = {}
    if donation_ids:
        for party in (
            db.query(BizResearchDonationParty)
            .filter(
                BizResearchDonationParty.research_donation_id.in_(donation_ids),
                BizResearchDonationParty.del_flag == 0,
            )
            .all()
        ):
            donation_party_map[party.research_donation_id] = (
                party.mother_name or "",
                party.newborn_name or "",
            )
    for item in items:
        if item.get("workOrderType") == "temporary":
            item["receptionSource"] = ""
            item["receptionSourceLabel"] = ""
            item["motherName"] = ""
            item["newbornName"] = ""
            continue
        rec = _resolve_reception(item)
        source = _effective_reception_source(item, rec)
        item["receptionSource"] = source
        item["receptionSourceLabel"] = _reception_source_label(source)
        if source == "research_donation":
            _did = (
                (getattr(rec, "research_donation_id", None) if rec else None) or ""
            ).strip() or (item.get("collectionProtocolId") or "").strip()
            _mother, _newborn = donation_party_map.get(_did or "", ("", ""))
            item["motherName"] = _mother
            item["newbornName"] = _newborn
        else:
            _pid = (
                (getattr(rec, "collection_protocol_id", None) if rec else None) or ""
            ).strip() or (item.get("collectionProtocolId") or "").strip()
            _mother, _newborn = cell_party_map.get(_pid or "", ("", ""))
            item["motherName"] = _mother
            item["newbornName"] = _newborn
        if not rec:
            item.setdefault("receptionTime", "")
            item.setdefault("handlerName", "")
            item.setdefault("sampleTypeName", "")
            item.setdefault("labStorageCode", "")
            continue
        item["receptionTime"] = _fmt_dt(rec.reception_time)
        item["handlerName"] = rec.handler_name or ""
        item["labStorageCode"] = rec.lab_storage_code or ""
        item["sampleTypeName"] = _sample_type_name_for_item(
            source=source,
            donation_sample_id=_effective_donation_sample_id(item, rec),
            protocol_sample_id=_effective_protocol_sample_id(item, rec),
            type_map=type_map,
            protocol_samples=protocol_samples,
            donation_samples=donation_samples,
        )
        apply_special_to_item_from_samples(
            item,
            source=source,
            protocol_sample_id=_effective_protocol_sample_id(item, rec),
            donation_sample_id=_effective_donation_sample_id(item, rec),
            protocol_samples=protocol_samples,
            donation_samples=donation_samples,
        )
        if source == "research_donation":
            _did = (
                (getattr(rec, "research_donation_id", None) or "").strip()
                or (item.get("collectionProtocolId") or "").strip()
            )
            _st = donation_status_map.get(_did or "", "")
            _vt = donation_void_type_map.get(_did or "", "")
            item["protocolStatus"] = _st
            item["voidType"] = _vt
            from app.core.protocol_status_display import format_protocol_status_label

            item["protocolStatusLabel"] = format_protocol_status_label(
                _st, _vt, kind="donation", db=db, tenant_no=tenant_no_for_status or None
            )
        else:
            _pid = (
                (getattr(rec, "collection_protocol_id", None) or "").strip()
                or (item.get("collectionProtocolId") or "").strip()
            )
            _st = proto_status_map.get(_pid or "", "")
            _vt = proto_void_type_map.get(_pid or "", "")
            item["protocolStatus"] = _st
            item["voidType"] = _vt
            from app.core.protocol_status_display import format_protocol_status_label

            item["protocolStatusLabel"] = format_protocol_status_label(
                _st, _vt, kind="protocol", db=db, tenant_no=tenant_no_for_status or None
            )
    return _enrich_protocol_sibling_samples(db, items)


def _protocol_group_key_for_item(item: Dict[str, Any]) -> str:
    wo_type = (item.get("workOrderType") or "inbound").strip() or "inbound"
    if wo_type != "inbound":
        return ""
    source = (item.get("receptionSource") or "cell_storage").strip() or "cell_storage"
    proto_id = (item.get("collectionProtocolId") or "").strip()
    if proto_id:
        if source == "research_donation":
            return f"donation:{proto_id}"
        return f"protocol:{proto_id}"
    protocol_no = (item.get("protocolNo") or "").strip()
    if protocol_no:
        return f"protocol-no:{protocol_no}"
    return ""


def _enrich_protocol_sibling_samples(
    db: Session, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """入库列表：补充同协议下全部样本（供卡片标题栏「同协议样本」展示）。"""
    if not items:
        return items
    cell_proto_ids: List[str] = []
    donation_ids: List[str] = []
    for item in items:
        if (item.get("workOrderType") or "inbound").strip() not in ("", "inbound"):
            continue
        pid = (item.get("collectionProtocolId") or "").strip()
        if not pid:
            continue
        source = (item.get("receptionSource") or "cell_storage").strip() or "cell_storage"
        if source == "research_donation":
            if pid not in donation_ids:
                donation_ids.append(pid)
        elif pid not in cell_proto_ids:
            cell_proto_ids.append(pid)

    siblings_by_key: Dict[str, List[Dict[str, str]]] = {}

    if cell_proto_ids:
        proto_samples = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.collection_protocol_id.in_(cell_proto_ids),
                BizProtocolSample.del_flag == 0,
            )
            .all()
        )
        by_proto: Dict[str, List[BizProtocolSample]] = defaultdict(list)
        type_ids: List[str] = []
        tenant_no = ""
        for sample in proto_samples:
            pid = (sample.collection_protocol_id or "").strip()
            if pid:
                by_proto[pid].append(sample)
            if sample.sample_type_id and sample.sample_type_id not in type_ids:
                type_ids.append(sample.sample_type_id)
            if not tenant_no and (sample.tenant_no or "").strip():
                tenant_no = (sample.tenant_no or "").strip()
        type_map = _load_sample_type_name_map(db, type_ids)
        sample_codes = list(
            {
                (s.sample_code or "").strip()
                for samples in by_proto.values()
                for s in samples
                if (s.sample_code or "").strip()
            }
        )
        lab_map = (
            sample_reception_svc.get_lab_storage_code_map_by_sample_codes(
                db, tenant_no, sample_codes, reception_source="cell_storage"
            )
            if tenant_no and sample_codes
            else {}
        )
        for pid, samples in by_proto.items():
            entries: List[Dict[str, str]] = []
            for sample in sorted(samples, key=lambda row: (row.sample_code or "")):
                code = (sample.sample_code or "").strip()
                if not code:
                    continue
                entries.append(
                    {
                        "sampleCode": code,
                        "labStorageCode": lab_map.get(code, ""),
                        "sampleTypeName": type_map.get(sample.sample_type_id or "", ""),
                    }
                )
            if len(entries) > 1:
                siblings_by_key[f"protocol:{pid}"] = entries

    if donation_ids:
        donation_samples = (
            db.query(BizResearchDonationSample)
            .filter(
                BizResearchDonationSample.research_donation_id.in_(donation_ids),
                BizResearchDonationSample.del_flag == 0,
            )
            .all()
        )
        by_donation: Dict[str, List[BizResearchDonationSample]] = defaultdict(list)
        type_ids = []
        tenant_no = ""
        for sample in donation_samples:
            did = (sample.research_donation_id or "").strip()
            if did:
                by_donation[did].append(sample)
            if sample.sample_type_id and sample.sample_type_id not in type_ids:
                type_ids.append(sample.sample_type_id)
            if not tenant_no and (sample.tenant_no or "").strip():
                tenant_no = (sample.tenant_no or "").strip()
        type_map = _load_sample_type_name_map(db, type_ids)
        sample_codes = list(
            {
                (s.sample_code or "").strip()
                for samples in by_donation.values()
                for s in samples
                if (s.sample_code or "").strip()
            }
        )
        lab_map = (
            sample_reception_svc.get_lab_storage_code_map_by_sample_codes(
                db, tenant_no, sample_codes, reception_source="research_donation"
            )
            if tenant_no and sample_codes
            else {}
        )
        for did, samples in by_donation.items():
            entries = []
            for sample in sorted(samples, key=lambda row: (row.sample_code or "")):
                code = (sample.sample_code or "").strip()
                if not code:
                    continue
                entries.append(
                    {
                        "sampleCode": code,
                        "labStorageCode": lab_map.get(code, ""),
                        "sampleTypeName": type_map.get(sample.sample_type_id or "", ""),
                    }
                )
            if len(entries) > 1:
                siblings_by_key[f"donation:{did}"] = entries

    for item in items:
        key = _protocol_group_key_for_item(item)
        siblings = siblings_by_key.get(key)
        if siblings:
            item["protocolSiblingSamples"] = siblings
    return items


def _enrich_production_work_order_refs(
    db: Session, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    if not items:
        return items
    pwo_ids = list(
        {x.get("productionWorkOrderId") for x in items if (x.get("productionWorkOrderId") or "").strip()}
    )
    pwo_map: Dict[str, str] = {}
    if pwo_ids:
        rows = (
            db.query(BizProductionWorkOrder.id, BizProductionWorkOrder.work_order_no)
            .filter(
                BizProductionWorkOrder.id.in_(pwo_ids),
                BizProductionWorkOrder.del_flag == 0,
            )
            .all()
        )
        pwo_map = {r.id: r.work_order_no or "" for r in rows}
    for item in items:
        pid = (item.get("productionWorkOrderId") or "").strip()
        item["productionWorkOrderNo"] = pwo_map.get(pid, "") if pid else ""
    return items


def _gen_daily_no(db: Session, tenant_no: str, prefix: str) -> str:
    day = datetime.now().strftime("%Y%m%d")
    like = f"{prefix}{day}%"
    cnt = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.work_order_no.like(like),
        )
        .count()
    )
    return f"{prefix}{day}{cnt + 1:04d}"


def _user_nick_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
    ids = list({x for x in user_ids if x})
    if not ids:
        return {}
    rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
    return {u.id: (u.nick_name or "").strip() for u in rows}


def _wo_to_dict(
    wo: BizQcWorkOrder,
    *,
    outbound_info: Optional[Dict[str, Any]] = None,
    configured_by_name: str = "",
) -> Dict[str, Any]:
    wo_type = getattr(wo, "work_order_type", None) or "inbound"
    recv_source_raw = getattr(wo, "reception_source", None)
    if wo_type == "temporary":
        recv_source = (recv_source_raw or "").strip()
    else:
        recv_source = (recv_source_raw or "cell_storage").strip() or "cell_storage"
    biz_type = getattr(wo, "biz_type", None) or ""
    is_outbound = wo_type == "outbound"
    ob = outbound_info or {}
    return {
        "id": wo.id,
        "workOrderNo": wo.work_order_no,
        "workOrderType": wo_type,
        "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": recv_source,
        "receptionSourceLabel": _reception_source_label(recv_source) if recv_source else "",
        "qcBizType": _work_order_qc_biz_type(wo),
        "qcBizTypeLabel": _qc_biz_type_label(_work_order_qc_biz_type(wo)),
        "infusionScheduleId": getattr(wo, "infusion_schedule_id", None) or None,
        "bizType": biz_type if is_outbound else "",
        "bizTypeLabel": _OUTBOUND_BIZ_TYPE_LABELS.get(biz_type, "") if is_outbound else "",
        "applicationNo": ob.get("applicationNo", "") if is_outbound else "",
        "applicationName": ob.get("applicationName", "") if is_outbound else "",
        "channelName": ob.get("channelName", "") if is_outbound else "",
        "infusionTime": ob.get("infusionTime", "") if is_outbound else "",
        "cellProductId": getattr(wo, "cell_product_id", None) or "",
        "externalTargetCode": getattr(wo, "external_target_code", None) or "",
        "cellProductCode": "",
        "cellProductName": "",
        "cellProductTypeName": "",
        "cellProductPassage": "",
        "cellProductSpecName": "",
        "triggerSource": getattr(wo, "trigger_source", None) or "reception",
        "triggerSourceLabel": _TRIGGER_SOURCE_LABELS.get(
            getattr(wo, "trigger_source", None) or "reception", ""
        ),
        "productionWorkOrderId": getattr(wo, "production_work_order_id", None) or "",
        "productionWorkOrderNo": "",
        "productionTaskId": getattr(wo, "production_task_id", None) or "",
        "productionStepOutputId": getattr(wo, "production_step_output_id", None) or "",
        "receptionTime": "",
        "handlerName": "",
        "sampleTypeName": "",
        "workOrderStatus": wo.work_order_status,
        "workOrderStatusLabel": _WORK_ORDER_STATUS_LABELS.get(
            wo.work_order_status or "", wo.work_order_status or ""
        ),
        "qcConclusion": (
            (getattr(wo, "conclusion", None) or "").strip()
            if (getattr(wo, "conclusion", None) or "").strip() in _VALID_QC_WORK_ORDER_CONCLUSIONS
            else ""
        ),
        "qcConclusionLabel": work_order_qc_conclusion_label(getattr(wo, "conclusion", None) or ""),
        "hasLab": wo.has_lab == "1",
        "hasThirdParty": wo.has_third_party == "1",
        "configuredTime": _fmt_dt(wo.configured_time),
        "configuredBy": wo.configured_by or "",
        "configuredByName": configured_by_name or wo.configured_by or "",
        "thirdPartyFeeTotal": _dec(wo.third_party_fee_total),
        "remark": wo.remark or "",
        "createTime": _fmt_dt(wo.create_time),
        "updateTime": _fmt_dt(wo.update_time),
    }


def create_work_order_from_reception(
    db: Session,
    *,
    reception,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> Optional[BizQcWorkOrder]:
    """接收结论为 js 时创建待配置工单（flush，不 commit）。"""
    if (reception.accept_conclusion or "").strip() != "js":
        return None
    exists = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.reception_id == reception.id,
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.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 = BizQcWorkOrder(
            id=generate_uuid(),
            work_order_no=_gen_daily_no(db, tenant_no, "ZJ"),
            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",
            qc_biz_type="inbound",
            trigger_source="reception",
            work_order_status="dpz",
            has_lab="0",
            has_third_party="0",
            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 = BizQcWorkOrder(
            id=generate_uuid(),
            work_order_no=_gen_daily_no(db, tenant_no, "ZJ"),
            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",
            qc_biz_type="inbound",
            trigger_source="reception",
            work_order_status="dpz",
            has_lab="0",
            has_third_party="0",
            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 _valid_qc_dispatch_output_type_ids(db: Session, tenant_no: str) -> set:
    rows = (
        db.query(BizOutputType)
        .filter(
            BizOutputType.tenant_no == tenant_no,
            BizOutputType.del_flag == 0,
            BizOutputType.type_name == _VALID_QC_DISPATCH_OUTPUT_TYPE_NAME,
        )
        .all()
    )
    return {r.id for r in rows}


def _resolve_cell_product_id_by_barcode(
    db: Session, barcode: Optional[str], tenant_no: str
) -> Optional[str]:
    code = (barcode or "").strip()
    if not code:
        return None
    cp = (
        db.query(BizCellProduct)
        .filter(
            BizCellProduct.product_code == code,
            BizCellProduct.tenant_no == tenant_no,
            BizCellProduct.del_flag == 0,
        )
        .first()
    )
    return cp.id if cp else None


def create_qc_work_order_from_production(
    db: Session,
    *,
    production_work_order_id: str,
    production_step_output_id: str,
    remark: str = "",
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> BizQcWorkOrder:
    """生产过程派发质检工单（flush，不 commit）。"""
    pwo = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == production_work_order_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if not pwo:
        raise ValueError("生产工单不存在")
    wo_type = (getattr(pwo, "work_order_type", None) or "inbound").strip()
    if wo_type not in ("inbound", "outbound", "temporary"):
        raise ValueError("仅入库、出库或临时生产工单可派发质检")
    status = (pwo.work_order_status or "").strip()
    if status == "yzf":
        raise ValueError("已作废的生产工单不可派发质检")
    if status not in ("jxz", "ywc"):
        raise ValueError("仅进行中或已完成的生产工单可派发质检")

    step_output_id = (production_step_output_id or "").strip()
    if not step_output_id:
        raise ValueError("请选择步骤产出物")

    options = list_step_output_dispatch_options_for_production_work_order(
        db, production_work_order_id=production_work_order_id, tenant_no=tenant_no
    )
    option_map = {o["id"]: o for o in options}
    selected = option_map.get(step_output_id)
    if not selected:
        raise ValueError("所选步骤产出物不在该生产工单的可派发列表中")

    step_output = (
        db.query(BizProductionTaskStepOutput)
        .filter(
            BizProductionTaskStepOutput.id == step_output_id,
            BizProductionTaskStepOutput.tenant_no == tenant_no,
            BizProductionTaskStepOutput.del_flag == 0,
        )
        .first()
    )
    if not step_output:
        raise ValueError("步骤产出物不存在")

    step = (
        db.query(BizProductionTaskStep)
        .filter(
            BizProductionTaskStep.id == step_output.step_id,
            BizProductionTaskStep.tenant_no == tenant_no,
            BizProductionTaskStep.del_flag == 0,
        )
        .first()
    )
    if not step:
        raise ValueError("关联工艺步骤不存在")

    task = (
        db.query(BizProductionTask)
        .filter(
            BizProductionTask.id == step.task_id,
            BizProductionTask.work_order_id == pwo.id,
            BizProductionTask.tenant_no == tenant_no,
            BizProductionTask.del_flag == 0,
        )
        .first()
    )
    if not task:
        raise ValueError("关联生产任务不存在")

    cp_id = _resolve_cell_product_id_by_barcode(db, step_output.output_barcode, tenant_no)
    task_id = task.id

    common_fields = dict(
        trigger_source="production",
        cell_product_id=cp_id,
        production_work_order_id=pwo.id,
        production_task_id=task_id,
        production_step_output_id=step_output_id,
        work_order_status="dpz",
        has_lab="0",
        has_third_party="0",
        tenant_no=tenant_no,
        remark=(remark or "").strip() or None,
        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,
    )

    if wo_type == "outbound":
        schedule_id = (getattr(pwo, "infusion_schedule_id", None) or "").strip()
        if not schedule_id:
            raise ValueError("出库生产工单缺少输注计划关联")
        biz_type = (getattr(pwo, "biz_type", None) or "pilot_service").strip() or "pilot_service"
        if biz_type not in _OUTBOUND_BIZ_TYPE_LABELS:
            biz_type = "pilot_service"
        wo = BizQcWorkOrder(
            id=generate_uuid(),
            work_order_no=_gen_daily_no(db, tenant_no, "CKZJ"),
            work_order_type="outbound",
            reception_id=None,
            protocol_sample_id=None,
            collection_protocol_id=None,
            protocol_no=None,
            sample_code=None,
            reception_source="outbound",
            qc_biz_type="outbound",
            biz_type=biz_type,
            infusion_schedule_id=schedule_id,
            **common_fields,
        )
    elif wo_type == "temporary":
        wo = BizQcWorkOrder(
            id=generate_uuid(),
            work_order_no=_gen_daily_no(db, tenant_no, "LSZJ"),
            work_order_type="temporary",
            reception_id=None,
            protocol_sample_id=None,
            collection_protocol_id=None,
            protocol_no=None,
            sample_code=None,
            reception_source=None,
            qc_biz_type="temporary",
            **common_fields,
        )
    else:
        wo = BizQcWorkOrder(
            id=generate_uuid(),
            work_order_no=_gen_daily_no(db, tenant_no, "ZJ"),
            work_order_type="inbound",
            reception_id=None,
            protocol_sample_id=pwo.protocol_sample_id,
            collection_protocol_id=pwo.collection_protocol_id,
            protocol_no=pwo.protocol_no or None,
            sample_code=pwo.sample_code or None,
            reception_source=getattr(pwo, "reception_source", "cell_storage") or "cell_storage",
            qc_biz_type="inbound",
            **common_fields,
        )
    db.add(wo)
    db.flush()
    return wo


def list_step_output_dispatch_options_for_production_work_order(
    db: Session,
    *,
    production_work_order_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    """汇总生产工单下可派发的步骤产出物：类型为有效物且有条码。"""
    pwo = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == production_work_order_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if not pwo:
        raise ValueError("生产工单不存在")

    valid_type_ids = _valid_qc_dispatch_output_type_ids(db, tenant_no)
    if not valid_type_ids:
        return []

    tasks = (
        db.query(BizProductionTask)
        .filter(
            BizProductionTask.work_order_id == pwo.id,
            BizProductionTask.tenant_no == tenant_no,
            BizProductionTask.del_flag == 0,
        )
        .all()
    )
    if not tasks:
        return []

    task_map = {t.id: t for t in tasks}
    task_ids = list(task_map.keys())
    steps = (
        db.query(BizProductionTaskStep)
        .filter(
            BizProductionTaskStep.task_id.in_(task_ids),
            BizProductionTaskStep.tenant_no == tenant_no,
            BizProductionTaskStep.del_flag == 0,
        )
        .all()
    )
    if not steps:
        return []

    step_map = {s.id: s for s in steps}
    step_ids = list(step_map.keys())
    type_rows = (
        db.query(BizOutputType)
        .filter(
            BizOutputType.id.in_(list(valid_type_ids)),
            BizOutputType.del_flag == 0,
        )
        .all()
    )
    type_name_map = {t.id: t.type_name or "" for t in type_rows}

    outputs = (
        db.query(BizProductionTaskStepOutput)
        .filter(
            BizProductionTaskStepOutput.step_id.in_(step_ids),
            BizProductionTaskStepOutput.tenant_no == tenant_no,
            BizProductionTaskStepOutput.del_flag == 0,
            BizProductionTaskStepOutput.output_type_id.in_(list(valid_type_ids)),
        )
        .order_by(
            BizProductionTaskStepOutput.order_num.asc(),
            BizProductionTaskStepOutput.create_time.asc(),
        )
        .all()
    )

    items: List[Dict[str, Any]] = []
    for out in outputs:
        barcode = (out.output_barcode or "").strip()
        if not barcode:
            continue
        step = step_map.get(out.step_id)
        if not step:
            continue
        task = task_map.get(step.task_id)
        if not task:
            continue
        type_id = (out.output_type_id or "").strip()
        cell_product_id = _resolve_cell_product_id_by_barcode(db, barcode, tenant_no)
        items.append(
            {
                "id": out.id,
                "stepOutputId": out.id,
                "taskId": task.id,
                "taskNo": task.task_no or "",
                "taskName": task.task_name or "",
                "stepId": step.id,
                "processName": step.process_name or "",
                "outputTypeId": type_id,
                "outputTypeName": type_name_map.get(type_id, _VALID_QC_DISPATCH_OUTPUT_TYPE_NAME),
                "outputName": (out.output_name or "").strip(),
                "outputBarcode": barcode,
                "cellProductId": cell_product_id or "",
            }
        )
    return items


def list_cell_product_options_for_production_work_order(
    db: Session,
    *,
    production_work_order_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    """兼容旧接口名：返回步骤产出物派发选项。"""
    return list_step_output_dispatch_options_for_production_work_order(
        db,
        production_work_order_id=production_work_order_id,
        tenant_no=tenant_no,
    )


def _protocol_sample_ids_for_sample_type(
    db: Session,
    tenant_no: str,
    sample_type_id: str,
    reception_source: Optional[str] = None,
) -> List[str]:
    stid = (sample_type_id or "").strip()
    if not stid or not tenant_no:
        return []
    ids: List[str] = []
    rs = (reception_source or "").strip()
    if not rs or rs == "cell_storage":
        cell_rows = (
            db.query(BizProtocolSample.id)
            .filter(
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
                BizProtocolSample.sample_type_id == stid,
            )
            .all()
        )
        ids.extend([r[0] for r in cell_rows if r[0]])
    if not rs or rs == "research_donation":
        donation_rows = (
            db.query(BizResearchDonationSample.id)
            .filter(
                BizResearchDonationSample.tenant_no == tenant_no,
                BizResearchDonationSample.del_flag == 0,
                BizResearchDonationSample.sample_type_id == stid,
            )
            .all()
        )
        ids.extend([r[0] for r in donation_rows if r[0]])
    return ids


def list_work_orders(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    work_order_status: Optional[str] = None,
    work_order_type: Optional[str] = None,
    biz_type: Optional[str] = None,
    reception_source: Optional[str] = None,
    trigger_source: Optional[str] = None,
    sample_type_id: Optional[str] = None,
    create_time_start: Optional[str] = None,
    create_time_end: Optional[str] = None,
    infusion_time_start: Optional[str] = None,
    infusion_time_end: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(BizQcWorkOrder).filter(BizQcWorkOrder.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizQcWorkOrder.tenant_no == tenant_no)
        wt = (work_order_type or "").strip()
        if wt:
            q0 = q0.filter(BizQcWorkOrder.work_order_type == wt)
        else:
            q0 = q0.filter(
                ~and_(
                    BizQcWorkOrder.work_order_type == "outbound",
                    BizQcWorkOrder.biz_type == "clinical_application",
                )
            )
        bt = (biz_type or "").strip()
        if bt:
            q0 = q0.filter(BizQcWorkOrder.biz_type == bt)
        rs = (reception_source or "").strip()
        if rs:
            q0 = q0.filter(BizQcWorkOrder.reception_source == rs)
        ts = (trigger_source or "").strip()
        if ts:
            if ts == "temp_manual":
                q0 = q0.filter(
                    BizQcWorkOrder.work_order_type == "temporary",
                    BizQcWorkOrder.trigger_source != "production",
                )
            elif ts == "temp_production":
                q0 = q0.filter(
                    BizQcWorkOrder.work_order_type == "temporary",
                    BizQcWorkOrder.trigger_source == "production",
                )
            elif ts == "manual" and wt == "temporary":
                q0 = q0.filter(BizQcWorkOrder.trigger_source != "production")
            elif ts == "production" and not wt:
                q0 = q0.filter(
                    BizQcWorkOrder.trigger_source == "production",
                    BizQcWorkOrder.work_order_type != "temporary",
                )
            else:
                q0 = q0.filter(BizQcWorkOrder.trigger_source == ts)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            keyword_conditions = [
                BizQcWorkOrder.work_order_no.like(like),
                BizQcWorkOrder.sample_code.like(like),
                BizQcWorkOrder.protocol_no.like(like),
                BizQcWorkOrder.external_target_code.like(like),
            ]
            if tenant_no:
                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(BizQcWorkOrder.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(BizQcWorkOrder.sample_code.in_(lab_codes))
            if tenant_no and (wt == "outbound" or not wt):
                outbound_ids = _outbound_qc_work_order_ids_by_keyword(db, tenant_no, kw)
                if outbound_ids:
                    keyword_conditions.append(BizQcWorkOrder.id.in_(outbound_ids))
            if tenant_no and (not wt or wt == "inbound"):
                inbound_mother_ids = _inbound_qc_work_order_ids_by_mother_name(
                    db, tenant_no, kw, reception_source=rs or None
                )
                if inbound_mother_ids:
                    keyword_conditions.append(BizQcWorkOrder.id.in_(inbound_mother_ids))
            q0 = q0.filter(or_(*keyword_conditions))
        st_raw = (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(BizQcWorkOrder.work_order_status.in_(statuses))
        use_infusion_time = wt == "outbound" and (
            (infusion_time_start or "").strip() or (infusion_time_end or "").strip()
        )
        if use_infusion_time:
            q0 = q0.join(
                BizPilotInfusionSchedule,
                eq_unicode(BizQcWorkOrder.infusion_schedule_id, BizPilotInfusionSchedule.id),
            ).filter(
                BizPilotInfusionSchedule.del_flag == 0,
                BizPilotInfusionSchedule.tenant_no == tenant_no,
            )
            if (infusion_time_start or "").strip():
                q0 = q0.filter(
                    BizPilotInfusionSchedule.infusion_time
                    >= _normalize_date_bound(infusion_time_start.strip())
                )
            if (infusion_time_end or "").strip():
                q0 = q0.filter(
                    BizPilotInfusionSchedule.infusion_time
                    <= _normalize_date_bound(infusion_time_end.strip(), end_of_day=True)
                )
        else:
            if create_time_start:
                q0 = q0.filter(
                    BizQcWorkOrder.create_time >= _normalize_date_bound(create_time_start)
                )
            if create_time_end:
                q0 = q0.filter(
                    BizQcWorkOrder.create_time
                    <= _normalize_date_bound(create_time_end, end_of_day=True)
                )
        stid = (sample_type_id or "").strip()
        if stid and tenant_no:
            ps_ids = _protocol_sample_ids_for_sample_type(db, tenant_no, stid, rs or None)
            if not ps_ids:
                q0 = q0.filter(BizQcWorkOrder.id.in_([]))
            else:
                q0 = q0.filter(
                    BizQcWorkOrder.work_order_type == "inbound",
                    BizQcWorkOrder.protocol_sample_id.in_(ps_ids),
                )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizQcWorkOrder, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizQcWorkOrder.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    q_page = _filtered()
    if uses_inbound_lab_storage_list_sort(reception_source):
        q_page = q_page.order_by(*inbound_lab_storage_order_by_clauses())
    else:
        q_page = q_page.order_by(BizQcWorkOrder.create_time.desc())
    rows = (
        q_page
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = [_wo_to_dict(x) for x in rows]
    items = _enrich_work_order_list_items(db, items)
    if tenant_no:
        items = _enrich_work_order_qc_targets(db, tenant_no, items)
        items = _enrich_production_work_order_refs(db, items)
        items = _enrich_outbound_work_orders(db, tenant_no, items)
        items = enrich_work_order_fee_summaries(db, items, tenant_no)
    items = _enrich_unfinished_badge_counts(db, items)
    items = _enrich_work_order_qc_conclusions(db, items)
    return items, int(total)


def list_work_orders_by_sample(
    db: Session,
    *,
    sample_code: str,
    reception_source: str,
    tenant_no: Optional[str],
    user_info: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
    """按样本编码精确查询入库质检工单（同协议样本跳转插入卡）。"""
    code = (sample_code or "").strip()
    rs = (reception_source or "").strip()
    if not code or not rs:
        return []
    q = db.query(BizQcWorkOrder).filter(
        BizQcWorkOrder.del_flag == 0,
        BizQcWorkOrder.work_order_type == "inbound",
        BizQcWorkOrder.sample_code == code,
        BizQcWorkOrder.reception_source == rs,
    )
    if tenant_no:
        q = q.filter(BizQcWorkOrder.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizQcWorkOrder, user_info, db, tenant_no)
    rows = q.order_by(BizQcWorkOrder.create_time.desc()).all()
    items = [_wo_to_dict(x) for x in rows]
    items = _enrich_work_order_list_items(db, items)
    if tenant_no:
        items = _enrich_work_order_qc_targets(db, tenant_no, items)
        items = _enrich_production_work_order_refs(db, items)
        items = _enrich_outbound_work_orders(db, tenant_no, items)
        items = enrich_work_order_fee_summaries(db, items, tenant_no)
    items = _enrich_unfinished_badge_counts(db, items)
    items = _enrich_work_order_qc_conclusions(db, items)
    return items


def build_work_order_list_summaries(
    db: Session,
    work_orders: List[BizQcWorkOrder],
    tenant_no: Optional[str],
) -> Dict[str, Dict[str, Any]]:
    """批量构建列表级工单摘要（与 list_work_orders 单项字段一致，非详情接口）。"""
    if not work_orders:
        return {}
    unique: Dict[str, BizQcWorkOrder] = {}
    for wo in work_orders:
        wid = (wo.id or "").strip()
        if wid and wid not in unique:
            unique[wid] = wo
    items = [_wo_to_dict(wo) for wo in unique.values()]
    items = _enrich_work_order_list_items(db, items)
    if tenant_no:
        items = _enrich_work_order_qc_targets(db, tenant_no, items)
        items = _enrich_production_work_order_refs(db, items)
        items = _enrich_outbound_work_orders(db, tenant_no, items)
        items = enrich_work_order_fee_summaries(db, items, tenant_no)
    items = _enrich_unfinished_badge_counts(db, items)
    items = _enrich_work_order_qc_conclusions(db, items)
    return {item["id"]: item for item in items if item.get("id")}


_QC_TAB_COUNT_KEYS = (
    "all",
    "inbound_cell_storage",
    "inbound_research_donation",
    "outbound_pilot_service",
    "outbound_basic_experiment",
    "temporary",
)


def _empty_qc_tab_counts() -> Dict[str, int]:
    return {k: 0 for k in _QC_TAB_COUNT_KEYS}


def _resolve_wo_tab_key(
    wo_type: Optional[str],
    reception_source: Optional[str],
    biz_type: Optional[str],
) -> str:
    wt = (wo_type or "inbound").strip() or "inbound"
    if wt == "temporary":
        return "temporary"
    if wt == "outbound":
        bt = (biz_type or "pilot_service").strip() or "pilot_service"
        if bt == "basic_experiment":
            return "outbound_basic_experiment"
        if bt == "clinical_application":
            return "outbound_clinical_application"
        return "outbound_pilot_service"
    rs = (reception_source or "cell_storage").strip() or "cell_storage"
    return "inbound_research_donation" if rs == "research_donation" else "inbound_cell_storage"


def _unfinished_work_order_filter(q):
    return q.filter(
        BizQcWorkOrder.work_order_status.in_(("dpz", "ypz")),
        ~and_(
            BizQcWorkOrder.work_order_type == "outbound",
            BizQcWorkOrder.biz_type == "clinical_application",
        ),
    )


def _enrich_unfinished_badge_counts(
    db: Session, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """列表项补充未完成卡片角标：待配置=1，已配置=未完成质检任务数（至少 1）。"""
    if not items:
        return items
    unfinished_ids = [
        x["id"]
        for x in items
        if (x.get("id") or "").strip() and x.get("workOrderStatus") in ("dpz", "ypz")
    ]
    task_counts: Dict[str, int] = {}
    if unfinished_ids:
        rows = (
            db.query(BizQcTask.work_order_id, func.count(BizQcTask.id))
            .filter(
                BizQcTask.del_flag == 0,
                BizQcTask.work_order_id.in_(unfinished_ids),
                BizQcTask.task_status.notin_(("ywc", "yzf")),
            )
            .group_by(BizQcTask.work_order_id)
            .all()
        )
        task_counts = {wo_id: int(cnt or 0) for wo_id, cnt in rows}
    for item in items:
        status = item.get("workOrderStatus") or ""
        if status == "dpz":
            item["unfinishedBadgeCount"] = 1
        elif status == "ypz":
            pending = task_counts.get(item.get("id") or "", 0)
            item["unfinishedBadgeCount"] = max(pending, 1)
        else:
            item["unfinishedBadgeCount"] = 0
    return items


def _enrich_work_order_qc_conclusions(
    db: Session, items: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """列表项补充工单级质检结论：优先落库 conclusion，否则由任务结论派生。"""
    if not items:
        return items
    need_derive_ids = [
        (x.get("id") or "").strip()
        for x in items
        if (x.get("id") or "").strip()
        and (x.get("qcConclusion") or "").strip() not in _VALID_QC_WORK_ORDER_CONCLUSIONS
    ]
    conclusions_by_wo: Dict[str, List[Optional[str]]] = defaultdict(list)
    if need_derive_ids:
        rows = (
            db.query(BizQcTask.work_order_id, BizQcTask.conclusion)
            .filter(
                BizQcTask.del_flag == 0,
                BizQcTask.work_order_id.in_(need_derive_ids),
            )
            .all()
        )
        for wo_id, conclusion in rows:
            wid = (wo_id or "").strip()
            if wid:
                conclusions_by_wo[wid].append(conclusion)
    for item in items:
        stored = (item.get("qcConclusion") or "").strip()
        if stored in _VALID_QC_WORK_ORDER_CONCLUSIONS:
            item["qcConclusion"] = stored
            item["qcConclusionLabel"] = work_order_qc_conclusion_label(stored)
            continue
        wid = (item.get("id") or "").strip()
        fields = build_qc_conclusion_fields(conclusions_by_wo.get(wid, []))
        item["qcConclusion"] = fields["qcConclusion"]
        item["qcConclusionLabel"] = fields["qcConclusionLabel"]
    return items


def _aggregate_qc_tab_counts(rows) -> Dict[str, int]:
    counts = _empty_qc_tab_counts()
    for wo_type, rs, bt, cnt in rows:
        tab_key = _resolve_wo_tab_key(wo_type, rs, bt)
        if tab_key == "outbound_clinical_application":
            continue
        n = int(cnt or 0)
        counts[tab_key] = counts.get(tab_key, 0) + n
        counts["all"] += n
    return counts


def count_unfinished_work_orders_by_tab(
    db: Session,
    *,
    tenant_no: str,
    user_info=None,
) -> Dict[str, int]:
    """按 Tab 统计未完成质检工单数（待配置 + 已配置）。"""
    q = db.query(BizQcWorkOrder).filter(
        BizQcWorkOrder.del_flag == 0,
        BizQcWorkOrder.tenant_no == tenant_no,
    )
    q = _unfinished_work_order_filter(q)
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizQcWorkOrder, user_info, db, tenant_no)
    rows = (
        q.with_entities(
            BizQcWorkOrder.work_order_type,
            BizQcWorkOrder.reception_source,
            BizQcWorkOrder.biz_type,
            func.count(BizQcWorkOrder.id),
        )
        .group_by(
            BizQcWorkOrder.work_order_type,
            BizQcWorkOrder.reception_source,
            BizQcWorkOrder.biz_type,
        )
        .all()
    )
    return _aggregate_qc_tab_counts(rows)


def count_mine_unfinished_work_orders_by_tab(
    db: Session,
    *,
    tenant_no: str,
    assignee_user_id: str,
) -> Dict[str, int]:
    """我的质检任务：按 Tab 统计当前用户负责且未完成的质检工单数。"""
    q = (
        db.query(
            BizQcWorkOrder.work_order_type,
            BizQcWorkOrder.reception_source,
            BizQcWorkOrder.biz_type,
            func.count(func.distinct(BizQcWorkOrder.id)),
        )
        .join(
            BizQcTask,
            and_(
                BizQcTask.work_order_id == BizQcWorkOrder.id,
                BizQcTask.del_flag == 0,
            ),
        )
        .join(
            BizQcTaskAssignee,
            and_(
                eq_unicode(BizQcTaskAssignee.task_id, BizQcTask.id),
                unicode_ci(BizQcTaskAssignee.user_id) == assignee_user_id,
                BizQcTaskAssignee.del_flag == 0,
            ),
        )
        .filter(
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.tenant_no == tenant_no,
        )
    )
    q = _unfinished_work_order_filter(q)
    rows = (
        q.group_by(
            BizQcWorkOrder.work_order_type,
            BizQcWorkOrder.reception_source,
            BizQcWorkOrder.biz_type,
        )
        .all()
    )
    return _aggregate_qc_tab_counts(rows)


def get_work_order(
    db: Session, row_id: str, tenant_no: Optional[str] = None
) -> Optional[BizQcWorkOrder]:
    q = db.query(BizQcWorkOrder).filter(BizQcWorkOrder.id == row_id, BizQcWorkOrder.del_flag == 0)
    if tenant_no:
        q = q.filter(BizQcWorkOrder.tenant_no == tenant_no)
    return q.first()


def _lab_lines(db: Session, work_order_id: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcWorkOrderLabItem)
        .filter(
            BizQcWorkOrderLabItem.work_order_id == work_order_id,
            BizQcWorkOrderLabItem.del_flag == 0,
        )
        .order_by(BizQcWorkOrderLabItem.order_num.asc())
        .all()
    )
    return [
        {
            "id": r.id,
            "labTestItemId": r.lab_test_item_id,
            "itemCode": r.item_code,
            "itemName": r.item_name,
        }
        for r in rows
    ]


def _third_lines(db: Session, work_order_id: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.work_order_id == work_order_id,
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .order_by(BizQcWorkOrderThirdItem.order_num.asc())
        .all()
    )
    return [
        {
            "id": r.id,
            "thirdPartyItemId": r.third_party_item_id,
            "companyId": r.company_id,
            "companyName": r.company_name,
            "itemCode": r.item_code,
            "itemName": r.item_name,
            "feeAmount": _dec(r.fee_amount),
        }
        for r in rows
    ]


def _qc_info_item(db: Session, wo: BizQcWorkOrder, tenant_no: str) -> Dict[str, Any]:
    from app.crud import qc_inspection_batch as batch_svc
    from app.crud import qc_task_result as result_svc

    tasks = task_svc.list_tasks_by_work_order(db, wo.id, tenant_no)
    task_ids = [str(t.get("id") or "").strip() for t in tasks if t.get("id")]
    results_map = (
        result_svc.list_saved_results_by_task_ids(db, task_ids, tenant_no) if task_ids else {}
    )
    for t in tasks:
        tt = (t.get("taskType") or "").strip()
        ts = (t.get("taskStatus") or "").strip()
        tc = (t.get("conclusion") or "").strip()
        t["taskTypeLabel"] = _TASK_TYPE_LABELS.get(tt, tt)
        t["taskStatusLabel"] = _TASK_STATUS_LABELS.get(ts, ts)
        t["conclusionLabel"] = t.get("conclusionLabel") or _CONCLUSION_LABELS.get(tc, tc or "—")
        tid = str(t.get("id") or "").strip()
        t["results"] = results_map.get(tid, []) if tid else []
    wo_status = (wo.work_order_status or "").strip()
    biz = _work_order_qc_biz_type(wo)
    return {
        **_wo_to_dict(wo),
        "workOrderStatusLabel": _WORK_ORDER_STATUS_LABELS.get(wo_status, wo_status),
        "qcBizTypeLabel": _qc_biz_type_label(biz),
        "labItems": _lab_lines(db, wo.id),
        "thirdPartyItems": _third_lines(db, wo.id),
        "inspectionBatches": batch_svc.list_inspection_batches(db, wo.id, tenant_no),
        "tasks": tasks,
    }


def get_qc_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]]:
    """按样本批量返回门禁质检工单摘要（trigger_source=reception），供协议/捐赠查看弹窗使用。

    细胞存储：传入 sample_code_by_id 时，按租户 + 样本编码关联。
    科研捐赠等：仅传 sample_ids 时，仍按 protocol_sample_id（存捐赠样本 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(BizQcWorkOrder)
            .filter(
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
                BizQcWorkOrder.reception_source == "cell_storage",
                BizQcWorkOrder.trigger_source == "reception",
                BizQcWorkOrder.sample_code.in_(codes),
            )
            .all()
        )
        code_to_wo: Dict[str, BizQcWorkOrder] = {}
        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] = _qc_info_item(db, wo, tenant_no)
        return out

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


def _process_qc_info_items_from_wos(
    db: Session, wos: List[BizQcWorkOrder], tenant_no: str
) -> List[Dict[str, Any]]:
    if not wos:
        return []
    items = [_qc_info_item(db, wo, tenant_no) for wo in wos]
    items = _enrich_work_order_list_items(db, items)
    items = _enrich_work_order_qc_targets(db, tenant_no, items)
    items = _enrich_production_work_order_refs(db, items)
    return items


def get_process_qc_info_map_for_samples(
    db: Session,
    sample_ids: List[str],
    tenant_no: str,
    *,
    sample_code_by_id: Optional[Dict[str, str]] = None,
) -> Dict[str, List[Dict[str, Any]]]:
    """按样本批量返回生产过程触发的入库质检工单摘要。"""
    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(BizQcWorkOrder)
            .filter(
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
                BizQcWorkOrder.work_order_type == "inbound",
                BizQcWorkOrder.trigger_source == "production",
                BizQcWorkOrder.sample_code.in_(codes),
            )
            .order_by(BizQcWorkOrder.create_time.asc())
            .all()
        )
        code_to_wos: Dict[str, List[BizQcWorkOrder]] = {}
        for wo in wos:
            code = (wo.sample_code or "").strip()
            if code:
                code_to_wos.setdefault(code, []).append(wo)
        out: Dict[str, List[Dict[str, Any]]] = {}
        for sid, raw_code in sample_code_by_id.items():
            code = (raw_code or "").strip()
            if not code:
                continue
            batch = code_to_wos.get(code)
            if batch:
                out[sid] = _process_qc_info_items_from_wos(db, batch, tenant_no)
        return out

    if not sample_ids:
        return {}
    wos = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.protocol_sample_id.in_(sample_ids),
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.work_order_type == "inbound",
            BizQcWorkOrder.trigger_source == "production",
        )
        .order_by(BizQcWorkOrder.create_time.asc())
        .all()
    )
    sid_to_wos: Dict[str, List[BizQcWorkOrder]] = {}
    for wo in wos:
        sid = (wo.protocol_sample_id or "").strip()
        if sid:
            sid_to_wos.setdefault(sid, []).append(wo)
    out: Dict[str, List[Dict[str, Any]]] = {}
    for sid, batch in sid_to_wos.items():
        out[sid] = _process_qc_info_items_from_wos(db, batch, tenant_no)
    return out


def list_process_qc_info_for_sample(
    db: Session,
    *,
    sample_code: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    """按样本编码返回生产过程触发的入库质检工单摘要（供入库放行查看等）。"""
    code = (sample_code or "").strip()
    if not code or not tenant_no:
        return []
    wos = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.work_order_type == "inbound",
            BizQcWorkOrder.trigger_source == "production",
            BizQcWorkOrder.sample_code == code,
        )
        .order_by(BizQcWorkOrder.create_time.asc())
        .all()
    )
    return _process_qc_info_items_from_wos(db, wos, tenant_no)


def get_schedule_gate_qc_info(
    db: Session,
    *,
    schedule_id: str,
    tenant_no: str,
) -> Optional[Dict[str, Any]]:
    """出库计划自动质检（trigger_source=schedule），供输注计划/出库放行查看。"""
    sid = (schedule_id or "").strip()
    if not sid or not tenant_no:
        return None
    wo = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.work_order_type == "outbound",
            BizQcWorkOrder.infusion_schedule_id == sid,
            BizQcWorkOrder.trigger_source == "schedule",
        )
        .order_by(BizQcWorkOrder.create_time.asc())
        .first()
    )
    if not wo:
        return None
    return _qc_info_item(db, wo, tenant_no)


def list_process_qc_info_for_schedule(
    db: Session,
    *,
    schedule_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    """按输注计划返回生产过程触发的出库质检工单摘要。"""
    sid = (schedule_id or "").strip()
    if not sid or not tenant_no:
        return []
    wos = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.work_order_type == "outbound",
            BizQcWorkOrder.infusion_schedule_id == sid,
            BizQcWorkOrder.trigger_source == "production",
        )
        .order_by(BizQcWorkOrder.create_time.asc())
        .all()
    )
    return _process_qc_info_items_from_wos(db, wos, tenant_no)


def list_process_qc_info_for_production_work_order(
    db: Session,
    *,
    production_work_order_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    """按生产工单返回生产过程派发的质检工单摘要。"""
    pid = (production_work_order_id or "").strip()
    if not pid or not tenant_no:
        return []
    wos = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.production_work_order_id == pid,
            BizQcWorkOrder.trigger_source == "production",
        )
        .order_by(BizQcWorkOrder.create_time.asc())
        .all()
    )
    return _process_qc_info_items_from_wos(db, wos, tenant_no)


def _gate_qc_info_for_inbound_production_work_order(
    db: Session, pwo, tenant_no: str
) -> Optional[Dict[str, Any]]:
    """入库生产工单关联的样本接收质检（trigger_source=reception）。"""
    from app.models.models import BizProductionWorkOrder

    if not isinstance(pwo, BizProductionWorkOrder):
        return None
    source = (getattr(pwo, "reception_source", None) or "cell_storage").strip()
    if source not in ("cell_storage", "research_donation"):
        return None

    q = db.query(BizQcWorkOrder).filter(
        BizQcWorkOrder.tenant_no == tenant_no,
        BizQcWorkOrder.del_flag == 0,
        BizQcWorkOrder.trigger_source == "reception",
    )
    wo = None
    if source == "cell_storage":
        code = (pwo.sample_code or "").strip()
        if code:
            wo = q.filter(
                BizQcWorkOrder.reception_source == "cell_storage",
                BizQcWorkOrder.sample_code == code,
            ).first()
    else:
        psid = (pwo.protocol_sample_id or "").strip()
        if psid:
            wo = q.filter(BizQcWorkOrder.protocol_sample_id == psid).first()
        if not wo:
            code = (pwo.sample_code or "").strip()
            if code:
                wo = q.filter(BizQcWorkOrder.sample_code == code).first()
    if not wo and pwo.reception_id:
        wo = q.filter(BizQcWorkOrder.reception_id == pwo.reception_id).first()
    if not wo:
        return None
    return _qc_info_item(db, wo, tenant_no)


def get_production_work_order_qc_info(
    db: Session,
    *,
    production_work_order_id: str,
    tenant_no: str,
) -> Dict[str, Any]:
    """生产工单质检只读摘要：入库含样本接收质检 + 本工单派发的生产过程质检。"""
    from app.models.models import BizProductionWorkOrder, BizSampleReception

    pid = (production_work_order_id or "").strip()
    if not pid or not tenant_no:
        raise ValueError("生产工单不存在")
    pwo = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == pid,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if not pwo:
        raise ValueError("生产工单不存在")

    wo_type = (getattr(pwo, "work_order_type", None) or "inbound").strip()
    reception_source = (getattr(pwo, "reception_source", None) or "cell_storage").strip()
    show_gate_section = (
        wo_type == "inbound"
        and reception_source in ("cell_storage", "research_donation")
    )

    gate_qc_info = None
    sample_reception = None
    if show_gate_section:
        gate_qc_info = _gate_qc_info_for_inbound_production_work_order(db, pwo, tenant_no)
        rec = None
        if pwo.reception_id:
            rec = (
                db.query(BizSampleReception)
                .filter(
                    BizSampleReception.id == pwo.reception_id,
                    BizSampleReception.del_flag == 0,
                )
                .first()
            )
        if rec:
            sample_reception = {"acceptConclusion": rec.accept_conclusion or ""}

    process_qc_info_list = list_process_qc_info_for_production_work_order(
        db, production_work_order_id=pwo.id, tenant_no=tenant_no
    )

    return {
        "showGateSection": show_gate_section,
        "gateQcInfo": gate_qc_info,
        "processQcInfoList": process_qc_info_list,
        "sampleReception": sample_reception,
    }


def _resolve_reception_id_for_work_order_detail(
    db: Session, wo: BizQcWorkOrder, tenant_no: str
) -> Optional[str]:
    """详情接口解析接收记录 ID：优先 reception_id，生产过程工单按 sample_code 回查。"""
    rid = (wo.reception_id or "").strip()
    if rid:
        return rid
    code = (wo.sample_code or "").strip()
    if not code:
        return None
    q = db.query(BizSampleReception).filter(
        BizSampleReception.tenant_no == tenant_no,
        BizSampleReception.del_flag == 0,
        BizSampleReception.sample_code == code,
    )
    source = (getattr(wo, "reception_source", None) or "").strip()
    if source:
        q = q.filter(BizSampleReception.reception_source == source)
    rec = q.order_by(BizSampleReception.reception_time.desc()).first()
    return rec.id if rec else None


def get_work_order_detail(
    db: Session, row_id: str, tenant_no: str
) -> Dict[str, Any]:
    wo = get_work_order(db, row_id, tenant_no=tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    out = _wo_to_dict(wo)
    enriched_list = _enrich_work_order_list_items(db, [out])
    if tenant_no:
        enriched_list = _enrich_work_order_qc_targets(db, tenant_no, enriched_list)
    enriched_list = _enrich_work_order_qc_conclusions(db, enriched_list)
    enriched = enriched_list[0]
    out.update(
        {
            "receptionSource": enriched.get("receptionSource", out.get("receptionSource", "")),
            "receptionTime": enriched.get("receptionTime", ""),
            "handlerName": enriched.get("handlerName", ""),
            "labStorageCode": enriched.get("labStorageCode", ""),
            "sampleTypeName": enriched.get("sampleTypeName", ""),
            "receptionSourceLabel": enriched.get("receptionSourceLabel", ""),
            "motherName": enriched.get("motherName", ""),
            "newbornName": enriched.get("newbornName", ""),
            "protocolStatus": enriched.get("protocolStatus", ""),
            "protocolStatusLabel": enriched.get("protocolStatusLabel", ""),
            "cellProductCode": enriched.get("cellProductCode", ""),
            "cellProductName": enriched.get("cellProductName", ""),
            "cellProductTypeName": enriched.get("cellProductTypeName", ""),
            "cellProductPassage": enriched.get("cellProductPassage", ""),
            "cellProductSpecName": enriched.get("cellProductSpecName", ""),
            "qcConclusion": enriched.get("qcConclusion", ""),
            "qcConclusionLabel": enriched.get("qcConclusionLabel", ""),
        }
    )
    resolved_reception_id = _resolve_reception_id_for_work_order_detail(db, wo, tenant_no)
    reception_detail = (
        sample_reception_svc.get_reception_by_id(db, resolved_reception_id, tenant_no)
        if resolved_reception_id
        else None
    )
    if reception_detail:
        out["reception"] = reception_detail
        out["productionReception"] = reception_detail.get("productionReception")
    else:
        rec = (
            db.query(BizSampleReception)
            .filter(
                BizSampleReception.id == resolved_reception_id,
                BizSampleReception.del_flag == 0,
            )
            .first()
            if resolved_reception_id
            else None
        )
        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 "",
            "labStorageCode": (rec.lab_storage_code 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": enriched.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
            ),
        }
    out["labItems"] = _lab_lines(db, wo.id)
    out["thirdPartyItems"] = _third_lines(db, wo.id)
    out["tasks"] = task_svc.list_tasks_by_work_order(db, wo.id, tenant_no)
    from app.crud import qc_inspection_batch as batch_svc

    out["inspectionBatches"] = batch_svc.list_inspection_batches(db, wo.id, tenant_no)
    enriched_fee = enrich_work_order_fee_summaries(db, [out], tenant_no)
    if enriched_fee:
        out["paymentSummary"] = enriched_fee[0].get("paymentSummary")
        out["feeSummary"] = enriched_fee[0].get("feeSummary")
    if tenant_no:
        _enrich_production_work_order_refs(db, [out])
    return out


def _soft_delete_tasks_by_source_line_ids(
    db: Session, source_line_ids: List[str], tenant_no: str
) -> None:
    """软删指定明细行关联的质检任务及其检测结果。"""
    if not source_line_ids:
        return
    tasks = (
        db.query(BizQcTask)
        .filter(
            BizQcTask.source_line_id.in_(source_line_ids),
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
        )
        .all()
    )
    if not tasks:
        return
    now = datetime.now()
    task_ids = [t.id for t in tasks]
    assignee_svc.soft_delete_assignees_by_task_ids(db, task_ids, tenant_no)
    db.query(BizQcTaskResult).filter(
        BizQcTaskResult.task_id.in_(task_ids),
        BizQcTaskResult.tenant_no == tenant_no,
        BizQcTaskResult.del_flag == 0,
    ).update(
        {"del_flag": 1, "update_time": now},
        synchronize_session="fetch",
    )
    for t in tasks:
        t.del_flag = 1
        t.update_time = now
        db.add(t)


def _get_task_by_source_line_id(
    db: Session, source_line_id: str, tenant_no: str
) -> Optional[BizQcTask]:
    return (
        db.query(BizQcTask)
        .filter(
            BizQcTask.source_line_id == source_line_id,
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
        )
        .first()
    )


def _apply_item_assignees(
    db: Session,
    task: Optional[BizQcTask],
    item_master_id: str,
    assignees_map: Optional[Dict[str, List[str]]],
    *,
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str,
    item_label: str,
) -> None:
    if not task:
        return
    user_ids = (assignees_map or {}).get(item_master_id) or []
    if not user_ids:
        raise ValueError(f"请为{item_label}指定负责人")
    assignee_svc.sync_assignees_for_task(
        db,
        task.id,
        user_ids,
        tenant_no=tenant_no,
        operator_id=user_id,
        create_dept_id=create_dept_id,
    )


def _sync_work_order_lab_items(
    db: Session,
    wo: BizQcWorkOrder,
    *,
    order_source: str,
    lab_item_ids: List[str],
    lab_item_assignees: Optional[Dict[str, List[str]]] = None,
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str,
    seq_start: int = 0,
) -> int:
    """比对同步实验室项目明细：保留已有任务数据，仅增删变更项。"""
    now = datetime.now()
    existing_rows = (
        db.query(BizQcWorkOrderLabItem)
        .filter(
            BizQcWorkOrderLabItem.work_order_id == wo.id,
            BizQcWorkOrderLabItem.order_source == order_source,
            BizQcWorkOrderLabItem.del_flag == 0,
        )
        .all()
    )
    existing_by_item_id = {r.lab_test_item_id: r for r in existing_rows}
    new_id_set = set(lab_item_ids)
    old_id_set = set(existing_by_item_id.keys())

    remove_line_ids = [existing_by_item_id[i].id for i in old_id_set - new_id_set]
    for item_id in old_id_set - new_id_set:
        line = existing_by_item_id[item_id]
        line.del_flag = 1
        line.update_by = user_id
        line.update_time = now
        db.add(line)
    _soft_delete_tasks_by_source_line_ids(db, remove_line_ids, tenant_no)

    seq = seq_start
    for lid in lab_item_ids:
        seq += 1
        if lid in existing_by_item_id and lid in new_id_set:
            line = existing_by_item_id[lid]
            line.order_num = seq
            line.update_by = user_id
            line.update_time = now
            db.add(line)
            task = _get_task_by_source_line_id(db, line.id, tenant_no)
            if task:
                task.order_num = seq
                task.update_by = user_id
                task.update_time = now
                db.add(task)
            else:
                task = task_svc.create_task_for_line(
                    db,
                    work_order=wo,
                    task_type="sysz",
                    source_line_id=line.id,
                    item_code=line.item_code,
                    item_name=line.item_name,
                    tenant_no=tenant_no,
                    user_id=user_id,
                    create_dept_id=create_dept_id,
                    seq=seq,
                    order_source=order_source,
                )
            _apply_item_assignees(
                db,
                task,
                lid,
                lab_item_assignees,
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=create_dept_id,
                item_label="实验室自检项目",
            )
            continue

        item = lab_item_svc.get_lab_item(db, lid, tenant_no=tenant_no)
        if not item or item.status != "1":
            raise ValueError("实验室项目无效或已停用")
        line = BizQcWorkOrderLabItem(
            id=generate_uuid(),
            work_order_id=wo.id,
            order_source=order_source,
            lab_test_item_id=item.id,
            item_code=item.item_code,
            item_name=item.item_name,
            tenant_no=tenant_no,
            order_num=seq,
            create_by=user_id,
            create_dept_id=create_dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(line)
        db.flush()
        task = task_svc.create_task_for_line(
            db,
            work_order=wo,
            task_type="sysz",
            source_line_id=line.id,
            item_code=line.item_code,
            item_name=line.item_name,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id,
            seq=seq,
            order_source=order_source,
        )
        _apply_item_assignees(
            db,
            task,
            lid,
            lab_item_assignees,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id,
            item_label="实验室自检项目",
        )
    return seq


def _sync_work_order_third_party_items(
    db: Session,
    wo: BizQcWorkOrder,
    *,
    order_source: str,
    third_party_item_ids: List[str],
    third_party_item_assignees: Optional[Dict[str, List[str]]] = None,
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str,
    seq_start: int = 0,
) -> Tuple[int, Decimal]:
    """比对同步第三方项目明细：保留已有任务数据，仅增删变更项。"""
    now = datetime.now()
    existing_rows = (
        db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.work_order_id == wo.id,
            BizQcWorkOrderThirdItem.order_source == order_source,
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .all()
    )
    existing_by_item_id = {r.third_party_item_id: r for r in existing_rows}
    new_id_set = set(third_party_item_ids)
    old_id_set = set(existing_by_item_id.keys())

    remove_line_ids = [existing_by_item_id[i].id for i in old_id_set - new_id_set]
    for item_id in old_id_set - new_id_set:
        line = existing_by_item_id[item_id]
        line.del_flag = 1
        line.update_by = user_id
        line.update_time = now
        db.add(line)
    _soft_delete_tasks_by_source_line_ids(db, remove_line_ids, tenant_no)

    fee_total = Decimal("0")
    seq = seq_start
    for tid in third_party_item_ids:
        seq += 1
        if tid in existing_by_item_id and tid in new_id_set:
            line = existing_by_item_id[tid]
            line.order_num = seq
            line.update_by = user_id
            line.update_time = now
            db.add(line)
            fee_total += line.fee_amount or Decimal("0")
            task = _get_task_by_source_line_id(db, line.id, tenant_no)
            if task:
                task.order_num = seq
                task.update_by = user_id
                task.update_time = now
                db.add(task)
            else:
                task = task_svc.create_task_for_line(
                    db,
                    work_order=wo,
                    task_type="dsfjc",
                    source_line_id=line.id,
                    item_code=line.item_code,
                    item_name=line.item_name,
                    tenant_no=tenant_no,
                    user_id=user_id,
                    create_dept_id=create_dept_id,
                    seq=seq,
                    company_id=line.company_id,
                    company_name=line.company_name,
                    fee_amount=line.fee_amount,
                    order_source=order_source,
                )
            _apply_item_assignees(
                db,
                task,
                tid,
                third_party_item_assignees,
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=create_dept_id,
                item_label="第三方检测项目",
            )
            continue

        item = third_item_svc.get_item(db, tid, tenant_no=tenant_no)
        if not item or item.status != "1":
            raise ValueError("第三方项目无效或已停用")
        comp = get_company(db, item.company_id, tenant_no=tenant_no)
        if not comp:
            raise ValueError("第三方公司不存在")
        fee = item.fee_amount or Decimal("0")
        fee_total += fee
        line = BizQcWorkOrderThirdItem(
            id=generate_uuid(),
            work_order_id=wo.id,
            order_source=order_source,
            third_party_item_id=item.id,
            company_id=comp.id,
            company_name=comp.company_name,
            item_code=item.item_code,
            item_name=item.item_name,
            fee_amount=fee,
            tenant_no=tenant_no,
            order_num=seq,
            create_by=user_id,
            create_dept_id=create_dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(line)
        db.flush()
        task = task_svc.create_task_for_line(
            db,
            work_order=wo,
            task_type="dsfjc",
            source_line_id=line.id,
            item_code=line.item_code,
            item_name=line.item_name,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id,
            seq=seq,
            company_id=comp.id,
            company_name=comp.company_name,
            fee_amount=fee,
            order_source=order_source,
        )
        _apply_item_assignees(
            db,
            task,
            tid,
            third_party_item_assignees,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id,
            item_label="第三方检测项目",
        )
    return seq, fee_total


def configure_work_order(
    db: Session,
    work_order_id: str,
    *,
    has_lab: bool,
    has_third_party: bool,
    lab_item_ids: List[str],
    third_party_item_ids: List[str],
    lab_item_assignees: Optional[Dict[str, List[str]]] = None,
    third_party_item_assignees: Optional[Dict[str, List[str]]] = None,
    batches: Optional[List[Dict[str, Any]]] = None,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> BizQcWorkOrder:
    from app.crud import qc_inspection_batch as batch_svc

    wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
    if not wo:
        raise ValueError("工单不存在")

    if batches is not None:
        payload_batches = batches
    else:
        payload_batches = batch_svc.normalize_legacy_configure_to_batches(
            {
                "hasLab": has_lab,
                "hasThirdParty": has_third_party,
                "labItemIds": lab_item_ids,
                "thirdPartyItemIds": third_party_item_ids,
                "labItemAssignees": lab_item_assignees,
                "thirdPartyItemAssignees": third_party_item_assignees,
            }
        )
    return batch_svc.configure_work_order_batches(
        db,
        wo,
        payload_batches,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=create_dept_id,
    )


def try_complete_work_order(db: Session, work_order_id: str, tenant_no: str) -> None:
    pending = (
        db.query(BizQcTask)
        .filter(
            BizQcTask.work_order_id == work_order_id,
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
            BizQcTask.task_status != "ywc",
        )
        .count()
    )
    if pending:
        return
    wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
    if wo and wo.work_order_status == "ypz":
        wo.work_order_status = "ywc"
        task_conclusions = [
            c
            for (c,) in db.query(BizQcTask.conclusion)
            .filter(
                BizQcTask.work_order_id == work_order_id,
                BizQcTask.tenant_no == tenant_no,
                BizQcTask.del_flag == 0,
            )
            .all()
        ]
        derived = build_qc_conclusion_fields(task_conclusions)["qcConclusion"]
        wo.conclusion = derived or None
        wo.update_time = datetime.now()
        db.add(wo)
        from app.crud.pilot_outbound_status_sync import sync_schedule_from_qc_work_order

        sync_schedule_from_qc_work_order(db, wo)
    db.commit()


def try_reopen_work_order(db: Session, work_order_id: str, tenant_no: str) -> None:
    """任务从已完成回退时，若工单已完成则恢复为已配置。"""
    wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
    if not wo or (wo.work_order_status or "").strip() != "ywc":
        return
    wo.work_order_status = "ypz"
    wo.conclusion = None
    wo.update_time = datetime.now()
    db.add(wo)
    from app.crud.pilot_outbound_status_sync import sync_schedule_from_qc_work_order

    sync_schedule_from_qc_work_order(db, wo)
    db.commit()


# ---------- 临时质检工单 ----------


def _temp_lab_lines(db: Session, work_order_id: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcWorkOrderLabItem)
        .filter(
            BizQcWorkOrderLabItem.work_order_id == work_order_id,
            BizQcWorkOrderLabItem.order_source == "temporary",
            BizQcWorkOrderLabItem.del_flag == 0,
        )
        .order_by(BizQcWorkOrderLabItem.order_num.asc())
        .all()
    )
    return [
        {
            "id": r.id,
            "labTestItemId": r.lab_test_item_id,
            "itemCode": r.item_code,
            "itemName": r.item_name,
        }
        for r in rows
    ]


def _temp_third_lines(db: Session, work_order_id: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.work_order_id == work_order_id,
            BizQcWorkOrderThirdItem.order_source == "temporary",
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .order_by(BizQcWorkOrderThirdItem.order_num.asc())
        .all()
    )
    return [
        {
            "id": r.id,
            "thirdPartyItemId": r.third_party_item_id,
            "companyId": r.company_id,
            "companyName": r.company_name,
            "itemCode": r.item_code,
            "itemName": r.item_name,
            "feeAmount": _dec(r.fee_amount),
        }
        for r in rows
    ]


def _temp_tasks(db: Session, work_order_id: str, tenant_no: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcTask)
        .filter(
            BizQcTask.work_order_id == work_order_id,
            BizQcTask.order_source == "temporary",
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
        )
        .order_by(BizQcTask.order_num.asc())
        .all()
    )
    items = []
    for t in rows:
        d = task_svc._task_to_dict(t, "", db, tenant_no)
        tt = (t.task_type or "").strip()
        ts = (t.task_status or "").strip()
        tc = (t.conclusion or "").strip()
        d["taskTypeLabel"] = _TASK_TYPE_LABELS.get(tt, tt)
        d["taskStatusLabel"] = _TASK_STATUS_LABELS.get(ts, ts)
        d["conclusionLabel"] = _CONCLUSION_LABELS.get(tc, tc or "—")
        items.append(d)
    return items


def create_temp_qc_work_order(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    cell_product_id: str = "",
    external_target_code: str = "",
    remark: str = "",
) -> BizQcWorkOrder:
    """创建临时质检工单（work_order_type='temporary'，LSZJ前缀编号）"""
    cp_id, ext_code = resolve_temp_qc_target(
        cell_product_id=cell_product_id,
        external_target_code=external_target_code,
    )
    if cp_id:
        cp_id = _validate_cell_product_id(db, cp_id, tenant_no)
    wo = BizQcWorkOrder(
        id=generate_uuid(),
        work_order_no=_gen_daily_no(db, tenant_no, "LSZJ"),
        work_order_type="temporary",
        reception_id=None,
        protocol_sample_id=None,
        collection_protocol_id=None,
        protocol_no=None,
        sample_code=None,
        reception_source=None,
        qc_biz_type="temporary",
        trigger_source="manual",
        cell_product_id=cp_id,
        external_target_code=ext_code,
        work_order_status="dpz",
        has_lab="0",
        has_third_party="0",
        tenant_no=tenant_no,
        remark=remark,
        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 update_qc_work_order(
    db: Session,
    work_order_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    cell_product_id: Optional[str] = None,
    external_target_code: Optional[str] = None,
    work_order_status: Optional[str] = None,
    qc_conclusion: Optional[str] = None,
    remark: Optional[str] = None,
) -> BizQcWorkOrder:
    """编辑质检工单：入库/出库可改备注、状态与质检结论；临时可改备注与质检对象"""
    wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    wo_type = (getattr(wo, "work_order_type", None) or "inbound").strip()
    if wo_type == "temporary":
        # 编辑弹窗同时传两侧（一侧为空串）；备注-only 更新两侧均为 None，不改对象
        if cell_product_id is not None and external_target_code is not None:
            cp_id, ext_code = resolve_temp_qc_target(
                cell_product_id=cell_product_id,
                external_target_code=external_target_code,
            )
            if cp_id:
                cp_id = _validate_cell_product_id(db, cp_id, tenant_no)
            wo.cell_product_id = cp_id
            wo.external_target_code = ext_code
    elif cell_product_id is not None:
        pass
    if work_order_status is not None:
        st = (work_order_status or "").strip()
        if st not in _VALID_QC_WORK_ORDER_STATUSES:
            raise ValueError("工单状态无效")
        wo.work_order_status = st
    if qc_conclusion is not None:
        conc = (qc_conclusion or "").strip()
        if conc and conc not in _VALID_QC_WORK_ORDER_CONCLUSIONS:
            raise ValueError("质检结论无效")
        wo.conclusion = conc or None
    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 update_temp_qc_work_order(
    db: Session,
    work_order_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    cell_product_id: Optional[str] = None,
    work_order_status: Optional[str] = None,
    qc_conclusion: Optional[str] = None,
    remark: Optional[str] = None,
) -> BizQcWorkOrder:
    """编辑临时质检工单（备注、质检对象、状态、质检结论）"""
    return update_qc_work_order(
        db,
        work_order_id,
        tenant_no=tenant_no,
        user_id=user_id,
        cell_product_id=cell_product_id,
        work_order_status=work_order_status,
        qc_conclusion=qc_conclusion,
        remark=remark,
    )


def list_temp_qc_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]:
    def _filtered():
        q0 = db.query(BizQcWorkOrder).filter(
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.work_order_type == "temporary",
        )
        if tenant_no:
            q0 = q0.filter(BizQcWorkOrder.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizQcWorkOrder.work_order_no.like(like),
                    BizQcWorkOrder.external_target_code.like(like),
                )
            )
        st = (work_order_status or "").strip()
        if st:
            q0 = q0.filter(BizQcWorkOrder.work_order_status == st)
        if user_info:
            from app.core.data_scope import apply_data_scope

            q0 = apply_data_scope(q0, BizQcWorkOrder, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizQcWorkOrder.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizQcWorkOrder.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = [_wo_to_dict(x) for x in rows]
    if tenant_no:
        items = _enrich_work_order_qc_targets(db, tenant_no, items)
    items = _enrich_work_order_qc_conclusions(db, items)
    return items, int(total)


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


def get_temp_qc_work_order_detail(
    db: Session, row_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    wo = get_temp_qc_work_order(db, row_id, tenant_no)
    if not wo:
        return None
    out = _wo_to_dict(wo)
    enriched_list = _enrich_work_order_qc_targets(db, tenant_no, [out])
    out = enriched_list[0]
    out["labItems"] = _temp_lab_lines(db, wo.id)
    out["thirdPartyItems"] = _temp_third_lines(db, wo.id)
    out["tasks"] = _temp_tasks(db, wo.id, tenant_no)
    return out


def configure_temp_qc_work_order(
    db: Session,
    work_order_id: str,
    *,
    has_lab: bool,
    has_third_party: bool,
    lab_item_ids: List[str],
    third_party_item_ids: List[str],
    batches: Optional[List[Dict[str, Any]]] = None,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> BizQcWorkOrder:
    wo = get_temp_qc_work_order(db, work_order_id, tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    from app.crud import qc_inspection_batch as batch_svc

    if batches is not None:
        payload_batches = batches
    else:
        payload_batches = batch_svc.normalize_legacy_configure_to_batches(
            {
                "hasLab": has_lab,
                "hasThirdParty": has_third_party,
                "labItemIds": lab_item_ids,
                "thirdPartyItemIds": third_party_item_ids,
            }
        )
    return batch_svc.configure_work_order_batches(
        db,
        wo,
        payload_batches,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=create_dept_id,
    )


def try_complete_temp_qc_work_order(
    db: Session, work_order_id: str, tenant_no: str
) -> None:
    pending = (
        db.query(BizQcTask)
        .filter(
            BizQcTask.work_order_id == work_order_id,
            BizQcTask.order_source == "temporary",
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
            BizQcTask.task_status != "ywc",
        )
        .count()
    )
    if pending:
        return
    wo = get_temp_qc_work_order(db, work_order_id, tenant_no)
    if wo and wo.work_order_status == "ypz":
        wo.work_order_status = "ywc"
        wo.update_time = datetime.now()
        db.add(wo)
        db.commit()


def delete_temp_qc_work_order(
    db: Session, work_order_id: str, tenant_no: str, user_id: Optional[str]
) -> None:
    wo = get_temp_qc_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(BizQcTask)
        .filter(
            BizQcTask.work_order_id == work_order_id,
            BizQcTask.order_source == "temporary",
            BizQcTask.del_flag == 0,
        )
        .all()
    )
    for task in tasks:
        results = (
            db.query(BizQcTaskResult)
            .filter(
                BizQcTaskResult.task_id == task.id,
                BizQcTaskResult.del_flag == 0,
            )
            .all()
        )
        for r in results:
            r.del_flag = 1
            r.update_by = user_id
            r.update_time = now
        task.del_flag = 1
        task.update_by = user_id
        task.update_time = now

    lab_items = (
        db.query(BizQcWorkOrderLabItem)
        .filter(
            BizQcWorkOrderLabItem.work_order_id == work_order_id,
            BizQcWorkOrderLabItem.order_source == "temporary",
            BizQcWorkOrderLabItem.del_flag == 0,
        )
        .all()
    )
    for item in lab_items:
        item.del_flag = 1
        item.update_by = user_id
        item.update_time = now

    third_items = (
        db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.work_order_id == work_order_id,
            BizQcWorkOrderThirdItem.order_source == "temporary",
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .all()
    )
    for item in third_items:
        item.del_flag = 1
        item.update_by = user_id
        item.update_time = now

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

    db.commit()


def build_outbound_pilot_context_by_schedule(
    db: Session, schedule_id: str, tenant_no: str
) -> Dict[str, Any]:
    """按输注计划 ID 组装出库中试服务上下文（申请单/患者/方案/计划/产品），供质检与生产工单详情复用。"""
    out: Dict[str, Any] = {
        "application": None,
        "patient": None,
        "infusionPlan": None,
        "infusionSchedule": None,
        "products": [],
    }
    if not schedule_id:
        return out

    sched = (
        db.query(BizPilotInfusionSchedule)
        .filter(
            BizPilotInfusionSchedule.id == schedule_id,
            BizPilotInfusionSchedule.tenant_no == tenant_no,
            BizPilotInfusionSchedule.del_flag == 0,
        )
        .first()
    )
    if sched:
        out["infusionSchedule"] = {
            "infusionTime": _fmt_dt(sched.infusion_time),
            "infusionLocation": sched.infusion_location or "",
            "infusionMethod": sched.infusion_method or "",
            "cellType": sched.cell_type or "",
            "specText": sched.spec_text or "",
            "scheduleStatus": sched.schedule_status or "",
            "infusionStatus": sched.infusion_status or "",
            "qcStatus": sched.qc_status or "",
            "cancelReason": sched.cancel_reason or "",
            "amount": _dec(sched.amount) if sched.amount is not None else None,
        }

        plan = (
            db.query(BizPilotInfusionPlan)
            .filter(
                BizPilotInfusionPlan.id == (sched.infusion_plan_id or ""),
                BizPilotInfusionPlan.tenant_no == tenant_no,
                BizPilotInfusionPlan.del_flag == 0,
            )
            .first()
        )
        if plan:
            out["infusionPlan"] = {
                "cellType": plan.cell_type or "",
                "specText": plan.spec_text or "",
                "totalCourseCount": plan.total_course_count,
                "courseInterval": plan.course_interval or "",
                "infusionRoute": plan.infusion_route or "",
                "infusionStartDate": plan.infusion_start_date or "",
            }

            app = (
                db.query(BizPilotApplication)
                .filter(
                    BizPilotApplication.id == (plan.pilot_application_id or ""),
                    BizPilotApplication.tenant_no == tenant_no,
                    BizPilotApplication.del_flag == 0,
                )
                .first()
            )
            if app:
                app_biz = (app.biz_type or "").strip() or "pilot_service"
                out["application"] = {
                    "applicationNo": app.application_no or "",
                    "applicationName": app.application_name or "",
                    "applicationStatus": app.application_status or "",
                    "channelName": app.channel_name or "",
                    "bizType": app_biz,
                    "bizTypeLabel": _OUTBOUND_BIZ_TYPE_LABELS.get(app_biz, app_biz),
                }
                if app_biz == "basic_experiment":
                    out["application"]["projectTitle"] = app.project_title or ""
                    out["application"]["customerName"] = app.customer_name or ""
                    out["application"]["customerContact"] = app.customer_contact or ""
                    out["application"]["customerPhone"] = app.customer_phone or ""
                else:
                    patient = (
                        db.query(BizPilotPatient)
                        .filter(
                            BizPilotPatient.pilot_application_id == app.id,
                            BizPilotPatient.tenant_no == tenant_no,
                            BizPilotPatient.del_flag == 0,
                        )
                        .first()
                    )
                    if patient:
                        out["patient"] = {
                            "patientName": patient.patient_name or "",
                            "sex": patient.sex or "",
                            "age": str(patient.age) if patient.age is not None else "",
                            "phoneNo": patient.phone_no or "",
                            "idCardNo": patient.id_card_no or "",
                            "emergencyContact": patient.emergency_contact or "",
                            "emergencyPhone": patient.emergency_phone or "",
                            "clinicalDiagnosis": patient.clinical_diagnosis or "",
                            "diagnosisAttachments": patient.diagnosis_attachments or "",
                        }

    products = (
        db.query(BizPilotInfusionScheduleProduct)
        .filter(
            BizPilotInfusionScheduleProduct.infusion_schedule_id == schedule_id,
            BizPilotInfusionScheduleProduct.tenant_no == tenant_no,
            BizPilotInfusionScheduleProduct.del_flag == 0,
        )
        .all()
    )
    out["products"] = [
        {
            "projectCode": p.project_code or "",
            "projectName": p.project_name or "",
            "cellTypeName": p.cell_type_name or "",
            "specText": p.spec_text or "",
            "agreementPrice": _dec(p.agreement_price),
            "actualPrice": _dec(p.actual_price),
            "quantity": p.quantity,
            "productStatus": p.product_status or "",
        }
        for p in products
    ]
    return out


def get_outbound_qc_detail(
    db: Session, work_order_id: str, tenant_no: str
) -> Dict[str, Any]:
    wo = get_work_order(db, work_order_id, tenant_no=tenant_no)
    if not wo:
        raise ValueError("工单不存在")
    wo_type = getattr(wo, "work_order_type", None) or "inbound"
    if wo_type != "outbound":
        raise ValueError("该工单不是出库质检类型")

    configured_by = wo.configured_by or ""
    configured_by_name = ""
    if configured_by:
        configured_by_name = _user_nick_name_map(db, [configured_by]).get(configured_by, "")

    schedule_id = getattr(wo, "infusion_schedule_id", None) or ""
    pilot_ctx = build_outbound_pilot_context_by_schedule(db, schedule_id, tenant_no)

    out: Dict[str, Any] = {
        "workOrder": _wo_to_dict(wo, configured_by_name=configured_by_name),
        **pilot_ctx,
        "tasks": [],
    }

    from app.crud import qc_inspection_batch as batch_svc

    out["tasks"] = task_svc.list_tasks_by_work_order(db, wo.id, tenant_no)
    out["inspectionBatches"] = batch_svc.list_inspection_batches(db, wo.id, tenant_no)
    enriched_fee = enrich_work_order_fee_summaries(db, [out["workOrder"]], tenant_no)
    if enriched_fee:
        out["workOrder"]["feeSummary"] = enriched_fee[0].get("feeSummary")
        out["feeSummary"] = enriched_fee[0].get("feeSummary")
    return out


def create_qc_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",
    qc_biz_type: str = "inbound",
    remark: str = "",
) -> BizQcWorkOrder:
    """手动创建质检工单（不关联接收记录）"""
    wo = BizQcWorkOrder(
        id=generate_uuid(),
        work_order_no=_gen_daily_no(db, tenant_no, "ZJ"),
        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,
        qc_biz_type=qc_biz_type,
        work_order_status="dpz",
        has_lab="0",
        has_third_party="0",
        tenant_no=tenant_no,
        remark=remark,
        create_by=user_id,
        create_dept_id=create_dept_id,
    )
    db.add(wo)
    db.flush()
    return wo


def create_outbound_qc_work_order(
    db: Session,
    *,
    schedule_id: str,
    cell_product_id: str = "",
    tenant_no: str = "",
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    remark: str = "",
    biz_type: Optional[str] = None,
) -> BizQcWorkOrder:
    resolved_biz_type = (biz_type or "").strip() or resolve_outbound_biz_type_for_schedule(
        db, schedule_id, tenant_no
    )
    if resolved_biz_type not in _OUTBOUND_BIZ_TYPE_LABELS:
        resolved_biz_type = "pilot_service"

    existing = (
        db.query(BizQcWorkOrder)
        .filter(
            BizQcWorkOrder.infusion_schedule_id == schedule_id,
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
            BizQcWorkOrder.trigger_source == "schedule",
        )
        .first()
    )
    if existing:
        if (existing.biz_type or "") != resolved_biz_type:
            existing.biz_type = resolved_biz_type
            db.flush()
        return existing
    wo = BizQcWorkOrder(
        id=generate_uuid(),
        work_order_no=_gen_daily_no(db, tenant_no, "CKZJ"),
        work_order_type="outbound",
        reception_id=None,
        protocol_sample_id=None,
        collection_protocol_id=None,
        protocol_no=None,
        sample_code=None,
        reception_source="outbound",
        qc_biz_type="outbound",
        biz_type=resolved_biz_type,
        infusion_schedule_id=schedule_id,
        trigger_source="schedule",
        cell_product_id=cell_product_id or None,
        work_order_status="dpz",
        has_lab="0",
        has_third_party="0",
        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_qc_work_order(db: Session, work_order_id: str, tenant_no: str, user_id: Optional[str]) -> None:
    """软删除质检工单及其关联的任务、任务结果、检测项目和第三方检测项目"""
    wo = db.query(BizQcWorkOrder).filter(
        and_(
            BizQcWorkOrder.id == work_order_id,
            BizQcWorkOrder.tenant_no == tenant_no,
            BizQcWorkOrder.del_flag == 0,
        )
    ).first()
    if not wo:
        raise ValueError("工单不存在")
    if wo.work_order_status == "ywc":
        raise ValueError("已完成的工单不能删除")

    now = datetime.now()

    # 1. 软删除所有关联任务及其结果
    tasks = db.query(BizQcTask).filter(
        and_(BizQcTask.work_order_id == work_order_id, BizQcTask.del_flag == 0)
    ).all()
    from app.models.qc_param import BizQcTaskResult

    for task in tasks:
        results = db.query(BizQcTaskResult).filter(
            and_(BizQcTaskResult.task_id == task.id, BizQcTaskResult.del_flag == 0)
        ).all()
        for r in results:
            r.del_flag = 1
            r.update_by = user_id
            r.update_time = now
        task.del_flag = 1
        task.update_by = user_id
        task.update_time = now

    # 2. 软删除检测项目
    lab_items = db.query(BizQcWorkOrderLabItem).filter(
        and_(BizQcWorkOrderLabItem.work_order_id == work_order_id, BizQcWorkOrderLabItem.del_flag == 0)
    ).all()
    for item in lab_items:
        item.del_flag = 1
        item.update_by = user_id
        item.update_time = now

    # 3. 软删除第三方检测项目
    third_items = db.query(BizQcWorkOrderThirdItem).filter(
        and_(BizQcWorkOrderThirdItem.work_order_id == work_order_id, BizQcWorkOrderThirdItem.del_flag == 0)
    ).all()
    for item in third_items:
        item.del_flag = 1
        item.update_by = user_id
        item.update_time = now

    # 4. 软删除工单本体
    wo.del_flag = 1
    wo.update_by = user_id
    wo.update_time = now

    db.commit()