archived-MoviePilot/app/agent/tools/impl/query_system_settings.py

"""统一查询系统设置工具。"""

import json
from typing import Optional, Type

from pydantic import BaseModel, Field

from app.agent.tools.base import MoviePilotTool
from app.agent.tools.impl._system_setting_utils import (
    SettingSpec,
    list_setting_specs,
    resolve_setting_spec,
)
from app.core.config import settings
from app.db.systemconfig_oper import SystemConfigOper
from app.log import logger


class QuerySystemSettingsInput(BaseModel):
    """查询系统设置工具的输入参数模型。"""

    explanation: str = Field(
        ...,
        description="Clear explanation of why this tool is being used in the current context",
    )
    setting_key: Optional[str] = Field(
        None,
        description=(
            "Exact setting key to query. Supports Settings field names like 'APP_DOMAIN' or 'TMDB_API_KEY', "
            "SystemConfigKey values like 'Downloaders' or 'MediaServers', enum names, and some single-key aliases "
            "such as 'downloaders', 'directories', 'search_sites', 'subscribe_sites', 'site_auth', 'ai_agent', "
            "and 'custom_identifiers'."
        ),
    )
    group: Optional[str] = Field(
        "all",
        description=(
            "Optional group filter when setting_key is not provided. Supports 'all', 'settings', 'systemconfig', "
            "and category aliases such as 'downloaders', 'media_servers', 'notifications', 'notification_switches', "
            "'storages', 'directories', 'search_sites', 'subscribe_sites', 'site_auth', 'ai_agent', 'filter_rules', "
            "'subscribe_defaults', 'plugins', and 'custom_identifiers'. Chinese aliases are also accepted."
        ),
    )
    keyword: Optional[str] = Field(
        None,
        description=(
            "Optional keyword used to fuzzy match setting keys, group names, or labels when listing settings."
        ),
    )
    include_values: Optional[bool] = Field(
        None,
        description=(
            "Whether to include full setting values. Default behavior: when a single setting is matched it returns the full value; "
            "when multiple settings are matched it returns summaries only unless this is explicitly set to true."
        ),
    )


class QuerySystemSettingsTool(MoviePilotTool):
    name: str = "query_system_settings"
    description: str = (
        "Query system settings across both the basic Settings module and all SystemConfig-backed categories. "
        "Use this tool to inspect downloaders, media servers, notification channels, storages, directories, search-site ranges, "
        "subscribe-site ranges, site auth params, AI agent config, and any other system setting before making changes."
    )
    require_admin: bool = True
    args_schema: Type[BaseModel] = QuerySystemSettingsInput

    def get_tool_message(self, **kwargs) -> Optional[str]:
        """根据查询参数生成友好的提示消息。"""

        setting_key = kwargs.get("setting_key")
        group = kwargs.get("group", "all")
        keyword = kwargs.get("keyword")
        if setting_key:
            return f"查询系统设置: {setting_key}"
        if keyword:
            return f"筛选系统设置: {group} / {keyword}"
        return f"查询系统设置分组: {group}"

    @staticmethod
    def _load_setting_value(spec: SettingSpec):
        if spec.source == "settings":
            return getattr(settings, spec.key)
        return SystemConfigOper().get(spec.key)

    @staticmethod
    def _summarize_value(value) -> dict:
        summary = {
            "has_value": value is not None,
            "value_type": type(value).__name__,
        }
        if isinstance(value, list):
            summary["item_count"] = len(value)
            if value:
                summary["item_type"] = type(value[0]).__name__
        elif isinstance(value, dict):
            keys = list(value.keys())
            summary["item_count"] = len(keys)
            summary["keys_preview"] = keys[:10]
            if len(keys) > 10:
                summary["keys_truncated"] = True
        elif isinstance(value, str):
            summary["length"] = len(value)
            preview = value[:200]
            if preview:
                summary["value_preview"] = preview
                if len(value) > len(preview):
                    summary["value_truncated"] = True
        elif value is not None:
            summary["value_preview"] = value
        return summary

    async def run(
        self,
        setting_key: Optional[str] = None,
        group: Optional[str] = "all",
        keyword: Optional[str] = None,
        include_values: Optional[bool] = None,
        **kwargs,
    ) -> str:
        logger.info(
            "执行工具: %s, setting_key=%s, group=%s, keyword=%s",
            self.name,
            setting_key,
            group,
            keyword,
        )

        try:
            if setting_key:
                spec = resolve_setting_spec(setting_key)
                if not spec:
                    return json.dumps(
                        {
                            "success": False,
                            "message": f"系统设置项 '{setting_key}' 不存在",
                        },
                        ensure_ascii=False,
                    )
                specs = [spec]
            else:
                specs = list_setting_specs(group=group, keyword=keyword)
                if not specs:
                    return json.dumps(
                        {
                            "success": False,
                            "message": "没有找到匹配的系统设置项",
                        },
                        ensure_ascii=False,
                    )

            should_include_values = (
                include_values if include_values is not None else len(specs) == 1
            )
            settings_payload = []
            for spec in specs:
                value = self._load_setting_value(spec)
                item = {
                    "setting_key": spec.key,
                    "source": spec.source,
                    "group": spec.group,
                    "label": spec.label,
                }
                item.update(self._summarize_value(value))
                if should_include_values:
                    item["value"] = value
                settings_payload.append(item)

            return json.dumps(
                {
                    "success": True,
                    "matched_count": len(settings_payload),
                    "include_values": should_include_values,
                    "settings": settings_payload,
                },
                ensure_ascii=False,
                indent=2,
                default=str,
            )
        except Exception as e:
            logger.error(f"查询系统设置失败: {e}", exc_info=True)
            return json.dumps(
                {"success": False, "message": f"查询系统设置时发生错误: {str(e)}"},
                ensure_ascii=False,
            )