shanxia2.0/backend/app/services/workflow_engine.py

"""工作流引擎 —— 读取节点与连线执行审批流转"""
from __future__ import annotations

import json
import logging
import re
from datetime import datetime
from typing import Any, Dict, List, Optional

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.crud import workflow as wf_crud
from app.models.models import (
    SysUser,
    SysUserRole,
    WfNode,
    WfProcessInstance,
    WfTask,
    WfTransition,
)
from app.services.workflow_callback import trigger_callback

logger = logging.getLogger("sqxp.workflow_engine")

_FORBIDDEN_SQL_KEYWORDS = re.compile(
    r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|TRUNCATE)\b", re.IGNORECASE
)


class WorkflowEngine:
    def __init__(self, db: Session):
        self.db = db

    # ------------------------------------------------------------------
    # start_process
    # ------------------------------------------------------------------

    def start_process(
        self,
        *,
        process_def_id: str,
        business_type: Optional[str] = None,
        business_id: Optional[str] = None,
        starter_user_id: Optional[str] = None,
        variables: Optional[Dict[str, Any]] = None,
        callback_url: Optional[str] = None,
        tenant_no: Optional[str] = None,
        user_id: Optional[str] = None,
    ) -> WfProcessInstance:
        version = wf_crud.get_latest_version(self.db, process_def_id)
        if not version:
            raise ValueError("流程定义没有可用版本，请先发布")
        instance = wf_crud.create_instance(
            self.db,
            process_version_id=version.id,
            business_type=business_type,
            business_id=business_id,
            starter_user_id=starter_user_id,
            variables=variables,
            callback_url=callback_url,
            tenant_no=tenant_no,
            user_id=user_id,
        )
        nodes = wf_crud.get_nodes_by_version(self.db, version.id)
        start_node = next((n for n in nodes if n.node_type == "start"), None)
        if not start_node:
            raise ValueError("流程缺少开始节点")
        self._advance(instance.id, start_node.node_key, tenant_no or "shanxia2.0")
        self.db.refresh(instance)
        return instance

    # ------------------------------------------------------------------
    # approve_task
    # ------------------------------------------------------------------

    def approve_task(self, task_id: str, comment: str, tenant_no: str) -> WfProcessInstance:
        task = wf_crud.get_task(self.db, task_id)
        if not task:
            raise ValueError("任务不存在")
        if task.status != "pending":
            raise ValueError(f"任务状态为 {task.status}，无法审批")

        task.status = "approved"
        task.comment_text = comment
        task.update_time = datetime.now()
        self.db.add(task)
        self.db.commit()

        instance = wf_crud.get_instance(self.db, task.instance_id)
        if not instance:
            raise ValueError("流程实例不存在")

        version_id = instance.process_version_id
        nodes = wf_crud.get_nodes_by_version(self.db, version_id)
        current_node = next(
            (n for n in nodes if n.node_key == task.node_key), None
        )
        config = self._parse_node_config(current_node)
        pass_type = config.get("passType", "any")

        if pass_type == "any":
            others = wf_crud.get_pending_tasks_by_node(
                self.db, task.instance_id, task.node_key
            )
            for t in others:
                t.status = "withdrawn"
                t.update_time = datetime.now()
                self.db.add(t)
            self.db.commit()
            self._advance(instance.id, task.node_key, tenant_no)
        else:
            remaining = wf_crud.get_pending_tasks_by_node(
                self.db, task.instance_id, task.node_key
            )
            if not remaining:
                self._advance(instance.id, task.node_key, tenant_no)

        self.db.refresh(instance)
        return instance

    # ------------------------------------------------------------------
    # reject_task
    # ------------------------------------------------------------------

    def reject_task(self, task_id: str, comment: str, tenant_no: str) -> WfProcessInstance:
        task = wf_crud.get_task(self.db, task_id)
        if not task:
            raise ValueError("任务不存在")
        if task.status != "pending":
            raise ValueError(f"任务状态为 {task.status}，无法驳回")

        task.status = "rejected"
        task.comment_text = comment
        task.update_time = datetime.now()
        self.db.add(task)
        self.db.commit()

        others = wf_crud.get_pending_tasks_by_node(
            self.db, task.instance_id, task.node_key
        )
        for t in others:
            t.status = "withdrawn"
            t.update_time = datetime.now()
            self.db.add(t)
        self.db.commit()

        instance = wf_crud.get_instance(self.db, task.instance_id)
        if instance:
            wf_crud.update_instance_status(self.db, instance.id, "rejected")
            self.db.refresh(instance)
            trigger_callback(self.db, instance, "rejected")

        return instance

    # ------------------------------------------------------------------
    # _advance (internal)
    # ------------------------------------------------------------------

    def _advance(
        self, instance_id: str, from_node_key: str, tenant_no: str
    ) -> None:
        instance = wf_crud.get_instance(self.db, instance_id)
        if not instance:
            return
        if instance.status != "running":
            return

        version_id = instance.process_version_id
        nodes = wf_crud.get_nodes_by_version(self.db, version_id)
        transitions = wf_crud.get_transitions_by_version(self.db, version_id)
        variables = self._parse_variables(instance)

        current_node = next(
            (n for n in nodes if n.node_key == from_node_key), None
        )
        if not current_node:
            return

        out_edges = [
            t for t in transitions if t.from_node_key == from_node_key
        ]
        if not out_edges:
            return

        if current_node.node_type in ("start", "approval"):
            target_key = out_edges[0].to_node_key
        elif current_node.node_type == "condition":
            target_key = self._evaluate_condition(out_edges, variables)
        else:
            target_key = out_edges[0].to_node_key

        next_node = next((n for n in nodes if n.node_key == target_key), None)
        if next_node:
            self._execute_node(instance_id, next_node, variables, tenant_no)

    # ------------------------------------------------------------------
    # _execute_node
    # ------------------------------------------------------------------

    def _execute_node(
        self,
        instance_id: str,
        node: WfNode,
        variables: Dict[str, Any],
        tenant_no: str,
    ) -> None:
        if node.node_type == "end":
            wf_crud.update_instance_status(self.db, instance_id, "completed")
            instance = wf_crud.get_instance(self.db, instance_id)
            if instance:
                trigger_callback(self.db, instance, "completed")
            return

        if node.node_type == "approval":
            config = self._parse_node_config(node)
            user_ids = self._resolve_approvers(config, variables, tenant_no)
            for uid in user_ids:
                wf_crud.create_task(
                    self.db,
                    instance_id=instance_id,
                    node_key=node.node_key,
                    assignee_user_id=uid,
                    tenant_no=tenant_no,
                )
            return

        if node.node_type == "condition":
            self._advance(instance_id, node.node_key, tenant_no)
            return

        self._advance(instance_id, node.node_key, tenant_no)

    # ------------------------------------------------------------------
    # _resolve_approvers
    # ------------------------------------------------------------------

    def _resolve_approvers(
        self,
        config: Dict[str, Any],
        variables: Dict[str, Any],
        tenant_no: str,
    ) -> List[str]:
        strategy = config.get("strategy", "users")

        if strategy == "users":
            return config.get("userIds", [])

        if strategy == "roles":
            role_ids = config.get("roleIds", [])
            if not role_ids:
                return []
            rows = (
                self.db.query(SysUser.id)
                .join(SysUserRole, SysUserRole.user_id == SysUser.id)
                .filter(
                    SysUserRole.role_id.in_(role_ids),
                    SysUser.del_flag == 0,
                    SysUser.status == "1",
                )
                .all()
            )
            return [r.id for r in rows]

        if strategy == "sql":
            sql_template = config.get("sql", "")
            if not sql_template or not sql_template.strip().upper().startswith("SELECT"):
                raise ValueError("SQL 审批人策略必须以 SELECT 开头")
            if _FORBIDDEN_SQL_KEYWORDS.search(sql_template):
                raise ValueError("SQL 审批人策略包含禁止的关键字")
            try:
                result = self.db.execute(text(sql_template), variables or {})
                return [str(row[0]) for row in result.fetchall()]
            except Exception as exc:
                logger.exception("SQL 审批人解析失败: %s", exc)
                raise ValueError(f"SQL 审批人解析失败: {exc}") from exc

        return []

    # ------------------------------------------------------------------
    # _evaluate_condition
    # ------------------------------------------------------------------

    def _evaluate_condition(
        self, out_edges: List[WfTransition], variables: Dict[str, Any]
    ) -> str:
        unconditional_fallback = out_edges[-1].to_node_key if out_edges else ""

        for edge in out_edges:
            cond_str = edge.condition_expr
            if not cond_str:
                continue
            try:
                rules = json.loads(cond_str)
            except (json.JSONDecodeError, TypeError):
                continue
            if self._check_rules(rules, variables):
                return edge.to_node_key

        return unconditional_fallback

    # ------------------------------------------------------------------
    # _check_rules (recursive)
    # ------------------------------------------------------------------

    def _check_rules(self, rules: Any, variables: Dict[str, Any]) -> bool:
        if isinstance(rules, list):
            return any(self._check_rules(r, variables) for r in rules)

        if isinstance(rules, dict):
            if "and" in rules:
                return all(self._check_rules(r, variables) for r in rules["and"])
            if "or" in rules:
                return any(self._check_rules(r, variables) for r in rules["or"])
            if "rules" in rules:
                return self._check_rules(rules["rules"], variables)
            if "field" in rules:
                actual = self._get_variable(rules["field"], variables)
                op = rules.get("op", "==")
                expected = rules.get("value")
                return self._compare(actual, op, expected)

        return False

    # ------------------------------------------------------------------
    # _compare
    # ------------------------------------------------------------------

    @staticmethod
    def _compare(actual: Any, op: str, expected: Any) -> bool:
        try:
            if op == "==":
                return str(actual) == str(expected)
            if op == "!=":
                return str(actual) != str(expected)
            if op == ">":
                return float(actual) > float(expected)
            if op == "<":
                return float(actual) < float(expected)
            if op == ">=":
                return float(actual) >= float(expected)
            if op == "<=":
                return float(actual) <= float(expected)
            if op == "in":
                return str(actual) in ([str(e) for e in expected] if isinstance(expected, list) else str(expected))
            if op == "not_in":
                return str(actual) not in ([str(e) for e in expected] if isinstance(expected, list) else str(expected))
            if op == "contains":
                return str(expected) in str(actual)
        except (ValueError, TypeError):
            return False
        return False

    # ------------------------------------------------------------------
    # _get_variable (dot-notation)
    # ------------------------------------------------------------------

    @staticmethod
    def _get_variable(field: str, variables: Dict[str, Any]) -> Any:
        parts = field.split(".")
        current: Any = variables
        for part in parts:
            if isinstance(current, dict):
                current = current.get(part)
            else:
                return None
        return current

    # ------------------------------------------------------------------
    # helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _parse_node_config(node: Optional[WfNode]) -> Dict[str, Any]:
        if not node or not node.config_json:
            return {}
        try:
            return json.loads(node.config_json)
        except (json.JSONDecodeError, TypeError):
            return {}

    @staticmethod
    def _parse_variables(instance: WfProcessInstance) -> Dict[str, Any]:
        if not instance.variables_json:
            return {}
        try:
            return json.loads(instance.variables_json)
        except (json.JSONDecodeError, TypeError):
            return {}