zkxw/backend/app/crud/qc_inspection_batch.py

"""质检工单检测批次 CRUD"""
from __future__ import annotations

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

from sqlalchemy.orm import Session

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.models.models import (
    BizQcBatchAssignee,
    BizQcTask,
    BizQcWorkOrder,
    BizQcWorkOrderInspectionBatch,
    BizQcWorkOrderLabItem,
    BizQcWorkOrderThirdItem,
    SysUser,
)

_CONFIGURABLE_WORK_ORDER_STATUSES = frozenset({"dpz", "ypz", "ywc", "yzf"})


def _is_voided_work_order(wo: BizQcWorkOrder) -> bool:
    return (wo.work_order_status or "").strip() == "yzf"


def _dec(v) -> float:
    if v is None:
        return 0.0
    return float(v)


def _unique_ids(ids: List[str]) -> List[str]:
    seen: Set[str] = set()
    out: List[str] = []
    for x in ids:
        v = (x or "").strip()
        if not v or v in seen:
            continue
        seen.add(v)
        out.append(v)
    return out


def _validate_batch_payload(batch: Dict[str, Any], index: int) -> None:
    label = f"批次 {index + 1}"
    batch_name = (batch.get("batchName") or batch.get("batch_name") or "").strip()
    lab_ids = _unique_ids(batch.get("labItemIds") or batch.get("lab_item_ids") or [])
    third_ids = _unique_ids(batch.get("thirdPartyItemIds") or batch.get("third_party_item_ids") or [])
    assignees = batch.get("assigneeUserIds") or batch.get("assignee_user_ids") or []
    if not batch_name:
        raise ValueError(f"{label}：请填写批次名称")
    if len(lab_ids) != len(set(lab_ids)):
        raise ValueError(f"{label}：实验室项目不能重复")
    if len(third_ids) != len(set(third_ids)):
        raise ValueError(f"{label}：第三方项目不能重复")
    if not assignees:
        raise ValueError(f"{label}：请指定负责人")


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


def _next_batch_no(db: Session, wo: BizQcWorkOrder, tenant_no: str) -> str:
    prefix = f"{wo.work_order_no}-B"
    rows = (
        db.query(BizQcWorkOrderInspectionBatch.batch_no)
        .filter(
            BizQcWorkOrderInspectionBatch.work_order_id == wo.id,
            BizQcWorkOrderInspectionBatch.tenant_no == tenant_no,
            BizQcWorkOrderInspectionBatch.del_flag == 0,
        )
        .all()
    )
    max_seq = 0
    for (bn,) in rows:
        s = (bn or "")[len(prefix) :]
        if s.isdigit():
            max_seq = max(max_seq, int(s))
    return f"{prefix}{max_seq + 1:02d}"


def _allocate_batch_name(db: Session, wo_id: str, tenant_no: str, day: datetime) -> str:
    base = day.strftime("%Y%m%d") + "检测"
    rows = (
        db.query(BizQcWorkOrderInspectionBatch.batch_name)
        .filter(
            BizQcWorkOrderInspectionBatch.work_order_id == wo_id,
            BizQcWorkOrderInspectionBatch.tenant_no == tenant_no,
            BizQcWorkOrderInspectionBatch.del_flag == 0,
        )
        .all()
    )
    same_day = [n for (n,) in rows if (n or "").startswith(base)]
    if not same_day:
        return base
    if base not in same_day:
        return base
    suffix = 2
    while f"{base}-{suffix}" in same_day:
        suffix += 1
    return f"{base}-{suffix}"


