mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
599 lines
24 KiB
Python
599 lines
24 KiB
Python
"""
|
||
管理工作流实例及动作执行生命周期。
|
||
"""
|
||
|
||
import threading
|
||
from time import monotonic, sleep
|
||
from typing import Any, Dict, List, Optional, Tuple, cast
|
||
|
||
from pydantic import BaseModel
|
||
|
||
from app.application.workflow import (
|
||
WorkflowExecutionOwner,
|
||
WorkflowSnapshot,
|
||
get_configured_workflow_query,
|
||
)
|
||
from app.foundation.reflection import ModuleHelper
|
||
from app.foundation.singleton import Singleton
|
||
from app.runtime.events import Event, eventmanager
|
||
from app.runtime.log import logger
|
||
from app.runtime.stop import runtime_stop_state
|
||
from app.schemas.types import EventType
|
||
from app.schemas.workflow import Action, ActionContext, ActionResult
|
||
|
||
_WORKFLOW_STOP_TIMEOUT_SECONDS = 10.0
|
||
|
||
|
||
class _ActionDeadlineToken:
|
||
"""
|
||
把单次动作截止时间与工作流取消信号组合为协作停止令牌。
|
||
"""
|
||
|
||
def __init__(self, workflow_id: int, deadline: float, cancel_token: Optional[Any]) -> None:
|
||
"""
|
||
保存单调时钟截止时间和上层工作流令牌。
|
||
"""
|
||
self.workflow_id = workflow_id
|
||
self.deadline = deadline
|
||
self.cancel_token = cancel_token
|
||
|
||
def is_cancelled(self) -> bool:
|
||
"""
|
||
到达截止时间或收到上层停止请求时通知支持取消的动作退出。
|
||
"""
|
||
return bool(
|
||
monotonic() >= self.deadline
|
||
or (self.cancel_token and self.cancel_token.is_cancelled())
|
||
or runtime_stop_state.is_workflow_stopped(self.workflow_id)
|
||
)
|
||
|
||
|
||
class WorkflowManager(metaclass=Singleton):
|
||
"""
|
||
工作流管理器
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
"""
|
||
创建动作、事件触发器和活动执行 owner 注册表。
|
||
"""
|
||
# 所有动作定义
|
||
self._lock = threading.RLock()
|
||
self._actions: Dict[str, Any] = {}
|
||
self._event_workflows: Dict[str, List[int]] = {}
|
||
self._accepting_executions = True
|
||
self._executions: Dict[int, WorkflowExecutionOwner] = {}
|
||
self.init()
|
||
|
||
def init(self) -> None:
|
||
"""
|
||
扫描并登记工作流动作与事件触发器。
|
||
"""
|
||
|
||
def filter_func(obj: object) -> bool:
|
||
"""
|
||
只接纳在工作流动作包中定义的具体动作类。
|
||
"""
|
||
if not isinstance(obj, type):
|
||
return False
|
||
if not hasattr(obj, 'execute') or not hasattr(obj, "name"):
|
||
return False
|
||
if obj.__name__ == "BaseAction":
|
||
return False
|
||
return obj.__module__.startswith("app.workflow.actions")
|
||
|
||
# 加载所有动作
|
||
self._actions = {}
|
||
actions = ModuleHelper.load(
|
||
"app.workflow.actions",
|
||
filter_func=lambda _, obj: filter_func(obj)
|
||
)
|
||
for action in actions:
|
||
logger.debug(f"加载动作: {action.__name__}")
|
||
try:
|
||
self._actions[action.__name__] = action
|
||
except Exception as err:
|
||
logger.error(f"加载动作失败: {action.__name__} - {err}")
|
||
|
||
# 加载工作流事件触发器
|
||
self.load_workflow_events()
|
||
|
||
def register_execution(self, owner: WorkflowExecutionOwner) -> bool:
|
||
"""
|
||
在同一锁内登记执行，拒绝停机封口和同一工作流的重复准入。
|
||
"""
|
||
with self._lock:
|
||
if not self._accepting_executions:
|
||
return False
|
||
if any(execution.workflow_id == owner.workflow_id for execution in self._executions.values()):
|
||
return False
|
||
self._executions[id(owner)] = owner
|
||
return True
|
||
|
||
def unregister_execution(self, owner: WorkflowExecutionOwner) -> None:
|
||
"""
|
||
仅在执行及其节点线程池真实终止后释放 owner。
|
||
"""
|
||
with self._lock:
|
||
self._executions.pop(id(owner), None)
|
||
|
||
def stop(
|
||
self,
|
||
timeout: float = _WORKFLOW_STOP_TIMEOUT_SECONDS,
|
||
) -> bool:
|
||
"""
|
||
封口工作流入口并有限等待全部活动执行。
|
||
|
||
:param timeout: 等待活动执行真实终止的最长秒数
|
||
:return: 全部执行终止并安全释放动作注册表时返回 True
|
||
"""
|
||
with self._lock:
|
||
self._accepting_executions = False
|
||
event_type_values = tuple(self._event_workflows)
|
||
for event_type_str in event_type_values:
|
||
self.remove_workflow_event(event_type_str=event_type_str)
|
||
with self._lock:
|
||
self._event_workflows = {}
|
||
executions = tuple(self._executions.values())
|
||
|
||
converged = True
|
||
for execution in executions:
|
||
try:
|
||
execution.request_stop()
|
||
except Exception as err:
|
||
converged = False
|
||
logger.error("请求停止工作流执行失败：%s", err)
|
||
|
||
deadline = monotonic() + max(0.0, timeout)
|
||
for execution in executions:
|
||
try:
|
||
if not execution.wait_stopped(
|
||
timeout=max(0.0, deadline - monotonic()),
|
||
):
|
||
converged = False
|
||
except Exception as err:
|
||
converged = False
|
||
logger.error("等待工作流执行停止失败：%s", err)
|
||
with self._lock:
|
||
converged = converged and not self._executions
|
||
if converged:
|
||
self._actions = {}
|
||
return converged
|
||
|
||
def execute(self, workflow_id: int, action: Action, context: Optional[ActionContext] = None,
|
||
inputs: Optional[dict[str, Any]] = None,
|
||
runtime: Optional[dict[str, Any]] = None,
|
||
cancel_token: Optional[Any] = None) -> ActionResult:
|
||
"""
|
||
执行工作流动作
|
||
"""
|
||
if not context:
|
||
context = ActionContext()
|
||
action_type = action.type
|
||
if not action_type or action_type not in self._actions:
|
||
logger.error(f"未找到动作: {action.type} - {action.name}")
|
||
return ActionResult(success=False, message=" ", context=context)
|
||
|
||
retry_config = self._get_retry_config(action)
|
||
max_attempts = int(retry_config["max_attempts"])
|
||
interval = retry_config["interval"]
|
||
backoff = retry_config["backoff"]
|
||
action_result = ActionResult(success=False, message="", context=context)
|
||
|
||
for attempt in range(1, max_attempts + 1):
|
||
if self._is_cancelled(workflow_id, cancel_token):
|
||
return ActionResult(success=False, message="工作流已取消", context=context)
|
||
runtime_data = {
|
||
**(runtime or {}),
|
||
"attempt": attempt,
|
||
"max_attempts": max_attempts,
|
||
"cancel_token": cancel_token,
|
||
}
|
||
action_result = self._execute_action_once(
|
||
workflow_id=workflow_id,
|
||
action=action,
|
||
context=context,
|
||
inputs=inputs or {},
|
||
runtime=runtime_data,
|
||
cancel_token=cancel_token
|
||
)
|
||
action_result.attempts = attempt
|
||
context = action_result.context or context
|
||
if action_result.success:
|
||
logger.info(f"{action.name} 执行成功")
|
||
return action_result
|
||
if attempt < max_attempts and not self._is_cancelled(workflow_id, cancel_token):
|
||
wait_seconds = interval * (backoff ** (attempt - 1))
|
||
logger.info(f"{action.name} 执行失败，{wait_seconds} 秒后重试（{attempt}/{max_attempts}）...")
|
||
self._sleep_with_cancel(workflow_id, wait_seconds, cancel_token)
|
||
|
||
logger.error(f"{action.name} 执行失败！")
|
||
return action_result
|
||
|
||
def excute(self, workflow_id: int, action: Action,
|
||
context: Optional[ActionContext] = None) -> Tuple[bool, str, ActionContext]:
|
||
"""
|
||
执行工作流动作，兼容历史拼写错误的方法名。
|
||
"""
|
||
action_result = self.execute(workflow_id=workflow_id, action=action, context=context)
|
||
return bool(action_result.success), action_result.message or "", action_result.context or context or ActionContext()
|
||
|
||
@staticmethod
|
||
def _normalize_action_result(result: Any, action_obj: Any, fallback_context: ActionContext) -> ActionResult:
|
||
"""
|
||
将旧版动作上下文与新版结构化结果统一为动作执行结果。
|
||
"""
|
||
if isinstance(result, ActionResult):
|
||
result.context = result.context or fallback_context
|
||
if result.message is None:
|
||
result.message = action_obj.message
|
||
return result
|
||
return ActionResult(
|
||
success=action_obj.success,
|
||
message=action_obj.message,
|
||
context=result or fallback_context
|
||
)
|
||
|
||
def _execute_action_once(self, workflow_id: int, action: Action, context: ActionContext,
|
||
inputs: dict[str, Any], runtime: dict[str, Any],
|
||
cancel_token: Optional[Any]) -> ActionResult:
|
||
"""
|
||
运行单次尝试，保留同步动作真实返回后再判断失败和超时的语义。
|
||
"""
|
||
action_type = action.type
|
||
if not action_type or action_type not in self._actions:
|
||
return ActionResult(success=False, message="未找到动作", context=context)
|
||
action_obj = self._actions[action_type](action.id)
|
||
logger.info(f"执行动作: {action.id} - {action.name}")
|
||
try:
|
||
action_result = self._run_action_with_loop(
|
||
workflow_id=workflow_id,
|
||
action=action,
|
||
action_obj=action_obj,
|
||
context=context,
|
||
inputs=inputs,
|
||
runtime=runtime,
|
||
cancel_token=cancel_token
|
||
)
|
||
except Exception as err:
|
||
logger.error(f"{action.name} 执行失败: {err}")
|
||
return ActionResult(success=False, message=f"{err}", context=context)
|
||
return action_result
|
||
|
||
def _run_action_with_loop(self, workflow_id: int, action: Action, action_obj: Any,
|
||
context: ActionContext, inputs: dict[str, Any], runtime: dict[str, Any],
|
||
cancel_token: Optional[Any]) -> ActionResult:
|
||
"""
|
||
循环前后均检查截止时间，并向动作传递协作令牌；不分离阻塞节点线程。
|
||
"""
|
||
timeout = self._get_action_timeout(action)
|
||
deadline = monotonic() + timeout if timeout else None
|
||
action_token = _ActionDeadlineToken(workflow_id, deadline, cancel_token) if deadline is not None else cancel_token
|
||
runtime = {**runtime, "cancel_token": action_token, "deadline": deadline}
|
||
loop = self._get_action_data_value(action, "loop")
|
||
loop_interval = self._get_action_data_value(action, "loop_interval")
|
||
action_result = ActionResult(success=False, context=context)
|
||
while True:
|
||
if self._is_cancelled(workflow_id, cancel_token):
|
||
return ActionResult(success=False, message="工作流已取消", context=action_result.context or context)
|
||
if deadline is not None and monotonic() >= deadline:
|
||
return ActionResult(success=False, message=f"动作执行超时（{timeout}秒）", context=action_result.context or context)
|
||
action_result = self._call_action(
|
||
workflow_id=workflow_id,
|
||
action=action,
|
||
action_obj=action_obj,
|
||
context=action_result.context or context,
|
||
inputs=inputs,
|
||
runtime=runtime
|
||
)
|
||
if self._is_cancelled(workflow_id, cancel_token):
|
||
return ActionResult(success=False, message="工作流已取消", context=action_result.context or context)
|
||
if deadline is not None and monotonic() >= deadline:
|
||
return ActionResult(success=False, message=f"动作执行超时（{timeout}秒）", context=action_result.context or context)
|
||
if not (loop and loop_interval and not action_obj.done):
|
||
return action_result
|
||
wait_seconds = float(loop_interval)
|
||
if deadline is not None:
|
||
wait_seconds = min(wait_seconds, max(0.0, deadline - monotonic()))
|
||
logger.info(f"{action.name} 等待 {wait_seconds} 秒后继续执行 ...")
|
||
self._sleep_with_cancel(workflow_id, wait_seconds, action_token)
|
||
|
||
def _call_action(self, workflow_id: int, action: Action, action_obj: Any,
|
||
context: ActionContext, inputs: dict[str, Any],
|
||
runtime: dict[str, Any]) -> ActionResult:
|
||
"""
|
||
兼容新旧动作入口并把真实同步返回值转换为统一结果。
|
||
"""
|
||
if hasattr(action_obj, "execute_with_inputs"):
|
||
result = action_obj.execute_with_inputs(workflow_id, action.data, inputs, runtime, context)
|
||
else:
|
||
result = action_obj.execute(workflow_id, action.data, context)
|
||
return self._normalize_action_result(result, action_obj, context)
|
||
|
||
@staticmethod
|
||
def _get_action_data_value(action: Action, key: str) -> Any:
|
||
"""
|
||
读取兼容旧版存放在 data 中的动作参数。
|
||
"""
|
||
data = action.data or {}
|
||
return data.get(key) if isinstance(data, dict) else None
|
||
|
||
def _get_action_timeout(self, action: Action) -> Optional[int]:
|
||
"""
|
||
读取新旧动作配置中的单次尝试超时秒数。
|
||
"""
|
||
timeout = action.timeout or self._get_action_data_value(action, "timeout")
|
||
return int(timeout) if timeout else None
|
||
|
||
def _get_retry_config(self, action: Action) -> dict[str, float | int]:
|
||
"""
|
||
归一化动作重试次数、间隔和退避配置。
|
||
"""
|
||
retry_config = action.retry or self._get_action_data_value(action, "retry") or {}
|
||
if isinstance(retry_config, BaseModel):
|
||
retry_config = retry_config.model_dump(exclude_none=True)
|
||
if not isinstance(retry_config, dict):
|
||
retry_config = {}
|
||
return {
|
||
"max_attempts": max(int(retry_config.get("max_attempts") or 1), 1),
|
||
"interval": max(float(retry_config.get("interval") or 0), 0),
|
||
"backoff": max(float(retry_config.get("backoff") or 1), 1),
|
||
}
|
||
|
||
@staticmethod
|
||
def _is_cancelled(workflow_id: int, cancel_token: Optional[Any]) -> bool:
|
||
"""
|
||
组合调用方取消令牌和全局工作流停止状态。
|
||
"""
|
||
if cancel_token and cancel_token.is_cancelled():
|
||
return True
|
||
return runtime_stop_state.is_workflow_stopped(workflow_id)
|
||
|
||
def _sleep_with_cancel(self, workflow_id: int, seconds: float, cancel_token: Optional[Any]) -> None:
|
||
"""
|
||
分段等待，并在取消或截止时立即停止继续等待。
|
||
"""
|
||
deadline = monotonic() + seconds
|
||
while monotonic() < deadline:
|
||
if self._is_cancelled(workflow_id, cancel_token):
|
||
return
|
||
sleep(max(0.0, min(0.1, deadline - monotonic())))
|
||
|
||
def list_actions(self) -> List[dict[str, Any]]:
|
||
"""
|
||
获取所有动作
|
||
"""
|
||
return [
|
||
{
|
||
"type": key,
|
||
"name": action.name,
|
||
"description": action.description,
|
||
"contract": action.get_contract(),
|
||
"data": {
|
||
"label": action.name,
|
||
**action.data
|
||
}
|
||
} for key, action in self._actions.items()
|
||
]
|
||
|
||
def get_action_contract(self, action_type: str) -> dict[str, Any]:
|
||
"""
|
||
获取动作输入输出契约。
|
||
"""
|
||
action = self._actions.get(action_type)
|
||
if not action or not hasattr(action, "get_contract"):
|
||
return {}
|
||
return cast(dict[str, Any], action.get_contract())
|
||
|
||
def update_workflow_event(self, workflow: WorkflowSnapshot) -> None:
|
||
"""
|
||
根据工作流当前配置更新事件触发器。
|
||
|
||
:param workflow: 待更新的工作流快照
|
||
"""
|
||
# 工作流可能切换触发事件，先按工作流ID从所有事件映射中移除。
|
||
self.remove_workflow_event(workflow_id=workflow.id)
|
||
# 如果工作流是事件触发类型且未被禁用
|
||
if workflow.trigger_type == "event" and workflow.state != 'P':
|
||
# 注册事件触发器
|
||
self.register_workflow_event(workflow.id, workflow.event_type)
|
||
|
||
def load_workflow_events(self, workflow_id: Optional[int] = None) -> None:
|
||
"""
|
||
从持久化快照加载工作流触发事件。
|
||
|
||
:param workflow_id: 指定工作流标识；None 加载全部已启用的事件工作流
|
||
"""
|
||
workflows = []
|
||
if workflow_id:
|
||
workflow = get_configured_workflow_query().get_sync(workflow_id)
|
||
if workflow:
|
||
workflows = [workflow]
|
||
else:
|
||
workflows = get_configured_workflow_query().list_event_enabled()
|
||
try:
|
||
for workflow in workflows:
|
||
self.update_workflow_event(workflow)
|
||
except Exception as e:
|
||
logger.error(f"加载事件触发工作流失败: {e}")
|
||
|
||
def register_workflow_event(
|
||
self, workflow_id: int, event_type_str: Optional[str]
|
||
) -> None:
|
||
"""
|
||
为有效事件类型注册工作流触发器。
|
||
|
||
:param workflow_id: 待触发的工作流标识
|
||
:param event_type_str: 事件类型值；为空时不登记
|
||
"""
|
||
if not event_type_str:
|
||
return
|
||
try:
|
||
event_type = EventType(event_type_str)
|
||
except ValueError:
|
||
logger.error(f"无效的事件类型: {event_type_str}")
|
||
return
|
||
if event_type in EventType:
|
||
with self._lock:
|
||
if event_type.value not in self._event_workflows:
|
||
self._event_workflows[event_type.value] = []
|
||
eventmanager.add_event_listener(event_type, self._handle_event)
|
||
# 记录工作流事件触发器
|
||
if workflow_id not in self._event_workflows[event_type.value]:
|
||
self._event_workflows[event_type.value].append(workflow_id)
|
||
logger.info(f"已注册工作流 {workflow_id} 事件触发器: {event_type.value}")
|
||
|
||
def remove_workflow_event(
|
||
self, workflow_id: Optional[int] = None,
|
||
event_type_str: Optional[str] = None,
|
||
) -> None:
|
||
"""
|
||
移除指定工作流或事件类型对应的触发器。
|
||
|
||
:param workflow_id: 指定工作流标识；None 清空匹配事件的工作流
|
||
:param event_type_str: 指定事件类型；None 遍历已登记的全部事件
|
||
"""
|
||
event_type_values = [event_type_str] if event_type_str else list(self._event_workflows.keys())
|
||
for event_type_value in event_type_values:
|
||
try:
|
||
event_type = EventType(event_type_value)
|
||
except ValueError:
|
||
logger.error(f"无效的事件类型: {event_type_value}")
|
||
continue
|
||
with self._lock:
|
||
workflow_ids = self._event_workflows.get(event_type.value)
|
||
if not workflow_ids:
|
||
continue
|
||
if workflow_id is None:
|
||
workflow_ids.clear()
|
||
elif workflow_id in workflow_ids:
|
||
workflow_ids.remove(workflow_id)
|
||
if not workflow_ids:
|
||
self._event_workflows.pop(event_type.value, None)
|
||
eventmanager.remove_event_listener(event_type, self._handle_event)
|
||
logger.info(f"已移除工作流 {workflow_id or ''} 事件触发器")
|
||
|
||
def _handle_event(self, event: Event) -> None:
|
||
"""
|
||
收到事件后触发已登记且匹配的工作流。
|
||
|
||
:param event: 当前事件及其载荷
|
||
"""
|
||
try:
|
||
event_type = event.event_type
|
||
event_type_str = str(getattr(event_type, "value", event_type))
|
||
with self._lock:
|
||
if event_type_str not in self._event_workflows:
|
||
return
|
||
workflow_ids = self._event_workflows[event_type_str].copy()
|
||
for workflow_id in workflow_ids:
|
||
self._trigger_workflow(workflow_id, event)
|
||
except Exception as e:
|
||
logger.error(f"处理工作流事件失败: {e}")
|
||
|
||
def _trigger_workflow(self, workflow_id: int, event: Event) -> None:
|
||
"""
|
||
确认工作流状态与事件条件后发出执行事件。
|
||
|
||
:param workflow_id: 候选工作流标识
|
||
:param event: 触发执行的原始事件
|
||
"""
|
||
try:
|
||
# 检查工作流是否存在且启用
|
||
workflow = get_configured_workflow_query().get_sync(workflow_id)
|
||
if not workflow or workflow.state == 'P':
|
||
return
|
||
|
||
# 检查事件条件
|
||
if not self._check_event_conditions(workflow, event):
|
||
logger.debug(f"工作流 {workflow.name} 事件条件不匹配，跳过执行")
|
||
return
|
||
|
||
# 检查工作流是否正在运行
|
||
if workflow.state == 'R':
|
||
logger.warning(f"工作流 {workflow.name} 正在运行中，跳过重复触发")
|
||
return
|
||
|
||
event_type = event.event_type
|
||
logger.info(
|
||
f"事件 {getattr(event_type, 'value', event_type)} 触发工作流: {workflow.name}"
|
||
)
|
||
|
||
# 发送工作流执行事件以启动工作流
|
||
eventmanager.send_event(EventType.WorkflowExecute, {
|
||
"workflow_id": workflow_id,
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.error(f"触发工作流 {workflow_id} 失败: {e}")
|
||
|
||
def _check_event_conditions(self, workflow: Any, event: Event) -> bool:
|
||
"""
|
||
检查事件是否满足工作流的触发条件
|
||
"""
|
||
if not workflow.event_conditions:
|
||
return True
|
||
|
||
conditions = workflow.event_conditions
|
||
event_data: Any = event.event_data
|
||
if isinstance(event_data, BaseModel):
|
||
event_data = event_data.model_dump()
|
||
if not isinstance(event_data, dict):
|
||
return False
|
||
|
||
# 检查字段匹配条件
|
||
for field, expected_value in conditions.items():
|
||
if field not in event_data:
|
||
return False
|
||
actual_value = event_data[field]
|
||
# 支持多种条件匹配方式
|
||
if isinstance(expected_value, dict):
|
||
# 复杂条件匹配
|
||
if not self._check_complex_condition(actual_value, expected_value):
|
||
return False
|
||
else:
|
||
# 简单值匹配
|
||
if actual_value != expected_value:
|
||
return False
|
||
return True
|
||
|
||
@staticmethod
|
||
def _check_complex_condition(
|
||
actual_value: Any, condition: dict[str, Any]
|
||
) -> bool:
|
||
"""
|
||
检查复杂条件匹配
|
||
支持的操作符：equals, not_equals, contains, not_contains, in, not_in, regex
|
||
"""
|
||
for operator, expected_value in condition.items():
|
||
if operator == "equals":
|
||
if actual_value != expected_value:
|
||
return False
|
||
elif operator == "not_equals":
|
||
if actual_value == expected_value:
|
||
return False
|
||
elif operator == "contains":
|
||
if expected_value not in str(actual_value):
|
||
return False
|
||
elif operator == "not_contains":
|
||
if expected_value in str(actual_value):
|
||
return False
|
||
elif operator == "in":
|
||
if actual_value not in expected_value:
|
||
return False
|
||
elif operator == "not_in":
|
||
if actual_value in expected_value:
|
||
return False
|
||
elif operator == "regex":
|
||
import re
|
||
if not re.search(expected_value, str(actual_value)):
|
||
return False
|
||
return True
|
||
|
||
def get_event_workflows(self) -> dict[str, list[int]]:
|
||
"""
|
||
获取所有事件触发的工作流
|
||
"""
|
||
with self._lock:
|
||
return self._event_workflows.copy()
|