MoviePilot/app/agent/prompt/__init__.py

"""
提示词管理器
"""

import shutil
from dataclasses import dataclass, field
from pathlib import Path
from string import Formatter
from time import strftime
from typing import Any, Dict, Optional

import yaml

from app.adapters.system.host import SystemUtils
from app.agent.llm.capability import AgentCapabilityManager
from app.agent.shell import resolve_agent_shell
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
from app.schemas.notification import (
    ChannelCapabilities,
    ChannelCapability,
    ChannelCapabilityManager,
)
from app.schemas.types import NotificationChannel

SYSTEM_TASKS_FILE = "System Tasks.yaml"
SYSTEM_TASKS_SCHEMA_VERSION = 2
COMMON_SHELL_COMMANDS = (
    "ssh",
    "sshpass",
    "scp",
    "sftp",
    "git",
    "gh",
    "rg",
    "fd",
    "jq",
    "yq",
    "curl",
    "wget",
    "docker",
    "docker-compose",
    "python",
    "python3",
    "ffmpeg",
    "ffprobe",
    "mediainfo",
    "rclone",
    "aria2c",
    "yt-dlp",
)

_AGENT_OUTPUT_LANGUAGE_NAMES = {
    "zh-CN": "Simplified Chinese",
    "zh-TW": "Traditional Chinese",
    "en-US": "English",
    "ja-JP": "Japanese",
    "ko-KR": "Korean",
}
_AGENT_OUTPUT_LANGUAGE_ALIASES = {
    "zh": "zh-CN",
    "zh-cn": "zh-CN",
    "zh-tw": "zh-TW",
    "en": "en-US",
    "en-us": "en-US",
    "ja": "ja-JP",
    "ja-jp": "ja-JP",
    "ko": "ko-KR",
    "ko-kr": "ko-KR",
}


class PromptConfigError(ValueError):
    """
    程序内置提示词定义加载异常。
    """


@dataclass
class SystemTaskTypeDefinition:
    """
    单个后台系统任务定义。

    :param header: 任务提示的标题模板
    :param objective: 任务目标说明模板
    :param context_title: 上下文段落标题；None 时不提供标题
    :param context_lines: 上下文模板行，默认独立空列表
    :param steps_title: 执行步骤段落标题；None 时不提供标题
    :param steps: 按执行顺序排列的步骤模板，默认独立空列表
    :param task_rules: 此类任务额外遵守的规则，默认独立空列表
    :param empty_result: 无结果时的回复模板；未定义时为 None
    """

    header: str
    objective: str
    context_title: Optional[str] = None
    context_lines: list[str] = field(default_factory=list)
    steps_title: Optional[str] = None
    steps: list[str] = field(default_factory=list)
    task_rules: list[str] = field(default_factory=list)
    empty_result: Optional[str] = None


@dataclass
class SystemTasksDefinition:
    """
    程序内置后台系统任务定义。

    :param path: 系统任务配置的来源路径，用于报告配置问题
    :param version: 配置文件声明的格式版本号
    :param shared_rules: 所有系统任务共享的行为规则
    :param task_types: 按任务类型名称索引的已验证提示模板定义
    """

    path: Path
    version: int
    shared_rules: list[str]
    task_types: dict[str, SystemTaskTypeDefinition]


class PromptManager:
    """
    提示词管理器
    """

    def __init__(self, prompts_dir: Optional[str] = None) -> None:
        """
        初始化提示词来源与运行信息缓存。
        """
        if prompts_dir is None:
            self.prompts_dir = Path(__file__).parent
        else:
            self.prompts_dir = Path(prompts_dir)
        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_command_names_cache: Optional[list[str]] = None

    def load_prompt(self, prompt_name: str) -> str:
        """
        加载指定的提示词
        """
        if prompt_name in self.prompts_cache:
            return self.prompts_cache[prompt_name]

        prompt_file = self.prompts_dir / prompt_name
        try:
            with open(prompt_file, "r", encoding="utf-8", errors="replace") as f:
                content = f.read().strip()
            # 缓存提示词
            self.prompts_cache[prompt_name] = content
            logger.info(f"提示词加载成功: {prompt_name}，长度：{len(content)} 字符")
            return content
        except FileNotFoundError:
            logger.error(f"提示词文件不存在: {prompt_file}")
            raise
        except Exception as e:
            logger.error(f"加载提示词失败: {prompt_name}, 错误: {e}")
            raise

    def get_agent_prompt(self, channel: Optional[str] = None) -> str:
        """
        获取智能体提示词
        :param channel: 消息渠道（Telegram、微信、Slack等）
        :return: 提示词内容
        """
        # 基础提示词包含领域任务、代码工作流、授权边界和渠道能力约束。
        # 根层运行时配置由 RuntimeConfigMiddleware 在每次模型调用前动态注入，
        # 这样人格切换可以在同一轮 Agent 执行里立即生效。
        base_prompt = self.load_prompt("System Core Prompt.txt")

        # 识别渠道
        markdown_spec = ""
        msg_channel = (
            next((c for c in NotificationChannel if c.value.lower() == channel.lower()), None) if channel else None
        )
        # 获取渠道能力说明
        if msg_channel:
            caps = ChannelCapabilityManager.get_capabilities(msg_channel)
            if caps:
                markdown_spec = self._generate_formatting_instructions(caps)
        button_choice_spec = self._generate_button_choice_instructions(msg_channel)
        rich_message_spec = self._generate_rich_message_instructions(msg_channel)

        # MoviePilot系统信息
        moviepilot_info = self._get_moviepilot_info()
        voice_reply_spec = self._generate_voice_reply_instructions()

        # 始终替换占位符，避免后续 .format() 时因残留花括号报 KeyError
        base_prompt = base_prompt.format(
            markdown_spec=markdown_spec,
            moviepilot_info=moviepilot_info,
            voice_reply_spec=voice_reply_spec,
            button_choice_spec=button_choice_spec,
            rich_message_spec=rich_message_spec,
        )

        base_prompt = f"{base_prompt}\n\n{self._get_output_language_instruction()}"

        return base_prompt

    @staticmethod
    def _get_output_language_instruction() -> str:
        """
        根据系统设置生成内置智能助手的回复语言约束。
        """
        configured = str(get_runtime_setting("AI_AGENT_OUTPUT_LANGUAGE") or "zh-CN").strip()
        normalized = configured.casefold()
        language_code = _AGENT_OUTPUT_LANGUAGE_ALIASES.get(normalized, configured)
        language_name = _AGENT_OUTPUT_LANGUAGE_NAMES.get(language_code, "Simplified Chinese")
        return (
            "Output language policy:\n"
            f"- Always reply to the user in {language_name}.\n"
            "- Keep this language for explanations, conclusions, and user-facing messages unless the user explicitly asks for another language. "
            "Preserve code, file paths, media titles, proper nouns, and tool arguments when they need to remain in their original form."
        )

    def load_system_tasks_definition(self) -> SystemTasksDefinition:
        """
        加载程序内置的后台系统任务定义。
        """
        system_tasks_path = self.prompts_dir / SYSTEM_TASKS_FILE
        try:
            stat = system_tasks_path.stat()
        except FileNotFoundError as err:
            logger.error(f"系统任务定义文件不存在: {system_tasks_path}")
            raise PromptConfigError(f"系统任务定义文件不存在: {system_tasks_path}") from err

        signature = (stat.st_mtime_ns, stat.st_size)
        if self._system_tasks_signature == signature and self._system_tasks_cache is not None:
            return self._system_tasks_cache

        try:
            content = system_tasks_path.read_text(encoding="utf-8", errors="replace")
        # 将文件读取边界的异常统一转换为带来源路径的配置错误，供上层报告。
        except Exception as err:  # noqa: BLE001
            logger.error(f"读取系统任务定义失败: {system_tasks_path}, 错误: {err}")
            raise PromptConfigError(f"读取系统任务定义失败 {system_tasks_path}: {err}") from err

        try:
            data = yaml.safe_load(content) or {}
        except yaml.YAMLError as err:
            raise PromptConfigError(f"YAML 解析失败 {system_tasks_path}: {err}") from err
        if not isinstance(data, dict):
            raise PromptConfigError(f"YAML 根节点必须是映射类型: {system_tasks_path}")

        definition = self._parse_system_tasks_definition(system_tasks_path, data)
        self._system_tasks_signature = signature
        self._system_tasks_cache = definition
        return definition

    def render_system_task_message(
        self,
        task_type: str,
        *,
        template_context: Optional[dict[str, Any]] = None,
        extra_rules: Optional[list[str]] = None,
    ) -> str:
        """
        根据程序内置 YAML 渲染后台系统任务提示词。
        """
        system_tasks = self.load_system_tasks_definition()
        task_definition = system_tasks.task_types.get(task_type)
        if not task_definition:
            raise PromptConfigError(f"未定义的后台系统任务类型: {task_type}")

        rendered_context = self._render_template_lines(
            task_definition.context_lines,
            template_context,
            task_type,
            "context_lines",
        )
        rendered_steps = self._render_template_lines(
            task_definition.steps,
            template_context,
            task_type,
            "steps",
        )
        rendered_task_rules = self._render_template_lines(
            task_definition.task_rules,
            template_context,
            task_type,
            "task_rules",
        )

        sections = [
            self._render_template_text(
                task_definition.header,
                template_context,
                task_type,
                "header",
            ).strip(),
            self._render_template_text(
                task_definition.objective,
                template_context,
                task_type,
                "objective",
            ).strip(),
        ]
        if rendered_context:
            sections.append(
                self._format_titled_lines(
                    task_definition.context_title or "Task context",
                    rendered_context,
                )
            )
        if rendered_steps:
            sections.append(
                self._format_titled_lines(
                    task_definition.steps_title or "Follow these steps",
                    rendered_steps,
                )
            )

        rules = list(system_tasks.shared_rules)
        if task_definition.empty_result:
            rules.append(task_definition.empty_result)
        rules.extend(rendered_task_rules)
        if extra_rules:
            rules.extend(rule.strip() for rule in extra_rules if rule and rule.strip())
        if rules:
            sections.append(self._format_numbered_rules("IMPORTANT", rules))
        return "\n\n".join(section for section in sections if section).strip()

    def _get_moviepilot_info(self) -> str:
        """
        获取需要常驻注入的最小 MoviePilot 运行信息。
        """
        # 保留日期用于提供“今天是哪天”的稳定上下文，但不再注入秒级时间，
        # 避免每次请求都生成不同的 system prompt，影响 provider 侧 cache 命中率。
        info_lines = [
            f"- 当前日期: {strftime('%Y-%m-%d')}",
            f"- 运行环境: {SystemUtils.platform} {'docker' if SystemUtils.is_docker() else ''}",
            "- 详细运行状态和数据库通过 `query_doctor_report` 或 `execute_command` 查询；配置值先加载对应 Skill，再通过 `moviepilot_api` 的配置 operation 查询。",
            "- Python 命令: 项目 `venv` 或 Docker 的 `VENV_PATH` 存在时，Agent 会把它放在命令 PATH 最前，并让裸 `python`/`python3` 自动使用专用入口 `moviepilot-python`；入口不存在时回退到环境中的标准 Python（Windows 对应 `Scripts`）。请直接使用 `python -m ...`，尤其是下载器、SMB、媒体服务器等需要系统权限的操作。安装依赖优先使用项目 `uv pip ...` 或绑定环境中的 `python -m pip ...`，不要切换到系统 Python 或使用 `sudo pip`。",
        ]
        try:
            shell = resolve_agent_shell(cwd=str(get_runtime_setting("ROOT_PATH")))
        except (OSError, ValueError):
            # 命令解释器配置错误不应阻断其余业务工具和对话。
            info_lines.append("- 默认命令解释器不可用；需要执行命令时，在 `execute_command` 中显式选择可用的 `shell`。")
        else:
            if shell:
                info_lines.append(shell.prompt_guidance())
        path_lines = self._get_runtime_path_lines()
        if path_lines:
            info_lines.extend(
                [
                    "- 关键运行路径（必要时可用文件/命令工具读取，避免扫描无关目录）:",
                    *path_lines,
                ]
            )
        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("- 搜索文件或文本时优先使用 `rg` / `rg --files`，不适合或不可用时再使用其他命令。")

        return "\n".join(info_lines)

    @staticmethod
    def _get_runtime_path_lines() -> list[str]:
        """
        返回基础系统提示词需要常驻注入的全局运行路径。
        """
        paths = {
            "项目根目录": get_runtime_setting("ROOT_PATH"),
            "配置目录": get_runtime_setting("CONFIG_PATH"),
            "临时目录": get_runtime_setting("TEMP_PATH"),
        }
        return [f"  - {label}: `{path}`" for label, path in paths.items()]

    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 = [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_command_names_cache(self) -> None:
        """
        清理可用命令名称缓存，供测试或运行时手动刷新使用。
        """
        self._available_shell_command_names_cache = None

    @staticmethod
    def _generate_formatting_instructions(caps: ChannelCapabilities) -> str:
        """
        根据渠道能力动态生成格式指令
        """
        instructions = []
        if ChannelCapability.MARKDOWN not in caps.capabilities:
            instructions.append("- Formatting: Use plain text only. The channel does NOT support Markdown.")
            instructions.append(
                "- Emphasis: Use clear wording; do not use Markdown headings, emphasis markers, code fences, or link syntax."
            )
            instructions.append(
                "- Lists: Prefer short paragraphs; when a list helps, use plain numbered lines such as 1、 and 2、."
            )
            instructions.append("- Links: Paste URLs directly as text.")
        return "\n".join(instructions)

    @staticmethod
    def _generate_voice_reply_instructions() -> str:
        """
        根据语音能力约束回复方式，避免语音和文本重复交付。
        """
        if not AgentCapabilityManager.supports_audio_output():
            return "Audio output is disabled; do not call `send_voice_message`."
        return (
            "Use normal text replies by default. Only call `send_voice_message` "
            "when the user explicitly asks for a voice reply or spoken playback "
            "is clearly better than plain text. `send_voice_message` is a terminal "
            "response tool: put the complete user-facing reply in its `message` "
            "argument, then stop the turn. Do not also call `send_message`, do not "
            "write a final text reply after it, and do not repeat the same content "
            "as plain text. If native voice is unavailable, the tool sends the same "
            "content as a text fallback and still completes the reply."
        )

    @staticmethod
    def _generate_rich_message_instructions(
        channel: Optional[NotificationChannel] = None,
    ) -> str:
        """
        根据渠道生成 Telegram Rich Message 回复提示。
        """
        if channel != NotificationChannel.Telegram:
            return ""
        return (
            "- Telegram final replies: Prefer the `send_message` tool with its "
            "`rich_message` argument. Put the complete reply in that argument using "
            "GitHub-style Markdown; headings, lists, tables, blockquotes, code blocks, "
            "and links are converted to Telegram Rich Message blocks. Do not also set "
            "`message`, `title`, or `image_url` for the same reply. Use a normal plain "
            "reply only when the response is very short and has no useful structure."
        )

    @staticmethod
    def _generate_button_choice_instructions(
        channel: Optional[NotificationChannel] = None,
    ) -> str:
        """
        按渠道能力选择按钮交互或文字提问，明确终结本轮的交付约束。
        """
        if (
            channel
            and ChannelCapabilityManager.supports_buttons(channel)
            and ChannelCapabilityManager.supports_callbacks(channel)
        ):
            return (
                "- User questions: If you need the user to choose from a few clear options, "
                "call `ask_user_choice` to send button options. `ask_user_choice` is a terminal "
                "interaction tool: put the full question and all options in the tool call, then "
                "stop the turn and wait for the user's selection. The selected value will come back "
                "as the user's next message. Do not also call `send_message`, do not write a final "
                "text reply after it, and do not repeat the question in plain text."
            )
        return "- User questions: When you truly need user input, ask briefly in plain text."

    def _parse_system_tasks_definition(
        self,
        path: Path,
        data: dict[str, Any],
    ) -> SystemTasksDefinition:
        """
        把 YAML 结构转换成系统任务定义对象。
        """
        version = self._normalize_positive_int(data.get("version"), "version", default=1)
        if version < SYSTEM_TASKS_SCHEMA_VERSION:
            raise PromptConfigError(
                f"{path} 的 version={version} 过旧，"
                f"当前要求 System Tasks schema v{SYSTEM_TASKS_SCHEMA_VERSION} 或更高版本"
            )

        shared_rules = self._normalize_string_list(data.get("shared_rules"), "shared_rules")
        if not shared_rules:
            raise PromptConfigError(f"{path} 缺少 shared_rules")

        raw_task_types = data.get("task_types")
        if not isinstance(raw_task_types, dict) or not raw_task_types:
            raise PromptConfigError(f"{path} 缺少 task_types 映射")

        task_types: dict[str, SystemTaskTypeDefinition] = {}
        for key, raw in raw_task_types.items():
            if not isinstance(raw, dict):
                raise PromptConfigError(f"task_types.{key} 必须是映射")

            header = str(raw.get("header") or "").strip()
            objective = str(raw.get("objective") or "").strip()
            if not header or not objective:
                raise PromptConfigError(f"task_types.{key} 缺少 header 或 objective")

            task_types[str(key)] = SystemTaskTypeDefinition(
                header=header,
                objective=objective,
                context_title=str(raw.get("context_title") or "").strip() or None,
                context_lines=self._normalize_string_list(
                    raw.get("context_lines"),
                    f"task_types.{key}.context_lines",
                ),
                steps_title=str(raw.get("steps_title") or "").strip() or None,
                steps=self._normalize_string_list(
                    raw.get("steps"),
                    f"task_types.{key}.steps",
                ),
                task_rules=self._normalize_string_list(
                    raw.get("task_rules"),
                    f"task_types.{key}.task_rules",
                ),
                empty_result=str(raw.get("empty_result") or "").strip() or None,
            )
        return SystemTasksDefinition(
            path=path,
            version=version,
            shared_rules=shared_rules,
            task_types=task_types,
        )

    @classmethod
    def _render_template_text(
        cls,
        text: str,
        template_context: Optional[dict[str, Any]],
        task_type: str,
        field_name: str,
    ) -> str:
        if not text:
            return ""

        formatter = Formatter()
        required_fields = {placeholder_name for _, placeholder_name, _, _ in formatter.parse(text) if placeholder_name}
        if not required_fields:
            return text

        context = cls._normalize_template_context(template_context)
        missing_fields = sorted(f for f in required_fields if f not in context)
        if missing_fields:
            raise PromptConfigError(
                f"系统任务定义 `{task_type}` 的 `{field_name}` 缺少变量: " + ", ".join(f"`{f}`" for f in missing_fields)
            )

        # 这里统一做字符串替换，让 YAML 成为后台任务文案的唯一行为来源。
        return text.format_map(context)

    @classmethod
    def _render_template_lines(
        cls,
        items: list[str],
        template_context: Optional[dict[str, Any]],
        task_type: str,
        field_name: str,
    ) -> list[str]:
        return [
            cls._render_template_text(
                item,
                template_context,
                task_type,
                f"{field_name}[{index}]",
            ).rstrip()
            for index, item in enumerate(items, start=1)
            if item and item.rstrip()
        ]

    @staticmethod
    def _normalize_template_context(
        template_context: Optional[dict[str, Any]],
    ) -> dict[str, str]:
        if not template_context:
            return {}
        return {str(key): "" if value is None else str(value) for key, value in template_context.items()}

    @staticmethod
    def _format_numbered_rules(title: str, items: list[str]) -> str:
        return "\n".join([f"{title}:"] + [f"{index}. {item}" for index, item in enumerate(items, start=1)])

    @staticmethod
    def _format_titled_lines(title: str, items: list[str]) -> str:
        cleaned = [item.rstrip() for item in items if item and item.rstrip()]
        return "\n".join([f"{title}:"] + cleaned)

    @staticmethod
    def _normalize_positive_int(
        value: Any,
        field_name: str,
        *,
        default: int,
    ) -> int:
        if value in (None, ""):
            return default
        try:
            normalized = int(value)
        except (TypeError, ValueError) as err:
            raise PromptConfigError(f"{field_name} 必须是正整数") from err
        if normalized <= 0:
            raise PromptConfigError(f"{field_name} 必须是正整数")
        return normalized

    @staticmethod
    def _normalize_string_list(values: Any, field_name: str) -> list[str]:
        if values is None:
            return []
        if not isinstance(values, list):
            raise PromptConfigError(f"{field_name} 必须是字符串数组")
        normalized: list[str] = []
        for value in values:
            text = str(value).strip()
            if text:
                normalized.append(text)
        return normalized

    def clear_cache(self) -> None:
        """
        清空缓存
        """
        self.prompts_cache.clear()
        self._system_tasks_cache = None
        self._system_tasks_signature = None
        logger.info("提示词缓存已清空")


prompt_manager = PromptManager()