#!/usr/bin/env python3 """为 UC 开头实验室入库编码的入库生产工单补派 5 步脐带MSC原代任务。 默认 dry-run；加 --apply 才写库。会先幂等确保内置模版存在。 用法（在 backend 目录、已配置可连目标库的 .env）: python -m scripts.backfill_uc_msc_production_tasks python -m scripts.backfill_uc_msc_production_tasks --apply python -m scripts.backfill_uc_msc_production_tasks --tenant zkxw --apply """ from __future__ import annotations import argparse import os import sys from typing import Any, Dict, List, Optional, Tuple sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sqlalchemy import or_ from sqlalchemy.orm import Session from app.core.database import SessionLocal from app.crud import production_task_set as production_task_set_svc from app.models.models import ( BizProductionTask, BizProductionTaskSet, BizProductionWorkOrder, BizSampleReception, ) from app.services.uc_msc_builtin_template import ensure_uc_msc_builtin_template from app.utils.lab_storage_code_task_set import UC_MSC_BUILTIN_SET_NAME def _list_candidate_work_orders( db: Session, *, tenant_no: str ) -> List[Tuple[BizProductionWorkOrder, BizSampleReception]]: """UC 接收记录关联的入库工单，且当前无有效生产任务。""" q = ( db.query(BizProductionWorkOrder, BizSampleReception) .join( BizSampleReception, BizSampleReception.id == BizProductionWorkOrder.reception_id, ) .filter( BizProductionWorkOrder.tenant_no == tenant_no, BizProductionWorkOrder.del_flag == 0, BizProductionWorkOrder.work_order_type == "inbound", BizProductionWorkOrder.work_order_status.in_(("dpz", "jxz")), BizSampleReception.tenant_no == tenant_no, BizSampleReception.del_flag == 0, BizSampleReception.accept_conclusion == "js", BizSampleReception.lab_storage_code.isnot(None), BizSampleReception.lab_storage_code != "", or_( BizSampleReception.lab_storage_code.like("UC%"), BizSampleReception.lab_storage_code.like("uc%"), ), ) .order_by(BizSampleReception.reception_time.asc()) ) rows = q.all() out: List[Tuple[BizProductionWorkOrder, BizSampleReception]] = [] for wo, rec in rows: has_task = ( db.query(BizProductionTask.id) .filter( BizProductionTask.work_order_id == wo.id, BizProductionTask.tenant_no == tenant_no, BizProductionTask.del_flag == 0, ) .first() ) if has_task: continue code = (rec.lab_storage_code or "").strip().upper() if not code.startswith("UC"): continue out.append((wo, rec)) return out def _dispatch_one( db: Session, *, tenant_no: str, user_id: Optional[str], create_dept_id: str, wo: BizProductionWorkOrder, rec: BizSampleReception, set_id: str, ) -> Dict[str, Any]: result = production_task_set_svc.create_tasks_from_set( db, tenant_no=tenant_no, user_id=user_id, create_dept_id=create_dept_id or "", work_order_id=wo.id, set_id=set_id, ) return { "workOrderId": wo.id, "workOrderNo": wo.work_order_no, "sampleCode": wo.sample_code or rec.sample_code, "labStorageCode": rec.lab_storage_code, "createdCount": result.get("createdCount"), "tasks": [ {"id": t.get("id"), "taskName": t.get("taskName"), "routeId": t.get("routeId")} for t in (result.get("list") or []) ], } def run( *, tenant_no: str, apply: bool, user_id: Optional[str], create_dept_id: str, ) -> Dict[str, Any]: db = SessionLocal() try: ensure_info = ensure_uc_msc_builtin_template( db, tenant_no=tenant_no, user_id=user_id, create_dept_id=create_dept_id, ) set_row = ( db.query(BizProductionTaskSet) .filter( BizProductionTaskSet.tenant_no == tenant_no, BizProductionTaskSet.set_name == UC_MSC_BUILTIN_SET_NAME, BizProductionTaskSet.is_builtin == "1", BizProductionTaskSet.status == "1", BizProductionTaskSet.del_flag == 0, ) .first() ) if not set_row: raise RuntimeError("UC 内置任务组合创建失败") candidates = _list_candidate_work_orders(db, tenant_no=tenant_no) preview = [ { "workOrderId": wo.id, "workOrderNo": wo.work_order_no, "workOrderStatus": wo.work_order_status, "sampleCode": wo.sample_code or rec.sample_code, "labStorageCode": rec.lab_storage_code, "receptionId": rec.id, "receptionTime": str(rec.reception_time) if rec.reception_time else None, } for wo, rec in candidates ] if not apply: db.rollback() return { "dryRun": True, "template": ensure_info, "candidateCount": len(preview), "candidates": preview, "message": "dry-run：未写库。确认后加 --apply 执行补派。", } dispatched: List[Dict[str, Any]] = [] errors: List[Dict[str, Any]] = [] for wo, rec in candidates: try: dispatched.append( _dispatch_one( db, tenant_no=tenant_no, user_id=user_id, create_dept_id=create_dept_id, wo=wo, rec=rec, set_id=set_row.id, ) ) except Exception as exc: # noqa: BLE001 - 汇总后继续 errors.append( { "workOrderId": wo.id, "workOrderNo": wo.work_order_no, "labStorageCode": rec.lab_storage_code, "error": str(exc), } ) db.commit() return { "dryRun": False, "template": ensure_info, "candidateCount": len(preview), "dispatchedCount": len(dispatched), "dispatched": dispatched, "errorCount": len(errors), "errors": errors, } except Exception: db.rollback() raise finally: db.close() def main(argv: Optional[List[str]] = None) -> int: parser = argparse.ArgumentParser( description="回补 UC 入库编码生产工单的 5 步脐带MSC原代任务" ) parser.add_argument( "--tenant", default="zkxw", help="租户编号，默认 zkxw", ) parser.add_argument( "--apply", action="store_true", help="实际写库；默认仅 dry-run", ) parser.add_argument( "--user-id", default=None, help="写库 create_by/update_by，可选", ) parser.add_argument( "--create-dept-id", default="", help="写库 create_dept_id，可选", ) args = parser.parse_args(argv) result = run( tenant_no=(args.tenant or "zkxw").strip() or "zkxw", apply=bool(args.apply), user_id=(args.user_id or None), create_dept_id=(args.create_dept_id or ""), ) import json print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) if result.get("errorCount"): return 2 return 0 if __name__ == "__main__": raise SystemExit(main())