zkxw/backend/scripts/backfill_uc_msc_production_tasks.py

#!/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

等价一次性 SQL（与种子脚本分开）:
  docs/sql/migrations/2026-11-13-uc-msc-production-task-backfill.sql
"""
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())