zkxw/backend/app/crud/process.py

"""工艺库 CRUD（biz_process + biz_process_reagent/consumable/output）"""
from __future__ import annotations

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

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

from app.core.utils import generate_uuid
from app.models.models import (
    BizProcess,
    BizProcessCategory,
    BizProcessConsumable,
    BizProcessDataCollection,
    BizProcessOutput,
    BizProcessReagent,
)

_UNSET = object()


def _reagent_row_to_dict(r: BizProcessReagent) -> Dict[str, Any]:
    return {
        "id": r.id,
        "processId": r.process_id,
        "reagentCode": r.reagent_code,
        "reagentName": r.reagent_name,
        "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,
        "remark": r.remark or "",
        "orderNum": r.order_num if r.order_num is not None else 0,
    }


def _consumable_row_to_dict(c: BizProcessConsumable) -> Dict[str, Any]:
    return {
        "id": c.id,
        "processId": c.process_id,
        "consumableCode": c.consumable_code,
        "consumableName": c.consumable_name,
        "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 0,
        "remark": c.remark or "",
    }


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


def _data_collection_row_to_dict(d: BizProcessDataCollection) -> Dict[str, Any]:
    return {
        "id": d.id,
        "processId": d.process_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 "",
    }


def _row_to_dict(p: BizProcess) -> Dict[str, Any]:
    return {
        "id": p.id,
        "processCode": p.process_code,
        "categoryId": p.category_id or "",
        "processName": p.process_name,
        "description": p.description or "",
        "estimatedDuration": p.estimated_duration or "",
        "sopIds": p.sop_ids or "",
        "sopContent": p.sop_content or "",
        "tenantNo": p.tenant_no,
        "orderNum": p.order_num if p.order_num is not None else 0,
        "remark": p.remark or "",
        "createTime": p.create_time.strftime("%Y-%m-%d %H:%M:%S") if p.create_time else "",
        "updateTime": p.update_time.strftime("%Y-%m-%d %H:%M:%S") if p.update_time else "",
    }


def list_processes(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(BizProcess).filter(BizProcess.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizProcess.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizProcess.process_code.like(like),
                    BizProcess.process_name.like(like),
                )
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizProcess, user_info, db, tenant_no)
        return q0

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


def list_all_processes(db: Session, tenant_no: str, user_info=None) -> List[Dict[str, Any]]:
    q0 = db.query(BizProcess).filter(and_(BizProcess.tenant_no == tenant_no, BizProcess.del_flag == 0))
    if user_info:
        from app.core.data_scope import apply_data_scope
        q0 = apply_data_scope(q0, BizProcess, user_info, db, tenant_no)
    rows = q0.order_by(BizProcess.create_time.desc()).all()
    return [_row_to_dict(x) for x in rows]


def get_process(db: Session, process_id: str) -> Optional[BizProcess]:
    return (
        db.query(BizProcess)
        .filter(and_(BizProcess.id == process_id, BizProcess.del_flag == 0))
        .first()
    )


def get_process_detail(db: Session, process_id: str) -> Optional[Dict[str, Any]]:
    proc = get_process(db, process_id)
    if not proc:
        return None
    result = _row_to_dict(proc)

    reagents = (
        db.query(BizProcessReagent)
        .filter(and_(BizProcessReagent.process_id == process_id, BizProcessReagent.del_flag == 0))
        .order_by(BizProcessReagent.order_num.asc(), BizProcessReagent.create_time.asc(), BizProcessReagent.id.asc())
        .all()
    )
    consumables = (
        db.query(BizProcessConsumable)
        .filter(and_(BizProcessConsumable.process_id == process_id, BizProcessConsumable.del_flag == 0))
        .order_by(BizProcessConsumable.order_num.asc(), BizProcessConsumable.create_time.asc(), BizProcessConsumable.id.asc())
        .all()
    )
    outputs = (
        db.query(BizProcessOutput)
        .filter(and_(BizProcessOutput.process_id == process_id, BizProcessOutput.del_flag == 0))
        .order_by(BizProcessOutput.order_num.asc(), BizProcessOutput.create_time.asc(), BizProcessOutput.id.asc())
        .all()
    )

    result["reagents"] = [_reagent_row_to_dict(r) for r in reagents]
    result["consumables"] = [_consumable_row_to_dict(c) for c in consumables]
    result["outputs"] = [_output_row_to_dict(o) for o in outputs]

    data_collections = (
        db.query(BizProcessDataCollection)
        .filter(and_(BizProcessDataCollection.process_id == process_id, BizProcessDataCollection.del_flag == 0))
        .order_by(BizProcessDataCollection.order_num.asc(), BizProcessDataCollection.create_time.asc())
        .all()
    )
    result["dataCollections"] = [_data_collection_row_to_dict(d) for d in data_collections]

    return result


def process_code_exists(db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
    q = db.query(BizProcess).filter(
        and_(BizProcess.process_code == code.strip(), BizProcess.tenant_no == tenant_no, BizProcess.del_flag == 0)
    )
    if exclude_id:
        q = q.filter(BizProcess.id != exclude_id)
    return q.first() is not None


def _create_sub_reagent(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
    obj = BizProcessReagent(
        id=generate_uuid(),
        process_id=process_id,
        reagent_code=(data.get("reagent_code") or "").strip(),
        reagent_name=(data.get("reagent_name") or "").strip(),
        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,
        tenant_no=tenant_no,
        order_num=data.get("order_num", 0) if data.get("order_num") is not None else 0,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)


def _create_sub_consumable(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
    obj = BizProcessConsumable(
        id=generate_uuid(),
        process_id=process_id,
        consumable_code=(data.get("consumable_code") or "").strip(),
        consumable_name=(data.get("consumable_name") or "").strip(),
        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", 0),
        remark=(data.get("remark") or "").strip() or None,
        tenant_no=tenant_no,
        order_num=data.get("order_num", 0) if data.get("order_num") is not None else 0,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)


def _create_sub_output(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
    obj = BizProcessOutput(
        id=generate_uuid(),
        process_id=process_id,
        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,
        tenant_no=tenant_no,
        order_num=data.get("order_num", 0) if data.get("order_num") is not None else 0,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)


def _sub_item_with_order(data: Dict[str, Any], index: int) -> Dict[str, Any]:
    """保存子表时按列表下标写入 order_num，避免同批 create_time 导致查询顺序漂移。"""
    item = dict(data)
    item["order_num"] = index
    return item


def _create_sub_data_collection(db: Session, process_id: str, tenant_no: str, user_id: Optional[str], now: datetime, data: Dict[str, Any]) -> None:
    obj = BizProcessDataCollection(
        id=generate_uuid(),
        process_id=process_id,
        collection_name=(data.get("collection_name") or "").strip(),
        unit=(data.get("unit") or "").strip() or None,
        tenant_no=tenant_no,
        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,
    )
    db.add(obj)


def create_process(
    db: Session,
    *,
    process_code: str,
    process_name: str,
    category_id: Optional[str] = None,
    description: Optional[str] = None,
    equipment_ids: Optional[Union[str, List[str]]] = None,
    estimated_duration: Optional[str] = None,
    sop_ids: Optional[str] = None,
    sop_content: Optional[str] = None,
    reagents: Optional[List[Dict[str, Any]]] = None,
    consumables: Optional[List[Dict[str, Any]]] = None,
    outputs: Optional[List[Dict[str, Any]]] = None,
    data_collections: Optional[List[Dict[str, Any]]] = None,
    tenant_no: str = "zkxw",
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizProcess:
    pc = (process_code or "").strip()
    if not pc:
        raise ValueError("工艺编码不能为空")
    if process_code_exists(db, pc, tenant_no):
        raise ValueError("工艺编码已存在")
    now = datetime.now()
    obj = BizProcess(
        id=generate_uuid(),
        process_code=pc,
        process_name=(process_name or "").strip() or pc,
        category_id=(category_id or "").strip() or None,
        description=(description or "").strip() or None,
        equipment_ids=None,
        equipment_name=None,
        estimated_duration=(estimated_duration or "").strip() or None,
        sop_ids=(sop_ids or "").strip() or None,
        sop_content=(sop_content or "").strip() or None,
        tenant_no=tenant_no,
        order_num=order_num if order_num is not None else 0,
        remark=(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,
    )
    db.add(obj)
    db.flush()

    if reagents:
        for idx, r in enumerate(reagents):
            _create_sub_reagent(db, obj.id, tenant_no, user_id, now, _sub_item_with_order(r, idx))
    if consumables:
        for idx, c in enumerate(consumables):
            _create_sub_consumable(db, obj.id, tenant_no, user_id, now, _sub_item_with_order(c, idx))
    if outputs:
        for idx, o in enumerate(outputs):
            _create_sub_output(db, obj.id, tenant_no, user_id, now, _sub_item_with_order(o, idx))
    if data_collections:
        for d in data_collections:
            _create_sub_data_collection(db, obj.id, tenant_no, user_id, now, d)

    db.commit()
    db.refresh(obj)
    return obj


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


def update_process(
    db: Session,
    process_id: str,
    *,
    process_code: Optional[str] = None,
    process_name: Optional[str] = None,
    category_id: Optional[str] = None,
    description: Optional[str] = None,
    equipment_ids: Any = _UNSET,
    estimated_duration: Optional[str] = None,
    sop_ids: Optional[str] = None,
    sop_content: Optional[str] = None,
    reagents: Optional[List[Dict[str, Any]]] = None,
    consumables: Optional[List[Dict[str, Any]]] = None,
    outputs: Optional[List[Dict[str, Any]]] = None,
    data_collections: Optional[List[Dict[str, Any]]] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizProcess:
    obj = get_process(db, process_id)
    if not obj:
        raise ValueError("工艺不存在")
    if process_code is not None:
        new_code = process_code.strip()
        if not new_code:
            raise ValueError("工艺编码不能为空")
        old_code = (obj.process_code or "").strip()
        if new_code != old_code and process_code_exists(db, new_code, obj.tenant_no, exclude_id=process_id):
            raise ValueError("工艺编码已存在")
        obj.process_code = new_code
    if process_name is not None:
        obj.process_name = (process_name or "").strip() or obj.process_name
    if category_id is not None:
        obj.category_id = (category_id or "").strip() or None
    if description is not None:
        obj.description = (description or "").strip() or None
    if estimated_duration is not None:
        obj.estimated_duration = (estimated_duration or "").strip() or None
    if sop_ids is not None:
        obj.sop_ids = (sop_ids or "").strip() or None
    if sop_content is not None:
        obj.sop_content = (sop_content or "").strip() or None
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = (remark or "").strip() or None
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)

    sub_updated = reagents is not None or consumables is not None or outputs is not None or data_collections is not None
    if sub_updated:
        now = datetime.now()
        if reagents is not None:
            _soft_delete_sub_records(db, BizProcessReagent, process_id, now)
            for idx, r in enumerate(reagents):
                _create_sub_reagent(db, process_id, obj.tenant_no, user_id, now, _sub_item_with_order(r, idx))
        if consumables is not None:
            _soft_delete_sub_records(db, BizProcessConsumable, process_id, now)
            for idx, c in enumerate(consumables):
                _create_sub_consumable(db, process_id, obj.tenant_no, user_id, now, _sub_item_with_order(c, idx))
        if outputs is not None:
            _soft_delete_sub_records(db, BizProcessOutput, process_id, now)
            for idx, o in enumerate(outputs):
                _create_sub_output(db, process_id, obj.tenant_no, user_id, now, _sub_item_with_order(o, idx))
        if data_collections is not None:
            _soft_delete_sub_records(db, BizProcessDataCollection, process_id, now)
            for d in data_collections:
                _create_sub_data_collection(db, process_id, obj.tenant_no, user_id, now, d)

    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_process(db: Session, process_id: str) -> None:
    obj = get_process(db, process_id)
    if not obj:
        raise ValueError("工艺不存在")
    now = datetime.now()
    obj.del_flag = 1
    obj.update_time = now
    db.add(obj)

    _soft_delete_sub_records(db, BizProcessReagent, process_id, now)
    _soft_delete_sub_records(db, BizProcessConsumable, process_id, now)
    _soft_delete_sub_records(db, BizProcessOutput, process_id, now)
    _soft_delete_sub_records(db, BizProcessDataCollection, process_id, now)

    db.commit()


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