feat: 新增两个插件生命周期事件钩子 & 为插件metadata.yaml引入dependencies字段 #3182
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
你好 - 我已经审阅了你的修改，它们看起来很棒！
AI 代理的提示
请解决此代码审查中的评论：
## 个人评论
### 评论 1
<location> `astrbot/core/star/star_manager.py:865-866` </location>
<code_context>
return plugin_info
+
+ async def _trigger_star_lifecycle_event(
+ self, event_type: EventType, star_metadata: StarMetadata
+ ):
+ """
</code_context>
<issue_to_address>
**suggestion:** 考虑支持不指定 target_star_name 的处理程序。
目前，没有 target_star_name 的处理程序会被跳过，从而阻止全局生命周期挂钩。考虑触发这些处理程序以获得更广泛的事件覆盖。
</issue_to_address>
### 评论 2
<location> `astrbot/core/star/star_manager.py:900-902` </location>
<code_context>
os.path.join(self.plugin_store_path, root_dir_name)
if not is_reserved
else os.path.join(self.reserved_plugin_path, root_dir_name)
</code_context>
<issue_to_address>
**suggestion (code-quality):** 交换 if 表达式的 if/else 分支以消除否定 ([`swap-if-expression`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/swap-if-expression))
```suggestion
os.path.join(self.reserved_plugin_path, root_dir_name) if is_reserved else os.path.join(self.plugin_store_path, root_dir_name)
```
<br/><details><summary>Explanation</summary>否定条件比肯定条件更难阅读，因此最好尽可能避免使用它们。通过交换 `if` 和 `else` 条件，我们可以反转条件并使其变为肯定。
</details>
</issue_to_address>
### 评论 3
<location> `astrbot/core/star/star_manager.py:335` </location>
<code_context>
async def load(self, plugin_modules=None):
"""载入插件。
当 specified_module_path 或者 specified_dir_name 不为 None 时，只载入指定的插件。
Args:
specified_module_path (str, optional): 指定要加载的插件模块路径。例如: "data.plugins.my_plugin.main"
specified_dir_name (str, optional): 指定要加载的插件目录名。例如: "my_plugin"
Returns:
tuple: (success, error_message)
- success (bool): 是否全部加载成功
- error_message (str|None): 错误信息，成功时为 None
"""
inactivated_plugins = await sp.global_get("inactivated_plugins", [])
inactivated_llm_tools = await sp.global_get("inactivated_llm_tools", [])
alter_cmd = await sp.global_get("alter_cmd", {})
if plugin_modules is None:
return False, "未找到任何插件模块"
logger.info(
f"正在按顺序加载插件: {[plugin_module['pname'] for plugin_module in plugin_modules]}"
)
fail_rec = ""
# 导入插件模块，并尝试实例化插件类
for plugin_module in plugin_modules:
try:
module_str = plugin_module["module"]
# module_path = plugin_module['module_path']
root_dir_name = plugin_module["pname"] # 插件的目录名
reserved = plugin_module.get(
"reserved", False
) # 是否是保留插件。目前在 packages/ 目录下的都是保留插件。保留插件不可以卸载。
path = "data.plugins." if not reserved else "packages."
path += root_dir_name + "." + module_str
logger.info(f"正在载入插件 {root_dir_name} ...")
# 尝试导入模块
try:
module = __import__(path, fromlist=[module_str])
except (ModuleNotFoundError, ImportError):
# 尝试安装依赖
await self._check_plugin_dept_update(target_plugin=root_dir_name)
module = __import__(path, fromlist=[module_str])
except Exception as e:
logger.error(traceback.format_exc())
logger.error(f"插件 {root_dir_name} 导入失败。原因：{str(e)}")
continue
# 检查 _conf_schema.json
plugin_config = None
plugin_dir_path = (
os.path.join(self.plugin_store_path, root_dir_name)
if not reserved
else os.path.join(self.reserved_plugin_path, root_dir_name)
)
plugin_schema_path = os.path.join(
plugin_dir_path, self.conf_schema_fname
)
if os.path.exists(plugin_schema_path):
# 加载插件配置
with open(plugin_schema_path, "r", encoding="utf-8") as f:
plugin_config = AstrBotConfig(
config_path=os.path.join(
self.plugin_config_path, f"{root_dir_name}_config.json"
),
schema=json.loads(f.read()),
)
if path in star_map:
# 通过 __init__subclass__ 注册插件
metadata = star_map[path]
try:
# yaml 文件的元数据优先
metadata_yaml = self._load_plugin_metadata(
plugin_path=plugin_dir_path
)
if metadata_yaml:
metadata.name = metadata_yaml.name
metadata.author = metadata_yaml.author
metadata.desc = metadata_yaml.desc
metadata.version = metadata_yaml.version
metadata.repo = metadata_yaml.repo
except Exception as e:
logger.warning(
f"插件 {root_dir_name} 元数据载入失败: {str(e)}。使用默认元数据。"
)
logger.info(metadata)
metadata.config = plugin_config
if path not in inactivated_plugins:
# 只有没有禁用插件时才实例化插件类
if plugin_config and metadata.star_cls_type:
try:
metadata.star_cls = metadata.star_cls_type(
context=self.context, config=plugin_config
)
except TypeError as _:
metadata.star_cls = metadata.star_cls_type(
context=self.context
)
elif metadata.star_cls_type:
metadata.star_cls = metadata.star_cls_type(
context=self.context
)
await self._trigger_star_lifecycle_event(
EventType.OnStarActivatedEvent, metadata
)
else:
logger.info(f"插件 {metadata.name} 已被禁用。")
metadata.module = module
metadata.root_dir_name = root_dir_name
metadata.reserved = reserved
assert metadata.module_path is not None, (
f"插件 {metadata.name} 的模块路径为空。"
)
# 绑定 handler
related_handlers = (
star_handlers_registry.get_handlers_by_module_name(
metadata.module_path
)
)
for handler in related_handlers:
handler.handler = functools.partial(
handler.handler,
metadata.star_cls, # type: ignore
)
# 绑定 llm_tool handler
for func_tool in llm_tools.func_list:
if isinstance(func_tool, HandoffTool):
need_apply = []
sub_tools = func_tool.agent.tools
for sub_tool in sub_tools:
if isinstance(sub_tool, FunctionTool):
need_apply.append(sub_tool)
else:
need_apply = [func_tool]
for ft in need_apply:
if (
ft.handler
and ft.handler.__module__ == metadata.module_path
):
ft.handler_module_path = metadata.module_path
ft.handler = functools.partial(
ft.handler,
metadata.star_cls, # type: ignore
)
if ft.name in inactivated_llm_tools:
ft.active = False
else:
# v3.4.0 以前的方式注册插件
logger.debug(
f"插件 {path} 未通过装饰器注册。尝试通过旧版本方式载入。"
)
classes = self._get_classes(module)
if path not in inactivated_plugins:
# 只有没有禁用插件时才实例化插件类
if plugin_config:
try:
obj = getattr(module, classes[0])(
context=self.context, config=plugin_config
) # 实例化插件类
except TypeError as _:
obj = getattr(module, classes[0])(
context=self.context
) # 实例化插件类
else:
obj = getattr(module, classes[0])(
context=self.context
) # 实例化插件类
metadata = self._load_plugin_metadata(
plugin_path=plugin_dir_path, plugin_obj=obj
)
if not metadata:
raise Exception(f"无法找到插件 {plugin_dir_path} 的元数据。")
metadata.star_cls = obj
metadata.config = plugin_config
metadata.module = module
metadata.root_dir_name = root_dir_name
metadata.reserved = reserved
metadata.star_cls_type = obj.__class__
metadata.module_path = path
star_map[path] = metadata
star_registry.append(metadata)
# 禁用/启用插件
if metadata.module_path in inactivated_plugins:
metadata.activated = False
assert metadata.module_path is not None, (
f"插件 {metadata.name} 的模块路径为空。"
)
full_names = []
for handler in star_handlers_registry.get_handlers_by_module_name(
metadata.module_path
):
full_names.append(handler.handler_full_name)
# 检查并且植入自定义的权限过滤器（alter_cmd）
if (
metadata.name in alter_cmd
and handler.handler_name in alter_cmd[metadata.name]
):
cmd_type = alter_cmd[metadata.name][handler.handler_name].get(
"permission", "member"
)
found_permission_filter = False
for filter_ in handler.event_filters:
if isinstance(filter_, PermissionTypeFilter):
if cmd_type == "admin":
filter_.permission_type = PermissionType.ADMIN
else:
filter_.permission_type = PermissionType.MEMBER
found_permission_filter = True
break
if not found_permission_filter:
handler.event_filters.append(
PermissionTypeFilter(
PermissionType.ADMIN
if cmd_type == "admin"
else PermissionType.MEMBER
)
)
logger.debug(
f"插入权限过滤器 {cmd_type} 到 {metadata.name} 的 {handler.handler_name} 方法。"
)
metadata.star_handler_full_names = full_names
# 执行 initialize() 方法
if hasattr(metadata.star_cls, "initialize") and metadata.star_cls:
await metadata.star_cls.initialize()
except BaseException as e:
logger.error(f"----- 插件 {root_dir_name} 载入失败 -----")
errors = traceback.format_exc()
for line in errors.split("\n"):
logger.error(f"| {line}")
logger.error("----------------------------------")
fail_rec += f"加载 {root_dir_name} 插件时出现问题，原因 {str(e)}。\n"
# 清除 pip.main 导致的多余的 logging handlers
for handler in logging.root.handlers[:]:
logging.root.removeHandler(handler)
if not fail_rec:
return True, None
else:
self.failed_plugin_info = fail_rec
return False, fail_rec
</code_context>
<issue_to_address>
**issue (code-quality):** 我们发现了这些问题：
- 从 except 子句中移除冗余异常 ([`remove-redundant-exception`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-exception/))
- 使用命名表达式简化赋值和条件 ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- 将长度为一的异常元组替换为异常 ([`simplify-single-exception-tuple`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/simplify-single-exception-tuple/))
- 在 PluginManager.load 中发现低代码质量 - 1% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
<br/><details><summary>Explanation</summary>
此函数的质量分数低于 25% 的质量阈值。
此分数是方法长度、认知复杂度和工作内存的组合。
您如何解决这个问题？
重构此函数以使其更短、更具可读性可能值得。
- 通过将部分功能提取到自己的函数中来减少函数长度。这是您可以做的最重要的事情 - 理想情况下，一个函数应该少于 10 行。
- 减少嵌套，例如通过引入守卫子句来提前返回。
- 确保变量的作用域严格，以便使用相关概念的代码在函数中紧密地放在一起，而不是分散开来。
</details>
</issue_to_address>
### 评论 4
<location> `astrbot/core/star/star_manager.py:613` </location>
<code_context>
async def install_plugin(self, repo_url: str, proxy=""):
"""从仓库 URL 安装插件
从指定的仓库 URL 下载并安装插件，然后加载该插件到系统中
Args:
repo_url (str): 要安装的插件仓库 URL
proxy (str, optional): 用于下载的代理服务器。默认为空字符串。
Returns:
dict | None: 安装成功时返回包含插件信息的字典:
- repo: 插件的仓库 URL
- readme: README.md 文件的内容(如果存在)
如果找不到插件元数据则返回 None。
"""
async with self._pm_lock:
plugin_path = await self.updator.install(repo_url, proxy)
# reload the plugin
dir_name = os.path.basename(plugin_path)
plugin_modules = await self._get_load_order(specified_dir_name=dir_name)
await self.batch_reload(plugin_modules=plugin_modules)
# Get the plugin metadata to return repo info
plugin = self.context.get_registered_star(dir_name)
if not plugin:
# Try to find by other name if directory name doesn't match plugin name
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
# Extract README.md content if exists
readme_content = None
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if os.path.exists(readme_path):
try:
with open(readme_path, "r", encoding="utf-8") as f:
readme_content = f.read()
except Exception as e:
logger.warning(
f"读取插件 {dir_name} 的 README.md 文件失败: {str(e)}"
)
plugin_info = None
if plugin:
plugin_info = {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
return plugin_info
</code_context>
<issue_to_address>
**issue (code-quality):** 我们发现了这些问题：
- 将变量的默认值设置移至 `else` 分支 ([`introduce-default-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/introduce-default-else/))
- 用 if 表达式替换 if 语句 ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
- 内联立即返回的变量 ([`inline-immediately-returned-variable`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/inline-immediately-returned-variable/))
</issue_to_address>
### 评论 5
<location> `astrbot/core/star/star_manager.py:970-971` </location>
<code_context>
def _build_star_graph(self):
plugin_modules = self._get_plugin_modules()
if plugin_modules is None:
return None
G = nx.DiGraph()
for plugin_module in plugin_modules:
root_dir_name = plugin_module["pname"]
is_reserved = plugin_module.get("reserved", False)
plugin_dir_path = self._get_plugin_dir_path(root_dir_name, is_reserved)
G.add_node(root_dir_name, data=plugin_module)
try:
metadata = self._load_plugin_metadata(plugin_dir_path)
if metadata:
for dep_name in metadata.dependencies:
G.add_edge(root_dir_name, dep_name)
except Exception:
pass
# 过滤不存在的依赖(出边没有data, 就删除指向的节点)
nodes_to_remove = []
for node_name in list(G.nodes()):
for neighbor in list(G.neighbors(node_name)):
if G.nodes[neighbor].get("data") is None:
nodes_to_remove.append(neighbor)
logger.warning(
f"插件 {node_name} 声明依赖 {neighbor}, 但该插件未被发现，跳过加载。"
)
for node in nodes_to_remove:
G.remove_node(node)
return G
</code_context>
<issue_to_address>
**suggestion (code-quality):** 使用命名表达式简化赋值和条件 ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
```suggestion
if metadata := self._load_plugin_metadata(plugin_dir_path):
```
</issue_to_address>
### 评论 6
<location> `astrbot/core/star/star_manager.py:996-997` </location>
<code_context>
async def batch_reload(self, specified_module_path=None, plugin_modules=None):
if not plugin_modules:
plugin_modules = await self._get_load_order(
specified_module_path=specified_module_path
)
for plugin_module in plugin_modules:
specified_module_path = self._build_module_path(plugin_module)
smd = star_map.get(specified_module_path)
if smd:
try:
await self._terminate_plugin(smd)
except Exception as e:
logger.warning(traceback.format_exc())
logger.warning(
f"插件 {smd.name} 未被正常终止: {str(e)}, 可能会导致该插件运行不正常。"
)
await self._unbind_plugin(smd.name, specified_module_path)
return await self.load(plugin_modules=plugin_modules)
</code_context>
<issue_to_address>
**suggestion (code-quality):** 使用命名表达式简化赋值和条件 ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
```suggestion
if smd := star_map.get(specified_module_path):
```
</issue_to_address>帮助我更有用！请点击每个评论上的 👍 或 👎，我将使用反馈来改进您的评论。
Original comment in English
Hey there - I've reviewed your changes and they look great!
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `astrbot/core/star/star_manager.py:865-866` </location>
<code_context>
return plugin_info
+
+ async def _trigger_star_lifecycle_event(
+ self, event_type: EventType, star_metadata: StarMetadata
+ ):
+ """
</code_context>
<issue_to_address>
**suggestion:** Consider supporting handlers that do not specify a target_star_name.
Handlers without a target_star_name are currently skipped, preventing global lifecycle hooks. Consider triggering these handlers for broader event coverage.
</issue_to_address>
### Comment 2
<location> `astrbot/core/star/star_manager.py:900-902` </location>
<code_context>
os.path.join(self.plugin_store_path, root_dir_name)
if not is_reserved
else os.path.join(self.reserved_plugin_path, root_dir_name)
</code_context>
<issue_to_address>
**suggestion (code-quality):** Swap if/else branches of if expression to remove negation ([`swap-if-expression`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/swap-if-expression))
```suggestion
os.path.join(self.reserved_plugin_path, root_dir_name) if is_reserved else os.path.join(self.plugin_store_path, root_dir_name)
```
<br/><details><summary>Explanation</summary>Negated conditions are more difficult to read than positive ones, so it is best
to avoid them where we can. By swapping the `if` and `else` conditions around we
can invert the condition and make it positive.
</details>
</issue_to_address>
### Comment 3
<location> `astrbot/core/star/star_manager.py:335` </location>
<code_context>
async def load(self, plugin_modules=None):
"""载入插件。
当 specified_module_path 或者 specified_dir_name 不为 None 时，只载入指定的插件。
Args:
specified_module_path (str, optional): 指定要加载的插件模块路径。例如: "data.plugins.my_plugin.main"
specified_dir_name (str, optional): 指定要加载的插件目录名。例如: "my_plugin"
Returns:
tuple: (success, error_message)
- success (bool): 是否全部加载成功
- error_message (str|None): 错误信息，成功时为 None
"""
inactivated_plugins = await sp.global_get("inactivated_plugins", [])
inactivated_llm_tools = await sp.global_get("inactivated_llm_tools", [])
alter_cmd = await sp.global_get("alter_cmd", {})
if plugin_modules is None:
return False, "未找到任何插件模块"
logger.info(
f"正在按顺序加载插件: {[plugin_module['pname'] for plugin_module in plugin_modules]}"
)
fail_rec = ""
# 导入插件模块，并尝试实例化插件类
for plugin_module in plugin_modules:
try:
module_str = plugin_module["module"]
# module_path = plugin_module['module_path']
root_dir_name = plugin_module["pname"] # 插件的目录名
reserved = plugin_module.get(
"reserved", False
) # 是否是保留插件。目前在 packages/ 目录下的都是保留插件。保留插件不可以卸载。
path = "data.plugins." if not reserved else "packages."
path += root_dir_name + "." + module_str
logger.info(f"正在载入插件 {root_dir_name} ...")
# 尝试导入模块
try:
module = __import__(path, fromlist=[module_str])
except (ModuleNotFoundError, ImportError):
# 尝试安装依赖
await self._check_plugin_dept_update(target_plugin=root_dir_name)
module = __import__(path, fromlist=[module_str])
except Exception as e:
logger.error(traceback.format_exc())
logger.error(f"插件 {root_dir_name} 导入失败。原因：{str(e)}")
continue
# 检查 _conf_schema.json
plugin_config = None
plugin_dir_path = (
os.path.join(self.plugin_store_path, root_dir_name)
if not reserved
else os.path.join(self.reserved_plugin_path, root_dir_name)
)
plugin_schema_path = os.path.join(
plugin_dir_path, self.conf_schema_fname
)
if os.path.exists(plugin_schema_path):
# 加载插件配置
with open(plugin_schema_path, "r", encoding="utf-8") as f:
plugin_config = AstrBotConfig(
config_path=os.path.join(
self.plugin_config_path, f"{root_dir_name}_config.json"
),
schema=json.loads(f.read()),
)
if path in star_map:
# 通过 __init__subclass__ 注册插件
metadata = star_map[path]
try:
# yaml 文件的元数据优先
metadata_yaml = self._load_plugin_metadata(
plugin_path=plugin_dir_path
)
if metadata_yaml:
metadata.name = metadata_yaml.name
metadata.author = metadata_yaml.author
metadata.desc = metadata_yaml.desc
metadata.version = metadata_yaml.version
metadata.repo = metadata_yaml.repo
except Exception as e:
logger.warning(
f"插件 {root_dir_name} 元数据载入失败: {str(e)}。使用默认元数据。"
)
logger.info(metadata)
metadata.config = plugin_config
if path not in inactivated_plugins:
# 只有没有禁用插件时才实例化插件类
if plugin_config and metadata.star_cls_type:
try:
metadata.star_cls = metadata.star_cls_type(
context=self.context, config=plugin_config
)
except TypeError as _:
metadata.star_cls = metadata.star_cls_type(
context=self.context
)
elif metadata.star_cls_type:
metadata.star_cls = metadata.star_cls_type(
context=self.context
)
await self._trigger_star_lifecycle_event(
EventType.OnStarActivatedEvent, metadata
)
else:
logger.info(f"插件 {metadata.name} 已被禁用。")
metadata.module = module
metadata.root_dir_name = root_dir_name
metadata.reserved = reserved
assert metadata.module_path is not None, (
f"插件 {metadata.name} 的模块路径为空。"
)
# 绑定 handler
related_handlers = (
star_handlers_registry.get_handlers_by_module_name(
metadata.module_path
)
)
for handler in related_handlers:
handler.handler = functools.partial(
handler.handler,
metadata.star_cls, # type: ignore
)
# 绑定 llm_tool handler
for func_tool in llm_tools.func_list:
if isinstance(func_tool, HandoffTool):
need_apply = []
sub_tools = func_tool.agent.tools
for sub_tool in sub_tools:
if isinstance(sub_tool, FunctionTool):
need_apply.append(sub_tool)
else:
need_apply = [func_tool]
for ft in need_apply:
if (
ft.handler
and ft.handler.__module__ == metadata.module_path
):
ft.handler_module_path = metadata.module_path
ft.handler = functools.partial(
ft.handler,
metadata.star_cls, # type: ignore
)
if ft.name in inactivated_llm_tools:
ft.active = False
else:
# v3.4.0 以前的方式注册插件
logger.debug(
f"插件 {path} 未通过装饰器注册。尝试通过旧版本方式载入。"
)
classes = self._get_classes(module)
if path not in inactivated_plugins:
# 只有没有禁用插件时才实例化插件类
if plugin_config:
try:
obj = getattr(module, classes[0])(
context=self.context, config=plugin_config
) # 实例化插件类
except TypeError as _:
obj = getattr(module, classes[0])(
context=self.context
) # 实例化插件类
else:
obj = getattr(module, classes[0])(
context=self.context
) # 实例化插件类
metadata = self._load_plugin_metadata(
plugin_path=plugin_dir_path, plugin_obj=obj
)
if not metadata:
raise Exception(f"无法找到插件 {plugin_dir_path} 的元数据。")
metadata.star_cls = obj
metadata.config = plugin_config
metadata.module = module
metadata.root_dir_name = root_dir_name
metadata.reserved = reserved
metadata.star_cls_type = obj.__class__
metadata.module_path = path
star_map[path] = metadata
star_registry.append(metadata)
# 禁用/启用插件
if metadata.module_path in inactivated_plugins:
metadata.activated = False
assert metadata.module_path is not None, (
f"插件 {metadata.name} 的模块路径为空。"
)
full_names = []
for handler in star_handlers_registry.get_handlers_by_module_name(
metadata.module_path
):
full_names.append(handler.handler_full_name)
# 检查并且植入自定义的权限过滤器（alter_cmd）
if (
metadata.name in alter_cmd
and handler.handler_name in alter_cmd[metadata.name]
):
cmd_type = alter_cmd[metadata.name][handler.handler_name].get(
"permission", "member"
)
found_permission_filter = False
for filter_ in handler.event_filters:
if isinstance(filter_, PermissionTypeFilter):
if cmd_type == "admin":
filter_.permission_type = PermissionType.ADMIN
else:
filter_.permission_type = PermissionType.MEMBER
found_permission_filter = True
break
if not found_permission_filter:
handler.event_filters.append(
PermissionTypeFilter(
PermissionType.ADMIN
if cmd_type == "admin"
else PermissionType.MEMBER
)
)
logger.debug(
f"插入权限过滤器 {cmd_type} 到 {metadata.name} 的 {handler.handler_name} 方法。"
)
metadata.star_handler_full_names = full_names
# 执行 initialize() 方法
if hasattr(metadata.star_cls, "initialize") and metadata.star_cls:
await metadata.star_cls.initialize()
except BaseException as e:
logger.error(f"----- 插件 {root_dir_name} 载入失败 -----")
errors = traceback.format_exc()
for line in errors.split("\n"):
logger.error(f"| {line}")
logger.error("----------------------------------")
fail_rec += f"加载 {root_dir_name} 插件时出现问题，原因 {str(e)}。\n"
# 清除 pip.main 导致的多余的 logging handlers
for handler in logging.root.handlers[:]:
logging.root.removeHandler(handler)
if not fail_rec:
return True, None
else:
self.failed_plugin_info = fail_rec
return False, fail_rec
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Remove redundant exceptions from an except clause ([`remove-redundant-exception`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-exception/))
- Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- Replace length-one exception tuple with exception ([`simplify-single-exception-tuple`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/simplify-single-exception-tuple/))
- Low code quality found in PluginManager.load - 1% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
<br/><details><summary>Explanation</summary>
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.</details>
</issue_to_address>
### Comment 4
<location> `astrbot/core/star/star_manager.py:613` </location>
<code_context>
async def install_plugin(self, repo_url: str, proxy=""):
"""从仓库 URL 安装插件
从指定的仓库 URL 下载并安装插件，然后加载该插件到系统中
Args:
repo_url (str): 要安装的插件仓库 URL
proxy (str, optional): 用于下载的代理服务器。默认为空字符串。
Returns:
dict | None: 安装成功时返回包含插件信息的字典:
- repo: 插件的仓库 URL
- readme: README.md 文件的内容(如果存在)
如果找不到插件元数据则返回 None。
"""
async with self._pm_lock:
plugin_path = await self.updator.install(repo_url, proxy)
# reload the plugin
dir_name = os.path.basename(plugin_path)
plugin_modules = await self._get_load_order(specified_dir_name=dir_name)
await self.batch_reload(plugin_modules=plugin_modules)
# Get the plugin metadata to return repo info
plugin = self.context.get_registered_star(dir_name)
if not plugin:
# Try to find by other name if directory name doesn't match plugin name
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
# Extract README.md content if exists
readme_content = None
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if os.path.exists(readme_path):
try:
with open(readme_path, "r", encoding="utf-8") as f:
readme_content = f.read()
except Exception as e:
logger.warning(
f"读取插件 {dir_name} 的 README.md 文件失败: {str(e)}"
)
plugin_info = None
if plugin:
plugin_info = {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
return plugin_info
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Move setting of default value for variable into `else` branch ([`introduce-default-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/introduce-default-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
- Inline variable that is immediately returned ([`inline-immediately-returned-variable`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/inline-immediately-returned-variable/))
</issue_to_address>
### Comment 5
<location> `astrbot/core/star/star_manager.py:970-971` </location>
<code_context>
def _build_star_graph(self):
plugin_modules = self._get_plugin_modules()
if plugin_modules is None:
return None
G = nx.DiGraph()
for plugin_module in plugin_modules:
root_dir_name = plugin_module["pname"]
is_reserved = plugin_module.get("reserved", False)
plugin_dir_path = self._get_plugin_dir_path(root_dir_name, is_reserved)
G.add_node(root_dir_name, data=plugin_module)
try:
metadata = self._load_plugin_metadata(plugin_dir_path)
if metadata:
for dep_name in metadata.dependencies:
G.add_edge(root_dir_name, dep_name)
except Exception:
pass
# 过滤不存在的依赖(出边没有data, 就删除指向的节点)
nodes_to_remove = []
for node_name in list(G.nodes()):
for neighbor in list(G.neighbors(node_name)):
if G.nodes[neighbor].get("data") is None:
nodes_to_remove.append(neighbor)
logger.warning(
f"插件 {node_name} 声明依赖 {neighbor}, 但该插件未被发现，跳过加载。"
)
for node in nodes_to_remove:
G.remove_node(node)
return G
</code_context>
<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
```suggestion
if metadata := self._load_plugin_metadata(plugin_dir_path):
```
</issue_to_address>
### Comment 6
<location> `astrbot/core/star/star_manager.py:996-997` </location>
<code_context>
async def batch_reload(self, specified_module_path=None, plugin_modules=None):
if not plugin_modules:
plugin_modules = await self._get_load_order(
specified_module_path=specified_module_path
)
for plugin_module in plugin_modules:
specified_module_path = self._build_module_path(plugin_module)
smd = star_map.get(specified_module_path)
if smd:
try:
await self._terminate_plugin(smd)
except Exception as e:
logger.warning(traceback.format_exc())
logger.warning(
f"插件 {smd.name} 未被正常终止: {str(e)}, 可能会导致该插件运行不正常。"
)
await self._unbind_plugin(smd.name, specified_module_path)
return await self.load(plugin_modules=plugin_modules)
</code_context>
<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
```suggestion
if smd := star_map.get(specified_module_path):
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+865
to
+866
| async def _trigger_star_lifecycle_event( | ||
| self, event_type: EventType, star_metadata: StarMetadata |
Contributor
There was a problem hiding this comment.
suggestion: 考虑支持不指定 target_star_name 的处理程序。
目前，没有 target_star_name 的处理程序会被跳过，从而阻止全局生命周期挂钩。考虑触发这些处理程序以获得更广泛的事件覆盖。
Original comment in English
suggestion: Consider supporting handlers that do not specify a target_star_name.
Handlers without a target_star_name are currently skipped, preventing global lifecycle hooks. Consider triggering these handlers for broader event coverage.
Comment on lines
+900
to
+902
| os.path.join(self.plugin_store_path, root_dir_name) | ||
| if not is_reserved | ||
| else os.path.join(self.reserved_plugin_path, root_dir_name) |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): 交换 if 表达式的 if/else 分支以消除否定 (swap-if-expression)
Suggested change
| os.path.join(self.plugin_store_path, root_dir_name) | |
| if not is_reserved | |
| else os.path.join(self.reserved_plugin_path, root_dir_name) | |
| os.path.join(self.reserved_plugin_path, root_dir_name) if is_reserved else os.path.join(self.plugin_store_path, root_dir_name) | |
Explanation
否定条件比肯定条件更难阅读，因此最好尽可能避免使用它们。通过交换if 和 else 条件，我们可以反转条件并使其变为肯定。
Original comment in English
suggestion (code-quality): Swap if/else branches of if expression to remove negation (swap-if-expression)
Suggested change
| os.path.join(self.plugin_store_path, root_dir_name) | |
| if not is_reserved | |
| else os.path.join(self.reserved_plugin_path, root_dir_name) | |
| os.path.join(self.reserved_plugin_path, root_dir_name) if is_reserved else os.path.join(self.plugin_store_path, root_dir_name) | |
Explanation
Negated conditions are more difficult to read than positive ones, so it is bestto avoid them where we can. By swapping the
if and else conditions around wecan invert the condition and make it positive.
| return result | ||
|
|
||
| async def load(self, specified_module_path=None, specified_dir_name=None): | ||
| async def load(self, plugin_modules=None): |
Contributor
There was a problem hiding this comment.
issue (code-quality): 我们发现了这些问题：
- 从 except 子句中移除冗余异常 (
remove-redundant-exception) - 使用命名表达式简化赋值和条件 (
use-named-expression) - 将长度为一的异常元组替换为异常 (
simplify-single-exception-tuple) - 在 PluginManager.load 中发现低代码质量 - 1% (
low-code-quality)
Explanation
此函数的质量分数低于 25% 的质量阈值。
此分数是方法长度、认知复杂度和工作内存的组合。
您如何解决这个问题？
重构此函数以使其更短、更具可读性可能值得。
- 通过将部分功能提取到自己的函数中来减少函数长度。这是您可以做的最重要的事情 - 理想情况下，一个函数应该少于 10 行。
- 减少嵌套，例如通过引入守卫子句来提前返回。
- 确保变量的作用域严格，以便使用相关概念的代码在函数中紧密地放在一起，而不是分散开来。
Original comment in English
issue (code-quality): We've found these issues:
- Remove redundant exceptions from an except clause (
remove-redundant-exception) - Use named expression to simplify assignment and conditional (
use-named-expression) - Replace length-one exception tuple with exception (
simplify-single-exception-tuple) - Low code quality found in PluginManager.load - 1% (
low-code-quality)
Explanation
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines. - Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.
| 如果找不到插件元数据则返回 None。 | ||
| """ | ||
| async with self._pm_lock: | ||
| plugin_path = await self.updator.install(repo_url, proxy) |
Contributor
There was a problem hiding this comment.
issue (code-quality): 我们发现了这些问题：
- 将变量的默认值设置移至
else分支 (introduce-default-else) - 用 if 表达式替换 if 语句 (
assign-if-exp) - 内联立即返回的变量 (
inline-immediately-returned-variable)
Original comment in English
issue (code-quality): We've found these issues:
- Move setting of default value for variable into
elsebranch (introduce-default-else) - Replace if statement with if expression (
assign-if-exp) - Inline variable that is immediately returned (
inline-immediately-returned-variable)
Comment on lines
+970
to
+971
| metadata = self._load_plugin_metadata(plugin_dir_path) | ||
| if metadata: |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): 使用命名表达式简化赋值和条件 (use-named-expression)
Suggested change
| metadata = self._load_plugin_metadata(plugin_dir_path) | |
| if metadata: | |
| if metadata := self._load_plugin_metadata(plugin_dir_path): |
Original comment in English
suggestion (code-quality): Use named expression to simplify assignment and conditional (use-named-expression)
Suggested change
| metadata = self._load_plugin_metadata(plugin_dir_path) | |
| if metadata: | |
| if metadata := self._load_plugin_metadata(plugin_dir_path): |
Comment on lines
+996
to
+997
| smd = star_map.get(specified_module_path) | ||
| if smd: |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): 使用命名表达式简化赋值和条件 (use-named-expression)
Suggested change
| smd = star_map.get(specified_module_path) | |
| if smd: | |
| if smd := star_map.get(specified_module_path): |
Original comment in English
suggestion (code-quality): Use named expression to simplify assignment and conditional (use-named-expression)
Suggested change
| smd = star_map.get(specified_module_path) | |
| if smd: | |
| if smd := star_map.get(specified_module_path): |
Member
|
第一个问题， 第二个问题， |
Contributor
Author
|
都还未支持，当时还没考虑到自动安装依赖插件和具体的版本管理等。 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
此PR为重新提交#1925
Motivation
为可能的插件联动做铺垫，方便监控插件生命周期
Modifications
举例来说：有A插件，其功能一定程度上依赖B插件
则A先于B初始化，待B初始化（启用）时，就能触发A中的on_star_activated钩子
此外，系统插件优先初始化
Verification Steps / 验证步骤
需按上述示例简单编写几个测试插件，启动时日志中可见加载顺序
Screenshots or Test Results / 运行截图或测试结果
Compatibility & Breaking Changes / 兼容性与破坏性变更
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Sourcery 摘要
添加插件激活/停用钩子以及依赖驱动的初始化，以实现插件间的互联和受控的加载顺序
新功能：
改进：
构建：
Original summary in English
Summary by Sourcery
Add plugin activation/deactivation hooks and dependency-driven initialization to enable plugin inter-linking and controlled load order
New Features:
Enhancements:
Build: