zkxw/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,
    WfNode,
    WfProcessInstance,
    WfTask,
    WfTransition,
)
from app.services.web_auth_roles import list_user_ids_with_any_effective_role_ids
from app.services.workflow_callback import trigger_callback, validate_callback_url

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("流程定义没有可用版本，请先发布")
        safe_callback = validate_callback_url(callback_url)
        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=safe_callback,
            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 "zkxw")
        self.db.refresh(instance)
        return instance

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

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

        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,
        *,
        actor_user_id: Optional[str] = None,
    ) -> WfProcessInstance:
        task = wf_crud.get_task(self.db, task_id)
        if not task:
            raise ValueError("任务不存在")
        if task.status != "pending":
            raise ValueError(f"任务状态为 {task.status}，无法驳回")
        self._assert_task_action_allowed(task, tenant_no, actor_user_id)

        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

    def _assert_task_action_allowed(
        self,
        task: WfTask,
        tenant_no: str,
        actor_user_id: Optional[str],
    ) -> None:
        """仅办理人本人可审批/驳回；并校验任务所属实例租户。"""
        instance = wf_crud.get_instance(self.db, task.instance_id)
        if not instance:
            raise ValueError("流程实例不存在")
        inst_tenant = (instance.tenant_no or "").strip()
        if inst_tenant and tenant_no and inst_tenant != tenant_no:
            raise ValueError("无权操作其他租户的审批任务")
        if not actor_user_id:
            raise ValueError("无法识别当前用户，禁止审批")
        assignee = (task.assignee_user_id or "").strip()
        if not assignee or assignee != actor_user_id:
            raise ValueError("仅任务办理人可审批或驳回")

    # ------------------------------------------------------------------
    # _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 []
            user_ids = list_user_ids_with_any_effective_role_ids(
                self.db, tenant_no=tenant_no, role_ids=role_ids
            )
            if not user_ids:
                return []
            rows = (
                self.db.query(SysUser.id)
                .filter(
                    SysUser.id.in_(user_ids),
                    SysUser.tenant_no == tenant_no,
                    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 {}