""" 提供工作流定义、运行状态和执行控制接口。 """ from typing import Any, List, Literal, Optional from fastapi import Depends, Response from app.adapters.external.server import MoviePilotServerHelper from app.api.dependencies.auth import ( get_current_active_manage_user, get_current_active_manage_user_async, ) from app.api.dependencies.workflow import ( get_workflow_definition_command, get_workflow_mutation_command, get_workflow_query_service, ) from app.api.response import ( COLLECTION_TOTAL_HEADER, COLLECTION_TOTAL_OPENAPI_KEY, CompatibleCountParam, CompatiblePageParam, ResponseAPIRouter, resolve_compatible_pagination, ) from app.application.plugin.runtime import get_plugin_manager from app.application.workflow import ( WorkflowDefinitionCommand, WorkflowMutationCommand, WorkflowQueryService, get_workflow_manager, ) from app.chain.workflow import WorkflowChain from app.schemas.common import JsonObject as _SchemaJsonObject from app.schemas.response import Response as _SchemaResponse from app.schemas.types import EVENT_TYPE_NAMES, EventType from app.schemas.workflow import NameValueOption as _SchemaNameValueOption from app.schemas.workflow import PluginWorkflowActionGroup as _SchemaPluginWorkflowActionGroup from app.schemas.workflow import Workflow as _SchemaWorkflow from app.schemas.workflow import WorkflowActionDefinition as _SchemaWorkflowActionDefinition from app.schemas.workflow import WorkflowShare as _SchemaWorkflowShare router = ResponseAPIRouter() @router.get( "/", summary="所有工作流", response_model=List[_SchemaWorkflow], openapi_extra={COLLECTION_TOTAL_OPENAPI_KEY: True}, ) async def list_workflows( response: Response = None, # type: ignore[assignment] query: WorkflowQueryService = Depends(get_workflow_query_service), _: Any = Depends(get_current_active_manage_user_async), page: CompatiblePageParam = None, count: CompatibleCountParam = None, ) -> Any: """ 获取工作流列表 """ page, count = resolve_compatible_pagination(page, count) if response is not None: response.headers[COLLECTION_TOTAL_HEADER] = str(await query.count()) return await query.list(page=page, count=count) @router.get( "/agent", summary="查询 Agent 可用工作流", response_model=List[_SchemaJsonObject], openapi_extra={COLLECTION_TOTAL_OPENAPI_KEY: True}, ) async def list_agent_workflows( response: Response = None, # type: ignore[assignment] state: Literal["W", "R", "P", "S", "F", "all"] = "all", name: Optional[str] = None, trigger_type: Literal["timer", "event", "manual", "all"] = "all", query: WorkflowQueryService = Depends(get_workflow_query_service), _: Any = Depends(get_current_active_manage_user_async), page: CompatiblePageParam = None, count: CompatibleCountParam = None, ) -> List[dict[str, Any]]: """ 按旧 Agent 过滤与字段投影返回工作流列表,避免输出完整动作上下文。 """ state_filter = None if state == "all" else state trigger_filter = None if trigger_type == "all" else trigger_type page, count = resolve_compatible_pagination(page, count) if response is not None: response.headers[COLLECTION_TOTAL_HEADER] = str( await query.count( state=state_filter, name=name, trigger_type=trigger_filter, ) ) workflows = await query.list( state=state_filter, name=name, trigger_type=trigger_filter, page=page, count=count, ) results = [] for workflow in workflows: normalized_trigger = workflow.trigger_type or "timer" results.append( { "id": workflow.id, "name": workflow.name, "description": workflow.description, "trigger_type": normalized_trigger, "state": workflow.state, "run_count": workflow.run_count, "timer": workflow.timer, "event_type": workflow.event_type, "add_time": workflow.add_time, "last_time": workflow.last_time, "current_action": workflow.current_action, } ) return results @router.post("/", summary="创建工作流", response_model=_SchemaResponse[None]) async def create_workflow( workflow: _SchemaWorkflow, command: WorkflowDefinitionCommand = Depends(get_workflow_definition_command), _: Any = Depends(get_current_active_manage_user_async), ) -> Any: """ 创建工作流 """ result = await command.create(workflow.model_dump(exclude={"id"})) return _SchemaResponse(success=result.success, message=result.message) @router.get( "/plugin/actions", summary="查询插件动作", response_model=List[_SchemaPluginWorkflowActionGroup], ) def list_plugin_actions( plugin_id: Optional[str] = None, _: Any = Depends(get_current_active_manage_user), page: CompatiblePageParam = None, count: CompatibleCountParam = None, ) -> Any: """ 获取所有动作 """ return get_plugin_manager().get_plugin_actions(plugin_id) @router.get( "/actions", summary="所有动作", response_model=List[_SchemaWorkflowActionDefinition], ) async def list_actions(_: Any = Depends(get_current_active_manage_user_async), page: CompatiblePageParam = None, count: CompatibleCountParam = None) -> Any: """ 获取所有动作 """ return get_workflow_manager().list_actions() @router.get( "/event_types", summary="获取所有事件类型", response_model=List[_SchemaNameValueOption], ) async def get_event_types(_: Any = Depends(get_current_active_manage_user_async), page: CompatiblePageParam = None, count: CompatibleCountParam = None) -> Any: """ 获取所有事件类型 """ return [ { "title": EVENT_TYPE_NAMES.get(event_type, event_type.name), "value": event_type.value, } for event_type in EventType ] @router.post("/share", summary="分享工作流", response_model=_SchemaResponse[None]) async def workflow_share( workflow: _SchemaWorkflowShare, _: Any = Depends(get_current_active_manage_user_async) ) -> Any: """ 分享工作流 """ if not workflow.id or not workflow.share_title or not workflow.share_user: return _SchemaResponse( success=False, message="请填写工作流ID、分享标题和分享人" ) state, errmsg = await MoviePilotServerHelper.async_workflow_share_by_id( workflow_id=workflow.id, share_title=workflow.share_title or "", share_comment=workflow.share_comment or "", share_user=workflow.share_user or "", ) return _SchemaResponse(success=state, message=errmsg) @router.delete("/share/{share_id}", summary="删除分享", response_model=_SchemaResponse[None]) async def workflow_share_delete( share_id: int, _: Any = Depends(get_current_active_manage_user_async) ) -> Any: """ 删除分享 """ state, errmsg = await MoviePilotServerHelper.async_workflow_share_delete_by_id(share_id=share_id) return _SchemaResponse(success=state, message=errmsg) @router.post("/fork", summary="复用工作流", response_model=_SchemaResponse[None]) async def workflow_fork( workflow: _SchemaWorkflowShare, command: WorkflowDefinitionCommand = Depends(get_workflow_definition_command), _: Any = Depends(get_current_active_manage_user_async), ) -> Any: """ 复用工作流 """ result = await command.fork(workflow.model_dump(), share_id=workflow.id) return _SchemaResponse(success=result.success, message=result.message) @router.get( "/shares", summary="查询分享的工作流", response_model=List[_SchemaWorkflowShare] ) async def workflow_shares( name: Optional[str] = None, page: Optional[int] = 1, count: Optional[int] = 30, _: Any = Depends(get_current_active_manage_user_async), ) -> Any: """ 查询分享的工作流 """ return await MoviePilotServerHelper.async_get_workflow_shares(name=name, page=page, count=count) @router.post( "/{workflow_id}/run", summary="执行工作流", response_model=_SchemaResponse[None] ) def run_workflow( workflow_id: int, from_begin: Optional[bool] = True, _: Any = Depends(get_current_active_manage_user), ) -> Any: """ 执行工作流 """ state, errmsg = WorkflowChain().process(workflow_id, from_begin=from_begin) if not state: return _SchemaResponse(success=False, message=errmsg) return _SchemaResponse(success=True) @router.post( "/{workflow_id}/start", summary="启用工作流", response_model=_SchemaResponse[None] ) def start_workflow( workflow_id: int, command: WorkflowMutationCommand = Depends(get_workflow_mutation_command), _: Any = Depends(get_current_active_manage_user), ) -> Any: """ 启用工作流 """ result = command.start(workflow_id) return _SchemaResponse(success=result.success, message=result.message) @router.post( "/{workflow_id}/pause", summary="停用工作流", response_model=_SchemaResponse[None] ) def pause_workflow( workflow_id: int, command: WorkflowMutationCommand = Depends(get_workflow_mutation_command), _: Any = Depends(get_current_active_manage_user), ) -> Any: """ 停用工作流 """ result = command.pause(workflow_id) return _SchemaResponse(success=result.success, message=result.message) @router.post( "/{workflow_id}/reset", summary="重置工作流", response_model=_SchemaResponse[None] ) async def reset_workflow( workflow_id: int, command: WorkflowDefinitionCommand = Depends(get_workflow_definition_command), _: Any = Depends(get_current_active_manage_user_async), ) -> Any: """ 重置工作流 """ result = await command.reset(workflow_id) return _SchemaResponse(success=result.success, message=result.message) @router.get("/{workflow_id}", summary="工作流详情", response_model=_SchemaWorkflow) async def get_workflow( workflow_id: int, query: WorkflowQueryService = Depends(get_workflow_query_service), _: Any = Depends(get_current_active_manage_user_async), ) -> Any: """ 获取工作流详情 """ return await query.get(workflow_id) @router.put("/{workflow_id}", summary="更新工作流", response_model=_SchemaResponse[None]) def update_workflow( workflow_id: int, workflow: _SchemaWorkflow, command: WorkflowMutationCommand = Depends(get_workflow_mutation_command), _: Any = Depends(get_current_active_manage_user), ) -> Any: """ 更新工作流,路径 ID 是本次更新的唯一目标。 """ workflow_data = workflow.model_dump() workflow_data["id"] = workflow_id result = command.update(workflow_data) return _SchemaResponse(success=result.success, message=result.message) @router.delete("/{workflow_id}", summary="删除工作流", response_model=_SchemaResponse[None]) def delete_workflow( workflow_id: int, command: WorkflowMutationCommand = Depends(get_workflow_mutation_command), _: Any = Depends(get_current_active_manage_user), ) -> Any: """ 删除工作流 """ result = command.delete(workflow_id) return _SchemaResponse(success=result.success, message=result.message)