mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-05 23:47:41 +08:00
Refactor plugin cache handling and update related docs
This commit is contained in:
jxxghp
@@ -1,7 +1,6 @@
|
||||
"""提示词管理器"""
|
||||
|
||||
import shutil
|
||||
import socket
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from string import Formatter
|
||||
@@ -24,8 +23,6 @@ from app.utils.system import SystemUtils
|
||||
SYSTEM_TASKS_FILE = "System Tasks.yaml"
|
||||
SYSTEM_TASKS_SCHEMA_VERSION = 2
|
||||
COMMON_SHELL_COMMANDS = (
|
||||
# 只探测会明显改变 Agent 执行策略的可选能力。基础命令、语言运行时、
|
||||
# 包管理器、服务管理器和数据库客户端默认不做启动探测，减少 which 扫描量。
|
||||
"ssh",
|
||||
"scp",
|
||||
"sftp",
|
||||
@@ -91,7 +88,7 @@ class PromptManager:
|
||||
self.prompts_cache: Dict[str, str] = {}
|
||||
self._system_tasks_cache: Optional[SystemTasksDefinition] = None
|
||||
self._system_tasks_signature: Optional[tuple[int, int]] = None
|
||||
self._available_shell_commands_cache: Optional[list[tuple[str, str]]] = None
|
||||
self._available_shell_command_names_cache: Optional[list[str]] = None
|
||||
|
||||
def load_prompt(self, prompt_name: str) -> str:
|
||||
"""
|
||||
@@ -281,90 +278,62 @@ class PromptManager:
|
||||
|
||||
def _get_moviepilot_info(self) -> str:
|
||||
"""
|
||||
获取MoviePilot系统信息，用于注入到系统提示词中
|
||||
获取需要常驻注入的最小 MoviePilot 运行信息。
|
||||
"""
|
||||
# 获取主机名和IP地址
|
||||
try:
|
||||
hostname = socket.gethostname()
|
||||
ip_address = socket.gethostbyname(hostname)
|
||||
except Exception: # noqa
|
||||
hostname = "localhost"
|
||||
ip_address = "127.0.0.1"
|
||||
|
||||
# 配置文件和日志文件目录
|
||||
config_path = str(settings.CONFIG_PATH)
|
||||
log_path = str(settings.LOG_PATH)
|
||||
|
||||
# API地址构建
|
||||
api_port = settings.PORT
|
||||
api_path = settings.API_V1_STR
|
||||
|
||||
# 数据库信息只保留非敏感连接摘要，凭据由内部工具自行读取。
|
||||
db_type = settings.DB_TYPE
|
||||
if db_type == "sqlite":
|
||||
db_info = "SQLite（本地配置目录数据库，路径由内部工具读取）"
|
||||
else:
|
||||
db_info = (
|
||||
f"PostgreSQL（{settings.DB_POSTGRESQL_TARGET}/"
|
||||
f"{settings.DB_POSTGRESQL_DATABASE}，凭据由内部工具读取）"
|
||||
)
|
||||
|
||||
# 保留日期用于提供“今天是哪天”的稳定上下文，但不再注入秒级时间，
|
||||
# 避免每次请求都生成不同的 system prompt，影响 provider 侧 cache 命中率。
|
||||
info_lines = [
|
||||
f"- 当前日期: {strftime('%Y-%m-%d')}",
|
||||
f"- 运行环境: {SystemUtils.platform} {'docker' if SystemUtils.is_docker() else ''}",
|
||||
f"- 主机名: {hostname}",
|
||||
f"- IP地址: {ip_address}",
|
||||
f"- API端口: {api_port}",
|
||||
f"- API路径: {api_path}",
|
||||
"- API认证: 由内部工具自动处理，不在提示词中暴露令牌",
|
||||
f"- 外网域名: {settings.APP_DOMAIN or '未设置'}",
|
||||
f"- 数据库类型: {db_type}",
|
||||
f"- 数据库: {db_info}",
|
||||
f"- 配置文件目录: {config_path}",
|
||||
f"- 日志文件目录: {log_path}",
|
||||
f"- 系统安装目录: {settings.ROOT_PATH}",
|
||||
f"- 插件安装目录: {settings.ROOT_PATH / 'app' / 'plugins'}",
|
||||
"- 详细运行状态、数据库、API 和配置值需要时通过 `query_doctor_report`、`query_system_settings` 或 `execute_command` 查询。",
|
||||
]
|
||||
|
||||
available_commands = self._get_available_shell_commands()
|
||||
if available_commands:
|
||||
info_lines.append("- 可用系统命令（可通过 `execute_command` 调用）:")
|
||||
path_lines = self._get_runtime_path_lines()
|
||||
if path_lines:
|
||||
info_lines.extend(
|
||||
f" - {command}: {path}" for command, path in available_commands
|
||||
[
|
||||
"- 关键运行路径（必要时可用文件/命令工具读取，避免扫描无关目录）:",
|
||||
*path_lines,
|
||||
]
|
||||
)
|
||||
# `rg` 同时覆盖文件枚举和文本检索，且比通用 shell 查找更适合
|
||||
# Agent 的代码阅读与定位场景；只有在它不可用或不适合时才退回其他工具。
|
||||
if any(command == "rg" for command, _ in available_commands):
|
||||
available_commands = self._get_available_shell_command_names()
|
||||
if available_commands:
|
||||
info_lines.append(
|
||||
"- 已安装的常用系统命令（仅列命令名，可通过 `execute_command` 调用）: "
|
||||
+ ", ".join(f"`{command}`" for command in available_commands)
|
||||
)
|
||||
if "rg" in available_commands:
|
||||
info_lines.append(
|
||||
"- When searching files or text, prefer `rg` / `rg --files`. Only fall back to other search tools when `rg` is unavailable or unsuitable."
|
||||
"- 搜索文件或文本时优先使用 `rg` / `rg --files`，不适合或不可用时再使用其他命令。"
|
||||
)
|
||||
|
||||
return "\n".join(info_lines)
|
||||
|
||||
def _get_available_shell_commands(self) -> list[tuple[str, str]]:
|
||||
"""
|
||||
探测 PATH 中已经安装的常用命令。
|
||||
@staticmethod
|
||||
def _get_runtime_path_lines() -> list[str]:
|
||||
"""返回基础系统提示词需要常驻注入的全局运行路径。"""
|
||||
paths = {
|
||||
"项目根目录": settings.ROOT_PATH,
|
||||
"配置目录": settings.CONFIG_PATH,
|
||||
"临时目录": settings.TEMP_PATH,
|
||||
"日志目录": settings.LOG_PATH,
|
||||
"主日志文件": settings.LOG_PATH / "moviepilot.log",
|
||||
}
|
||||
return [f" - {label}: `{path}`" for label, path in paths.items()]
|
||||
|
||||
这里只使用 shutil.which 做无副作用查找，不实际执行命令；执行权限、
|
||||
高风险操作确认和输出限制仍由 execute_command 工具负责。探测结果
|
||||
在进程内缓存，避免每次组装提示词都重复扫描 PATH。
|
||||
"""
|
||||
if self._available_shell_commands_cache is not None:
|
||||
return self._available_shell_commands_cache
|
||||
def _get_available_shell_command_names(self) -> list[str]:
|
||||
"""探测 PATH 中可用的常用命令名称，不把绝对路径注入提示词。"""
|
||||
if self._available_shell_command_names_cache is not None:
|
||||
return self._available_shell_command_names_cache
|
||||
|
||||
available_commands: list[tuple[str, str]] = []
|
||||
for command in COMMON_SHELL_COMMANDS:
|
||||
command_path = shutil.which(command)
|
||||
if command_path:
|
||||
available_commands.append((command, command_path))
|
||||
self._available_shell_commands_cache = available_commands
|
||||
available_commands = [
|
||||
command for command in COMMON_SHELL_COMMANDS if shutil.which(command)
|
||||
]
|
||||
self._available_shell_command_names_cache = available_commands
|
||||
return available_commands
|
||||
|
||||
def clear_available_shell_commands_cache(self) -> None:
|
||||
"""清理可用系统命令缓存，供测试或运行时手动刷新使用。"""
|
||||
self._available_shell_commands_cache = None
|
||||
def clear_available_shell_command_names_cache(self) -> None:
|
||||
"""清理可用命令名称缓存，供测试或运行时手动刷新使用。"""
|
||||
self._available_shell_command_names_cache = None
|
||||
|
||||
@staticmethod
|
||||
def _generate_formatting_instructions(caps: ChannelCapabilities) -> str:
|
||||
|
||||
Reference in New Issue
Block a user