fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-15 17:30:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(commands): improve type safety and add validation

Browse Source 
alter_cmd.py:
- Add explicit type annotations for alter_cmd_cfg
- Rename variables for clarity (scene_num -> scene_index, cmd_type -> permission_type)
- Add validation for permission_type parameter

conversation.py:
- Add type annotations and improve command handling

persona.py:
- Clean up type annotations

provider.py:
- Add type annotations and improve provider command handling

setunset.py:
- Add type annotations for configuration operations
This commit is contained in:
LIghtJUNction
2026-03-31 20:16:02 +08:00
parent 65841684de
commit 806c3dfade
5 changed files with 120 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
astrbot/builtin_stars/builtin_commands/commands/alter_cmd.py
View File

@@ -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}｡",
),
)

40
astrbot/builtin_stars/builtin_commands/commands/conversation.py
View File

@@ -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", {})

6
astrbot/builtin_stars/builtin_commands/commands/persona.py
View File

@@ -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,

49
astrbot/builtin_stars/builtin_commands/commands/provider.py
View File

@@ -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,

18
astrbot/builtin_stars/builtin_commands/commands/setunset.py
View File

@@ -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(