AstrBot/astrbot/dashboard/services/config_service.py

from __future__ import annotations

import asyncio
import copy
import inspect
import os
import traceback
from pathlib import Path
from typing import Any

from astrbot.core import file_token_service, logger
from astrbot.core.config.astrbot_config import AstrBotConfig
from astrbot.core.config.default import (
    CONFIG_METADATA_2,
    CONFIG_METADATA_3,
    CONFIG_METADATA_3_SYSTEM,
    DEFAULT_CONFIG,
    DEFAULT_VALUE_MAP,
)
from astrbot.core.config.i18n_utils import ConfigMetadataI18n
from astrbot.core.core_lifecycle import AstrBotCoreLifecycle
from astrbot.core.db import BaseDatabase
from astrbot.core.platform.register import platform_cls_map, platform_registry
from astrbot.core.provider.register import provider_registry
from astrbot.core.star.star import star_registry
from astrbot.core.utils.astrbot_path import get_astrbot_plugin_data_path
from astrbot.core.utils.totp import (
    is_totp_enabled,
    revoke_user_trusted_devices,
    set_pending_totp_secret,
    verify_configured_2fa_code,
)
from astrbot.core.utils.webhook_utils import ensure_platform_webhook_config
from astrbot.dashboard.async_utils import run_maybe_async
from astrbot.dashboard.responses import ApiError

PROTECTED_2FA_CONFIG_PATHS = (
    ("dashboard", "totp", "enable"),
    ("dashboard", "totp", "secret"),
    ("dashboard", "totp", "recovery_code_hash"),
)
MAX_FILE_BYTES = 500 * 1024 * 1024


def try_cast(value: Any, type_: str):
    if type_ == "int":
        try:
            return int(value)
        except (ValueError, TypeError):
            return None
    elif (
        type_ == "float"
        and isinstance(value, str)
        and value.replace(".", "", 1).isdigit()
    ) or (type_ == "float" and isinstance(value, int)):
        return float(value)
    elif type_ == "float":
        try:
            return float(value)
        except (ValueError, TypeError):
            return None


def _expect_type(value, expected_type, path_key, errors, expected_name=None) -> bool:
    if not isinstance(value, expected_type):
        errors.append(
            f"错误的类型 {path_key}: 期望是 {expected_name or expected_type.__name__}, "
            f"得到了 {type(value).__name__}"
        )
        return False
    return True


def _validate_template_list(value, meta, path_key, errors, validate_fn) -> None:
    if not _expect_type(value, list, path_key, errors, "list"):
        return

    templates = meta.get("templates")
    if not isinstance(templates, dict):
        templates = {}

    for idx, item in enumerate(value):
        item_path = f"{path_key}[{idx}]"
        if not _expect_type(item, dict, item_path, errors, "dict"):
            continue

        template_key = item.get("__template_key") or item.get("template")
        if not template_key:
            errors.append(f"缺少模板选择 {item_path}: 需要 __template_key")
            continue

        template_meta = templates.get(template_key)
        if not template_meta:
            errors.append(f"未知模板 {item_path}: {template_key}")
            continue

        validate_fn(
            item,
            template_meta.get("items", {}),
            path=f"{path_key}.templates.{template_key}.",
        )


def sanitize_path_segment(segment: str) -> str:
    cleaned = []
    for ch in segment:
        if (
            ("a" <= ch <= "z")
            or ("A" <= ch <= "Z")
            or ch.isdigit()
            or ch
            in {
                "-",
                "_",
            }
        ):
            cleaned.append(ch)
        else:
            cleaned.append("_")
    result = "".join(cleaned).strip("_")
    return result or "_"


def _config_key_to_folder(key_path: str) -> str:
    parts = [sanitize_path_segment(part) for part in key_path.split(".") if part]
    return "/".join(parts) if parts else "_"


def config_key_to_folder(key_path: str) -> str:
    return _config_key_to_folder(key_path)


def _normalize_rel_path(path: object) -> str | None:
    if not isinstance(path, str):
        return None
    rel = path.replace("\\", "/").lstrip("/")
    if not rel:
        return None
    parts = [part for part in rel.split("/") if part]
    if any(part in {".", ".."} for part in parts):
        return None
    if rel.startswith("../") or "/../" in rel:
        return None
    return "/".join(parts)


def normalize_rel_path(path: str | None) -> str | None:
    return _normalize_rel_path(path)


def _get_schema_item(schema: dict | None, key_path: str) -> dict | None:
    if not isinstance(schema, dict) or not key_path:
        return None
    if key_path in schema:
        return schema.get(key_path)

    parts = key_path.split(".")
    current = schema
    idx = 0
    while idx < len(parts):
        part = parts[idx]
        if part not in current:
            return None
        meta = current.get(part)
        if idx == len(parts) - 1:
            return meta
        if not isinstance(meta, dict) or meta.get("type") != "object":
            if not isinstance(meta, dict) or meta.get("type") != "template_list":
                return None
            if idx + 2 >= len(parts) or parts[idx + 1] != "templates":
                return None
            template_meta = meta.get("templates", {}).get(parts[idx + 2])
            if not isinstance(template_meta, dict):
                return None
            if idx + 2 == len(parts) - 1:
                return template_meta
            current = template_meta.get("items", {})
            idx += 3
            continue
        current = meta.get("items", {})
        idx += 1
    return None


def get_schema_item(schema: dict | None, key_path: str) -> dict | None:
    return _get_schema_item(schema, key_path)


def _sanitize_filename(name: str) -> str:
    cleaned = os.path.basename(name).strip()
    if not cleaned or cleaned in {".", ".."}:
        return ""
    for sep in (os.sep, os.altsep):
        if sep:
            cleaned = cleaned.replace(sep, "_")
    return cleaned


def sanitize_filename(name: str) -> str:
    return _sanitize_filename(name)


