mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 20:05:42 +08:00
87561a90d1
- Agent 代渠道用户调用内部 API 时，把会话渠道与渠道用户写入宿主签发令牌的签名 声明（agent_channel、agent_channel_user），不再新增可伪造的请求头。 - /commands/run 对 Telegram 等通知渠道只使用与请求渠道一致的签名渠道用户执行命令 和投递回执；缺少该声明时明确拒绝，不再把 MoviePilot 用户表主键当作渠道 chat id。 Web、WebAgent 与无渠道调用仍按 MoviePilot 用户 ID 处理。 - 只读公共资料借用宿主身份，不携带渠道发起人声明。 - 同步依赖基线与架构总览中的内部导入边数量。 Refs #7225 Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: jxxghp <jxxghp@gmail.com>
115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
"""
|
||
命令工具服务门面。
|
||
|
||
Agent 工具与 API 端点对命令注册表的操作统一经本模块调用，
|
||
Command 实现由 startup 组合根在生命周期启动阶段注册，避免 application 层
|
||
静态依赖顶层 command 模块。
|
||
|
||
依赖方向：
|
||
|
||
agent.tools / api.endpoints -> application.commands <- startup（注册 Command 类）
|
||
"""
|
||
|
||
from collections.abc import Callable
|
||
from typing import Any, Dict, Optional
|
||
|
||
from app.schemas.types import EventType, NotificationChannel
|
||
|
||
# Command 类：由 startup/initializers/command.py 在显式启动阶段注册。
|
||
_command_class: Any = None
|
||
|
||
|
||
def register_command_class(command_class: Any) -> Any:
|
||
"""
|
||
注册 Command 类，并返回先前实现供隔离测试或失败回滚。
|
||
"""
|
||
global _command_class
|
||
previous = _command_class
|
||
_command_class = command_class
|
||
return previous
|
||
|
||
|
||
def reset_command_class() -> None:
|
||
"""
|
||
清除 Command 实现，防止重复 lifespan 保留上一轮 provider。
|
||
"""
|
||
register_command_class(None)
|
||
|
||
|
||
def get_command_object() -> Any:
|
||
"""
|
||
返回命令注册表实例。
|
||
"""
|
||
if _command_class is None:
|
||
raise RuntimeError("命令服务未初始化：请先通过 register_command_class 注册 Command 类")
|
||
return _command_class()
|
||
|
||
|
||
def get_commands() -> Dict[str, Any]:
|
||
"""
|
||
返回全部已注册命令。
|
||
"""
|
||
commands: dict[str, object] = get_command_object().get_commands()
|
||
return commands
|
||
|
||
|
||
def get_command(name: str) -> Optional[Any]:
|
||
"""
|
||
按命令名查询注册表。
|
||
"""
|
||
return get_command_object().get(name)
|
||
|
||
|
||
def init_commands(plugin_id: Optional[str] = None) -> None:
|
||
"""
|
||
初始化命令（可指定单个插件）。
|
||
"""
|
||
get_command_object().init_commands(plugin_id)
|
||
|
||
|
||
def dispatch_command(
|
||
command: str,
|
||
*,
|
||
user_id: Optional[str],
|
||
channel: Optional[NotificationChannel] = None,
|
||
source: Optional[str] = None,
|
||
publish_event: Callable[[EventType, dict[str, Any]], Any],
|
||
) -> dict[str, Any]:
|
||
"""
|
||
校验斜杠命令并通过注入的事件端口异步触发执行。
|
||
|
||
:param command: 包含可选参数的命令文本
|
||
:param user_id: 渠道用户标识或 Web 会话用户标识；无真实目标时为 None
|
||
:param channel: 命令所属渠道；无渠道时为 None
|
||
:param source: 渠道来源实例；无来源时为 None
|
||
:param publish_event: 注入的命令事件发布入口
|
||
:return: 已触发命令的描述和可选插件标识
|
||
:raises ValueError: 命令为空或未注册
|
||
"""
|
||
normalized = command.strip()
|
||
if not normalized:
|
||
raise ValueError("命令不能为空")
|
||
if not normalized.startswith("/"):
|
||
normalized = f"/{normalized}"
|
||
command_name = normalized.split()[0]
|
||
matched = get_command(command_name)
|
||
# Command.get 使用空字典表示未注册命令，不能将其报告为执行成功。
|
||
if not matched:
|
||
available = [f"{name} - {info.get('description', '无描述')}" for name, info in get_commands().items()]
|
||
raise ValueError(f"命令 {command_name} 不存在" + (f"；可用命令: {'; '.join(available)}" if available else ""))
|
||
publish_event(
|
||
EventType.CommandExcute,
|
||
{
|
||
"cmd": normalized,
|
||
"user": user_id,
|
||
"channel": channel,
|
||
"source": source,
|
||
},
|
||
)
|
||
return {
|
||
"message": f"命令 {command_name} 已触发执行",
|
||
"command": normalized,
|
||
"command_desc": matched.get("description", ""),
|
||
**({"plugin_id": matched["pid"]} if matched.get("pid") else {}),
|
||
}
|