def sync_batch_assignees(
    db: Session,
    batch_id: str,
    user_ids: List[str],
    *,
    tenant_no: str,
    operator_id: Optional[str],
    create_dept_id: str = "",
) -> List[str]:
    normalized = assignee_svc.validate_assignee_user_ids(db, user_ids, tenant_no)
    now = datetime.now()
    existing = (
        db.query(BizQcBatchAssignee)
        .filter(
            BizQcBatchAssignee.batch_id == batch_id,
            BizQcBatchAssignee.tenant_no == tenant_no,
            BizQcBatchAssignee.del_flag == 0,
        )
        .all()
    )
    keep = set(normalized)
    for row in existing:
        if row.user_id not in keep:
            row.del_flag = 1
            row.update_by = operator_id
            row.update_time = now
    for idx, uid in enumerate(normalized):
        hit = next((r for r in existing if r.user_id == uid and r.del_flag == 0), None)
        if hit:
            continue
        db.add(
            BizQcBatchAssignee(
                id=generate_uuid(),
                batch_id=batch_id,
                user_id=uid,
                tenant_no=tenant_no,
                order_num=idx,
                create_by=operator_id,
                create_dept_id=create_dept_id or "",
                create_time=now,
                update_by=operator_id,
                update_time=now,
                del_flag=0,
            )
        )
    db.flush()
    return normalized


def _batch_assignee_user_ids(db: Session, batch_id: str, tenant_no: str) -> List[str]:
    rows = (
        db.query(BizQcBatchAssignee.user_id)
        .filter(
            BizQcBatchAssignee.batch_id == batch_id,
            BizQcBatchAssignee.tenant_no == tenant_no,
            BizQcBatchAssignee.del_flag == 0,
        )
        .order_by(BizQcBatchAssignee.order_num.asc())
        .all()
    )
    return [r[0] for r in rows]


def _assignee_names(db: Session, user_ids: List[str]) -> str:
    if not user_ids:
        return ""
    rows = (
        db.query(SysUser.id, SysUser.nick_name, SysUser.user_no)
        .filter(SysUser.id.in_(user_ids), SysUser.del_flag == 0)
        .all()
    )
    name_map = {r.id: (r.nick_name or r.user_no or r.id) for r in rows}
    return "、".join(name_map.get(uid, uid) for uid in user_ids)


def _prefetch_batch_side_data(
    db: Session,
    batches: List[BizQcWorkOrderInspectionBatch],
    tenant_no: str,
) -> Dict[str, Any]:
    """一次预取多批次的负责人 / 实验室项 / 第三方项 / 用户名。"""
    batch_ids = [b.id for b in batches if b and b.id]
    assignees_by_batch: Dict[str, List[str]] = {bid: [] for bid in batch_ids}
    lab_by_batch: Dict[str, List[str]] = {bid: [] for bid in batch_ids}
    third_by_batch: Dict[str, List[str]] = {bid: [] for bid in batch_ids}
    name_map: Dict[str, str] = {}
    if not batch_ids:
        return {
            "assignees_by_batch": assignees_by_batch,
            "lab_by_batch": lab_by_batch,
            "third_by_batch": third_by_batch,
            "name_map": name_map,
        }

    for batch_id, user_id in (
        db.query(BizQcBatchAssignee.batch_id, BizQcBatchAssignee.user_id)
        .filter(
            BizQcBatchAssignee.batch_id.in_(batch_ids),
            BizQcBatchAssignee.tenant_no == tenant_no,
            BizQcBatchAssignee.del_flag == 0,
        )
        .order_by(BizQcBatchAssignee.order_num.asc())
        .all()
    ):
        assignees_by_batch.setdefault(batch_id, []).append(user_id)

    for batch_id, lab_id in (
        db.query(BizQcWorkOrderLabItem.batch_id, BizQcWorkOrderLabItem.lab_test_item_id)
        .filter(
            BizQcWorkOrderLabItem.batch_id.in_(batch_ids),
            BizQcWorkOrderLabItem.del_flag == 0,
        )
        .order_by(BizQcWorkOrderLabItem.order_num.asc())
        .all()
    ):
        if lab_id:
            lab_by_batch.setdefault(batch_id, []).append(lab_id)

    for batch_id, third_id in (
        db.query(BizQcWorkOrderThirdItem.batch_id, BizQcWorkOrderThirdItem.third_party_item_id)
        .filter(
            BizQcWorkOrderThirdItem.batch_id.in_(batch_ids),
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .order_by(BizQcWorkOrderThirdItem.order_num.asc())
        .all()
    ):
        if third_id:
            third_by_batch.setdefault(batch_id, []).append(third_id)

    all_user_ids = _unique_ids(
        [uid for ids in assignees_by_batch.values() for uid in ids]
    )
    if all_user_ids:
        for row in (
            db.query(SysUser.id, SysUser.nick_name, SysUser.user_no)
            .filter(SysUser.id.in_(all_user_ids), SysUser.del_flag == 0)
            .all()
        ):
            name_map[row.id] = row.nick_name or row.user_no or row.id

    return {
        "assignees_by_batch": assignees_by_batch,
        "lab_by_batch": lab_by_batch,
        "third_by_batch": third_by_batch,
        "name_map": name_map,
    }


def _apply_assignees_to_batch_tasks(
    db: Session,
    batch_id: str,
    user_ids: List[str],
    *,
    tenant_no: str,
    operator_id: Optional[str],
    create_dept_id: str,
) -> None:
    tasks = (
        db.query(BizQcTask)
        .filter(
            BizQcTask.batch_id == batch_id,
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
        )
        .all()
    )
    for task in tasks:
        assignee_svc.sync_assignees_for_task(
            db,
            task.id,
            user_ids,
            tenant_no=tenant_no,
            operator_id=operator_id,
            create_dept_id=create_dept_id,
        )


def _batch_has_blocking_tasks(db: Session, batch_id: str, tenant_no: str) -> bool:
    return (
        db.query(BizQcTask)
        .filter(
            BizQcTask.batch_id == batch_id,
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
            BizQcTask.task_status.in_(["jxz", "ywc"]),
        )
        .first()
        is not None
    )


def _soft_delete_batch(
    db: Session,
    batch: BizQcWorkOrderInspectionBatch,
    *,
    tenant_no: str,
    user_id: Optional[str],
    force: bool = False,
) -> None:
    from app.crud.qc_work_order import _soft_delete_tasks_by_source_line_ids

    if not force and _batch_has_blocking_tasks(db, batch.id, tenant_no):
        raise ValueError(f"批次「{batch.batch_name}」存在进行中或已完成任务，无法删除")
    now = datetime.now()
    lab_lines = (
        db.query(BizQcWorkOrderLabItem)
        .filter(
            BizQcWorkOrderLabItem.batch_id == batch.id,
            BizQcWorkOrderLabItem.del_flag == 0,
        )
        .all()
    )
    third_lines = (
        db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.batch_id == batch.id,
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .all()
    )
    line_ids = [r.id for r in lab_lines + third_lines]
    for line in lab_lines + third_lines:
        line.del_flag = 1
        line.update_by = user_id
        line.update_time = now
    _soft_delete_tasks_by_source_line_ids(db, line_ids, tenant_no)
    assignees = (
        db.query(BizQcBatchAssignee)
        .filter(BizQcBatchAssignee.batch_id == batch.id, BizQcBatchAssignee.del_flag == 0)
        .all()
    )
    for a in assignees:
        a.del_flag = 1
        a.update_by = user_id
        a.update_time = now
    batch.del_flag = 1
    batch.update_by = user_id
    batch.update_time = now


def _sync_batch_lab_items(
    db: Session,
    wo: BizQcWorkOrder,
    batch: BizQcWorkOrderInspectionBatch,
    *,
    order_source: str,
    lab_item_ids: List[str],
    assignee_user_ids: List[str],
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str,
    seq_start: int,
) -> int:
    from app.crud.qc_work_order import _get_task_by_source_line_id, _soft_delete_tasks_by_source_line_ids

    now = datetime.now()
    existing_rows = (
        db.query(BizQcWorkOrderLabItem)
        .filter(
            BizQcWorkOrderLabItem.batch_id == batch.id,
            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())

    allow_force = _is_voided_work_order(wo)
    for item_id in old_id_set - new_id_set:
        line = existing_by_item_id[item_id]
        task = _get_task_by_source_line_id(db, line.id, tenant_no)
        if (
            not allow_force
            and task
            and (task.task_status or "") in ("jxz", "ywc")
        ):
            raise ValueError(f"项目「{line.item_name}」任务进行中或已完成，无法移除")
        line.del_flag = 1
        line.update_by = user_id
        line.update_time = now
        db.add(line)
    remove_line_ids = [existing_by_item_id[i].id for i in old_id_set - new_id_set]
    _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.batch_id = batch.id
                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,
                    batch_id=batch.id,
                )
            assignee_svc.sync_assignees_for_task(
                db,
                task.id,
                assignee_user_ids,
                tenant_no=tenant_no,
                operator_id=user_id,
                create_dept_id=create_dept_id,
            )
            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,
            batch_id=batch.id,
            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,
            batch_id=batch.id,
        )
        assignee_svc.sync_assignees_for_task(
            db,
            task.id,
            assignee_user_ids,
            tenant_no=tenant_no,
            operator_id=user_id,
            create_dept_id=create_dept_id,
        )
    return seq


def _sync_batch_third_items(
    db: Session,
    wo: BizQcWorkOrder,
    batch: BizQcWorkOrderInspectionBatch,
    *,
    order_source: str,
    third_party_item_ids: List[str],
    assignee_user_ids: List[str],
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str,
    seq_start: int,
) -> Tuple[int, Decimal]:
    from app.crud.qc_work_order import _get_task_by_source_line_id, _soft_delete_tasks_by_source_line_ids

    now = datetime.now()
    existing_rows = (
        db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.batch_id == batch.id,
            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())

    allow_force = _is_voided_work_order(wo)
    for item_id in old_id_set - new_id_set:
        line = existing_by_item_id[item_id]
        task = _get_task_by_source_line_id(db, line.id, tenant_no)
        if (
            not allow_force
            and task
            and (task.task_status or "") in ("jxz", "ywc")
        ):
            raise ValueError(f"项目「{line.item_name}」任务进行中或已完成，无法移除")
        line.del_flag = 1
        line.update_by = user_id
        line.update_time = now
        db.add(line)
    remove_line_ids = [existing_by_item_id[i].id for i in old_id_set - new_id_set]
    _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.batch_id = batch.id
                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,
                    batch_id=batch.id,
                )
            assignee_svc.sync_assignees_for_task(
                db,
                task.id,
                assignee_user_ids,
                tenant_no=tenant_no,
                operator_id=user_id,
                create_dept_id=create_dept_id,
            )
            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,
            batch_id=batch.id,
            third_party_item_id=item.id,
            company_id=item.company_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=line.company_id,
            company_name=line.company_name,
            fee_amount=line.fee_amount,
            order_source=order_source,
            batch_id=batch.id,
        )
        assignee_svc.sync_assignees_for_task(
            db,
            task.id,
            assignee_user_ids,
            tenant_no=tenant_no,
            operator_id=user_id,
            create_dept_id=create_dept_id,
        )
    return seq, fee_total


def _batch_item_ids(db: Session, batch_id: str) -> Tuple[List[str], List[str]]:
    lab_ids = [
        r.lab_test_item_id
        for r in db.query(BizQcWorkOrderLabItem)
        .filter(BizQcWorkOrderLabItem.batch_id == batch_id, BizQcWorkOrderLabItem.del_flag == 0)
        .order_by(BizQcWorkOrderLabItem.order_num.asc())
        .all()
    ]
    third_ids = [
        r.third_party_item_id
        for r in db.query(BizQcWorkOrderThirdItem)
        .filter(
            BizQcWorkOrderThirdItem.batch_id == batch_id,
            BizQcWorkOrderThirdItem.del_flag == 0,
        )
        .order_by(BizQcWorkOrderThirdItem.order_num.asc())
        .all()
    ]
    return lab_ids, third_ids


def _batch_to_dict(
    db: Session,
    batch: BizQcWorkOrderInspectionBatch,
    *,
    tenant_no: str,
    tasks: Optional[List[Dict[str, Any]]] = None,
    side_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    if side_data is not None:
        assignee_ids = list(side_data["assignees_by_batch"].get(batch.id, []))
        lab_ids = list(side_data["lab_by_batch"].get(batch.id, []))
        third_ids = list(side_data["third_by_batch"].get(batch.id, []))
        name_map = side_data.get("name_map") or {}
        assignee_names = "、".join(name_map.get(uid, uid) for uid in assignee_ids)
    else:
        assignee_ids = _batch_assignee_user_ids(db, batch.id, tenant_no)
        lab_ids, third_ids = _batch_item_ids(db, batch.id)
        assignee_names = _assignee_names(db, assignee_ids)
    task_list = tasks or []
    completed = sum(1 for t in task_list if (t.get("taskStatus") or "") == "ywc")
    return {
        "id": batch.id,
        "workOrderId": batch.work_order_id,
        "batchNo": batch.batch_no,
        "batchName": batch.batch_name,
        "orderNum": batch.order_num or 0,
        "assigneeUserIds": assignee_ids,
        "assigneeUserNames": assignee_names,
        "hasLab": bool(lab_ids),
        "hasThirdParty": bool(third_ids),
        "labItemIds": lab_ids,
        "thirdPartyItemIds": third_ids,
        "taskCount": len(task_list),
        "completedTaskCount": completed,
        "tasks": task_list,
    }


def _build_inspection_batch_list(
    db: Session,
    work_order_id: str,
    batches: List[BizQcWorkOrderInspectionBatch],
    all_tasks: List[Dict[str, Any]],
    tenant_no: str,
    *,
    side_data: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
    tasks_by_batch: Dict[str, List[Dict[str, Any]]] = {}
    orphan: List[Dict[str, Any]] = []
    batch_ids = {b.id for b in batches}
    for t in all_tasks:
        bid = t.get("batchId") or ""
        if bid and bid in batch_ids:
            tasks_by_batch.setdefault(bid, []).append(t)
        else:
            orphan.append(t)
    resolved_side = side_data
    if resolved_side is None and batches:
        resolved_side = _prefetch_batch_side_data(db, batches, tenant_no)
    out = [
        _batch_to_dict(
            db,
            b,
            tenant_no=tenant_no,
            tasks=tasks_by_batch.get(b.id, []),
            side_data=resolved_side,
        )
        for b in batches
    ]
    if orphan and out:
        out[0]["tasks"] = orphan + out[0]["tasks"]
        out[0]["taskCount"] = len(out[0]["tasks"])
        out[0]["completedTaskCount"] = sum(
            1 for t in out[0]["tasks"] if (t.get("taskStatus") or "") == "ywc"
        )
    elif orphan:
        out.append(
            {
                "id": "",
                "workOrderId": work_order_id,
                "batchNo": "—",
                "batchName": "历史任务",
                "orderNum": 0,
                "assigneeUserIds": [],
                "assigneeUserNames": "",
                "taskCount": len(orphan),
                "completedTaskCount": sum(1 for t in orphan if (t.get("taskStatus") or "") == "ywc"),
                "tasks": orphan,
            }
        )
    return out


def list_inspection_batches(
    db: Session, work_order_id: str, tenant_no: str, *, user_info=None
) -> List[Dict[str, Any]]:
    batches = (
        db.query(BizQcWorkOrderInspectionBatch)
        .filter(
            BizQcWorkOrderInspectionBatch.work_order_id == work_order_id,
            BizQcWorkOrderInspectionBatch.tenant_no == tenant_no,
            BizQcWorkOrderInspectionBatch.del_flag == 0,
        )
        .order_by(BizQcWorkOrderInspectionBatch.order_num.asc())
        .all()
    )
    all_tasks = task_svc.list_tasks_by_work_order(db, work_order_id, tenant_no)
    return _build_inspection_batch_list(db, work_order_id, batches, all_tasks, tenant_no)


def list_inspection_batches_by_work_order_ids(
    db: Session,
    work_order_ids: List[str],
    tenant_no: str,
    *,
    user_info=None,
) -> Dict[str, List[Dict[str, Any]]]:
    """批量返回多个工单的检测批次（含任务），供列表页一次加载。"""
    ids = _unique_ids(work_order_ids)
    if not ids or not tenant_no:
        return {}
    ids = ids[:100]
    batches = (
        db.query(BizQcWorkOrderInspectionBatch)
        .filter(
            BizQcWorkOrderInspectionBatch.work_order_id.in_(ids),
            BizQcWorkOrderInspectionBatch.tenant_no == tenant_no,
            BizQcWorkOrderInspectionBatch.del_flag == 0,
        )
        .order_by(
            BizQcWorkOrderInspectionBatch.work_order_id.asc(),
            BizQcWorkOrderInspectionBatch.order_num.asc(),
        )
        .all()
    )
    tasks_by_wo = task_svc.list_tasks_by_work_order_ids(db, ids, tenant_no)
    side_data = _prefetch_batch_side_data(db, batches, tenant_no)
    batches_by_wo: Dict[str, List[BizQcWorkOrderInspectionBatch]] = {}
    for batch in batches:
        batches_by_wo.setdefault(batch.work_order_id, []).append(batch)
    out: Dict[str, List[Dict[str, Any]]] = {}
    for wo_id in ids:
        out[wo_id] = _build_inspection_batch_list(
            db,
            wo_id,
            batches_by_wo.get(wo_id, []),
            tasks_by_wo.get(wo_id, []),
            tenant_no,
            side_data=side_data,
        )
    return out


def configure_work_order_batches(
    db: Session,
    wo: BizQcWorkOrder,
    batches: List[Dict[str, Any]],
    *,
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str = "",
) -> BizQcWorkOrder:
    status = (wo.work_order_status or "").strip()
    if status not in _CONFIGURABLE_WORK_ORDER_STATUSES:
        raise ValueError("当前工单状态不可配置检测")

    order_source = _order_source_for_wo(wo)
    now = datetime.now()
    is_voided = status == "yzf"
    is_first = status == "dpz"

    for idx, b in enumerate(batches):
        _validate_batch_payload(b, idx)

    existing_batches = {
        r.id: r
        for r in db.query(BizQcWorkOrderInspectionBatch)
        .filter(
            BizQcWorkOrderInspectionBatch.work_order_id == wo.id,
            BizQcWorkOrderInspectionBatch.tenant_no == tenant_no,
            BizQcWorkOrderInspectionBatch.del_flag == 0,
        )
        .all()
    }
    keep_batch_ids = {((b.get("id") or "").strip()) for b in batches if (b.get("id") or "").strip()}

    # 提交空 batches = 显式清空：强制软删批次及关联任务（含进行中/已完成）
    force_clear_all = not batches
    for bid, batch_row in existing_batches.items():
        if bid not in keep_batch_ids:
            _soft_delete_batch(
                db,
                batch_row,
                tenant_no=tenant_no,
                user_id=user_id,
                force=is_voided or force_clear_all,
            )

    fee_total = Decimal("0")
    has_lab = False
    has_third = False
    global_seq = 0

    for batch_idx, payload in enumerate(batches):
        batch_id = (payload.get("id") or "").strip()
        lab_ids = _unique_ids(payload.get("labItemIds") or payload.get("lab_item_ids") or [])
        third_ids = _unique_ids(payload.get("thirdPartyItemIds") or payload.get("third_party_item_ids") or [])
        has_lab_batch = bool(lab_ids) or bool(payload.get("hasLab") or payload.get("has_lab"))
        has_third_batch = bool(third_ids) or bool(payload.get("hasThirdParty") or payload.get("has_third_party"))
        assignee_ids = payload.get("assigneeUserIds") or payload.get("assignee_user_ids") or []
        normalized_assignees = assignee_svc.validate_assignee_user_ids(db, assignee_ids, tenant_no)
        batch_name = (payload.get("batchName") or payload.get("batch_name") or "").strip()

        if batch_id and batch_id in existing_batches:
            batch = existing_batches[batch_id]
            batch.order_num = batch_idx + 1
            if batch_name:
                batch.batch_name = batch_name
            batch.update_by = user_id
            batch.update_time = now
            db.add(batch)
        else:
            batch = BizQcWorkOrderInspectionBatch(
                id=generate_uuid(),
                work_order_id=wo.id,
                order_source=order_source,
                batch_no=_next_batch_no(db, wo, tenant_no),
                batch_name=batch_name or _allocate_batch_name(db, wo.id, tenant_no, now),
                tenant_no=tenant_no,
                order_num=batch_idx + 1,
                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(batch)
            db.flush()

        sync_batch_assignees(
            db,
            batch.id,
            normalized_assignees,
            tenant_no=tenant_no,
            operator_id=user_id,
            create_dept_id=create_dept_id,
        )

        if has_lab_batch:
            has_lab = True
            global_seq = _sync_batch_lab_items(
                db,
                wo,
                batch,
                order_source=order_source,
                lab_item_ids=lab_ids,
                assignee_user_ids=normalized_assignees,
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=create_dept_id,
                seq_start=global_seq,
            )
        else:
            global_seq = _sync_batch_lab_items(
                db,
                wo,
                batch,
                order_source=order_source,
                lab_item_ids=[],
                assignee_user_ids=normalized_assignees,
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=create_dept_id,
                seq_start=global_seq,
            )

        if has_third_batch:
            has_third = True
            global_seq, batch_fee = _sync_batch_third_items(
                db,
                wo,
                batch,
                order_source=order_source,
                third_party_item_ids=third_ids,
                assignee_user_ids=normalized_assignees,
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=create_dept_id,
                seq_start=global_seq,
            )
            fee_total += batch_fee
        else:
            global_seq, _ = _sync_batch_third_items(
                db,
                wo,
                batch,
                order_source=order_source,
                third_party_item_ids=[],
                assignee_user_ids=normalized_assignees,
                tenant_no=tenant_no,
                user_id=user_id,
                create_dept_id=create_dept_id,
                seq_start=global_seq,
            )

    wo.has_lab = "1" if has_lab else "0"
    wo.has_third_party = "1" if has_third else "0"
    if not is_voided:
        if not batches:
            # 空批次列表：清空全部检测批次/任务后回到待配置
            wo.work_order_status = "dpz"
        else:
            wo.work_order_status = "ypz"
            if is_first:
                wo.configured_time = now
                wo.configured_by = user_id
    wo.third_party_fee_total = fee_total if has_third else None
    wo.update_by = user_id
    wo.update_time = now
    db.add(wo)

    if not is_voided:
        from app.crud.pilot_outbound_status_sync import sync_schedule_from_qc_work_order

        sync_schedule_from_qc_work_order(db, wo, user_id=user_id)
    db.commit()

    if not is_voided and batches:
        from app.crud.qc_work_order import try_complete_work_order

        try_complete_work_order(db, wo.id, tenant_no)
    db.refresh(wo)
    return wo


def normalize_legacy_configure_to_batches(body: Dict[str, Any]) -> List[Dict[str, Any]]:
    """将扁平 labItemIds 配置转为单批次结构。

    注意：`batches: []` 表示显式清空，不得当作「未传 batches」走扁平兼容。
    """
    if "batches" in body:
        batches = body.get("batches")
        return list(batches) if batches is not None else []
    has_lab = bool(body.get("hasLab") or body.get("has_lab"))
    has_third = bool(body.get("hasThirdParty") or body.get("has_third_party"))
    lab_ids = body.get("labItemIds") or body.get("lab_item_ids") or []
    third_ids = body.get("thirdPartyItemIds") or body.get("third_party_item_ids") or []
    lab_assignees = body.get("labItemAssignees") or body.get("lab_item_assignees") or {}
    third_assignees = body.get("thirdPartyItemAssignees") or body.get("third_party_item_assignees") or {}
    assignee_set: Set[str] = set()
    for lid in lab_ids:
        for uid in lab_assignees.get(lid) or []:
            if uid:
                assignee_set.add(uid)
    for tid in third_ids:
        for uid in third_assignees.get(tid) or []:
            if uid:
                assignee_set.add(uid)
    return [
        {
            "assigneeUserIds": list(assignee_set),
            "hasLab": has_lab,
            "hasThirdParty": has_third,
            "labItemIds": lab_ids,
            "thirdPartyItemIds": third_ids,
        }
    ]