def validate_config(data, schema: dict, is_core: bool) -> tuple[list[str], dict]:
    errors = []

    def validate(data: dict, metadata: dict = schema, path="") -> None:
        for key, value in data.items():
            if key not in metadata:
                continue
            meta = metadata[key]
            if "type" not in meta:
                logger.debug(f"配置项 {path}{key} 没有类型定义, 跳过校验")
                continue
            if value is None:
                data[key] = DEFAULT_VALUE_MAP[meta["type"]]
                continue

            if meta["type"] == "template_list":
                _validate_template_list(value, meta, f"{path}{key}", errors, validate)
                continue

            if meta["type"] == "file":
                if not _expect_type(value, list, f"{path}{key}", errors, "list"):
                    continue
                for idx, item in enumerate(value):
                    if not isinstance(item, str):
                        errors.append(
                            f"Invalid type {path}{key}[{idx}]: expected string, got {type(item).__name__}",
                        )
                        continue
                    normalized = _normalize_rel_path(item)
                    if not normalized or not normalized.startswith("files/"):
                        errors.append(
                            f"Invalid file path {path}{key}[{idx}]: {item}",
                        )
                        continue
                    key_path = f"{path}{key}"
                    expected_folder = _config_key_to_folder(key_path)
                    expected_prefix = f"files/{expected_folder}/"
                    if not normalized.startswith(expected_prefix):
                        errors.append(
                            f"Invalid file path {path}{key}[{idx}]: {item}",
                        )
                        continue
                    value[idx] = normalized
                continue

            if meta["type"] == "list" and not isinstance(value, list):
                errors.append(
                    f"错误的类型 {path}{key}: 期望是 list, 得到了 {type(value).__name__}",
                )
            elif (
                meta["type"] == "list"
                and isinstance(value, list)
                and value
                and "items" in meta
                and isinstance(value[0], dict)
            ):
                for item in value:
                    validate(item, meta["items"], path=f"{path}{key}.")
            elif meta["type"] == "object" and isinstance(value, dict):
                validate(value, meta["items"], path=f"{path}{key}.")

            if meta["type"] == "int" and not isinstance(value, int):
                casted = try_cast(value, "int")
                if casted is None:
                    errors.append(
                        f"错误的类型 {path}{key}: 期望是 int, 得到了 {type(value).__name__}",
                    )
                data[key] = casted
            elif meta["type"] == "float" and not isinstance(value, float):
                casted = try_cast(value, "float")
                if casted is None:
                    errors.append(
                        f"错误的类型 {path}{key}: 期望是 float, 得到了 {type(value).__name__}",
                    )
                data[key] = casted
            elif meta["type"] == "bool" and not isinstance(value, bool):
                errors.append(
                    f"错误的类型 {path}{key}: 期望是 bool, 得到了 {type(value).__name__}",
                )
            elif meta["type"] in ["string", "text"] and not isinstance(value, str):
                errors.append(
                    f"错误的类型 {path}{key}: 期望是 string, 得到了 {type(value).__name__}",
                )
            elif meta["type"] == "list" and not isinstance(value, list):
                errors.append(
                    f"错误的类型 {path}{key}: 期望是 list, 得到了 {type(value).__name__}",
                )
            elif meta["type"] == "object" and not isinstance(value, dict):
                errors.append(
                    f"错误的类型 {path}{key}: 期望是 dict, 得到了 {type(value).__name__}",
                )

    if is_core:
        meta_all = {
            **schema["platform_group"]["metadata"],
            **schema["provider_group"]["metadata"],
            **schema["misc_config_group"]["metadata"],
        }
        validate(data, meta_all)
    else:
        validate(data, schema)

    return errors, data


def _log_computer_config_changes(
    old_config: dict,
    new_config: dict,
    *,
    log_info=None,
) -> None:
    log_info = log_info or logger.info
    old_ps = old_config.get("provider_settings", {})
    new_ps = new_config.get("provider_settings", {})

    old_runtime = old_ps.get("computer_use_runtime", "none")
    new_runtime = new_ps.get("computer_use_runtime", "none")
    if old_runtime != new_runtime:
        log_info(
            "[Computer] Config changed: computer_use_runtime %s -> %s",
            old_runtime,
            new_runtime,
        )

    old_sandbox = old_ps.get("sandbox", {})
    new_sandbox = new_ps.get("sandbox", {})
    all_keys = set(old_sandbox.keys()) | set(new_sandbox.keys())
    for key in sorted(all_keys):
        old_val = old_sandbox.get(key)
        new_val = new_sandbox.get(key)
        if old_val == new_val:
            continue
        if "token" in key or "secret" in key:
            old_display = "***" if old_val else "(empty)"
            new_display = "***" if new_val else "(empty)"
        else:
            old_display = old_val
            new_display = new_val
        log_info(
            "[Computer] Config changed: sandbox.%s %s -> %s",
            key,
            old_display,
            new_display,
        )


def _get_nested_value(data: dict, path: tuple[str, ...]) -> Any:
    current = data
    for key in path:
        if not isinstance(current, dict) or key not in current:
            return None
        current = current[key]
    return current


def _set_nested_value(data: dict, path: tuple[str, ...], value: Any) -> None:
    current = data
    for key in path[:-1]:
        next_value = current.get(key)
        if not isinstance(next_value, dict):
            next_value = {}
            current[key] = next_value
        current = next_value
    current[path[-1]] = value


def _protected_2fa_config_changed(old_config: dict, new_config: dict) -> bool:
    return any(
        _get_nested_value(old_config, path) != _get_nested_value(new_config, path)
        for path in PROTECTED_2FA_CONFIG_PATHS
    )


async def _validate_neo_connectivity(post_config: dict) -> str | None:
    ps = post_config.get("provider_settings", {})
    runtime = ps.get("computer_use_runtime", "none")
    sandbox = ps.get("sandbox", {})
    booter = sandbox.get("booter", "")

    if runtime != "sandbox" or booter != "shipyard_neo":
        return None

    endpoint = sandbox.get("shipyard_neo_endpoint", "").rstrip("/")
    if not endpoint:
        return "⚠️ Shipyard Neo endpoint 未设置"

    access_token = sandbox.get("shipyard_neo_access_token", "")
    if not access_token:
        from astrbot.core.computer.computer_client import _discover_bay_credentials

        access_token = _discover_bay_credentials(endpoint)

    if not access_token:
        return (
            "⚠️ 未找到 Bay API Key。请填写访问令牌，"
            "或确保 Bay 的 credentials.json 可被自动发现。"
        )

    import aiohttp

    health_url = f"{endpoint}/health"
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(
                health_url,
                timeout=aiohttp.ClientTimeout(total=5),
            ) as resp:
                if resp.status != 200:
                    return (
                        f"⚠️ Bay 健康检查失败 (HTTP {resp.status})，"
                        f"请确认 Bay 正在运行: {endpoint}"
                    )
    except Exception:
        return f"⚠️ 无法连接 Bay ({endpoint})，请确认 Bay 已启动。"

    return None


def save_config(
    post_config: dict,
    config: AstrBotConfig,
    is_core: bool = False,
) -> None:
    if is_core:
        _log_computer_config_changes(dict(config), post_config)

    try:
        if is_core:
            errors, post_config = validate_config(
                post_config,
                CONFIG_METADATA_2,
                is_core,
            )
        else:
            errors, post_config = validate_config(
                post_config,
                getattr(config, "schema", {}),
                is_core,
            )
    except BaseException as exc:
        logger.error(traceback.format_exc())
        logger.warning(f"验证配置时出现异常: {exc}")
        raise ValueError(f"验证配置时出现异常: {exc}")
    if errors:
        raise ValueError(f"格式校验未通过: {errors}")

    config.save_config(post_config)


