"""质检工单检测批次 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.crud.user_display import user_display_name_map from app.models.models import ( BizQcBatchAssignee, BizQcTask, BizQcWorkOrder, BizQcWorkOrderInspectionBatch, BizQcWorkOrderLabItem, BizQcWorkOrderThirdItem, ) _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 not assignees: raise ValueError(f"{label}：请指定负责人") if not lab_ids and not third_ids: 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}：第三方项目不能重复") 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 "" name_map = user_display_name_map(db, user_ids) return "、".join(name_map.get(uid) or 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: name_map = user_display_name_map(db, all_user_ids) 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 full_replace = bool(batches) and bool(existing_batches) and keep_batch_ids.isdisjoint( existing_batches.keys() ) force_remove = is_voided or force_clear_all or full_replace 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=force_remove, ) 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, } ]