diff --git a/astrbot/builtin_stars/builtin_commands/commands/alter_cmd.py b/astrbot/builtin_stars/builtin_commands/commands/alter_cmd.py index 3ed2ac5ed..7d6e2a25a 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/alter_cmd.py +++ b/astrbot/builtin_stars/builtin_commands/commands/alter_cmd.py @@ -18,7 +18,9 @@ class AlterCmdCommands(CommandParserMixin): """更新reset命令在特定场景下的权限设置""" from astrbot.api import sp - alter_cmd_cfg = await sp.global_get("alter_cmd", {}) or {} + alter_cmd_cfg: dict[str, dict[str, dict[str, str]]] = ( + await sp.global_get("alter_cmd", {}) or {} + ) plugin_cfg = alter_cmd_cfg.get("astrbot", {}) reset_cfg = plugin_cfg.get("reset", {}) reset_cfg[scene_key] = perm_type @@ -47,7 +49,9 @@ class AlterCmdCommands(CommandParserMixin): if cmd_name == "reset" and cmd_type == "config": from astrbot.api import sp - alter_cmd_cfg = await sp.global_get("alter_cmd", {}) or {} + alter_cmd_cfg: dict[str, dict[str, dict[str, str]]] = ( + await sp.global_get("alter_cmd", {}) or {} + ) plugin_ = alter_cmd_cfg.get("astrbot", {}) reset_cfg = plugin_.get("reset", {}) @@ -86,8 +90,8 @@ class AlterCmdCommands(CommandParserMixin): ) return - scene_num = int(scene_num) - scene = RstScene.from_index(scene_num) + scene_index = int(scene_num) + scene = RstScene.from_index(scene_index) scene_key = scene.key await self.update_reset_permission(scene_key, perm_type) @@ -107,7 +111,12 @@ class AlterCmdCommands(CommandParserMixin): # 查找指令 cmd_name = " ".join(token.tokens[1:-1]) - cmd_type = token.get(-1) + permission_type = token.get(-1) + if permission_type not in ["admin", "member"]: + await event.send( + MessageChain().message("指令类型错误,可选类型有 admin, member"), + ) + return found_command = None cmd_group = False for handler in star_handlers_registry: @@ -131,20 +140,25 @@ class AlterCmdCommands(CommandParserMixin): from astrbot.api import sp - alter_cmd_cfg = await sp.global_get("alter_cmd", {}) or {} - plugin_ = alter_cmd_cfg.get(found_plugin.name, {}) + stored_alter_cmd_cfg: dict[str, dict[str, dict[str, str]]] = ( + await sp.global_get("alter_cmd", {}) or {} + ) + if found_plugin.name is None: + await event.send(MessageChain().message("未找到指令对应的插件名称")) + return + plugin_ = stored_alter_cmd_cfg.get(found_plugin.name, {}) cfg = plugin_.get(found_command.handler_name, {}) - cfg["permission"] = cmd_type + cfg["permission"] = permission_type plugin_[found_command.handler_name] = cfg - alter_cmd_cfg[found_plugin.name] = plugin_ + stored_alter_cmd_cfg[found_plugin.name] = plugin_ - await sp.global_put("alter_cmd", alter_cmd_cfg) + await sp.global_put("alter_cmd", stored_alter_cmd_cfg) # 注入权限过滤器 found_permission_filter = False for filter_ in found_command.event_filters: if isinstance(filter_, PermissionTypeFilter): - if cmd_type == "admin": + if permission_type == "admin": from astrbot.api.event import filter filter_.permission_type = filter.PermissionType.ADMIN @@ -161,13 +175,13 @@ class AlterCmdCommands(CommandParserMixin): 0, PermissionTypeFilter( filter.PermissionType.ADMIN - if cmd_type == "admin" + if permission_type == "admin" else filter.PermissionType.MEMBER, ), ) cmd_group_str = "指令组" if cmd_group else "指令" await event.send( MessageChain().message( - f"已将「{cmd_name}」{cmd_group_str} 的权限级别调整为 {cmd_type}。", + f"已将「{cmd_name}」{cmd_group_str} 的权限级别调整为 {permission_type}。", ), ) diff --git a/astrbot/builtin_stars/builtin_commands/commands/conversation.py b/astrbot/builtin_stars/builtin_commands/commands/conversation.py index e52a3becc..a239f0cfe 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/conversation.py +++ b/astrbot/builtin_stars/builtin_commands/commands/conversation.py @@ -1,4 +1,5 @@ import datetime +from typing import Any, TypedDict, cast from astrbot.api import sp, star from astrbot.api.event import AstrMessageEvent, MessageEventResult @@ -21,6 +22,41 @@ THIRD_PARTY_AGENT_RUNNER_KEY = { THIRD_PARTY_AGENT_RUNNER_STR = ", ".join(THIRD_PARTY_AGENT_RUNNER_KEY.keys()) +class ResetPermissionConfig(TypedDict, total=False): + group_unique_on: str + group_unique_off: str + private: str + + +class AlterCmdPluginConfig(TypedDict, total=False): + reset: ResetPermissionConfig + + +def _normalize_alter_cmd_config(value: object) -> dict[str, AlterCmdPluginConfig]: + if not isinstance(value, dict): + return {} + + config: dict[str, AlterCmdPluginConfig] = {} + for plugin_name, raw_plugin_config in value.items(): + if not isinstance(plugin_name, str) or not isinstance(raw_plugin_config, dict): + continue + + plugin_config: AlterCmdPluginConfig = cast(AlterCmdPluginConfig, {}) + raw_reset = cast(dict[str, Any], raw_plugin_config).get("reset") + if isinstance(raw_reset, dict): + reset_config: ResetPermissionConfig = cast(ResetPermissionConfig, {}) + for key in ("group_unique_on", "group_unique_off", "private"): + permission = raw_reset.get(key) + if isinstance(permission, str): + reset_config[key] = permission + if reset_config: + plugin_config["reset"] = reset_config + + config[plugin_name] = plugin_config + + return config + + class ConversationCommands: def __init__(self, context: star.Context) -> None: self.context = context @@ -48,7 +84,9 @@ class ConversationCommands: scene = RstScene.get_scene(is_group, is_unique_session) - alter_cmd_cfg = await sp.get_async("global", "global", "alter_cmd", {}) or {} + alter_cmd_cfg = _normalize_alter_cmd_config( + await sp.get_async("global", "global", "alter_cmd", {}) + ) plugin_config = alter_cmd_cfg.get("astrbot", {}) reset_cfg = plugin_config.get("reset", {}) diff --git a/astrbot/builtin_stars/builtin_commands/commands/persona.py b/astrbot/builtin_stars/builtin_commands/commands/persona.py index b39eae3be..b0d7ecd31 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/persona.py +++ b/astrbot/builtin_stars/builtin_commands/commands/persona.py @@ -154,7 +154,7 @@ class PersonaCommands: message.set_result(MessageEventResult().message("请输入人格情景名")) return ps = parts[2].strip() - if persona := next( + if persona_info := next( builtins.filter( lambda persona: persona["name"] == ps, self.context.provider_manager.personas, @@ -162,7 +162,7 @@ class PersonaCommands: None, ): msg = f"人格{ps}的详细信息:\n" - msg += f"{persona['prompt']}\n" + msg += f"{persona_info['prompt']}\n" else: msg = f"人格{ps}不存在" message.set_result(MessageEventResult().message(msg)) @@ -186,7 +186,7 @@ class PersonaCommands: ), ) return - if persona := next( + if persona_info := next( builtins.filter( lambda persona: persona["name"] == ps, self.context.provider_manager.personas, diff --git a/astrbot/builtin_stars/builtin_commands/commands/provider.py b/astrbot/builtin_stars/builtin_commands/commands/provider.py index 41ea7986a..4943462d4 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/provider.py +++ b/astrbot/builtin_stars/builtin_commands/commands/provider.py @@ -4,12 +4,12 @@ import asyncio import time from collections.abc import Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Literal, Protocol, TypeAlias, TypedDict from astrbot import logger from astrbot.api import star from astrbot.api.event import AstrMessageEvent, MessageEventResult -from astrbot.core.provider.entities import ProviderType +from astrbot.core.provider.entities import ProviderMeta, ProviderType from astrbot.core.utils.error_redaction import safe_error if TYPE_CHECKING: @@ -31,6 +31,22 @@ class _ModelLookupConfig: max_concurrency: int +class ListedProvider(Protocol): + def meta(self) -> ProviderMeta: ... + + async def test(self) -> None: ... + + +class _ProviderDisplayEntry(TypedDict): + type: Literal["llm", "tts", "stt"] + info: str + mark: str + provider: ListedProvider + + +ReachabilityCheckResult: TypeAlias = tuple[bool, str | None, str | None] | BaseException + + class _ModelCache: def __init__(self) -> None: self._store: dict[tuple[str, str | None], tuple[float, list[str]]] = {} @@ -260,7 +276,7 @@ class ProviderCommands: def _log_reachability_failure( self, - provider, + provider: ListedProvider, provider_capability_type: ProviderType | None, err_code: str, err_reason: str, @@ -275,7 +291,9 @@ class ProviderCommands: err_reason, ) - async def _test_provider_capability(self, provider): + async def _test_provider_capability( + self, provider: ListedProvider + ) -> tuple[bool, str | None, str | None]: """测试单个 provider 的可用性""" meta = provider.meta() provider_capability_type = meta.provider_type @@ -395,7 +413,9 @@ class ProviderCommands: stts = self.context.get_all_stt_providers() # 构造待检测列表: [(provider, type_label), ...] - all_providers = [] + all_providers: list[ + tuple[ListedProvider, Literal["llm", "tts", "stt"]] + ] = [] all_providers.extend([(p, "llm") for p in llms]) all_providers.extend([(p, "tts") for p in ttss]) all_providers.extend([(p, "stt") for p in stts]) @@ -408,7 +428,7 @@ class ProviderCommands: "正在进行提供商可达性测试,请稍候..." ) ) - check_results = await asyncio.gather( + check_results: list[ReachabilityCheckResult] = await asyncio.gather( *[self._test_provider_capability(p) for p, _ in all_providers], return_exceptions=True, ) @@ -417,11 +437,12 @@ class ProviderCommands: check_results = [None for _ in all_providers] # 整合结果 - display_data = [] + display_data: list[_ProviderDisplayEntry] = [] for (p, p_type), reachable in zip(all_providers, check_results): meta = p.meta() id_ = meta.id error_code = None + reachable_flag: bool | None if isinstance(reachable, asyncio.CancelledError): raise reachable @@ -438,7 +459,7 @@ class ProviderCommands: elif isinstance(reachable, tuple): reachable_flag, error_code, _ = reachable else: - reachable_flag = reachable + reachable_flag = None # 根据类型构建显示名称 if p_type == "llm": @@ -519,8 +540,8 @@ class ProviderCommands: if idx2 > len(self.context.get_all_tts_providers()) or idx2 < 1: event.set_result(MessageEventResult().message("无效的提供商序号。")) return - provider = self.context.get_all_tts_providers()[idx2 - 1] - id_ = provider.meta().id + tts_provider = self.context.get_all_tts_providers()[idx2 - 1] + id_ = tts_provider.meta().id await self.context.provider_manager.set_provider( provider_id=id_, provider_type=ProviderType.TEXT_TO_SPEECH, @@ -534,8 +555,8 @@ class ProviderCommands: if idx2 > len(self.context.get_all_stt_providers()) or idx2 < 1: event.set_result(MessageEventResult().message("无效的提供商序号。")) return - provider = self.context.get_all_stt_providers()[idx2 - 1] - id_ = provider.meta().id + stt_provider = self.context.get_all_stt_providers()[idx2 - 1] + id_ = stt_provider.meta().id await self.context.provider_manager.set_provider( provider_id=id_, provider_type=ProviderType.SPEECH_TO_TEXT, @@ -546,8 +567,8 @@ class ProviderCommands: if idx > len(self.context.get_all_providers()) or idx < 1: event.set_result(MessageEventResult().message("无效的提供商序号。")) return - provider = self.context.get_all_providers()[idx - 1] - id_ = provider.meta().id + llm_provider = self.context.get_all_providers()[idx - 1] + id_ = llm_provider.meta().id await self.context.provider_manager.set_provider( provider_id=id_, provider_type=ProviderType.CHAT_COMPLETION, diff --git a/astrbot/builtin_stars/builtin_commands/commands/setunset.py b/astrbot/builtin_stars/builtin_commands/commands/setunset.py index 47d7230df..424393664 100644 --- a/astrbot/builtin_stars/builtin_commands/commands/setunset.py +++ b/astrbot/builtin_stars/builtin_commands/commands/setunset.py @@ -2,6 +2,16 @@ from astrbot.api import sp, star from astrbot.api.event import AstrMessageEvent, MessageEventResult +def _normalize_session_variables(value: object) -> dict[str, str]: + if not isinstance(value, dict): + return {} + return { + key: value + for key, value in value.items() + if isinstance(key, str) and isinstance(value, str) + } + + class SetUnsetCommands: def __init__(self, context: star.Context) -> None: self.context = context @@ -9,7 +19,9 @@ class SetUnsetCommands: async def set_variable(self, event: AstrMessageEvent, key: str, value: str) -> None: """设置会话变量""" uid = event.unified_msg_origin - session_var = await sp.session_get(uid, "session_variables", {}) or {} + session_var = _normalize_session_variables( + await sp.session_get(uid, "session_variables", {}) + ) session_var[key] = value await sp.session_put(uid, "session_variables", session_var) @@ -22,7 +34,9 @@ class SetUnsetCommands: async def unset_variable(self, event: AstrMessageEvent, key: str) -> None: """移除会话变量""" uid = event.unified_msg_origin - session_var = await sp.session_get(uid, "session_variables", {}) or {} + session_var = _normalize_session_variables( + await sp.session_get(uid, "session_variables", {}) + ) if key not in session_var: event.set_result(