class ConfigProfileService:
    def __init__(
        self,
        core_lifecycle: AstrBotCoreLifecycle,
        db: BaseDatabase | None = None,
    ) -> None:
        self.core_lifecycle = core_lifecycle
        self.acm = core_lifecycle.astrbot_config_mgr
        self.db = db

    def get_profile_schema(self) -> dict:
        return {
            "config": DEFAULT_CONFIG,
            "metadata": ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3),
        }

    def get_system_schema(self) -> dict:
        return {
            "config": self.acm.confs["default"],
            "metadata": ConfigMetadataI18n.convert_to_i18n_keys(
                CONFIG_METADATA_3_SYSTEM
            ),
        }

    def get_system_config(self) -> dict:
        return self.get_system_schema()

    def list_profiles(self) -> dict:
        return {"info_list": self.acm.get_conf_list()}

    async def create_profile(self, name: str | None, config: dict | None) -> dict:
        conf_id = self.acm.create_conf(name=name, config=config or DEFAULT_CONFIG)
        await self.core_lifecycle.reload_pipeline_scheduler(conf_id)
        return {"conf_id": conf_id}

    async def create_profile_from_dashboard_payload(
        self,
        payload: object,
    ) -> dict:
        data = payload if isinstance(payload, dict) else {}
        if not data:
            raise ValueError("缺少配置数据")
        return await self.create_profile(data.get("name"), data.get("config"))

    def get_profile(self, config_id: str) -> dict:
        if config_id not in self.acm.confs:
            raise ValueError(f"Config file {config_id} does not exist")
        return {
            "config": self.acm.confs[config_id],
            "metadata": ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3),
        }

    def get_profile_from_dashboard_query(
        self,
        *,
        config_id: str | None,
        system_config: bool,
    ) -> dict:
        if not config_id and not system_config:
            raise ValueError("缺少配置文件 ID")
        if system_config:
            return self.get_system_schema()
        if config_id is None:
            raise ValueError("abconf_id cannot be None")
        return self.get_profile(config_id)

    def get_profile_from_dashboard_args(self, args) -> dict:
        system_config = str(args.get("system_config", "0")).lower() == "1"
        return self.get_profile_from_dashboard_query(
            config_id=args.get("id"),
            system_config=system_config,
        )

    async def update_profile(
        self,
        config_id: str,
        config: dict,
        *,
        two_factor_code: str | None = None,
    ) -> str | None:
        if config_id not in self.acm.confs:
            raise ValueError(f"Config file {config_id} does not exist")
        config = copy.deepcopy(config)
        if config_id == "default":
            default_conf = getattr(self.acm, "default_conf", self.acm.confs["default"])
            for key in ("provider_sources", "provider", "platform"):
                config[key] = default_conf.get(key, [])

        current_config = self.acm.confs[config_id]
        protected_2fa_changed = _protected_2fa_config_changed(current_config, config)
        if (
            is_totp_enabled(current_config)
            and protected_2fa_changed
            and not await self._verify_config_2fa(current_config, two_factor_code)
        ):
            raise ApiError(
                "需要 TOTP 验证",
                status_code=401,
                data={"totp_required": True},
            )

        if not _get_nested_value(config, ("dashboard", "totp", "enable")):
            _set_nested_value(config, ("dashboard", "totp", "secret"), "")
            _set_nested_value(config, ("dashboard", "totp", "recovery_code_hash"), "")

        set_pending_totp_secret(None)
        save_config(config, self.acm.confs[config_id], is_core=True)
        if protected_2fa_changed and self.db is not None:
            await revoke_user_trusted_devices(self.db)
        await self.core_lifecycle.reload_pipeline_scheduler(config_id)
        warning = await _validate_neo_connectivity(config)
        if warning:
            return f"保存成功。{warning}"
        return "保存成功~"

    async def update_profile_from_dashboard_payload(
        self,
        payload: object,
        *,
        two_factor_code: str | None = None,
    ) -> str | None:
        data = payload if isinstance(payload, dict) else {}
        if not isinstance(payload, dict):
            raise ValueError("Invalid request payload")
        config = data.get("config")
        conf_id = data.get("conf_id")
        if not isinstance(config, dict):
            raise ValueError("Invalid config payload")
        return await self.update_profile(
            str(conf_id),
            config,
            two_factor_code=two_factor_code,
        )

    @staticmethod
    async def _verify_config_2fa(
        current_config: dict,
        two_factor_code: str | None,
    ) -> bool:
        code = (two_factor_code or "").strip()
        if not code:
            return False
        return bool(
            await verify_configured_2fa_code(
                current_config,
                code,
                include_pending=True,
                allow_recovery=False,
            )
        )

    def rename_profile(self, config_id: str, name: str | None) -> None:
        if not self.acm.update_conf_info(config_id, name=name):
            raise ValueError("Failed to update config profile")

    def rename_profile_from_dashboard_payload(self, payload: object) -> str:
        data = payload if isinstance(payload, dict) else {}
        if not data:
            raise ValueError("缺少配置数据")
        conf_id = data.get("id")
        if not conf_id:
            raise ValueError("缺少配置文件 ID")
        self.rename_profile(str(conf_id), name=data.get("name"))
        return "更新成功"

    def delete_profile(self, config_id: str) -> None:
        if not self.acm.delete_conf(config_id):
            raise ValueError("Failed to delete config profile")
        self.core_lifecycle.pipeline_scheduler_mapping.pop(config_id, None)

    def delete_profile_from_dashboard_payload(self, payload: object) -> str:
        data = payload if isinstance(payload, dict) else {}
        if not data:
            raise ValueError("缺少配置数据")
        conf_id = data.get("id")
        if not conf_id:
            raise ValueError("缺少配置文件 ID")
        self.delete_profile(str(conf_id))
        return "删除成功"


class ConfigRoutingService:
    def __init__(self, core_lifecycle: AstrBotCoreLifecycle) -> None:
        self.ucr = core_lifecycle.umop_config_router

    def list_routes(self) -> dict:
        return {"routing": self.ucr.umop_to_conf_id}

    async def replace_route_mapping(self, routing: dict[str, str]) -> None:
        await self.ucr.update_routing_data(routing)

    async def replace_routes(self, data: object) -> None:
        payload = data if isinstance(data, dict) else {}
        new_routing = payload.get("routing")
        if not isinstance(new_routing, dict):
            raise ValueError("缺少或错误的路由表数据")
        await self.replace_route_mapping(new_routing)

    async def replace_routes_from_dashboard_payload(self, payload: object) -> str:
        if not isinstance(payload, dict) or not payload:
            raise ValueError("缺少配置数据")
        await self.replace_routes(payload)
        return "更新成功"

    async def upsert_route(self, data: object) -> None:
        payload = data if isinstance(data, dict) else {}
        umo = payload.get("umo")
        conf_id = payload.get("conf_id")
        if not umo or not conf_id:
            raise ValueError("缺少 UMO 或配置文件 ID")
        await self.set_route(str(umo), str(conf_id))

    async def set_route(self, umo: str, config_id: str) -> None:
        if config_id == "default":
            await self.delete_route_by_umo(umo)
            return
        await self.ucr.update_route(umo, config_id)

    async def upsert_route_from_dashboard_payload(self, payload: object) -> str:
        if not isinstance(payload, dict) or not payload:
            raise ValueError("缺少配置数据")
        await self.upsert_route(payload)
        return "更新成功"

    async def delete_route(self, data: object) -> None:
        payload = data if isinstance(data, dict) else {}
        umo = payload.get("umo")
        if not umo:
            raise ValueError("缺少 UMO")
        await self.delete_route_by_umo(str(umo))

    async def delete_route_by_umo(self, umo: str) -> None:
        if umo in self.ucr.umop_to_conf_id:
            del self.ucr.umop_to_conf_id[umo]
            await self.ucr.update_routing_data(self.ucr.umop_to_conf_id)

    async def delete_route_from_dashboard_payload(self, payload: object) -> str:
        if not isinstance(payload, dict) or not payload:
            raise ValueError("缺少配置数据")
        await self.delete_route(payload)
        return "删除成功"


class ConfigDisplayService:
    def __init__(self, core_lifecycle: AstrBotCoreLifecycle) -> None:
        self.core_lifecycle = core_lifecycle
        self.config = core_lifecycle.astrbot_config
        self._logo_token_cache: dict[str, str] = {}

    async def get_configs(self, plugin_name: str | None = None) -> dict:
        if not plugin_name:
            return await self.get_astrbot_config()
        return self.get_plugin_config(plugin_name)

    async def get_configs_from_dashboard_args(self, args) -> dict:
        return await self.get_configs(args.get("plugin_name", None))

    async def get_astrbot_config(self) -> dict:
        metadata = copy.deepcopy(CONFIG_METADATA_2)
        platform_i18n = ConfigMetadataI18n.convert_to_i18n_keys(
            {
                "platform_group": {
                    "metadata": {
                        "platform": metadata["platform_group"]["metadata"]["platform"]
                    }
                }
            }
        )
        metadata["platform_group"]["metadata"]["platform"] = platform_i18n[
            "platform_group"
        ]["metadata"]["platform"]

        platform_default_tmpl = metadata["platform_group"]["metadata"]["platform"][
            "config_template"
        ]
        platform_i18n_translations = {}
        logo_registration_tasks = []

        for platform in platform_registry:
            if not platform.default_config_tmpl:
                continue

            platform_default_tmpl[platform.name] = copy.deepcopy(
                platform.default_config_tmpl
            )
            if platform.config_metadata:
                self.inject_platform_metadata_with_i18n(
                    platform,
                    metadata,
                    platform_i18n_translations,
                )
            if platform.logo_path:
                logo_registration_tasks.append(
                    self.register_platform_logo(platform, platform_default_tmpl),
                )

        if logo_registration_tasks:
            await asyncio.gather(*logo_registration_tasks, return_exceptions=True)

        provider_default_tmpl = metadata["provider_group"]["metadata"]["provider"][
            "config_template"
        ]
        for provider in provider_registry:
            if provider.default_config_tmpl:
                provider_default_tmpl[provider.type] = provider.default_config_tmpl

        return {
            "metadata": metadata,
            "config": self.config,
            "platform_i18n_translations": platform_i18n_translations,
        }

    def get_plugin_config(self, plugin_name: str) -> dict:
        result: dict = {"metadata": None, "config": None, "i18n": {}}

        for plugin_md in star_registry:
            if plugin_md.name != plugin_name:
                continue
            if not plugin_md.config:
                break
            result["config"] = plugin_md.config
            result["metadata"] = {
                plugin_name: {
                    "description": f"{plugin_name} 配置",
                    "type": "object",
                    "items": plugin_md.config.schema,
                },
            }
            result["i18n"] = plugin_md.i18n
            break

        return result

    async def register_platform_logo(self, platform, platform_default_tmpl) -> None:
        if not platform.logo_path:
            return

        try:
            cache_key = f"{platform.name}:{platform.logo_path}"
            if cache_key in self._logo_token_cache:
                self._set_platform_logo_token(
                    platform_default_tmpl,
                    platform.name,
                    self._logo_token_cache[cache_key],
                )
                logger.debug(f"Using cached logo token for platform {platform.name}")
                return

            platform_cls = platform_cls_map.get(platform.name)
            if not platform_cls:
                logger.warning(f"Platform class not found for {platform.name}")
                return

            module_file = inspect.getfile(platform_cls)
            plugin_dir = os.path.dirname(module_file)
            logo_file_path = os.path.join(plugin_dir, platform.logo_path)

            if not os.path.exists(logo_file_path):
                logger.warning(
                    f"Platform {platform.name} logo file not found: {logo_file_path}",
                )
                return

            logo_token = await file_token_service.register_file(
                logo_file_path,
                timeout=3600,
            )
            self._set_platform_logo_token(
                platform_default_tmpl,
                platform.name,
                logo_token,
            )
            self._logo_token_cache[cache_key] = logo_token
            logger.debug(f"Logo token registered for platform {platform.name}")

        except (ImportError, AttributeError) as exc:
            logger.warning(
                f"Failed to import required modules for platform {platform.name}: {exc}",
            )
        except OSError as exc:
            logger.warning(
                f"File system error for platform {platform.name} logo: {exc}"
            )
        except Exception as exc:
            logger.warning(
                f"Unexpected error registering logo for platform {platform.name}: {exc}",
            )

    @staticmethod
    def _set_platform_logo_token(
        platform_default_tmpl: dict,
        platform_name: str,
        logo_token: str,
    ) -> None:
        if platform_name not in platform_default_tmpl or not isinstance(
            platform_default_tmpl[platform_name],
            dict,
        ):
            platform_default_tmpl[platform_name] = {}
        platform_default_tmpl[platform_name]["logo_token"] = logo_token

    @staticmethod
    def inject_platform_metadata_with_i18n(
        platform,
        metadata,
        platform_i18n_translations: dict,
    ) -> None:
        metadata["platform_group"]["metadata"]["platform"].setdefault("items", {})
        platform_items_to_inject = copy.deepcopy(platform.config_metadata)

        if platform.i18n_resources:
            i18n_prefix = f"platform_group.platform.{platform.name}"

            for lang, lang_data in platform.i18n_resources.items():
                platform_i18n_translations.setdefault(lang, {}).setdefault(
                    "platform_group", {}
                ).setdefault("platform", {})[platform.name] = lang_data

            for field_key, field_value in platform_items_to_inject.items():
                for key in ("description", "hint", "labels"):
                    if key in field_value:
                        field_value[key] = f"{i18n_prefix}.{field_key}.{key}"

        metadata["platform_group"]["metadata"]["platform"]["items"].update(
            platform_items_to_inject
        )


class ConfigFileService:
    def __init__(self, core_lifecycle: AstrBotCoreLifecycle) -> None:
        self.core_lifecycle = core_lifecycle

    def get_plugin_metadata_by_name(self, plugin_name: str):
        for plugin_md in star_registry:
            if plugin_md.name == plugin_name:
                return plugin_md
        return None

    def resolve_config_file_scope(
        self,
        *,
        scope: str | None,
        name: str | None,
        key_path: str | None,
    ):
        scope = scope or "plugin"
        if scope != "plugin":
            raise ValueError(f"Unsupported scope: {scope}")
        if not name or not key_path:
            raise ValueError("Missing name or key parameter")

        metadata = self.get_plugin_metadata_by_name(name)
        if not metadata or not metadata.config:
            raise ValueError(f"Plugin {name} not found or has no config")

        return scope, name, key_path, metadata, metadata.config

    async def save_plugin_configs(
        self,
        post_configs: dict,
        plugin_name: str,
    ) -> None:
        metadata = self.get_plugin_metadata_by_name(plugin_name)
        if not metadata:
            raise ValueError(f"插件 {plugin_name} 不存在")
        if not metadata.config:
            raise ValueError(f"插件 {plugin_name} 没有注册配置")

        errors, post_configs = validate_config(
            post_configs,
            getattr(metadata.config, "schema", {}),
            is_core=False,
        )
        if errors:
            raise ValueError(f"格式校验未通过: {errors}")
        metadata.config.save_config(post_configs)
        await self.core_lifecycle.plugin_manager.reload(plugin_name)

    async def save_plugin_configs_from_dashboard_payload(
        self,
        payload: object,
        *,
        plugin_name: str,
    ) -> str:
        post_configs = payload if isinstance(payload, dict) else {}
        await self.save_plugin_configs(post_configs, plugin_name)
        return f"保存插件 {plugin_name} 成功~ 机器人正在热重载插件。"

    async def upload_config_file(
        self,
        *,
        scope: str | None,
        name: str | None,
        key_path: str | None,
        files,
    ) -> dict:
        _, name, key_path, _, config = self.resolve_config_file_scope(
            scope=scope,
            name=name,
            key_path=key_path,
        )
        meta = _get_schema_item(getattr(config, "schema", None), key_path)
        if not meta or meta.get("type") != "file":
            raise ValueError("Config item not found or not file type")
        if not files:
            raise ValueError("No files uploaded")

        allowed_exts = self._allowed_file_extensions(meta)
        plugin_root_path = self._plugin_root_path(name)
        plugin_root_path.mkdir(parents=True, exist_ok=True)

        uploaded: list[str] = []
        errors: list[str] = []
        folder = _config_key_to_folder(key_path)

        for file in files.values():
            filename = _sanitize_filename(file.filename or "")
            if not filename:
                errors.append("Invalid filename")
                continue

            file_size = getattr(file, "content_length", None)
            if isinstance(file_size, int) and file_size > MAX_FILE_BYTES:
                errors.append(f"File too large: {filename}")
                continue

            ext = os.path.splitext(filename)[1].lstrip(".").lower()
            if allowed_exts and ext not in allowed_exts:
                errors.append(f"Unsupported file type: {filename}")
                continue

            rel_path = f"files/{folder}/{filename}"
            save_path = self._safe_plugin_path(plugin_root_path, rel_path)
            if save_path is None:
                errors.append(f"Invalid path: {filename}")
                continue

            save_path.parent.mkdir(parents=True, exist_ok=True)
            await file.save(str(save_path))
            if save_path.is_file() and save_path.stat().st_size > MAX_FILE_BYTES:
                save_path.unlink()
                errors.append(f"File too large: {filename}")
                continue
            uploaded.append(rel_path)

        if not uploaded:
            raise ValueError(
                "Upload failed: " + ", ".join(errors) if errors else "Upload failed"
            )

        return {"uploaded": uploaded, "errors": errors}

    async def upload_config_file_from_dashboard_request(self, args, files) -> dict:
        return await self.upload_config_file(
            scope=args.get("scope"),
            name=args.get("name"),
            key_path=args.get("key"),
            files=files,
        )

    def delete_config_file(
        self,
        *,
        scope: str | None,
        name: str | None,
        rel_path: str | None,
    ) -> None:
        if not name:
            raise ValueError("Missing name parameter")
        if (scope or "plugin") != "plugin":
            raise ValueError(f"Unsupported scope: {scope}")

        rel_path = _normalize_rel_path(rel_path)
        if not rel_path or not rel_path.startswith("files/"):
            raise ValueError("Invalid path parameter")

        metadata = self.get_plugin_metadata_by_name(name)
        if not metadata:
            raise ValueError(f"Plugin {name} not found")

        plugin_root_path = self._plugin_root_path(name)
        target_path = self._safe_plugin_path(plugin_root_path, rel_path)
        if target_path is None:
            raise ValueError("Invalid path parameter")
        if target_path.is_file():
            target_path.unlink()

    def delete_config_file_from_dashboard_payload(
        self,
        *,
        scope: str | None,
        name: str | None,
        payload: object,
    ) -> str:
        data = payload if isinstance(payload, dict) else {}
        self.delete_config_file(
            scope=scope,
            name=name,
            rel_path=data.get("path"),
        )
        return "Deleted"

    def delete_config_file_from_dashboard_request(
        self,
        args,
        payload: object,
    ) -> str:
        return self.delete_config_file_from_dashboard_payload(
            scope=args.get("scope") or "plugin",
            name=args.get("name"),
            payload=payload,
        )

    def list_config_files(
        self,
        *,
        scope: str | None,
        name: str | None,
        key_path: str | None,
    ) -> dict:
        _, name, key_path, _, config = self.resolve_config_file_scope(
            scope=scope,
            name=name,
            key_path=key_path,
        )
        meta = _get_schema_item(getattr(config, "schema", None), key_path)
        if not meta or meta.get("type") != "file":
            raise ValueError("Config item not found or not file type")

        plugin_root_path = self._plugin_root_path(name)
        target_dir = self._safe_plugin_path(
            plugin_root_path,
            f"files/{_config_key_to_folder(key_path)}",
        )
        if target_dir is None:
            raise ValueError("Invalid path parameter")
        if not target_dir.exists() or not target_dir.is_dir():
            return {"files": []}

        files: list[str] = []
        for path in target_dir.rglob("*"):
            if not path.is_file():
                continue
            try:
                rel_path = path.relative_to(plugin_root_path).as_posix()
            except ValueError:
                continue
            if rel_path.startswith("files/"):
                files.append(rel_path)
        return {"files": files}

    def list_config_files_from_dashboard_args(self, args) -> dict:
        return self.list_config_files(
            scope=args.get("scope"),
            name=args.get("name"),
            key_path=args.get("key"),
        )

    @staticmethod
    def _allowed_file_extensions(meta: dict) -> list[str]:
        file_types = meta.get("file_types")
        if not isinstance(file_types, list):
            return []
        return [str(ext).lstrip(".").lower() for ext in file_types if str(ext).strip()]

    @staticmethod
    def _plugin_root_path(name: str) -> Path:
        storage_root_path = Path(get_astrbot_plugin_data_path()).resolve(strict=False)
        plugin_root_path = (storage_root_path / name).resolve(strict=False)
        try:
            plugin_root_path.relative_to(storage_root_path)
        except ValueError as exc:
            raise ValueError("Invalid name parameter") from exc
        return plugin_root_path

    @staticmethod
    def _safe_plugin_path(plugin_root_path: Path, rel_path: str) -> Path | None:
        target_path = (plugin_root_path / rel_path).resolve(strict=False)
        try:
            target_path.relative_to(plugin_root_path)
        except ValueError:
            return None
        return target_path


class BotConfigService:
    def __init__(self, core_lifecycle: AstrBotCoreLifecycle) -> None:
        self.core_lifecycle = core_lifecycle
        self.config = core_lifecycle.astrbot_config

    def list_bot_types(self) -> dict:
        bot_types = []
        for platform in platform_registry:
            bot_types.append(
                {
                    "type": platform.name,
                    "id": platform.id,
                    "description": platform.description,
                    "display_name": platform.adapter_display_name or platform.name,
                    "default_config": copy.deepcopy(platform.default_config_tmpl),
                    "schema": copy.deepcopy(platform.config_metadata or {}),
                    "support_streaming_message": platform.support_streaming_message,
                    "support_proactive_message": platform.support_proactive_message,
                }
            )
        return {"bot_types": bot_types}

    def list_bots(
        self, *, enabled: bool | None = None, type_: str | None = None
    ) -> dict:
        bots = []
        for bot in self.config.get("platform", []):
            if enabled is not None and bool(bot.get("enable", False)) != enabled:
                continue
            if type_ and bot.get("type") != type_:
                continue
            bots.append(copy.deepcopy(bot))
        return {"bots": bots}

    def get_bot(self, bot_id: str) -> dict:
        bot = self._find_bot(bot_id)
        if bot is None:
            raise ValueError(f"Bot {bot_id} not found")
        return {"bot": copy.deepcopy(bot)}

    def get_bot_stats(self) -> dict:
        return self.core_lifecycle.platform_manager.get_all_stats()

    async def create_bot(self, config: dict) -> None:
        bot_id = config.get("id")
        if not bot_id:
            raise ValueError("Bot config must have an 'id' field")
        if self._find_bot(bot_id) is not None:
            raise ValueError(f"Bot {bot_id} already exists")
        ensure_platform_webhook_config(config)
        self.config["platform"].append(config)
        save_config(self.config, self.config, is_core=True)
        await self.core_lifecycle.platform_manager.load_platform(config)

    async def update_bot(self, bot_id: str, config: dict) -> None:
        if config.get("id") != bot_id:
            raise ValueError("Bot id cannot be changed")
        ensure_platform_webhook_config(config)
        for idx, bot in enumerate(self.config.get("platform", [])):
            if bot.get("id") == bot_id:
                self.config["platform"][idx] = config
                save_config(self.config, self.config, is_core=True)
                await self.core_lifecycle.platform_manager.reload(config)
                return
        raise ValueError(f"Bot {bot_id} not found")

    async def set_bot_enabled(self, bot_id: str, enabled: bool) -> None:
        bot = self._find_bot(bot_id)
        if bot is None:
            raise ValueError(f"Bot {bot_id} not found")
        new_config = copy.deepcopy(bot)
        new_config["enable"] = enabled
        await self.update_bot(bot_id, new_config)

    async def delete_bot(self, bot_id: str) -> None:
        for idx, bot in enumerate(self.config.get("platform", [])):
            if bot.get("id") == bot_id:
                del self.config["platform"][idx]
                save_config(self.config, self.config, is_core=True)
                await self.core_lifecycle.platform_manager.terminate_platform(bot_id)
                return
        raise ValueError(f"Bot {bot_id} not found")

    def list_platforms_for_dashboard(self) -> dict:
        return {"platforms": self.list_bots()["bots"]}

    async def create_bot_from_dashboard_payload(self, payload: object) -> str:
        if not isinstance(payload, dict):
            raise ValueError("参数错误")
        await self.create_bot(payload)
        return "新增平台配置成功~"

    async def update_bot_from_dashboard_payload(self, payload: object) -> str:
        data = payload if isinstance(payload, dict) else {}
        origin_platform_id = data.get("id")
        new_config = data.get("config")
        if not origin_platform_id or not isinstance(new_config, dict):
            raise ValueError("参数错误")
        if origin_platform_id != new_config.get("id"):
            raise ValueError("机器人名称不允许修改")

        try:
            await self.update_bot(origin_platform_id, new_config)
        except ValueError as exc:
            if "not found" in str(exc):
                raise ValueError("未找到对应平台") from exc
            raise
        return "更新平台配置成功~"

    async def delete_bot_from_dashboard_payload(self, payload: object) -> str:
        data = payload if isinstance(payload, dict) else {}
        platform_id = data.get("id")
        if not platform_id:
            raise ValueError("缺少平台 ID")
        try:
            await self.delete_bot(str(platform_id))
        except ValueError as exc:
            if "not found" in str(exc):
                raise ValueError("未找到对应平台") from exc
            raise
        return "删除平台配置成功~"

    def _find_bot(self, bot_id: str) -> dict | None:
        for bot in self.config.get("platform", []):
            if bot.get("id") == bot_id:
                return bot
        return None


