zkxw/backend/app/crud/process_route.py

"""工艺路线 CRUD（biz_process_route + biz_process_route_step + 4张步骤子表）"""
from __future__ import annotations

import math
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

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

from app.core.utils import generate_uuid
from app.crud.equipment_snapshot import parse_equipment_ids, resolve_equipment_list_snapshot
from app.models.models import (
    BizProductionProductType,
    BizProcess,
    BizProcessReagent,
    BizProcessConsumable,
    BizProcessOutput,
    BizProcessDataCollection,
    BizProcessRoute,
    BizProcessRouteStep,
    BizProcessRouteStepReagent,
    BizProcessRouteStepConsumable,
    BizProcessRouteStepOutput,
    BizProcessRouteStepDataCollection,
)


# ---------------------------------------------------------------------------
# 序列化函数
# ---------------------------------------------------------------------------

def _route_row_to_dict(r: BizProcessRoute) -> Dict[str, Any]:
    return {
        "id": r.id,
        "routeCode": r.route_code,
        "routeName": r.route_name,
        "productTypeId": r.product_type_id or "",
        "version": r.version or "",
        "isCurrent": r.is_current if r.is_current is not None else "0",
        "equipmentIds": parse_equipment_ids(getattr(r, "equipment_ids", None)),
        "equipmentName": getattr(r, "equipment_name", None) or "",
        "tenantNo": r.tenant_no,
        "orderNum": r.order_num if r.order_num is not None else 0,
        "remark": r.remark or "",
        "isBuiltin": getattr(r, "is_builtin", None) or "0",
        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
    }


def _step_row_to_dict(s: BizProcessRouteStep) -> Dict[str, Any]:
    return {
        "id": s.id,
        "routeId": s.route_id,
        "processId": s.process_id or "",
        "stepOrder": s.step_order if s.step_order is not None else 0,
        "processCode": s.process_code or "",
        "categoryId": s.category_id or "",
        "processName": s.process_name or "",
        "description": s.description or "",
        "estimatedDuration": s.estimated_duration or "",
        "sopIds": s.sop_ids or "",
        "sopContent": s.sop_content or "",
        "remark": s.remark or "",
    }


def _step_reagent_row_to_dict(r: BizProcessRouteStepReagent) -> Dict[str, Any]:
    return {
        "id": r.id,
        "stepId": r.step_id,
        "reagentCode": r.reagent_code or "",
        "reagentName": r.reagent_name or "",
        "spec": r.spec or "",
        "standardUsage": r.standard_usage or "",
        "unit": r.unit or "",
        "isKeyMaterial": r.is_key_material if r.is_key_material is not None else "0",
        "orderNum": r.order_num if r.order_num is not None else 0,
        "remark": r.remark or "",
    }


def _step_consumable_row_to_dict(c: BizProcessRouteStepConsumable) -> Dict[str, Any]:
    return {
        "id": c.id,
        "stepId": c.step_id,
        "consumableCode": c.consumable_code or "",
        "consumableName": c.consumable_name or "",
        "specModel": c.spec_model or "",
        "standardUsage": c.standard_usage or "",
        "unit": c.unit or "",
        "isRequired": c.is_required if c.is_required is not None else "1",
        "orderNum": c.order_num if c.order_num is not None else 0,
        "remark": c.remark or "",
    }


def _step_output_row_to_dict(o: BizProcessRouteStepOutput) -> Dict[str, Any]:
    return {
        "id": o.id,
        "stepId": o.step_id,
        "outputTypeId": o.output_type_id or "",
        "outputName": o.output_name or "",
        "outputBarcode": o.output_barcode or "",
        "orderNum": o.order_num if o.order_num is not None else 0,
        "remark": o.remark or "",
    }


def _step_data_collection_row_to_dict(d: BizProcessRouteStepDataCollection) -> Dict[str, Any]:
    return {
        "id": d.id,
        "stepId": d.step_id,
        "collectionName": d.collection_name or "",
        "unit": d.unit or "",
        "orderNum": d.order_num if d.order_num is not None else 0,
        "remark": d.remark or "",
    }


