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(star): improve type safety and code quality

Browse Source 
session_llm_manager.py:
- Add SessionServiceConfig TypedDict for type safety
- Add _normalize_session_service_config helper for config validation

session_plugin_manager.py:
- Add type annotations and improve code structure

star_manager.py:
- Improve type annotations and code quality

Other star modules:
- Minor improvements
This commit is contained in:
LIghtJUNction
2026-03-31 20:17:13 +08:00
parent 585ffb3982
commit 7b16068a04
7 changed files with 101 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
astrbot/core/star/command_management.py
View File

@@ -156,7 +156,9 @@ async def update_command_permission(
raise ValueError("未找到指令所属插件")
# 1. Update Persistent Config (alter_cmd)
alter_cmd_cfg = await sp.global_get("alter_cmd", {}) or {}
alter_cmd_cfg: dict[str, dict[str, Any]] = (
await sp.global_get("alter_cmd", {}) or {}
)
plugin_ = alter_cmd_cfg.get(found_plugin.name, {})
cfg = plugin_.get(handler.handler_name, {})
cfg["permission"] = permission_type
@@ -487,7 +489,7 @@ def _set_filter_aliases(
filter_ref: CommandFilter | CommandGroupFilter,
aliases: list[str],
) -> None:
current_aliases = getattr(filter_ref, "alias", set())
current_aliases: set[str] = getattr(filter_ref, "alias", set())
if set(aliases) == current_aliases:
return
setattr(filter_ref, "alias", set(aliases))

6
astrbot/core/star/context.py
View File

@@ -40,7 +40,7 @@ from astrbot.core.subagent_orchestrator import SubAgentOrchestrator
from .filter.command import CommandFilter
from .filter.regex import RegexFilter
from .star import StarMetadata, star_map, star_registry
from .star import StarMetadata, star_registry
from .star_handler import EventType, StarHandlerMetadata, star_handlers_registry
logger = logging.getLogger("astrbot")
@@ -303,7 +303,7 @@ class Context:
Note:
注册的工具默认是激活状态｡
"""
return self.provider_manager.llm_tools.activate_llm_tool(name, star_map)
return self.provider_manager.llm_tools.activate_llm_tool(name)
def deactivate_llm_tool(self, name: str) -> bool:
"""停用一个已经注册的函数调用工具｡
@@ -564,6 +564,7 @@ class Context:
and ADAPTER_NAME_2_TYPE[name] & platform_type
):
return platform
return None
def get_platform_inst(self, platform_id: str) -> Platform | None:
"""获取指定 ID 的平台适配器实例｡
@@ -580,6 +581,7 @@ class Context:
for platform in self.platform_manager.platform_insts:
if platform.meta().id == platform_id:
return platform
return None
def get_db(self) -> BaseDatabase:
"""获取 AstrBot 数据库｡

2
astrbot/core/star/filter/command.py
View File

@@ -96,7 +96,7 @@ class CommandFilter(HandlerFilter):
param_type: dict[str, type | Any],
) -> dict[str, Any]:
"""将参数列表 params 根据 param_type 转换为参数字典｡"""
result = {}
result: dict[str, Any] = {}
param_items = list(param_type.items())
for i, (param_name, param_type_or_default_val) in enumerate(param_items):
is_greedy = param_type_or_default_val is GreedyStr

43
astrbot/core/star/session_llm_manager.py
View File

