From 7359cd174f45be28b11c1d9b9d53b63e1b3d4977 Mon Sep 17 00:00:00 2001 From: Rat0323 Date: Fri, 5 Jun 2026 09:46:20 +0800 Subject: [PATCH 1/5] fix(core): make StarTools.get_data_dir robust to avoid inspect-stack crash under submodules/debug environments --- astrbot/core/star/star_tools.py | 35 +++++++++++++++++++++++++++++---- 1 file changed, 31 insertions(+), 4 deletions(-) diff --git a/astrbot/core/star/star_tools.py b/astrbot/core/star/star_tools.py index fe5563b7dd..6baa369b3c 100644 --- a/astrbot/core/star/star_tools.py +++ b/astrbot/core/star/star_tools.py @@ -291,12 +291,39 @@ def get_data_dir(cls, plugin_name: str | None = None) -> Path: if not module: raise RuntimeError("无法获取调用者模块信息") + # 1. Try direct match in star_map metadata = star_map.get(module.__name__, None) - if not metadata: - raise RuntimeError(f"无法获取模块 {module.__name__} 的元数据信息") - - plugin_name = metadata.name + # 2. Try prefix match for submodule calls + if not metadata and "." in module.__name__: + caller_parts = module.__name__.split('.') + for mod_name, meta in star_map.items(): + mod_parts = mod_name.split('.') + if mod_parts and caller_parts and mod_parts[0] == caller_parts[0]: + metadata = meta + break + + if metadata: + plugin_name = metadata.name + else: + # 3. Try to resolve from file path if it resides in plugins/ + if hasattr(module, "__file__") and module.__file__: + try: + path_parts = Path(module.__file__).resolve().parts + if "plugins" in path_parts: + idx = path_parts.index("plugins") + if idx + 1 < len(path_parts): + plugin_name = path_parts[idx + 1] + except Exception: + pass + + # 4. Safe fallback to avoid breaking the bot + if not plugin_name: + import logging + logging.getLogger("astrbot").warning( + f"无法获取模块 {module.__name__} 的元数据信息，已安全回退到 'unknown_plugin'" + ) + plugin_name = 'unknown_plugin' if not plugin_name: raise ValueError("无法获取插件名称") From 17fdc1004046be11471fe19a1c079991b2453d30 Mon Sep 17 00:00:00 2001 From: Rat0323 Date: Fri, 5 Jun 2026 09:50:35 +0800 Subject: [PATCH 2/5] refactor: address review feedback, extract helper functions and clean up redundant checks --- astrbot/core/star/star_tools.py | 78 +++++++++++++++++++-------------- 1 file changed, 46 insertions(+), 32 deletions(-) diff --git a/astrbot/core/star/star_tools.py b/astrbot/core/star/star_tools.py index 6baa369b3c..66d9d87f7b 100644 --- a/astrbot/core/star/star_tools.py +++ b/astrbot/core/star/star_tools.py @@ -20,6 +20,7 @@ import inspect import os import uuid +import logging from collections.abc import Awaitable, Callable from pathlib import Path from typing import Any, ClassVar @@ -39,6 +40,46 @@ from astrbot.core.utils.astrbot_path import get_astrbot_data_path from astrbot.core.utils.io import ensure_dir +logger = logging.getLogger("astrbot") + + +def _resolve_plugin_from_star_map(module, star_map): + return star_map.get(module.__name__) + + +def _resolve_plugin_from_prefix(module, star_map): + if "." not in module.__name__: + return None + caller_parts = module.__name__.split('.') + # Prefer main modules or shorter paths deterministically + sorted_keys = sorted(star_map.keys(), key=lambda k: (not k.endswith('.main'), len(k))) + for mod_name in sorted_keys: + mod_parts = mod_name.split('.') + if mod_parts and caller_parts and mod_parts[0] == caller_parts[0]: + return star_map[mod_name] + return None + + +def _resolve_plugin_from_path(module): + if not (hasattr(module, "__file__") and module.__file__): + return None + try: + path_parts = Path(module.__file__).resolve().parts + if "plugins" in path_parts: + idx = path_parts.index("plugins") + if idx + 1 < len(path_parts): + return path_parts[idx + 1] + except Exception: + return None + return None + + +def _fallback_plugin_name(module): + logger.warning( + f"无法获取模块 {module.__name__} 的元数据信息，已安全回退到 'unknown_plugin'" + ) + return "unknown_plugin" + class StarTools: """提供给插件使用的便捷工具函数集合 @@ -291,42 +332,15 @@ def get_data_dir(cls, plugin_name: str | None = None) -> Path: if not module: raise RuntimeError("无法获取调用者模块信息") - # 1. Try direct match in star_map - metadata = star_map.get(module.__name__, None) - - # 2. Try prefix match for submodule calls - if not metadata and "." in module.__name__: - caller_parts = module.__name__.split('.') - for mod_name, meta in star_map.items(): - mod_parts = mod_name.split('.') - if mod_parts and caller_parts and mod_parts[0] == caller_parts[0]: - metadata = meta - break + metadata = ( + _resolve_plugin_from_star_map(module, star_map) + or _resolve_plugin_from_prefix(module, star_map) + ) if metadata: plugin_name = metadata.name else: - # 3. Try to resolve from file path if it resides in plugins/ - if hasattr(module, "__file__") and module.__file__: - try: - path_parts = Path(module.__file__).resolve().parts - if "plugins" in path_parts: - idx = path_parts.index("plugins") - if idx + 1 < len(path_parts): - plugin_name = path_parts[idx + 1] - except Exception: - pass - - # 4. Safe fallback to avoid breaking the bot - if not plugin_name: - import logging - logging.getLogger("astrbot").warning( - f"无法获取模块 {module.__name__} 的元数据信息，已安全回退到 'unknown_plugin'" - ) - plugin_name = 'unknown_plugin' - - if not plugin_name: - raise ValueError("无法获取插件名称") + plugin_name = _resolve_plugin_from_path(module) or _fallback_plugin_name(module) data_dir = Path( os.path.join(get_astrbot_data_path(), "plugin_data", plugin_name), From 78515ea32cad0cd5459b39a8c8baaa08affacb71 Mon Sep 17 00:00:00 2001 From: Rat0323 Date: Fri, 5 Jun 2026 10:01:32 +0800 Subject: [PATCH 3/5] fix(core): apply code review suggestions for strict package matching and relative path resolution --- astrbot/core/star/star_tools.py | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/astrbot/core/star/star_tools.py b/astrbot/core/star/star_tools.py index 66d9d87f7b..448ae4e1fd 100644 --- a/astrbot/core/star/star_tools.py +++ b/astrbot/core/star/star_tools.py @@ -50,12 +50,12 @@ def _resolve_plugin_from_star_map(module, star_map): def _resolve_plugin_from_prefix(module, star_map): if "." not in module.__name__: return None - caller_parts = module.__name__.split('.') + caller_name = module.__name__ # Prefer main modules or shorter paths deterministically sorted_keys = sorted(star_map.keys(), key=lambda k: (not k.endswith('.main'), len(k))) for mod_name in sorted_keys: - mod_parts = mod_name.split('.') - if mod_parts and caller_parts and mod_parts[0] == caller_parts[0]: + mod_package = mod_name.rpartition('.')[0] if "." in mod_name else mod_name + if caller_name == mod_package or caller_name.startswith(mod_package + "."): return star_map[mod_name] return None @@ -64,14 +64,12 @@ def _resolve_plugin_from_path(module): if not (hasattr(module, "__file__") and module.__file__): return None try: - path_parts = Path(module.__file__).resolve().parts - if "plugins" in path_parts: - idx = path_parts.index("plugins") - if idx + 1 < len(path_parts): - return path_parts[idx + 1] + from astrbot.core.utils.astrbot_path import get_astrbot_plugin_path + plugin_root = Path(get_astrbot_plugin_path()).resolve() + module_path = Path(module.__file__).resolve() + return module_path.relative_to(plugin_root).parts[0] except Exception: return None - return None def _fallback_plugin_name(module): From f635862ec116a811507586b13ba8868c8f70e6a0 Mon Sep 17 00:00:00 2001 From: Rat0323 Date: Fri, 5 Jun 2026 10:56:12 +0800 Subject: [PATCH 4/5] style: format code and sort imports using ruff to pass CI checks --- astrbot/core/star/star_tools.py | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/astrbot/core/star/star_tools.py b/astrbot/core/star/star_tools.py index 448ae4e1fd..58a6137b20 100644 --- a/astrbot/core/star/star_tools.py +++ b/astrbot/core/star/star_tools.py @@ -18,9 +18,9 @@ """ import inspect +import logging import os import uuid -import logging from collections.abc import Awaitable, Callable from pathlib import Path from typing import Any, ClassVar @@ -52,9 +52,11 @@ def _resolve_plugin_from_prefix(module, star_map): return None caller_name = module.__name__ # Prefer main modules or shorter paths deterministically - sorted_keys = sorted(star_map.keys(), key=lambda k: (not k.endswith('.main'), len(k))) + sorted_keys = sorted( + star_map.keys(), key=lambda k: (not k.endswith(".main"), len(k)) + ) for mod_name in sorted_keys: - mod_package = mod_name.rpartition('.')[0] if "." in mod_name else mod_name + mod_package = mod_name.rpartition(".")[0] if "." in mod_name else mod_name if caller_name == mod_package or caller_name.startswith(mod_package + "."): return star_map[mod_name] return None @@ -65,6 +67,7 @@ def _resolve_plugin_from_path(module): return None try: from astrbot.core.utils.astrbot_path import get_astrbot_plugin_path + plugin_root = Path(get_astrbot_plugin_path()).resolve() module_path = Path(module.__file__).resolve() return module_path.relative_to(plugin_root).parts[0] @@ -330,15 +333,16 @@ def get_data_dir(cls, plugin_name: str | None = None) -> Path: if not module: raise RuntimeError("无法获取调用者模块信息") - metadata = ( - _resolve_plugin_from_star_map(module, star_map) - or _resolve_plugin_from_prefix(module, star_map) - ) + metadata = _resolve_plugin_from_star_map( + module, star_map + ) or _resolve_plugin_from_prefix(module, star_map) if metadata: plugin_name = metadata.name else: - plugin_name = _resolve_plugin_from_path(module) or _fallback_plugin_name(module) + plugin_name = _resolve_plugin_from_path( + module + ) or _fallback_plugin_name(module) data_dir = Path( os.path.join(get_astrbot_data_path(), "plugin_data", plugin_name), From fe2b6274bd025bb9f27d54f1e656c71a1d3b94d5 Mon Sep 17 00:00:00 2001 From: Rat0323 <261020116+Rat0323@users.noreply.github.com> Date: Mon, 15 Jun 2026 08:02:21 +0800 Subject: [PATCH 5/5] fix(core): resolve StarTools data dir for plugin submodules --- astrbot/core/star/star_tools.py | 156 ++++++++++++++++++++++++-------- tests/unit/test_star_tools.py | 79 ++++++++++++++++ 2 files changed, 197 insertions(+), 38 deletions(-) create mode 100644 tests/unit/test_star_tools.py diff --git a/astrbot/core/star/star_tools.py b/astrbot/core/star/star_tools.py index 58a6137b20..0a390c4bfe 100644 --- a/astrbot/core/star/star_tools.py +++ b/astrbot/core/star/star_tools.py @@ -18,11 +18,11 @@ """ import inspect -import logging import os import uuid -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Mapping from pathlib import Path +from types import ModuleType from typing import Any, ClassVar from astrbot.api.platform import AstrBotMessage, MessageMember, MessageType @@ -36,50 +36,131 @@ AiocqhttpAdapter, ) from astrbot.core.star.context import Context -from astrbot.core.star.star import star_map -from astrbot.core.utils.astrbot_path import get_astrbot_data_path +from astrbot.core.star.star import StarMetadata, star_map +from astrbot.core.utils.astrbot_path import ( + get_astrbot_data_path, + get_astrbot_path, + get_astrbot_plugin_path, +) from astrbot.core.utils.io import ensure_dir -logger = logging.getLogger("astrbot") +_PLUGIN_MODULE_FLAGS = {"plugins", "builtin_stars"} + + +def _split_module_path(module_path: str | None) -> list[str]: + if not module_path: + return [] + return module_path.split(".") -def _resolve_plugin_from_star_map(module, star_map): - return star_map.get(module.__name__) +def _plugin_root_from_module_path(module_path: str | None) -> tuple[str, str] | None: + parts = _split_module_path(module_path) + for index, part in enumerate(parts): + if part in _PLUGIN_MODULE_FLAGS and index + 1 < len(parts): + return part, parts[index + 1] + return None + +def _metadata_root_dir_name( + metadata: StarMetadata, + module_path: str | None, +) -> str | None: + if metadata.root_dir_name: + return metadata.root_dir_name + + root_info = _plugin_root_from_module_path(metadata.module_path or module_path) + return root_info[1] if root_info else None + + +def _iter_star_metadata( + stars: Mapping[str, StarMetadata], +) -> list[tuple[str, StarMetadata]]: + seen: set[int] = set() + metadata_items: list[tuple[str, StarMetadata]] = [] + for module_path, metadata in reversed(tuple(stars.items())): + metadata_id = id(metadata) + if metadata_id in seen: + continue + seen.add(metadata_id) + metadata_items.append((module_path, metadata)) + return metadata_items + + +def _resolve_plugin_from_root_dir( + root_dir_name: str, + stars: Mapping[str, StarMetadata], + module_flag: str | None = None, +) -> StarMetadata | None: + for module_path, metadata in _iter_star_metadata(stars): + registered_module_path = metadata.module_path or module_path + registered_root = _plugin_root_from_module_path(registered_module_path) + if module_flag and registered_root and registered_root[0] != module_flag: + continue + if _metadata_root_dir_name(metadata, module_path) == root_dir_name: + return metadata + return None -def _resolve_plugin_from_prefix(module, star_map): - if "." not in module.__name__: + +def _resolve_plugin_from_registered_package( + module_path: str, + stars: Mapping[str, StarMetadata], +) -> StarMetadata | None: + root_info = _plugin_root_from_module_path(module_path) + if not root_info: return None - caller_name = module.__name__ - # Prefer main modules or shorter paths deterministically - sorted_keys = sorted( - star_map.keys(), key=lambda k: (not k.endswith(".main"), len(k)) + + module_flag, root_dir_name = root_info + return _resolve_plugin_from_root_dir(root_dir_name, stars, module_flag) + + +def _plugin_search_roots() -> tuple[tuple[str, Path], ...]: + return ( + ("plugins", Path(get_astrbot_plugin_path()).resolve()), + ( + "builtin_stars", + Path(get_astrbot_path()).resolve() / "astrbot" / "builtin_stars", + ), ) - for mod_name in sorted_keys: - mod_package = mod_name.rpartition(".")[0] if "." in mod_name else mod_name - if caller_name == mod_package or caller_name.startswith(mod_package + "."): - return star_map[mod_name] - return None -def _resolve_plugin_from_path(module): - if not (hasattr(module, "__file__") and module.__file__): +def _resolve_plugin_from_file_path( + module: ModuleType, + stars: Mapping[str, StarMetadata], +) -> StarMetadata | None: + module_file = getattr(module, "__file__", None) + if not module_file: return None - try: - from astrbot.core.utils.astrbot_path import get_astrbot_plugin_path - plugin_root = Path(get_astrbot_plugin_path()).resolve() - module_path = Path(module.__file__).resolve() - return module_path.relative_to(plugin_root).parts[0] + try: + module_path = Path(module_file).resolve() except Exception: return None + for module_flag, plugin_root in _plugin_search_roots(): + try: + relative_parts = module_path.relative_to(plugin_root).parts + except ValueError: + continue + + if relative_parts: + return _resolve_plugin_from_root_dir( + relative_parts[0], + stars, + module_flag, + ) + + return None + -def _fallback_plugin_name(module): - logger.warning( - f"无法获取模块 {module.__name__} 的元数据信息，已安全回退到 'unknown_plugin'" +def _resolve_plugin_metadata( + module: ModuleType, + stars: Mapping[str, StarMetadata], +) -> StarMetadata | None: + return ( + stars.get(module.__name__) + or _resolve_plugin_from_registered_package(module.__name__, stars) + or _resolve_plugin_from_file_path(module, stars) ) - return "unknown_plugin" class StarTools: @@ -333,16 +414,15 @@ def get_data_dir(cls, plugin_name: str | None = None) -> Path: if not module: raise RuntimeError("无法获取调用者模块信息") - metadata = _resolve_plugin_from_star_map( - module, star_map - ) or _resolve_plugin_from_prefix(module, star_map) + metadata = _resolve_plugin_metadata(module, star_map) + + if not metadata: + raise RuntimeError(f"无法获取模块 {module.__name__} 的元数据信息") - if metadata: - plugin_name = metadata.name - else: - plugin_name = _resolve_plugin_from_path( - module - ) or _fallback_plugin_name(module) + plugin_name = metadata.name + + if not plugin_name: + raise ValueError("无法获取插件名称") data_dir = Path( os.path.join(get_astrbot_data_path(), "plugin_data", plugin_name), diff --git a/tests/unit/test_star_tools.py b/tests/unit/test_star_tools.py new file mode 100644 index 0000000000..8b4578ea7f --- /dev/null +++ b/tests/unit/test_star_tools.py @@ -0,0 +1,79 @@ +from types import ModuleType + +import pytest + +from astrbot.core.star import star_tools +from astrbot.core.star.star import StarMetadata, star_map +from astrbot.core.star.star_tools import StarTools + + +@pytest.fixture(autouse=True) +def restore_star_map(): + original_map = dict(star_map) + star_map.clear() + try: + yield + finally: + star_map.clear() + star_map.update(original_map) + + +def make_module(name: str, file_path: str | None = None) -> ModuleType: + module = ModuleType(name) + if file_path: + module.__file__ = file_path + return module + + +def set_caller_module(monkeypatch: pytest.MonkeyPatch, module: ModuleType) -> None: + monkeypatch.setattr(star_tools.inspect, "getmodule", lambda _frame: module) + + +def test_get_data_dir_resolves_registered_plugin_submodule(monkeypatch, tmp_path): + data_path = tmp_path / "data" + monkeypatch.setattr(star_tools, "get_astrbot_data_path", lambda: str(data_path)) + set_caller_module( + monkeypatch, + make_module("data.plugins.demo_plugin.services.cache"), + ) + star_map["data.plugins.demo_plugin.main"] = StarMetadata( + name="demo", + module_path="data.plugins.demo_plugin.main", + root_dir_name="demo_plugin", + ) + + data_dir = StarTools.get_data_dir() + + assert data_dir == (data_path / "plugin_data" / "demo").resolve() + + +def test_get_data_dir_resolves_debug_module_from_plugin_path(monkeypatch, tmp_path): + data_path = tmp_path / "data" + plugin_root = tmp_path / "plugins" + debug_file = plugin_root / "demo_plugin" / "scripts" / "debug.py" + debug_file.parent.mkdir(parents=True) + debug_file.write_text("", encoding="utf-8") + monkeypatch.setattr(star_tools, "get_astrbot_data_path", lambda: str(data_path)) + monkeypatch.setattr(star_tools, "get_astrbot_plugin_path", lambda: str(plugin_root)) + monkeypatch.setattr(star_tools, "get_astrbot_path", lambda: str(tmp_path / "src")) + set_caller_module(monkeypatch, make_module("__main__", str(debug_file))) + star_map["data.plugins.demo_plugin.main"] = StarMetadata( + name="demo", + module_path="data.plugins.demo_plugin.main", + root_dir_name="demo_plugin", + ) + + data_dir = StarTools.get_data_dir() + + assert data_dir == (data_path / "plugin_data" / "demo").resolve() + + +def test_get_data_dir_keeps_unknown_module_failure(monkeypatch, tmp_path): + data_path = tmp_path / "data" + monkeypatch.setattr(star_tools, "get_astrbot_data_path", lambda: str(data_path)) + set_caller_module(monkeypatch, make_module("external.module")) + + with pytest.raises(RuntimeError, match="无法获取模块 external.module 的元数据信息"): + StarTools.get_data_dir() + + assert not (data_path / "plugin_data" / "unknown_plugin").exists()