# ---------------------------------------------------------------------------
# 路线主表 CRUD
# ---------------------------------------------------------------------------

def list_routes(db: Session, *, page: int = 1, page_size: int = 10,
                route_name: Optional[str] = None, route_code: Optional[str] = None,
                product_type_id: Optional[str] = None,
                tenant_no: Optional[str] = None,
                user_info=None) -> Tuple[List[Dict], int]:
    def _filtered():
        q = db.query(BizProcessRoute).filter(BizProcessRoute.del_flag == 0)
        if tenant_no:
            q = q.filter(BizProcessRoute.tenant_no == tenant_no)
        if route_name and route_name.strip():
            q = q.filter(BizProcessRoute.route_name.like(f"%{route_name.strip()}%"))
        if route_code and route_code.strip():
            q = q.filter(BizProcessRoute.route_code == route_code.strip())
        if product_type_id and product_type_id.strip():
            q = q.filter(BizProcessRoute.product_type_id == product_type_id.strip())
        if user_info:
            from app.core.data_scope import apply_data_scope
            q = apply_data_scope(q, BizProcessRoute, user_info, db, tenant_no)
        return q

    total = _filtered().order_by(None).with_entities(func.count(BizProcessRoute.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .outerjoin(
            BizProductionProductType,
            and_(
                BizProcessRoute.product_type_id == BizProductionProductType.id,
                BizProductionProductType.del_flag == 0,
            ),
        )
        .order_by(
            BizProductionProductType.product_type_name.is_(None),
            BizProductionProductType.product_type_name.asc(),
            BizProcessRoute.create_time.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_route_row_to_dict(x) for x in rows], int(total)


def get_route(db: Session, route_id: str) -> Optional[BizProcessRoute]:
    return db.query(BizProcessRoute).filter(
        and_(BizProcessRoute.id == route_id, BizProcessRoute.del_flag == 0)
    ).first()


def get_step(db: Session, step_id: str) -> Optional[BizProcessRouteStep]:
    return db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
    ).first()


def is_super_admin_user(user: Optional[Dict[str, Any]]) -> bool:
    return bool(user) and (user.get("role") or "") == "super_admin"


def assert_route_writable(
    db: Session, route_id: str, user: Optional[Dict[str, Any]] = None
) -> BizProcessRoute:
    """内置模版仅超级管理员可写；返回路线对象。"""
    route = get_route(db, route_id)
    if not route:
        raise ValueError("路线不存在")
    if (getattr(route, "is_builtin", None) or "0") == "1" and not is_super_admin_user(user):
        raise ValueError("内置模版仅超级管理员可修改")
    return route


def assert_step_route_writable(
    db: Session, step_id: str, user: Optional[Dict[str, Any]] = None
) -> BizProcessRouteStep:
    step = get_step(db, step_id)
    if not step:
        raise ValueError("步骤不存在")
    assert_route_writable(db, step.route_id, user)
    return step


def get_route_detail(db: Session, route_id: str) -> Optional[Dict[str, Any]]:
    route = get_route(db, route_id)
    if not route:
        return None
    result = _route_row_to_dict(route)

    steps = (
        db.query(BizProcessRouteStep)
        .filter(and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0))
        .order_by(BizProcessRouteStep.step_order.asc(), BizProcessRouteStep.create_time.asc())
        .all()
    )
    step_list = []
    for s in steps:
        sd = _step_row_to_dict(s)
        reagents = db.query(BizProcessRouteStepReagent).filter(
            and_(BizProcessRouteStepReagent.step_id == s.id, BizProcessRouteStepReagent.del_flag == 0)
        ).order_by(BizProcessRouteStepReagent.order_num.asc(), BizProcessRouteStepReagent.create_time.asc()).all()
        consumables = db.query(BizProcessRouteStepConsumable).filter(
            and_(BizProcessRouteStepConsumable.step_id == s.id, BizProcessRouteStepConsumable.del_flag == 0)
        ).order_by(BizProcessRouteStepConsumable.order_num.asc(), BizProcessRouteStepConsumable.create_time.asc()).all()
        outputs = db.query(BizProcessRouteStepOutput).filter(
            and_(BizProcessRouteStepOutput.step_id == s.id, BizProcessRouteStepOutput.del_flag == 0)
        ).order_by(BizProcessRouteStepOutput.order_num.asc(), BizProcessRouteStepOutput.create_time.asc()).all()
        data_collections = db.query(BizProcessRouteStepDataCollection).filter(
            and_(BizProcessRouteStepDataCollection.step_id == s.id, BizProcessRouteStepDataCollection.del_flag == 0)
        ).order_by(BizProcessRouteStepDataCollection.order_num.asc(), BizProcessRouteStepDataCollection.create_time.asc()).all()

        sd["reagents"] = [_step_reagent_row_to_dict(r) for r in reagents]
        sd["consumables"] = [_step_consumable_row_to_dict(c) for c in consumables]
        sd["outputs"] = [_step_output_row_to_dict(o) for o in outputs]
        sd["dataCollections"] = [_step_data_collection_row_to_dict(d) for d in data_collections]
        step_list.append(sd)

    result["steps"] = step_list
    return result