@@ -1,9 +1,37 @@
"""会话服务管理器 - 负责管理每个会话的LLM､TTS等服务的启停状态"""
from typing import TypedDict
from astrbot.core import logger, sp
from astrbot.core.platform.astr_message_event import AstrMessageEvent
class SessionServiceConfig(TypedDict, total=False):
llm_enabled: bool
tts_enabled: bool
session_enabled: bool
def _normalize_session_service_config(value: object) -> SessionServiceConfig:
if not isinstance(value, dict):
return {}
config: SessionServiceConfig = {}
llm_enabled = value.get("llm_enabled")
if isinstance(llm_enabled, bool):
config["llm_enabled"] = llm_enabled
tts_enabled = value.get("tts_enabled")
if isinstance(tts_enabled, bool):
config["tts_enabled"] = tts_enabled
session_enabled = value.get("session_enabled")
if isinstance(session_enabled, bool):
config["session_enabled"] = session_enabled
return config
class SessionServiceManager:
"""管理会话级别的服务启停状态,包括LLM和TTS"""
@@ -23,14 +51,13 @@ class SessionServiceManager:
"""
# 获取会话服务配置
session_services = (
session_services = _normalize_session_service_config(
await sp.get_async(
scope="umo",
scope_id=session_id,
key="session_service_config",
default={},
)
or {}
)
# 如果配置了该会话的LLM状态,返回该状态
@@ -50,14 +77,13 @@ class SessionServiceManager:
enabled: True表示启用,False表示禁用
"""
session_config = (
session_config = _normalize_session_service_config(
await sp.get_async(
scope="umo",
scope_id=session_id,
key="session_service_config",
default={},
)
or {}
)
session_config["llm_enabled"] = enabled
await sp.put_async(
@@ -97,14 +123,13 @@ class SessionServiceManager:
"""
# 获取会话服务配置
session_services = (
session_services = _normalize_session_service_config(
await sp.get_async(
scope="umo",
scope_id=session_id,
key="session_service_config",
default={},
)
or {}
)
# 如果配置了该会话的TTS状态,返回该状态
@@ -124,14 +149,13 @@ class SessionServiceManager:
enabled: True表示启用,False表示禁用
"""
session_config = (
session_config = _normalize_session_service_config(
await sp.get_async(
scope="umo",
scope_id=session_id,
key="session_service_config",
default={},
)
or {}
)
session_config["tts_enabled"] = enabled
await sp.put_async(
@@ -175,14 +199,13 @@ class SessionServiceManager:
"""
# 获取会话服务配置
session_services = (
session_services = _normalize_session_service_config(
await sp.get_async(
scope="umo",
scope_id=session_id,
key="session_service_config",
default={},
)
or {}
)
# 如果配置了该会话的整体状态,返回该状态

40
astrbot/core/star/session_plugin_manager.py
View File

@@ -1,9 +1,43 @@
"""会话插件管理器 - 负责管理每个会话的插件启停状态"""
from typing import TypedDict
from astrbot.core import logger, sp
from astrbot.core.platform.astr_message_event import AstrMessageEvent
class SessionPluginSettings(TypedDict, total=False):
enabled_plugins: list[str]
disabled_plugins: list[str]
def _normalize_session_plugin_config(value: object) -> dict[str, SessionPluginSettings]:
if not isinstance(value, dict):
return {}
config: dict[str, SessionPluginSettings] = {}
for session_id, raw_settings in value.items():
if not isinstance(session_id, str) or not isinstance(raw_settings, dict):
continue
settings: SessionPluginSettings = {}
enabled_plugins = raw_settings.get("enabled_plugins")
if isinstance(enabled_plugins, list) and all(
isinstance(plugin_name, str) for plugin_name in enabled_plugins
):
settings["enabled_plugins"] = enabled_plugins
disabled_plugins = raw_settings.get("disabled_plugins")
if isinstance(disabled_plugins, list) and all(
isinstance(plugin_name, str) for plugin_name in disabled_plugins
):
settings["disabled_plugins"] = disabled_plugins
config[session_id] = settings
return config
class SessionPluginManager:
"""管理会话级别的插件启停状态"""
@@ -23,14 +57,13 @@ class SessionPluginManager:
"""
# 获取会话插件配置
session_plugin_config = (
session_plugin_config = _normalize_session_plugin_config(
await sp.get_async(
scope="umo",
scope_id=session_id,
key="session_plugin_config",
default={},
)
or {}
)
session_config = session_plugin_config.get(session_id, {})
@@ -68,14 +101,13 @@ class SessionPluginManager:
session_id = event.unified_msg_origin
filtered_handlers = []
session_plugin_config = (
session_plugin_config = _normalize_session_plugin_config(
await sp.get_async(
scope="umo",
scope_id=session_id,
key="session_plugin_config",
default={},
)
or {}
)
session_config = session_plugin_config.get(session_id, {})
disabled_plugins = session_config.get("disabled_plugins", [])

33
astrbot/core/star/star_manager.py
View File

@@ -12,6 +12,7 @@ import sys
import tempfile
import traceback
from types import ModuleType
from typing import Any, cast
import anyio
import yaml
@@ -193,7 +194,7 @@ class PluginManager:
self._pm_lock = asyncio.Lock()
"""StarManager操作互斥锁"""
self.failed_plugin_dict = {}
self.failed_plugin_dict: dict[str, Any] = {}
"""加载失败插件的信息,用于后续可能的热重载"""
self.failed_plugin_info = ""
@@ -398,7 +399,7 @@ class PluginManager:
os.path.join(plugin_path, "metadata.yaml"),
encoding="utf-8",
) as f:
metadata = yaml.safe_load(f)
metadata = cast(dict[str, Any], yaml.safe_load(f))
elif plugin_obj and hasattr(plugin_obj, "info"):
# 使用 info() 函数
metadata = plugin_obj.info()
@@ -463,7 +464,7 @@ class PluginManager:
raise Exception("未找到 metadata.yaml,无法获取插件目录名｡")
with open(metadata_path, encoding="utf-8") as f:
metadata = yaml.safe_load(f)
metadata = cast(dict[str, Any], yaml.safe_load(f))
if not isinstance(metadata, dict):
raise Exception("metadata.yaml 格式错误｡")
@@ -780,9 +781,13 @@ class PluginManager:
- error_message (str|None): 错误信息,成功时为 None
"""
inactivated_plugins = await sp.global_get("inactivated_plugins", []) or []
inactivated_llm_tools = await sp.global_get("inactivated_llm_tools", []) or []
alter_cmd = await sp.global_get("alter_cmd", {}) or {}
inactivated_plugins: list[Any] = (
await sp.global_get("inactivated_plugins", []) or []
)
inactivated_llm_tools: list[Any] = (
await sp.global_get("inactivated_llm_tools", []) or []
)
alter_cmd: dict[str, Any] = await sp.global_get("alter_cmd", {}) or {}
plugin_modules = self._get_plugin_modules()
if plugin_modules is None:
@@ -1587,12 +1592,14 @@ class PluginManager:
await self._terminate_plugin(plugin)
# 加入到 shared_preferences 中
inactivated_plugins: list = await sp.global_get("inactivated_plugins", [])
inactivated_plugins: list[Any] = (
await sp.global_get("inactivated_plugins", []) or []
)
if plugin.module_path not in inactivated_plugins:
inactivated_plugins.append(plugin.module_path)
inactivated_llm_tools: list = list(
set(await sp.global_get("inactivated_llm_tools", [])),
inactivated_llm_tools: list[Any] = list(
set(await sp.global_get("inactivated_llm_tools", []) or []),
) # 后向兼容
# 禁用插件启用的 llm_tool
@@ -1682,8 +1689,12 @@ class PluginManager:
plugin = self.context.get_registered_star(plugin_name)
if plugin is None:
raise Exception(f"插件 {plugin_name} 不存在｡")
inactivated_plugins: list = await sp.global_get("inactivated_plugins", [])
inactivated_llm_tools: list = await sp.global_get("inactivated_llm_tools", [])
inactivated_plugins: list[Any] = (
await sp.global_get("inactivated_plugins", []) or []
)
inactivated_llm_tools: list[Any] = (
await sp.global_get("inactivated_llm_tools", []) or []
)
if plugin.module_path in inactivated_plugins:
inactivated_plugins.remove(plugin.module_path)
await sp.global_put("inactivated_plugins", inactivated_plugins)

4
astrbot/core/star/updator.py
View File

@@ -26,9 +26,7 @@ class PluginUpdator(RepoZipUpdator):
return plugin_path
async def update( # type: ignore[invalid-method-override]
self, plugin: StarMetadata, proxy=""
) -> str:
async def update(self, plugin: StarMetadata, proxy="") -> str:
repo_url = plugin.repo
if not repo_url: