mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 20:34:48 +08:00
556 lines
22 KiB
Python
556 lines
22 KiB
Python
"""
|
||
Agent 自主定时任务统一工具。
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from typing import Literal, Optional, Type, cast
|
||
|
||
import pytz
|
||
from pydantic import BaseModel, Field, model_validator
|
||
|
||
from app.agent.identity import AgentIdentityError, resolve_task_user
|
||
from app.agent.tools.base import MoviePilotTool
|
||
from app.agent.tools.tags import ToolTag
|
||
from app.application.agenttask import agent_task_run_to_dict, agent_task_to_dict
|
||
from app.application.security.user import UserSnapshot
|
||
from app.runtime.scheduling import TimerUtils
|
||
from app.runtime.settings import get_runtime_setting
|
||
|
||
AgentTaskAction = Literal["create", "list", "update", "run", "delete"]
|
||
AgentTaskTriggerType = Literal["date", "cron"]
|
||
|
||
|
||
class AgentTaskInput(BaseModel):
|
||
"""
|
||
统一管理 Agent 自主定时任务的输入参数。
|
||
"""
|
||
|
||
action: AgentTaskAction = Field(
|
||
...,
|
||
description="任务动作：create、list、update、run 或 delete；不知道整数 task_id 时应先 list 查询",
|
||
)
|
||
task_id: Optional[int] = Field(
|
||
None,
|
||
ge=1,
|
||
description="list、update、run 或 delete 使用的任务整数标识",
|
||
)
|
||
name: Optional[str] = Field(
|
||
None,
|
||
min_length=1,
|
||
max_length=100,
|
||
description="create 或 update 使用的任务名称",
|
||
)
|
||
content: Optional[str] = Field(
|
||
None,
|
||
min_length=1,
|
||
max_length=10000,
|
||
description="create 或 update 使用的完整任务指令",
|
||
)
|
||
trigger_type: Optional[AgentTaskTriggerType] = Field(
|
||
None,
|
||
description="date 表示未来执行一次，cron 表示周期执行",
|
||
)
|
||
trigger: Optional[str] = Field(
|
||
None,
|
||
min_length=1,
|
||
max_length=200,
|
||
description="ISO 8601 日期时间或标准五字段 cron 表达式",
|
||
)
|
||
delay_minutes: Optional[int] = Field(
|
||
None,
|
||
ge=1,
|
||
le=525600,
|
||
description="一次性 date 任务的相对延迟分钟数，与 trigger 二选一",
|
||
)
|
||
enabled: Optional[bool] = Field(
|
||
None,
|
||
description="list 时按启用状态筛选；update 时 False 暂停、True 恢复",
|
||
)
|
||
|
||
# Pydantic 校验器由运行时注册，保留第三方装饰器所需的动态类型兼容。
|
||
@model_validator(mode="after")
|
||
def validate_action(self) -> "AgentTaskInput":
|
||
"""
|
||
按动作校验必填字段和触发参数组合。
|
||
"""
|
||
if self.name is not None:
|
||
self.name = self.name.strip()
|
||
if not self.name:
|
||
raise ValueError("name 不能只包含空白字符")
|
||
if self.content is not None:
|
||
self.content = self.content.strip()
|
||
if not self.content:
|
||
raise ValueError("content 不能只包含空白字符")
|
||
|
||
if self.action == "create":
|
||
if self.name is None or self.content is None or self.trigger_type is None:
|
||
raise ValueError("create 必须提供 name、content 和 trigger_type")
|
||
self._validate_schedule(require_schedule=True, normalize_exact=True)
|
||
elif self.action == "update":
|
||
if self.task_id is None:
|
||
raise ValueError("update 必须提供 task_id")
|
||
has_schedule_update = any(
|
||
value is not None for value in (self.trigger_type, self.trigger, self.delay_minutes)
|
||
)
|
||
if has_schedule_update:
|
||
self._validate_schedule(require_schedule=True, normalize_exact=False)
|
||
if all(
|
||
value is None
|
||
for value in (
|
||
self.name,
|
||
self.content,
|
||
self.trigger_type,
|
||
self.enabled,
|
||
)
|
||
):
|
||
raise ValueError("update 至少需要提供一个要更新的字段")
|
||
elif self.action in {"run", "delete"} and self.task_id is None:
|
||
raise ValueError(f"{self.action} 必须提供 task_id")
|
||
return self
|
||
|
||
def _validate_schedule(
|
||
self,
|
||
*,
|
||
require_schedule: bool,
|
||
normalize_exact: bool,
|
||
) -> None:
|
||
"""
|
||
校验 date/cron 触发组合，并按需规范化绝对触发值。
|
||
"""
|
||
if require_schedule and self.trigger_type is None:
|
||
raise ValueError("修改触发配置时必须提供 trigger_type")
|
||
if self.trigger_type == "date":
|
||
if self.delay_minutes is not None:
|
||
self.trigger = None
|
||
return
|
||
if self.trigger is None:
|
||
raise ValueError("date 任务必须提供 trigger 或 delay_minutes")
|
||
elif self.trigger_type == "cron":
|
||
if self.trigger is None or self.delay_minutes is not None:
|
||
raise ValueError("cron 任务必须提供 trigger，且不能提供 delay_minutes")
|
||
if normalize_exact and self.trigger_type is not None:
|
||
normalized_type, normalized_trigger = TimerUtils.normalize_schedule_trigger(
|
||
trigger_type=self.trigger_type,
|
||
# 前置分支已拒绝缺失触发值，延迟任务也已提前返回。
|
||
trigger_value=cast(str, self.trigger),
|
||
timezone_name=get_runtime_setting("TZ"),
|
||
require_future=True,
|
||
)
|
||
self.trigger_type = cast(AgentTaskTriggerType, normalized_type)
|
||
self.trigger = normalized_trigger
|
||
|
||
|
||
class AgentTaskTool(MoviePilotTool):
|
||
"""
|
||
通过一个动作式接口管理当前用户的自主定时任务。
|
||
"""
|
||
|
||
name: str = Field(
|
||
default="agent_task",
|
||
description="供模型调用的稳定工具名称",
|
||
)
|
||
tags: list[str] = Field(
|
||
default=[ToolTag.Read, ToolTag.Write, ToolTag.AgentTask, ToolTag.Admin],
|
||
description="工具的能力和权限分类标签，初始化时会统一归一化",
|
||
)
|
||
description: str = Field(
|
||
default=(
|
||
"通过结构化动作管理持久化自主任务。"
|
||
"用户明确要求延迟或周期工作时用 create；"
|
||
"list 查看当前用户任务；"
|
||
"update 修改或暂停；"
|
||
"run 立即入队；"
|
||
"delete 永久删除。"
|
||
"任务标识是整数，与调度器作业标识不同。"
|
||
),
|
||
description="供模型理解工具用途、输入约束和调用边界的说明",
|
||
)
|
||
args_schema: Type[BaseModel] = Field(
|
||
default=AgentTaskInput,
|
||
description="工具输入参数的验证模型或 JSON Schema",
|
||
)
|
||
require_admin: bool = Field(
|
||
default=True,
|
||
description="是否仅允许系统管理员调用",
|
||
)
|
||
|
||
def get_tool_message(self, **kwargs: object) -> Optional[str]:
|
||
"""
|
||
生成统一任务操作的用户可见提示。
|
||
"""
|
||
action = str(kwargs.get("action") or "list")
|
||
action_name = {
|
||
"create": "创建",
|
||
"list": "查询",
|
||
"update": "更新",
|
||
"run": "立即执行",
|
||
"delete": "删除",
|
||
}.get(action, action)
|
||
target = kwargs.get("task_id") or kwargs.get("name") or ""
|
||
return f"{action_name}自主定时任务：{target}".rstrip("：")
|
||
|
||
@staticmethod
|
||
def _absolute_trigger(payload: AgentTaskInput) -> str:
|
||
"""
|
||
将相对分钟或绝对表达式转换为规范化触发值。
|
||
"""
|
||
trigger_value = payload.trigger
|
||
if payload.trigger_type == "date" and payload.delay_minutes is not None:
|
||
timezone = pytz.timezone(get_runtime_setting("TZ"))
|
||
trigger_value = (datetime.now(timezone) + timedelta(minutes=payload.delay_minutes)).isoformat(
|
||
timespec="seconds"
|
||
)
|
||
if payload.trigger_type is None:
|
||
raise ValueError("Agent 定时任务缺少 trigger_type")
|
||
if trigger_value is None:
|
||
raise ValueError("date 任务必须提供 trigger 或 delay_minutes")
|
||
_, normalized = TimerUtils.normalize_schedule_trigger(
|
||
trigger_type=payload.trigger_type,
|
||
trigger_value=trigger_value,
|
||
timezone_name=get_runtime_setting("TZ"),
|
||
require_future=True,
|
||
)
|
||
return normalized
|
||
|
||
def _create_task(
|
||
self, payload: AgentTaskInput, user: UserSnapshot,
|
||
) -> dict[str, object]:
|
||
"""
|
||
持久化已验证的执行主体，保留渠道会话归属并注册调度。
|
||
|
||
:param payload: 已验证的任务定义及调度参数
|
||
:param user: 由可信入口解析的启用账号，不接受模型指定的身份
|
||
:return: 注册调度后的任务展示数据
|
||
"""
|
||
from app.application.scheduling import update_agent_task_job
|
||
|
||
trigger_value = self._absolute_trigger(payload)
|
||
if payload.name is None or payload.content is None or payload.trigger_type is None:
|
||
raise ValueError("create 缺少必要参数")
|
||
chat = self.data.chat.get_sync(
|
||
session_id=self._session_id,
|
||
user_id=self._user_id,
|
||
)
|
||
task = self.data.tasks.add(
|
||
name=payload.name,
|
||
content=payload.content,
|
||
trigger_type=payload.trigger_type,
|
||
cron_expression=(trigger_value if payload.trigger_type == "cron" else None),
|
||
run_at=trigger_value if payload.trigger_type == "date" else None,
|
||
user_id=str(self._user_id),
|
||
username=user.name,
|
||
execution_user_id=user.id,
|
||
execution_user_identity=user.notification_identity,
|
||
session_id=str(self._session_id),
|
||
channel=self._channel or (chat.channel if chat else None),
|
||
source=self._source or (chat.source if chat else None),
|
||
original_chat_id=chat.original_chat_id if chat else None,
|
||
)
|
||
if task is None:
|
||
raise RuntimeError("Agent 定时任务创建后无法读取")
|
||
return agent_task_to_dict(
|
||
task,
|
||
next_run_at=update_agent_task_job(task.id),
|
||
timezone=get_runtime_setting("TZ"),
|
||
)
|
||
|
||
def _list_tasks(
|
||
self,
|
||
task_id: Optional[int],
|
||
enabled: Optional[bool],
|
||
) -> list[dict[str, object]]:
|
||
"""
|
||
读取当前用户任务及运行时下一次触发时间。
|
||
"""
|
||
from app.application.scheduling import get_agent_task_next_run
|
||
|
||
oper = self.data.tasks
|
||
if task_id is not None:
|
||
task = oper.get(task_id=task_id, user_id=str(self._user_id))
|
||
tasks = [task] if task else []
|
||
else:
|
||
tasks = oper.list(user_id=str(self._user_id), enabled=enabled)
|
||
result: list[dict[str, object]] = []
|
||
for task in tasks:
|
||
data = agent_task_to_dict(
|
||
task,
|
||
next_run_at=get_agent_task_next_run(task.id),
|
||
timezone=get_runtime_setting("TZ"),
|
||
)
|
||
if task_id is not None:
|
||
data["recent_runs"] = [
|
||
agent_task_run_to_dict(run)
|
||
for run in oper.list_runs(
|
||
task_id=task.id,
|
||
user_id=str(self._user_id),
|
||
limit=10,
|
||
)
|
||
]
|
||
result.append(data)
|
||
return result
|
||
|
||
def _update_task(
|
||
self,
|
||
payload: AgentTaskInput,
|
||
identity_payload: dict[str, object] | None = None,
|
||
) -> Optional[dict[str, object]]:
|
||
"""
|
||
更新当前用户任务并刷新运行时调度。
|
||
|
||
:param payload: 已验证的更新参数
|
||
:param identity_payload: 仅供旧任务一次性确认的身份字段；仓储拒绝覆盖已确认账号
|
||
:return: 更新后的展示数据、运行中提示，或未找到任务时返回 None
|
||
"""
|
||
from app.application.scheduling import update_agent_task_job
|
||
|
||
if payload.task_id is None:
|
||
raise ValueError("update 缺少 task_id")
|
||
oper = self.data.tasks
|
||
task = oper.get(task_id=payload.task_id, user_id=str(self._user_id))
|
||
if not task:
|
||
return None
|
||
if task.last_status == "running":
|
||
return {"error": f"Agent 定时任务 {payload.task_id} 正在执行，请稍后再修改"}
|
||
|
||
has_schedule_update = payload.trigger_type is not None
|
||
trigger_type = payload.trigger_type or task.trigger_type
|
||
trigger_value = payload.trigger
|
||
if trigger_type == "date" and payload.delay_minutes is not None:
|
||
timezone = pytz.timezone(get_runtime_setting("TZ"))
|
||
trigger_value = (datetime.now(timezone) + timedelta(minutes=payload.delay_minutes)).isoformat(
|
||
timespec="seconds"
|
||
)
|
||
if trigger_value is None:
|
||
trigger_value = task.cron_expression if trigger_type == "cron" else task.run_at
|
||
normalized_type = trigger_type
|
||
normalized_trigger = trigger_value
|
||
validates_existing_date_schedule = bool(
|
||
payload.enabled and task.last_status != "interrupted" and trigger_type == "date"
|
||
)
|
||
if has_schedule_update or validates_existing_date_schedule:
|
||
if trigger_value is None:
|
||
raise ValueError("Agent 定时任务缺少触发配置")
|
||
normalized_type, normalized_trigger = TimerUtils.normalize_schedule_trigger(
|
||
trigger_type=trigger_type,
|
||
trigger_value=trigger_value,
|
||
timezone_name=get_runtime_setting("TZ"),
|
||
require_future=bool(
|
||
trigger_type == "date"
|
||
and (
|
||
task.last_status == "interrupted"
|
||
or (task.enabled if payload.enabled is None else payload.enabled)
|
||
)
|
||
),
|
||
)
|
||
|
||
update_payload: dict[str, object] = dict(identity_payload or {})
|
||
if payload.name is not None:
|
||
update_payload["name"] = payload.name
|
||
if payload.content is not None:
|
||
update_payload["content"] = payload.content
|
||
if has_schedule_update:
|
||
update_payload.update(
|
||
{
|
||
"trigger_type": normalized_type,
|
||
"cron_expression": (normalized_trigger if normalized_type == "cron" else None),
|
||
"run_at": (normalized_trigger if normalized_type == "date" else None),
|
||
"last_status": "waiting",
|
||
"last_result": None,
|
||
}
|
||
)
|
||
if payload.enabled is not None:
|
||
update_payload["enabled"] = payload.enabled
|
||
if payload.enabled and task.last_status != "interrupted":
|
||
update_payload["last_status"] = "waiting"
|
||
|
||
updated = oper.update(
|
||
task_id=payload.task_id,
|
||
payload=update_payload,
|
||
user_id=str(self._user_id),
|
||
)
|
||
if not updated:
|
||
current = oper.get(task_id=payload.task_id, user_id=str(self._user_id))
|
||
if current and current.last_status == "running":
|
||
return {"error": (f"Agent 定时任务 {payload.task_id} 正在执行，请稍后再修改")}
|
||
return None
|
||
updated_task = oper.get(
|
||
task_id=payload.task_id,
|
||
user_id=str(self._user_id),
|
||
)
|
||
if updated_task is None:
|
||
return None
|
||
return agent_task_to_dict(
|
||
updated_task,
|
||
next_run_at=update_agent_task_job(payload.task_id),
|
||
timezone=get_runtime_setting("TZ"),
|
||
)
|
||
|
||
def _get_task_state(self, task_id: int) -> tuple[str, Optional[str]]:
|
||
"""
|
||
校验任务归属和状态，返回可执行性及任务名称。
|
||
"""
|
||
task = self.data.tasks.get(task_id=task_id, user_id=str(self._user_id))
|
||
if not task:
|
||
return "not_found", None
|
||
if not task.enabled:
|
||
return "disabled", task.name
|
||
if task.last_status == "running":
|
||
return "running", task.name
|
||
return "ready", task.name
|
||
|
||
def _delete_task(self, task_id: int) -> bool:
|
||
"""
|
||
删除当前用户任务并移除运行时调度。
|
||
"""
|
||
from app.application.scheduling import remove_agent_task_job
|
||
|
||
deleted = self.data.tasks.delete(
|
||
task_id=task_id,
|
||
user_id=str(self._user_id),
|
||
)
|
||
if deleted:
|
||
remove_agent_task_job(task_id)
|
||
return deleted
|
||
|
||
async def _execution_identity(self) -> UserSnapshot:
|
||
"""
|
||
从可信入口解析创建主体；渠道用户名和旧会话昵称不参与认证。
|
||
|
||
:return: 真实用户的身份与权限快照
|
||
:raises AgentIdentityError: 无法解析唯一启用的用户
|
||
"""
|
||
context_id = self._agent_context.get("execution_user_id")
|
||
chat = await self.run_blocking(
|
||
"db", self.data.chat.get_sync, session_id=self._session_id, user_id=self._user_id,
|
||
)
|
||
return await resolve_task_user(
|
||
self.data.users,
|
||
user_id=str(self._user_id or ""),
|
||
username=self._username or (chat.username if chat else None),
|
||
channel=self._channel or (chat.channel if chat else None),
|
||
source=self._source or (chat.source if chat else None),
|
||
is_admin=await self.is_admin_user(),
|
||
execution_user_id=context_id if isinstance(context_id, int) else None,
|
||
execution_user_identity=str(self._agent_context.get("execution_user_identity") or ""),
|
||
)
|
||
|
||
async def _update(self, payload: AgentTaskInput) -> str:
|
||
"""
|
||
更新任务；旧任务只允许从创建时的原会话确认执行主体，暂停不要求补绑。
|
||
|
||
:param payload: 已验证的任务更新参数
|
||
:return: 更新后的任务或不能更新的明确原因
|
||
"""
|
||
current = await self.run_blocking(
|
||
"db", self.data.tasks.get, task_id=payload.task_id, user_id=str(self._user_id),
|
||
)
|
||
identity_payload: dict[str, object] = {}
|
||
if current is not None and current.execution_user_id is None and payload.enabled is not False:
|
||
if (current.session_id, current.channel, current.source) != (
|
||
self._session_id, self._channel, self._source,
|
||
):
|
||
return "请从创建任务的原会话更新执行账号，或在原会话删除后重新创建。"
|
||
try:
|
||
user = await self._execution_identity()
|
||
except AgentIdentityError as error:
|
||
return str(error)
|
||
identity_payload = {
|
||
"execution_user_id": user.id,
|
||
"execution_user_identity": user.notification_identity,
|
||
"username": user.name,
|
||
}
|
||
task = await self.run_blocking("db", self._update_task, payload, identity_payload)
|
||
if not task:
|
||
return f"Agent 定时任务 {payload.task_id} 不存在或不属于当前用户"
|
||
if update_error := task.get("error"):
|
||
return str(update_error)
|
||
return json.dumps(task, ensure_ascii=False, indent=2)
|
||
|
||
# 具体参数由 args_schema 验证，执行入口保留与工具基类的动态分发兼容。
|
||
async def run( # type: ignore[override]
|
||
self,
|
||
action: AgentTaskAction,
|
||
task_id: Optional[int] = None,
|
||
name: Optional[str] = None,
|
||
content: Optional[str] = None,
|
||
trigger_type: Optional[AgentTaskTriggerType] = None,
|
||
trigger: Optional[str] = None,
|
||
delay_minutes: Optional[int] = None,
|
||
enabled: Optional[bool] = None,
|
||
**kwargs: object,
|
||
) -> str:
|
||
"""
|
||
执行一项任务创建、查询、更新、立即运行或删除操作。
|
||
"""
|
||
payload = AgentTaskInput(
|
||
action=action,
|
||
task_id=task_id,
|
||
name=name,
|
||
content=content,
|
||
trigger_type=trigger_type,
|
||
trigger=trigger,
|
||
delay_minutes=delay_minutes,
|
||
enabled=enabled,
|
||
)
|
||
if payload.action == "create":
|
||
if not get_runtime_setting("AI_AGENT_ENABLE"):
|
||
return "AI Agent 未启用，无法创建自主定时任务"
|
||
try:
|
||
user = await self._execution_identity()
|
||
except AgentIdentityError as error:
|
||
return str(error)
|
||
task = await self.run_blocking("db", self._create_task, payload, user)
|
||
return json.dumps(task, ensure_ascii=False, indent=2)
|
||
if payload.action == "list":
|
||
tasks = await self.run_blocking(
|
||
"db",
|
||
self._list_tasks,
|
||
payload.task_id,
|
||
payload.enabled,
|
||
)
|
||
return json.dumps(
|
||
{"total": len(tasks), "tasks": tasks},
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
if payload.action == "update":
|
||
return await self._update(payload)
|
||
if payload.action == "run":
|
||
return await self._run_now(payload.task_id)
|
||
return await self._delete(payload.task_id)
|
||
|
||
async def _run_now(self, task_id: Optional[int]) -> str:
|
||
"""
|
||
立即提交一项已启用且属于当前用户的任务。
|
||
"""
|
||
from app.application.scheduling import start_agent_task
|
||
|
||
if task_id is None:
|
||
raise ValueError("run 缺少 task_id")
|
||
status, task_name = await self.run_blocking(
|
||
"db",
|
||
self._get_task_state,
|
||
task_id,
|
||
)
|
||
if status == "not_found":
|
||
return f"Agent 定时任务 {task_id} 不存在或不属于当前用户"
|
||
if status == "disabled":
|
||
return f"Agent 定时任务 {task_id} 已暂停，请先恢复后再执行"
|
||
if status == "running":
|
||
return f"Agent 定时任务 {task_id} 正在执行，请勿重复触发"
|
||
if not start_agent_task(task_id):
|
||
return f"Agent 定时任务 {task_id} 尚未注册到运行时调度器，无法立即执行"
|
||
return f"Agent 定时任务 {task_id} 已提交立即执行：{task_name}。执行完成后将通过已配置的通知渠道广播结果"
|
||
|
||
async def _delete(self, task_id: Optional[int]) -> str:
|
||
"""
|
||
永久删除一项属于当前用户的自主定时任务。
|
||
"""
|
||
if task_id is None:
|
||
raise ValueError("delete 缺少 task_id")
|
||
deleted = await self.run_blocking("db", self._delete_task, task_id)
|
||
if not deleted:
|
||
return f"Agent 定时任务 {task_id} 不存在或不属于当前用户"
|
||
return f"Agent 定时任务 {task_id} 已删除"
|