def create_route(db: Session, *, route_code: str, route_name: str,
                 product_type_id: Optional[str] = None, version: Optional[str] = None,
                 remark: Optional[str] = None, tenant_no: str = "zkxw",
                 user_id: Optional[str] = None) -> BizProcessRoute:
    rc = (route_code or "").strip()
    if not rc:
        raise ValueError("路线编码不能为空")
    now = datetime.now()
    obj = BizProcessRoute(
        id=generate_uuid(),
        route_code=rc,
        route_name=(route_name or "").strip() or rc,
        product_type_id=(product_type_id or "").strip() or None,
        version=(version or "").strip() or None,
        is_current="0",
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        is_builtin="0",
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def update_route(db: Session, route_id: str, *, user_id: Optional[str] = None, **kwargs) -> BizProcessRoute:
    obj = get_route(db, route_id)
    if not obj:
        raise ValueError("路线不存在")
    for field in ["route_code", "route_name", "product_type_id", "version", "remark"]:
        val = kwargs.get(field)
        if val is not None:
            setattr(obj, field, (val or "").strip() or None if field != "route_code" else (val or "").strip())
    if "equipment_ids" in kwargs:
        eids, enames = resolve_equipment_list_snapshot(
            db, kwargs.get("equipment_ids"), obj.tenant_no or "zkxw"
        )
        obj.equipment_ids = eids
        obj.equipment_name = enames
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def set_current_route(db: Session, route_id: str, tenant_no: str) -> BizProcessRoute:
    obj = get_route(db, route_id)
    if not obj:
        raise ValueError("路线不存在")
    now = datetime.now()
    siblings = db.query(BizProcessRoute).filter(
        and_(BizProcessRoute.route_code == obj.route_code,
             BizProcessRoute.tenant_no == tenant_no,
             BizProcessRoute.del_flag == 0)
    ).all()
    for s in siblings:
        s.is_current = "0"
        s.update_time = now
        db.add(s)
    obj.is_current = "1"
    obj.update_time = now
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


# ---------------------------------------------------------------------------
# 步骤子表辅助
# ---------------------------------------------------------------------------

def _soft_delete_sub_records(db: Session, model_class, step_id: str, now: datetime) -> None:
    rows = db.query(model_class).filter(
        and_(model_class.step_id == step_id, model_class.del_flag == 0)
    ).all()
    for r in rows:
        r.del_flag = 1
        r.update_time = now
        db.add(r)


def _create_step_reagent(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepReagent(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        reagent_code=(data.get("reagent_code") or "").strip() or None,
        reagent_name=(data.get("reagent_name") or "").strip() or None,
        spec=(data.get("spec") or "").strip() or None,
        standard_usage=(data.get("standard_usage") or "").strip() or None,
        unit=(data.get("unit") or "").strip() or None,
        is_key_material=data.get("is_key_material", "0"),
        remark=(data.get("remark") or "").strip() or None,
        order_num=data.get("order_num", 0) or 0,
        create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


def _create_step_consumable(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepConsumable(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        consumable_code=(data.get("consumable_code") or "").strip() or None,
        consumable_name=(data.get("consumable_name") or "").strip() or None,
        spec_model=(data.get("spec_model") or "").strip() or None,
        standard_usage=(data.get("standard_usage") or "").strip() or None,
        unit=(data.get("unit") or "").strip() or None,
        is_required=data.get("is_required", "1"),
        remark=(data.get("remark") or "").strip() or None,
        order_num=data.get("order_num", 0) or 0,
        create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


def _create_step_output(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepOutput(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        output_type_id=(data.get("output_type_id") or "").strip() or None,
        output_name=(data.get("output_name") or "").strip() or None,
        output_barcode=(data.get("output_barcode") or "").strip() or None,
        remark=(data.get("remark") or "").strip() or None,
        order_num=data.get("order_num", 0) or 0,
        create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


def _create_step_data_collection(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepDataCollection(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        collection_name=(data.get("collection_name") or "").strip() or None,
        unit=(data.get("unit") or "").strip() or None,
        order_num=data.get("order_num", 0) or 0,
        remark=(data.get("remark") or "").strip() or None,
        create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


# ---------------------------------------------------------------------------
# 步骤 CRUD
# ---------------------------------------------------------------------------

def add_step(db: Session, *, route_id: str, process_id: str, step_order: Optional[int] = None,
             tenant_no: str = "zkxw", user_id: Optional[str] = None) -> Dict[str, Any]:
    route = get_route(db, route_id)
    if not route:
        raise ValueError("路线不存在")

    proc = db.query(BizProcess).filter(
        and_(BizProcess.id == process_id, BizProcess.del_flag == 0)
    ).first()
    if not proc:
        raise ValueError("工艺不存在")

    now = datetime.now()
    if step_order is None:
        max_order = db.query(func.max(BizProcessRouteStep.step_order)).filter(
            and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0)
        ).scalar()
        step_order = (max_order or 0) + 1

    step = BizProcessRouteStep(
        id=generate_uuid(), route_id=route_id, process_id=process_id,
        step_order=step_order,
        process_code=proc.process_code, category_id=proc.category_id,
        process_name=proc.process_name, description=proc.description,
        estimated_duration=proc.estimated_duration,
        sop_ids=proc.sop_ids,
        sop_content=proc.sop_content,
        tenant_no=tenant_no, order_num=0,
        remark=None, create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    )
    db.add(step)
    db.flush()

    src_reagents = db.query(BizProcessReagent).filter(
        and_(BizProcessReagent.process_id == process_id, BizProcessReagent.del_flag == 0)
    ).all()
    for r in src_reagents:
        _create_step_reagent(db, step.id, tenant_no, user_id, now, {
            "reagent_code": r.reagent_code, "reagent_name": r.reagent_name,
            "spec": r.spec, "standard_usage": r.standard_usage,
            "unit": r.unit, "is_key_material": r.is_key_material,
        })

    src_consumables = db.query(BizProcessConsumable).filter(
        and_(BizProcessConsumable.process_id == process_id, BizProcessConsumable.del_flag == 0)
    ).all()
    for c in src_consumables:
        _create_step_consumable(db, step.id, tenant_no, user_id, now, {
            "consumable_code": c.consumable_code, "consumable_name": c.consumable_name,
            "spec_model": c.spec_model, "standard_usage": c.standard_usage,
            "unit": c.unit, "is_required": c.is_required, "remark": c.remark,
        })

    src_outputs = db.query(BizProcessOutput).filter(
        and_(BizProcessOutput.process_id == process_id, BizProcessOutput.del_flag == 0)
    ).all()
    for o in src_outputs:
        _create_step_output(db, step.id, tenant_no, user_id, now, {
            "output_type_id": o.output_type_id, "output_name": o.output_name,
            "output_barcode": o.output_barcode,
        })

    src_collections = db.query(BizProcessDataCollection).filter(
        and_(BizProcessDataCollection.process_id == process_id, BizProcessDataCollection.del_flag == 0)
    ).all()
    for d in src_collections:
        _create_step_data_collection(db, step.id, tenant_no, user_id, now, {
            "collection_name": d.collection_name, "unit": d.unit, "order_num": d.order_num,
        })

    db.commit()
    detail = get_route_detail(db, route_id)
    return detail["steps"][-1] if detail and detail["steps"] else _step_row_to_dict(step)


def update_step(db: Session, step_id: str, *, user_id: Optional[str] = None, **kwargs) -> BizProcessRouteStep:
    step = db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
    ).first()
    if not step:
        raise ValueError("步骤不存在")
    for field in ["process_code", "category_id", "process_name", "description",
                   "estimated_duration", "sop_ids", "sop_content"]:
        val = kwargs.get(field)
        if val is not None:
            if field == "sop_content":
                setattr(step, field, val if str(val).strip() else None)
            else:
                setattr(step, field, (val or "").strip() or None)
    step.update_by = user_id
    step.update_time = datetime.now()
    db.add(step)
    db.commit()
    db.refresh(step)
    return step


def reorder_steps(db: Session, route_id: str, step_ids: List[str]) -> None:
    now = datetime.now()
    for idx, sid in enumerate(step_ids):
        step = db.query(BizProcessRouteStep).filter(
            and_(BizProcessRouteStep.id == sid, BizProcessRouteStep.route_id == route_id,
                 BizProcessRouteStep.del_flag == 0)
        ).first()
        if step:
            step.step_order = idx + 1
            step.update_time = now
            db.add(step)
    db.commit()


def copy_step(db: Session, step_id: str, *, tenant_no: str = "zkxw",
              user_id: Optional[str] = None) -> Dict[str, Any]:
    """复制步骤，插入到原步骤之后，名称追加（复制）"""
    src = db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
    ).first()
    if not src:
        raise ValueError("步骤不存在")

    now = datetime.now()
    new_order = src.step_order + 1

    # 后续步骤 step_order +1
    later_steps = db.query(BizProcessRouteStep).filter(
        and_(
            BizProcessRouteStep.route_id == src.route_id,
            BizProcessRouteStep.del_flag == 0,
            BizProcessRouteStep.step_order > src.step_order,
        )
    ).order_by(BizProcessRouteStep.step_order.asc()).all()
    for ls in later_steps:
        ls.step_order += 1
        ls.update_time = now
        db.add(ls)

    # 创建新步骤
    new_name = (src.process_name or "").strip()
    if not new_name.endswith("（复制）"):
        new_name += "（复制）"
    new_step = BizProcessRouteStep(
        id=generate_uuid(), route_id=src.route_id, process_id=src.process_id,
        step_order=new_order,
        process_code=src.process_code, category_id=src.category_id,
        process_name=new_name, description=src.description,
        estimated_duration=src.estimated_duration,
        sop_ids=src.sop_ids,
        sop_content=src.sop_content,
        tenant_no=tenant_no, order_num=0,
        remark=src.remark, create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    )
    db.add(new_step)
    db.flush()

    # 复制子表：试剂
    for r in db.query(BizProcessRouteStepReagent).filter(
        and_(BizProcessRouteStepReagent.step_id == step_id, BizProcessRouteStepReagent.del_flag == 0)
    ).order_by(
        BizProcessRouteStepReagent.order_num.asc(),
        BizProcessRouteStepReagent.create_time.asc(),
    ).all():
        _create_step_reagent(db, new_step.id, tenant_no, user_id, now, {
            "reagent_code": r.reagent_code, "reagent_name": r.reagent_name,
            "spec": r.spec, "standard_usage": r.standard_usage,
            "unit": r.unit, "is_key_material": r.is_key_material,
            "order_num": r.order_num,
            "remark": r.remark,
        })

    # 复制子表：耗材
    for c in db.query(BizProcessRouteStepConsumable).filter(
        and_(BizProcessRouteStepConsumable.step_id == step_id, BizProcessRouteStepConsumable.del_flag == 0)
    ).order_by(
        BizProcessRouteStepConsumable.order_num.asc(),
        BizProcessRouteStepConsumable.create_time.asc(),
    ).all():
        _create_step_consumable(db, new_step.id, tenant_no, user_id, now, {
            "consumable_code": c.consumable_code, "consumable_name": c.consumable_name,
            "spec_model": c.spec_model, "standard_usage": c.standard_usage,
            "unit": c.unit, "is_required": c.is_required,
            "order_num": c.order_num,
            "remark": c.remark,
        })

    # 复制子表：产出
    for o in db.query(BizProcessRouteStepOutput).filter(
        and_(BizProcessRouteStepOutput.step_id == step_id, BizProcessRouteStepOutput.del_flag == 0)
    ).order_by(
        BizProcessRouteStepOutput.order_num.asc(),
        BizProcessRouteStepOutput.create_time.asc(),
    ).all():
        _create_step_output(db, new_step.id, tenant_no, user_id, now, {
            "output_type_id": o.output_type_id, "output_name": o.output_name,
            "output_barcode": o.output_barcode,
            "order_num": o.order_num,
            "remark": o.remark,
        })

    # 复制子表：数据采集
    for d in db.query(BizProcessRouteStepDataCollection).filter(
        and_(BizProcessRouteStepDataCollection.step_id == step_id, BizProcessRouteStepDataCollection.del_flag == 0)
    ).order_by(
        BizProcessRouteStepDataCollection.order_num.asc(),
        BizProcessRouteStepDataCollection.create_time.asc(),
    ).all():
        _create_step_data_collection(db, new_step.id, tenant_no, user_id, now, {
            "collection_name": d.collection_name, "unit": d.unit,
            "order_num": d.order_num, "remark": d.remark,
        })

    db.commit()
    detail = get_route_detail(db, src.route_id)
    return detail["steps"][-1] if detail and detail["steps"] else _step_row_to_dict(new_step)


def soft_delete_step(db: Session, step_id: str) -> None:
    step = db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
    ).first()
    if not step:
        raise ValueError("步骤不存在")
    now = datetime.now()
    step.del_flag = 1
    step.update_time = now
    db.add(step)
    _soft_delete_sub_records(db, BizProcessRouteStepReagent, step_id, now)
    _soft_delete_sub_records(db, BizProcessRouteStepConsumable, step_id, now)
    _soft_delete_sub_records(db, BizProcessRouteStepOutput, step_id, now)
    _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, step_id, now)
    db.commit()


def soft_delete_route(db: Session, route_id: str) -> None:
    route = get_route(db, route_id)
    if not route:
        raise ValueError("路线不存在")
    now = datetime.now()
    route.del_flag = 1
    route.update_time = now
    db.add(route)

    steps = db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0)
    ).all()
    for s in steps:
        s.del_flag = 1
        s.update_time = now
        db.add(s)
        _soft_delete_sub_records(db, BizProcessRouteStepReagent, s.id, now)
        _soft_delete_sub_records(db, BizProcessRouteStepConsumable, s.id, now)
        _soft_delete_sub_records(db, BizProcessRouteStepOutput, s.id, now)
        _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, s.id, now)
    db.commit()


# ---------------------------------------------------------------------------
# 子表完整替换更新
# ---------------------------------------------------------------------------

def update_step_reagents(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepReagent, step_id, now)
    for it in items:
        _create_step_reagent(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def update_step_consumables(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepConsumable, step_id, now)
    for it in items:
        _create_step_consumable(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def update_step_outputs(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepOutput, step_id, now)
    for it in items:
        _create_step_output(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def update_step_data_collections(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, step_id, now)
    for it in items:
        _create_step_data_collection(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))