shanxia2.0/backend/app/crud/workflow.py

"""工作流 CRUD（wf_process_def / wf_process_version / wf_node / wf_transition / wf_process_instance / wf_task）"""
from __future__ import annotations

import json
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.delete_guard import assert_row_deletable_by_creator
from app.core.utils import generate_uuid
from app.models.models import (
    SysUser,
    WfNode,
    WfProcessDef,
    WfProcessInstance,
    WfProcessVersion,
    WfTask,
    WfTransition,
)


# ---------------------------------------------------------------------------
# 流程定义 wf_process_def
# ---------------------------------------------------------------------------


def _row_to_dict(t: WfProcessDef) -> Dict[str, Any]:
    return {
        "id": t.id,
        "processCode": t.process_code,
        "processName": t.process_name,
        "status": t.status,
        "tenantNo": t.tenant_no,
        "orderNum": t.order_num if t.order_num is not None else 0,
        "remark": t.remark or "",
        "createBy": t.create_by or "",
        "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
        "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
    }


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

    total = _filtered().order_by(None).with_entities(func.count(WfProcessDef.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(WfProcessDef.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    result = []
    for r in rows:
        d = _row_to_dict(r)
        ver = (
            db.query(WfProcessVersion)
            .filter(
                and_(
                    WfProcessVersion.process_def_id == r.id,
                    WfProcessVersion.is_latest == "1",
                    WfProcessVersion.del_flag == 0,
                )
            )
            .order_by(WfProcessVersion.version_no.desc())
            .first()
        )
        d["latestVersion"] = ver.version_no if ver else None
        result.append(d)
    return result, int(total)


def get_process_def(db: Session, def_id: str) -> Optional[WfProcessDef]:
    return (
        db.query(WfProcessDef)
        .filter(and_(WfProcessDef.id == def_id, WfProcessDef.del_flag == 0))
        .first()
    )


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


def create_process_def(
    db: Session,
    *,
    process_code: str,
    process_name: str,
    tenant_no: str,
    status: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> WfProcessDef:
    code = (process_code or "").strip()
    if not code:
        raise ValueError("流程编码不能为空")
    if process_code_exists(db, code, tenant_no):
        raise ValueError("流程编码已存在")
    obj = WfProcessDef(
        id=generate_uuid(),
        process_code=code,
        process_name=(process_name or "").strip(),
        status=status or "1",
        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=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def update_process_def(
    db: Session,
    def_id: str,
    *,
    process_code: Optional[str] = None,
    process_name: Optional[str] = None,
    status: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> WfProcessDef:
    obj = get_process_def(db, def_id)
    if not obj:
        raise ValueError("流程定义不存在")
    if process_code is not None:
        new_code = process_code.strip()
        if not new_code:
            raise ValueError("流程编码不能为空")
        tn = tenant_no or obj.tenant_no
        if new_code != obj.process_code and process_code_exists(db, new_code, tn, exclude_id=def_id):
            raise ValueError("流程编码已存在")
        obj.process_code = new_code
    if process_name is not None:
        obj.process_name = (process_name or "").strip()
    if status is not None:
        obj.status = status
    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)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_process_def(db: Session, def_id: str, *, user_info: dict = None) -> None:
    obj = get_process_def(db, def_id)
    if not obj:
        raise ValueError("流程定义不存在")
    assert_row_deletable_by_creator(obj, user_info)
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ---------------------------------------------------------------------------
# 流程版本 wf_process_version
# ---------------------------------------------------------------------------


def publish_version(
    db: Session,
    process_def_id: str,
    graph_json: str,
    nodes: List[Dict[str, Any]],
    transitions: List[Dict[str, Any]],
    tenant_no: str,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> WfProcessVersion:
    defn = get_process_def(db, process_def_id)
    if not defn:
        raise ValueError("流程定义不存在")

    db.query(WfProcessVersion).filter(
        and_(
            WfProcessVersion.process_def_id == process_def_id,
            WfProcessVersion.is_latest == "1",
            WfProcessVersion.del_flag == 0,
        )
    ).update({"is_latest": "0", "update_time": datetime.now()}, synchronize_session=False)

    max_ver = (
        db.query(func.max(WfProcessVersion.version_no))
        .filter(and_(WfProcessVersion.process_def_id == process_def_id, WfProcessVersion.del_flag == 0))
        .scalar()
    ) or 0
    new_ver_no = max_ver + 1

    ver = WfProcessVersion(
        id=generate_uuid(),
        process_def_id=process_def_id,
        version_no=new_ver_no,
        graph_json=graph_json,
        is_latest="1",
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(ver)
    db.flush()

    for idx, n in enumerate(nodes):
        node = WfNode(
            id=generate_uuid(),
            process_version_id=ver.id,
            node_key=n.get("nodeKey", ""),
            node_type=n.get("nodeType", ""),
            node_name=n.get("nodeName"),
            config_json=n.get("configJson"),
            tenant_no=tenant_no,
            order_num=idx,
            remark=None,
            create_by=user_id,
            create_dept_id="",
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
        db.add(node)

    for idx, t in enumerate(transitions):
        trans = WfTransition(
            id=generate_uuid(),
            process_version_id=ver.id,
            from_node_key=t.get("fromNodeKey", ""),
            to_node_key=t.get("toNodeKey", ""),
            condition_expr=t.get("conditionExpr"),
            tenant_no=tenant_no,
            order_num=idx,
            remark=None,
            create_by=user_id,
            create_dept_id="",
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
        db.add(trans)

    db.commit()
    db.refresh(ver)
    return ver


def get_latest_version(db: Session, process_def_id: str) -> Optional[WfProcessVersion]:
    return (
        db.query(WfProcessVersion)
        .filter(
            and_(
                WfProcessVersion.process_def_id == process_def_id,
                WfProcessVersion.is_latest == "1",
                WfProcessVersion.del_flag == 0,
            )
        )
        .first()
    )


def _version_to_dict(v: WfProcessVersion) -> Dict[str, Any]:
    return {
        "id": v.id,
        "versionNo": v.version_no,
        "isLatest": v.is_latest,
        "remark": v.remark or "",
        "createTime": v.create_time.strftime("%Y-%m-%d %H:%M:%S") if v.create_time else "",
        "updateTime": v.update_time.strftime("%Y-%m-%d %H:%M:%S") if v.update_time else "",
    }


def list_versions(
    db: Session, process_def_id: str, tenant_no: Optional[str] = None, user_info=None
) -> List[Dict[str, Any]]:
    q = db.query(WfProcessVersion).filter(
        and_(WfProcessVersion.process_def_id == process_def_id, WfProcessVersion.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(WfProcessVersion.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, WfProcessVersion, user_info, db, tenant_no)
    rows = q.order_by(WfProcessVersion.version_no.desc()).all()
    return [_version_to_dict(v) for v in rows]


def get_nodes_by_version(db: Session, version_id: str) -> List[WfNode]:
    return (
        db.query(WfNode)
        .filter(and_(WfNode.process_version_id == version_id, WfNode.del_flag == 0))
        .order_by(WfNode.order_num.asc())
        .all()
    )


def get_transitions_by_version(db: Session, version_id: str) -> List[WfTransition]:
    return (
        db.query(WfTransition)
        .filter(and_(WfTransition.process_version_id == version_id, WfTransition.del_flag == 0))
        .order_by(WfTransition.order_num.asc())
        .all()
    )


# ---------------------------------------------------------------------------
# 流程实例 wf_process_instance
# ---------------------------------------------------------------------------


def create_instance(
    db: Session,
    *,
    process_version_id: str,
    business_type: Optional[str] = None,
    business_id: Optional[str] = None,
    starter_user_id: Optional[str] = None,
    variables: Optional[Dict[str, Any]] = None,
    status: Optional[str] = None,
    callback_url: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_id: Optional[str] = None,
) -> WfProcessInstance:
    obj = WfProcessInstance(
        id=generate_uuid(),
        process_version_id=process_version_id,
        business_type=business_type,
        business_id=business_id,
        starter_user_id=starter_user_id,
        variables_json=json.dumps(variables, ensure_ascii=False) if variables else None,
        status=status or "running",
        callback_url=callback_url,
        tenant_no=tenant_no or "shanxia2.0",
        order_num=0,
        remark=None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def get_instance(db: Session, instance_id: str) -> Optional[WfProcessInstance]:
    return (
        db.query(WfProcessInstance)
        .filter(and_(WfProcessInstance.id == instance_id, WfProcessInstance.del_flag == 0))
        .first()
    )


def update_instance_status(db: Session, instance_id: str, status: str) -> None:
    obj = get_instance(db, instance_id)
    if not obj:
        raise ValueError("流程实例不存在")
    obj.status = status
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def list_my_instances(
    db: Session,
    user_id: str,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    from sqlalchemy import func as sa_func

    latest_per_biz = (
        db.query(
            WfProcessInstance.business_id,
            sa_func.max(WfProcessInstance.create_time).label("max_ct"),
        )
        .filter(
            and_(
                WfProcessInstance.starter_user_id == user_id,
                WfProcessInstance.del_flag == 0,
                WfProcessInstance.business_id != None,
                WfProcessInstance.business_id != "",
            )
        )
        .group_by(WfProcessInstance.business_id)
        .subquery()
    )

    q0 = (
        db.query(WfProcessInstance, WfProcessVersion, WfProcessDef)
        .join(WfProcessVersion, WfProcessInstance.process_version_id == WfProcessVersion.id)
        .join(WfProcessDef, WfProcessVersion.process_def_id == WfProcessDef.id)
        .join(
            latest_per_biz,
            and_(
                WfProcessInstance.business_id == latest_per_biz.c.business_id,
                WfProcessInstance.create_time == latest_per_biz.c.max_ct,
            ),
        )
        .filter(
            and_(
                WfProcessInstance.starter_user_id == user_id,
                WfProcessInstance.del_flag == 0,
                WfProcessVersion.del_flag == 0,
                WfProcessDef.del_flag == 0,
            )
        )
    )
    if tenant_no:
        q0 = q0.filter(WfProcessInstance.tenant_no == tenant_no)
    if status:
        q0 = q0.filter(WfProcessInstance.status == status)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(WfProcessDef.process_name.like(like), WfProcessInstance.business_id.like(like))
        )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q0 = apply_data_scope(q0, WfProcessInstance, user_info, db, tenant_no)

    total = q0.order_by(None).with_entities(func.count(WfProcessInstance.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q0.order_by(WfProcessInstance.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    result = []
    for inst, ver, defn in rows:
        result.append({
            "id": inst.id,
            "processName": defn.process_name,
            "businessType": inst.business_type,
            "businessId": inst.business_id,
            "starterUserId": inst.starter_user_id,
            "status": inst.status,
            "createTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
            "updateTime": inst.update_time.strftime("%Y-%m-%d %H:%M:%S") if inst.update_time else "",
        })
    return result, int(total)


def get_all_tasks_for_business(
    db: Session,
    business_type: str,
    business_id: str,
) -> List[Dict[str, Any]]:
    instances = (
        db.query(WfProcessInstance)
        .filter(
            and_(
                WfProcessInstance.business_type == business_type,
                WfProcessInstance.business_id == business_id,
                WfProcessInstance.del_flag == 0,
            )
        )
        .order_by(WfProcessInstance.create_time.asc())
        .all()
    )

    all_tasks: List[Dict[str, Any]] = []
    for inst in instances:
        node_name_map: Dict[str, str] = {}
        if inst.process_version_id:
            for n in get_nodes_by_version(db, inst.process_version_id):
                node_name_map[n.node_key] = n.node_name or n.node_key

        tasks = (
            db.query(WfTask)
            .filter(and_(WfTask.instance_id == inst.id, WfTask.del_flag == 0))
            .order_by(WfTask.create_time.asc())
            .all()
        )

        user_ids = {t.assignee_user_id for t in tasks if t.assignee_user_id}
        user_name_map: Dict[str, str] = {}
        if user_ids:
            rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(user_ids)).all()
            user_name_map = {r.id: r.nick_name for r in rows}

        for t in tasks:
            all_tasks.append({
                "id": t.id,
                "instanceId": inst.id,
                "instanceStatus": inst.status,
                "nodeKey": t.node_key,
                "nodeName": node_name_map.get(t.node_key, t.node_key or ""),
                "assigneeUserId": t.assignee_user_id,
                "assigneeUserName": user_name_map.get(t.assignee_user_id, t.assignee_user_id or ""),
                "status": t.status,
                "commentText": t.comment_text,
                "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
                "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
            })

    return all_tasks


# ---------------------------------------------------------------------------
# 流程任务 wf_task
# ---------------------------------------------------------------------------


def create_task(
    db: Session,
    *,
    instance_id: str,
    node_key: Optional[str] = None,
    assignee_user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_id: Optional[str] = None,
) -> WfTask:
    obj = WfTask(
        id=generate_uuid(),
        instance_id=instance_id,
        node_key=node_key,
        assignee_user_id=assignee_user_id,
        status="pending",
        tenant_no=tenant_no or "shanxia2.0",
        order_num=0,
        remark=None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def get_task(db: Session, task_id: str) -> Optional[WfTask]:
    return (
        db.query(WfTask)
        .filter(and_(WfTask.id == task_id, WfTask.del_flag == 0))
        .first()
    )


def get_pending_tasks_by_node(db: Session, instance_id: str, node_key: str) -> List[WfTask]:
    return (
        db.query(WfTask)
        .filter(
            and_(
                WfTask.instance_id == instance_id,
                WfTask.node_key == node_key,
                WfTask.status == "pending",
                WfTask.del_flag == 0,
            )
        )
        .all()
    )


def _task_row_to_dict(
    t: WfTask,
    process_name: Optional[str] = None,
    business_type: Optional[str] = None,
    business_id: Optional[str] = None,
    starter_user_id: Optional[str] = None,
    instance_status: Optional[str] = None,
    instance_create_time: Optional[str] = None,
) -> Dict[str, Any]:
    return {
        "id": t.id,
        "instanceId": t.instance_id,
        "nodeKey": t.node_key,
        "assigneeUserId": t.assignee_user_id,
        "status": t.status,
        "commentText": t.comment_text,
        "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
        "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
        "processName": process_name,
        "businessType": business_type,
        "businessId": business_id,
        "starterUserId": starter_user_id,
        "instanceStatus": instance_status,
        "instanceCreateTime": instance_create_time,
    }


def list_pending_tasks(
    db: Session,
    user_id: str,
    *,
    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]:
    q0 = (
        db.query(WfTask, WfProcessInstance, WfProcessVersion, WfProcessDef)
        .join(WfProcessInstance, WfTask.instance_id == WfProcessInstance.id)
        .join(WfProcessVersion, WfProcessInstance.process_version_id == WfProcessVersion.id)
        .join(WfProcessDef, WfProcessVersion.process_def_id == WfProcessDef.id)
        .filter(
            and_(
                WfTask.assignee_user_id == user_id,
                WfTask.status == "pending",
                WfTask.del_flag == 0,
                WfProcessInstance.del_flag == 0,
                WfProcessVersion.del_flag == 0,
                WfProcessDef.del_flag == 0,
            )
        )
    )
    if tenant_no:
        q0 = q0.filter(WfTask.tenant_no == tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(WfProcessDef.process_name.like(like), WfProcessDef.process_code.like(like))
        )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q0 = apply_data_scope(q0, WfTask, user_info, db, tenant_no)

    total = q0.order_by(None).with_entities(func.count(WfTask.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q0.order_by(WfTask.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    result = []
    for task, inst, ver, defn in rows:
        result.append(
            _task_row_to_dict(
                task,
                process_name=defn.process_name,
                business_type=inst.business_type,
                business_id=inst.business_id,
                starter_user_id=inst.starter_user_id,
                instance_status=inst.status,
                instance_create_time=inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
            )
        )
    return result, int(total)


def list_completed_tasks(
    db: Session,
    user_id: str,
    *,
    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]:
    q0 = (
        db.query(WfTask, WfProcessInstance, WfProcessVersion, WfProcessDef)
        .join(WfProcessInstance, WfTask.instance_id == WfProcessInstance.id)
        .join(WfProcessVersion, WfProcessInstance.process_version_id == WfProcessVersion.id)
        .join(WfProcessDef, WfProcessVersion.process_def_id == WfProcessDef.id)
        .filter(
            and_(
                WfTask.assignee_user_id == user_id,
                WfTask.status.in_(["approved", "rejected"]),
                WfTask.del_flag == 0,
                WfProcessInstance.del_flag == 0,
                WfProcessVersion.del_flag == 0,
                WfProcessDef.del_flag == 0,
            )
        )
    )
    if tenant_no:
        q0 = q0.filter(WfTask.tenant_no == tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(WfProcessDef.process_name.like(like), WfProcessDef.process_code.like(like))
        )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q0 = apply_data_scope(q0, WfTask, user_info, db, tenant_no)

    total = q0.order_by(None).with_entities(func.count(WfTask.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q0.order_by(WfTask.update_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    result = []
    for task, inst, ver, defn in rows:
        result.append(
            _task_row_to_dict(
                task,
                process_name=defn.process_name,
                business_type=inst.business_type,
                business_id=inst.business_id,
                starter_user_id=inst.starter_user_id,
                instance_status=inst.status,
                instance_create_time=inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
            )
        )
    return result, int(total)


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