class ProviderConfigService:
    CAPABILITY_TO_PROVIDER_TYPE = {
        "chat": "chat_completion",
        "agent": "agent_runner",
        "stt": "speech_to_text",
        "tts": "text_to_speech",
        "embedding": "embedding",
        "rerank": "rerank",
    }

    def __init__(self, core_lifecycle: AstrBotCoreLifecycle) -> None:
        self.core_lifecycle = core_lifecycle
        self.config = core_lifecycle.astrbot_config
        self.provider_manager = core_lifecycle.provider_manager

    def get_provider_schema(self) -> dict:
        provider_metadata = ConfigMetadataI18n.convert_to_i18n_keys(
            {
                "provider_group": {
                    "metadata": {
                        "provider": CONFIG_METADATA_2["provider_group"]["metadata"][
                            "provider"
                        ]
                    }
                }
            }
        )
        config_schema = {
            "provider": provider_metadata["provider_group"]["metadata"]["provider"]
        }
        provider_default_tmpl = config_schema["provider"]["config_template"]
        for provider in provider_registry:
            if provider.default_config_tmpl:
                provider_default_tmpl[provider.type] = provider.default_config_tmpl
        providers = copy.deepcopy(self.config.get("provider", []))
        from astrbot.core.utils.llm_metadata import LLM_METADATAS

        model_metadata = {}
        for provider in providers:
            model_id = provider.get("model")
            if isinstance(model_id, str) and model_id in LLM_METADATAS:
                model_metadata[model_id] = LLM_METADATAS[model_id]
        return {
            "config_schema": config_schema,
            "providers": providers,
            "provider_sources": self.config.get("provider_sources", []),
            "model_metadata": model_metadata,
        }

    def list_provider_sources(self) -> dict:
        return {
            "provider_sources": copy.deepcopy(self.config.get("provider_sources", []))
        }

    def get_provider_source(self, source_id: str) -> dict:
        source = self._find_provider_source(source_id)
        if source is None:
            raise ValueError(f"Provider source {source_id} not found")
        return {"provider_source": copy.deepcopy(source)}

    async def upsert_provider_source(self, source_id: str, config: dict) -> None:
        config = copy.deepcopy(config)
        next_source_id = str(config.get("id") or source_id).strip()
        if not next_source_id:
            raise ValueError("Provider source config must have an 'id' field")
        config["id"] = next_source_id
        sources = self.config.setdefault("provider_sources", [])

        for source in sources:
            if source.get("id") == next_source_id and next_source_id != source_id:
                raise ValueError(
                    f"Provider source ID '{next_source_id}' exists already, please try another ID."
                )

        for idx, source in enumerate(sources):
            if source.get("id") == source_id:
                old_source_id = source.get("id") or source_id
                sources[idx] = config
                affected_providers = self._move_providers_to_source(
                    old_source_id,
                    next_source_id,
                )
                save_config(self.config, self.config, is_core=True)
                self.provider_manager.provider_sources_config = sources
                await self._reload_providers(affected_providers)
                return

        sources.append(config)
        save_config(self.config, self.config, is_core=True)
        self.provider_manager.provider_sources_config = sources

    async def delete_provider_source(self, source_id: str) -> None:
        sources = self.config.get("provider_sources", [])
        next_sources = [source for source in sources if source.get("id") != source_id]
        if len(next_sources) == len(sources):
            raise ValueError(f"Provider source {source_id} not found")
        self.config["provider_sources"] = next_sources
        await self.provider_manager.delete_provider(provider_source_id=source_id)
        save_config(self.config, self.config, is_core=True)
        self.provider_manager.provider_sources_config = next_sources

    async def upsert_provider_source_from_dashboard_payload(
        self, payload: object
    ) -> str:
        if not isinstance(payload, dict) or not payload:
            raise ValueError("缺少配置数据")

        new_source_config = payload.get("config") or payload
        original_id = payload.get("original_id")
        if not original_id:
            raise ValueError("缺少 original_id")
        if not isinstance(new_source_config, dict):
            raise ValueError("缺少或错误的配置数据")
        if not new_source_config.get("id"):
            new_source_config["id"] = original_id

        await self.upsert_provider_source(str(original_id), new_source_config)
        return "更新 provider source 成功"

    async def delete_provider_source_from_dashboard_payload(
        self, payload: object
    ) -> str:
        if not isinstance(payload, dict) or not payload:
            raise ValueError("缺少配置数据")

        provider_source_id = payload.get("id")
        if not provider_source_id:
            raise ValueError("缺少 provider_source_id")

        try:
            await self.delete_provider_source(str(provider_source_id))
        except ValueError as exc:
            if "not found" in str(exc):
                raise ValueError("未找到对应的 provider source") from exc
            raise
        return "删除 provider source 成功"

    async def list_provider_source_models(self, source_id: str) -> dict:
        source = self._find_provider_source(source_id)
        if source is None:
            raise ValueError(f"Provider source {source_id} not found")

        from astrbot.core.provider import Provider
        from astrbot.core.provider.register import provider_cls_map
        from astrbot.core.utils.llm_metadata import LLM_METADATAS

        provider_type = source.get("type")
        if not provider_type:
            raise ValueError("Provider source missing type")
        try:
            self.provider_manager.dynamic_import_provider(provider_type)
        except ImportError as exc:
            raise ValueError(f"动态导入提供商适配器失败: {exc!s}") from exc
        provider_metadata = provider_cls_map.get(provider_type)
        cls_type = provider_metadata.cls_type if provider_metadata else None
        if cls_type is None or not issubclass(cls_type, Provider):
            raise ValueError(f"Provider source {source_id} does not support model list")

        inst = cls_type(source, {})
        init_fn = getattr(inst, "initialize", None)
        if callable(init_fn):
            await run_maybe_async(init_fn)
        try:
            models = await inst.get_models()
            models = models or []
            return {
                "models": models,
                "provider_source_id": source_id,
                "model_metadata": {
                    model_id: LLM_METADATAS[model_id]
                    for model_id in models
                    if model_id in LLM_METADATAS
                },
            }
        finally:
            terminate_fn = getattr(inst, "terminate", None)
            if callable(terminate_fn):
                await run_maybe_async(terminate_fn)

    async def list_provider_source_models_for_dashboard(
        self,
        source_id: str | None,
    ) -> dict:
        if not source_id:
            raise ValueError("缺少参数 source_id")
        data = await self.list_provider_source_models(source_id)
        data.pop("provider_source_id", None)
        return data

    async def list_provider_models(self, provider_id: str) -> dict:
        from astrbot.core.provider import Provider
        from astrbot.core.utils.llm_metadata import LLM_METADATAS

        provider = self.provider_manager.inst_map.get(provider_id)
        if not provider:
            raise ValueError(f"未找到 ID 为 {provider_id} 的提供商")
        if not isinstance(provider, Provider):
            raise ValueError(f"提供商 {provider_id} 类型不支持获取模型列表")

        models = await provider.get_models()
        models = models or []
        return {
            "models": models,
            "provider_id": provider_id,
            "model_metadata": {
                model_id: LLM_METADATAS[model_id]
                for model_id in models
                if model_id in LLM_METADATAS
            },
        }

    async def list_provider_models_for_dashboard(
        self,
        provider_id: str | None,
    ) -> dict:
        if not provider_id:
            raise ValueError("缺少参数 provider_id")
        return await self.list_provider_models(provider_id)

    async def list_provider_models_from_dashboard_args(self, args) -> dict:
        return await self.list_provider_models_for_dashboard(
            args.get("provider_id", None)
        )

    async def get_embedding_dimension(self, provider_config: dict | None) -> dict:
        if not provider_config:
            raise ValueError("缺少参数 provider_config")

        from astrbot.core.provider.provider import EmbeddingProvider
        from astrbot.core.provider.register import provider_cls_map

        provider_type = provider_config.get("type")
        if not provider_type:
            raise ValueError("provider_config 缺少 type 字段")

        if provider_type not in provider_cls_map:
            try:
                dynamic_import_provider = getattr(
                    self.provider_manager,
                    "dynamic_import_provider",
                    None,
                )
                if not callable(dynamic_import_provider):
                    raise ImportError(provider_type)
                dynamic_import_provider(provider_type)
            except ImportError as exc:
                raise ValueError(
                    "提供商适配器加载失败，请检查提供商类型配置或查看服务端日志"
                ) from exc

        provider_metadata = provider_cls_map.get(provider_type)
        cls_type = provider_metadata.cls_type if provider_metadata else None
        if not cls_type:
            raise ValueError(f"无法找到 {provider_type} 的类")

        inst = cls_type(provider_config, {})
        try:
            init_fn = getattr(inst, "initialize", None)
            if callable(init_fn):
                await run_maybe_async(init_fn)

            if not isinstance(inst, EmbeddingProvider):
                raise ValueError("提供商不是 EmbeddingProvider 类型")

            vec = await inst.get_embedding("echo")
            dim = len(vec)
            logger.info(
                f"检测到 {provider_config.get('id', 'unknown')} 的嵌入向量维度为 {dim}",
            )
            return {"embedding_dimensions": dim}
        finally:
            terminate_fn = getattr(inst, "terminate", None)
            if callable(terminate_fn):
                await run_maybe_async(terminate_fn)

    async def get_embedding_dimension_from_dashboard_payload(
        self,
        payload: object,
    ) -> dict:
        data = payload if isinstance(payload, dict) else {}
        return await self.get_embedding_dimension(data.get("provider_config"))

    def list_providers(
        self,
        *,
        capability: str | None = None,
        source_id: str | None = None,
        enabled: bool | None = None,
    ) -> dict:
        from astrbot.core.utils.llm_metadata import LLM_METADATAS

        provider_type = self._resolve_provider_type(capability)
        providers = []
        model_metadata = {}
        source_provider_type = {
            source["id"]: source.get("provider_type", "chat_completion")
            for source in self.provider_manager.provider_sources_config
        }
        for provider in self.provider_manager.providers_config:
            if source_id and provider.get("provider_source_id") != source_id:
                continue
            if enabled is not None and bool(provider.get("enable", False)) != enabled:
                continue
            effective_type = provider.get("provider_type")
            if not effective_type and provider.get("provider_source_id"):
                effective_type = source_provider_type.get(
                    provider.get("provider_source_id"), "chat_completion"
                )
            if provider_type and effective_type != provider_type:
                continue
            if provider.get("provider_source_id"):
                provider_response = self.provider_manager.get_merged_provider_config(
                    provider
                )
            else:
                provider_response = copy.deepcopy(provider)
            model_id = provider_response.get("model")
            if isinstance(model_id, str) and model_id in LLM_METADATAS:
                model_metadata[model_id] = LLM_METADATAS[model_id]
            providers.append(provider_response)
        return {"providers": providers, "model_metadata": model_metadata}

    def list_providers_for_dashboard_types(
        self, provider_type: str | None
    ) -> list[dict]:
        if not provider_type:
            raise ValueError("缺少参数 provider_type")

        provider_list = []
        seen_ids = set()
        for item in provider_type.split(","):
            providers = self.list_providers(capability=item)["providers"]
            for provider in providers:
                provider_id = provider.get("id")
                if provider_id in seen_ids:
                    continue
                seen_ids.add(provider_id)
                provider_list.append(provider)
        return provider_list

    def list_providers_from_dashboard_args(self, args) -> list[dict]:
        return self.list_providers_for_dashboard_types(args.get("provider_type", None))

    def get_provider(self, provider_id: str, *, merged: bool = False) -> dict:
        provider = self.provider_manager.get_provider_config_by_id(
            provider_id,
            merged=merged,
        )
        if provider is None:
            raise ValueError(f"Provider {provider_id} not found")
        provider_response = copy.deepcopy(provider)
        from astrbot.core.utils.llm_metadata import LLM_METADATAS

        model_id = provider_response.get("model")
        model_metadata = {}
        if isinstance(model_id, str) and model_id in LLM_METADATAS:
            model_metadata[model_id] = LLM_METADATAS[model_id]
        return {"provider": provider_response, "model_metadata": model_metadata}

    async def create_provider(self, config: dict, source_id: str | None = None) -> None:
        config = copy.deepcopy(config)
        if source_id:
            config["provider_source_id"] = source_id
        await self.provider_manager.create_provider(config)

    async def update_provider(self, provider_id: str, config: dict) -> None:
        if not config.get("id"):
            config["id"] = provider_id
        await self.provider_manager.update_provider(provider_id, config)

    async def set_provider_enabled(self, provider_id: str, enabled: bool) -> None:
        provider = self.provider_manager.get_provider_config_by_id(provider_id)
        if provider is None:
            raise ValueError(f"Provider {provider_id} not found")
        provider["enable"] = enabled
        await self.provider_manager.update_provider(provider_id, provider)

    async def delete_provider(self, provider_id: str) -> None:
        await self.provider_manager.delete_provider(provider_id=provider_id)

    async def create_provider_from_dashboard_payload(self, payload: object) -> str:
        if not isinstance(payload, dict):
            raise ValueError("参数错误")
        await self.create_provider(payload)
        return "新增服务提供商配置成功"

    async def update_provider_from_dashboard_payload(self, payload: object) -> str:
        data = payload if isinstance(payload, dict) else {}
        origin_provider_id = data.get("id")
        new_config = data.get("config")
        if not origin_provider_id or not isinstance(new_config, dict):
            raise ValueError("参数错误")

        await self.update_provider(origin_provider_id, new_config)
        return "更新成功，已经实时生效~"

    async def delete_provider_from_dashboard_payload(self, payload: object) -> str:
        data = payload if isinstance(payload, dict) else {}
        provider_id = data.get("id", "")
        if not provider_id:
            raise ValueError("缺少参数 id")

        await self.delete_provider(provider_id)
        return "删除成功，已经实时生效。"

    async def test_provider(self, provider_id: str) -> dict:
        target = self.provider_manager.inst_map.get(provider_id)
        if not target:
            raise ValueError(f"Provider {provider_id} not found")
        meta = target.meta()
        provider_type = getattr(meta, "provider_type", None)
        result = {
            "id": getattr(meta, "id", provider_id),
            "model": getattr(meta, "model", None),
            "type": getattr(provider_type, "value", None),
            "name": provider_id,
            "status": "unavailable",
            "error": None,
        }
        try:
            await target.test()
            result["status"] = "available"
        except Exception as exc:
            result["error"] = str(exc)
        return result

    async def test_provider_from_dashboard_args(self, args) -> dict:
        provider_id = args.get("id")
        if not provider_id:
            raise ValueError("Missing provider_id parameter")
        logger.info(f"API call: /config/provider/check_one id={provider_id}")
        return await self.test_provider(provider_id)

    def _find_provider_source(self, source_id: str) -> dict | None:
        for source in self.config.get("provider_sources", []):
            if source.get("id") == source_id:
                return source
        return None

    def _resolve_provider_type(self, capability: str | None) -> str | None:
        if not capability:
            return None
        return self.CAPABILITY_TO_PROVIDER_TYPE.get(capability, capability)

    async def _reload_providers_for_source(self, source_id: str) -> None:
        await self._reload_providers(
            [
                provider
                for provider in list(self.config.get("provider", []))
                if provider.get("provider_source_id") == source_id
            ]
        )

    def _move_providers_to_source(
        self,
        old_source_id: str,
        next_source_id: str,
    ) -> list[dict]:
        affected_providers = []
        for provider in self.config.get("provider", []):
            if provider.get("provider_source_id") == old_source_id:
                provider["provider_source_id"] = next_source_id
                affected_providers.append(provider)
        return affected_providers

    async def _reload_providers(self, providers: list[dict]) -> None:
        reload_fn = getattr(self.provider_manager, "reload", None)
        if not callable(reload_fn):
            return
        for provider in providers:
            await run_maybe_async(lambda provider=provider: reload_fn(provider))