mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-16 17:47:06 +08:00
d373a24320
Resolved 76 file conflicts across backend, frontend, and tests: - Core backend: agent, platform, provider, star, tools, cron, KB, utils - Dashboard backend: auth, config, plugin routes, server, runtime bootstrap - Frontend components: shared, extension, provider components - Frontend views: console, trace, cron, extension, platform, provider, etc. - Frontend infrastructure: layouts, router, stores, main.ts - Assets: MDI font subset (css, woff, woff2) - Tests: fixtures, plugin manager, anthropic provider, dashboard tests Strategy per file: - Most backend: merged master improvements (metrics, http_client -> default_headers, download_url params, plugin i18n, runtime guards, auth middleware) - Frontend views: adopted master dashboard-shell layout (v-container, is-dark theme) - InstalledPluginsTab: kept dev pinned-plugins feature (drag-sort, pin/unpin) - Binary assets: resolved via git checkout - Tests: merged both branches' assertions and mock updates
324 lines
11 KiB
Python
324 lines
11 KiB
Python
"""插件开发工具集
|
|
封装了许多常用的操作,方便插件开发者使用
|
|
|
|
说明:
|
|
|
|
主动发送消息: send_message(session, message_chain)
|
|
根据 session (unified_msg_origin) 主动发送消息, 前提是需要提前获得或构造 session
|
|
|
|
根据id直接主动发送消息: send_message_by_id(type, id, message_chain, platform="aiocqhttp")
|
|
根据 id (例如 qq 号, 群号等) 直接, 主动地发送消息
|
|
|
|
以上两种方式需要构造消息链, 也就是消息组件的列表
|
|
|
|
构造事件:
|
|
|
|
首先需要构造一个 AstrBotMessage 对象, 使用 create_message 方法
|
|
然后使用 create_event 方法提交事件到指定平台
|
|
"""
|
|
|
|
import inspect
|
|
import os
|
|
import uuid
|
|
from collections.abc import Awaitable, Callable
|
|
from pathlib import Path
|
|
from typing import Any, ClassVar
|
|
|
|
from astrbot.api.platform import AstrBotMessage, MessageMember, MessageType
|
|
from astrbot.core.message.components import BaseMessageComponent
|
|
from astrbot.core.message.message_event_result import MessageChain
|
|
from astrbot.core.platform.astr_message_event import MessageSesion
|
|
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.utils.io import ensure_dir
|
|
|
|
|
|
class StarTools:
|
|
"""提供给插件使用的便捷工具函数集合
|
|
这些方法封装了一些常用操作,使插件开发更加简单便捷!
|
|
"""
|
|
|
|
_context: ClassVar[Context | None] = None
|
|
|
|
@classmethod
|
|
def initialize(cls, context: Context) -> None:
|
|
"""初始化StarTools,设置context引用
|
|
|
|
Args:
|
|
context: 暴露给插件的上下文
|
|
|
|
"""
|
|
cls._context = context
|
|
|
|
@classmethod
|
|
async def send_message(
|
|
cls,
|
|
session: str | MessageSesion,
|
|
message_chain: MessageChain,
|
|
) -> bool:
|
|
"""根据session(unified_msg_origin)主动发送消息
|
|
|
|
Args:
|
|
session: 消息会话｡通过event.session或者event.unified_msg_origin获取
|
|
message_chain: 消息链
|
|
|
|
Returns:
|
|
bool: 是否找到匹配的平台
|
|
|
|
Raises:
|
|
ValueError: 当session为字符串且解析失败时抛出
|
|
|
|
Note:
|
|
qq_official(QQ官方API平台)不支持此方法
|
|
|
|
"""
|
|
if cls._context is None:
|
|
raise ValueError("StarTools not initialized")
|
|
return await cls._context.send_message(session, message_chain)
|
|
|
|
@classmethod
|
|
async def send_message_by_id(
|
|
cls,
|
|
type: str,
|
|
id: str,
|
|
message_chain: MessageChain,
|
|
platform: str = "aiocqhttp",
|
|
) -> None:
|
|
"""根据 id(例如qq号, 群号等) 直接, 主动地发送消息
|
|
|
|
Args:
|
|
type (str): 消息类型, 可选: PrivateMessage, GroupMessage
|
|
id (str): 目标ID, 例如QQ号, 群号等
|
|
message_chain (MessageChain): 消息链
|
|
platform (str): 可选的平台名称,默认平台(aiocqhttp), 目前只支持 aiocqhttp
|
|
|
|
"""
|
|
if cls._context is None:
|
|
raise ValueError("StarTools not initialized")
|
|
platforms = cls._context.platform_manager.get_insts()
|
|
if platform == "aiocqhttp":
|
|
from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_message_event import (
|
|
AiocqhttpMessageEvent,
|
|
)
|
|
from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_platform_adapter import (
|
|
AiocqhttpAdapter,
|
|
)
|
|
|
|
adapter = next(
|
|
(p for p in platforms if isinstance(p, AiocqhttpAdapter)),
|
|
None,
|
|
)
|
|
if adapter is None:
|
|
raise ValueError("未找到适配器: AiocqhttpAdapter")
|
|
await AiocqhttpMessageEvent.send_message(
|
|
bot=adapter.bot,
|
|
message_chain=message_chain,
|
|
is_group=(type == "GroupMessage"),
|
|
session_id=id,
|
|
)
|
|
else:
|
|
raise ValueError(f"不支持的平台: {platform}")
|
|
|
|
@classmethod
|
|
async def create_message(
|
|
cls,
|
|
type: str,
|
|
self_id: str,
|
|
session_id: str,
|
|
sender: MessageMember,
|
|
message: list[BaseMessageComponent],
|
|
message_str: str,
|
|
message_id: str = "",
|
|
raw_message: object = None,
|
|
group_id: str = "",
|
|
) -> AstrBotMessage:
|
|
"""创建一个AstrBot消息对象
|
|
|
|
Args:
|
|
type (str): 消息类型, 例如 "GroupMessage" "FriendMessage" "OtherMessage"
|
|
self_id (str): 机器人自身ID
|
|
session_id (str): 会话ID(通常为用户ID)(QQ号, 群号等)
|
|
sender (MessageMember): 发送者信息, 例如 MessageMember(user_id="123456", nickname="昵称")
|
|
message (List[BaseMessageComponent]): 消息组件列表, 也就是消息链, 这个不会发给 llm, 但是会经过其他处理
|
|
message_str (str): 消息字符串, 也就是纯文本消息, 也就是发送给 llm 的消息, 与消息链一致
|
|
|
|
message_id (str): 消息ID, 构造消息时可以随意填写也可不填
|
|
raw_message (object): 原始消息对象, 可以随意填写也可不填
|
|
group_id (str, optional): 群组ID, 如果为私聊则为空. Defaults to "".
|
|
|
|
Returns:
|
|
AstrBotMessage: 创建的消息对象
|
|
|
|
"""
|
|
abm = AstrBotMessage()
|
|
abm.type = MessageType(type)
|
|
abm.self_id = self_id
|
|
abm.session_id = session_id
|
|
if message_id == "":
|
|
message_id = uuid.uuid4().hex
|
|
abm.message_id = message_id
|
|
abm.sender = sender
|
|
abm.message = message
|
|
abm.message_str = message_str
|
|
abm.raw_message = raw_message
|
|
abm.group_id = group_id
|
|
return abm
|
|
|
|
@classmethod
|
|
async def create_event(
|
|
cls,
|
|
abm: AstrBotMessage,
|
|
platform: str = "aiocqhttp",
|
|
is_wake: bool = True,
|
|
) -> None:
|
|
"""创建并提交事件到指定平台
|
|
当有需要创建一个事件, 触发某些处理流程时, 使用该方法
|
|
|
|
Args:
|
|
abm (AstrBotMessage): 要提交的消息对象, 请先使用 create_message 创建
|
|
platform (str): 可选的平台名称,默认平台(aiocqhttp), 目前只支持 aiocqhttp
|
|
is_wake (bool): 是否标记为唤醒事件, 默认为 True, 只有唤醒事件才会被 llm 响应
|
|
|
|
"""
|
|
if cls._context is None:
|
|
raise ValueError("StarTools not initialized")
|
|
platforms = cls._context.platform_manager.get_insts()
|
|
if platform == "aiocqhttp":
|
|
from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_message_event import (
|
|
AiocqhttpMessageEvent,
|
|
)
|
|
from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_platform_adapter import (
|
|
AiocqhttpAdapter,
|
|
)
|
|
|
|
adapter = next(
|
|
(p for p in platforms if isinstance(p, AiocqhttpAdapter)),
|
|
None,
|
|
)
|
|
if adapter is None:
|
|
raise ValueError("未找到适配器: AiocqhttpAdapter")
|
|
event = AiocqhttpMessageEvent(
|
|
message_str=abm.message_str,
|
|
message_obj=abm,
|
|
platform_meta=adapter.metadata,
|
|
session_id=abm.session_id,
|
|
bot=adapter.bot,
|
|
)
|
|
event.is_wake = is_wake
|
|
adapter.commit_event(event)
|
|
else:
|
|
raise ValueError(f"不支持的平台: {platform}")
|
|
|
|
@classmethod
|
|
def activate_llm_tool(cls, name: str) -> bool:
|
|
"""激活一个已经注册的函数调用工具
|
|
注册的工具默认是激活状态
|
|
|
|
Args:
|
|
name (str): 工具名称
|
|
|
|
"""
|
|
if cls._context is None:
|
|
raise ValueError("StarTools not initialized")
|
|
return cls._context.activate_llm_tool(name)
|
|
|
|
@classmethod
|
|
def deactivate_llm_tool(cls, name: str) -> bool:
|
|
"""停用一个已经注册的函数调用工具
|
|
|
|
Args:
|
|
name (str): 工具名称
|
|
|
|
"""
|
|
if cls._context is None:
|
|
raise ValueError("StarTools not initialized")
|
|
return cls._context.deactivate_llm_tool(name)
|
|
|
|
@classmethod
|
|
def register_llm_tool(
|
|
cls,
|
|
name: str,
|
|
func_args: list,
|
|
desc: str,
|
|
func_obj: Callable[..., Awaitable[Any]],
|
|
) -> None:
|
|
"""为函数调用(function-calling/tools-use)添加工具
|
|
|
|
Args:
|
|
name (str): 工具名称
|
|
func_args (list): 函数参数列表
|
|
desc (str): 工具描述
|
|
func_obj (Awaitable): 函数对象,必须是异步函数
|
|
|
|
"""
|
|
if cls._context is None:
|
|
raise ValueError("StarTools not initialized")
|
|
cls._context.register_llm_tool(name, func_args, desc, func_obj)
|
|
|
|
@classmethod
|
|
def unregister_llm_tool(cls, name: str) -> None:
|
|
"""删除一个函数调用工具
|
|
如果再要启用,需要重新注册
|
|
|
|
Args:
|
|
name (str): 工具名称
|
|
|
|
"""
|
|
if cls._context is None:
|
|
raise ValueError("StarTools not initialized")
|
|
cls._context.unregister_llm_tool(name)
|
|
|
|
@classmethod
|
|
def get_data_dir(cls, plugin_name: str | None = None) -> Path:
|
|
"""返回插件数据目录的绝对路径｡
|
|
|
|
此方法会在 data/plugin_data 目录下为插件创建一个专属的数据目录｡如果未提供插件名称,
|
|
会自动从调用栈中获取插件信息｡
|
|
|
|
Args:
|
|
plugin_name: 可选的插件名称｡如果为None,将自动检测调用者的插件名称｡
|
|
|
|
Returns:
|
|
Path (Path): 插件数据目录的绝对路径,位于 data/plugin_data/{plugin_name}｡
|
|
|
|
Raises:
|
|
RuntimeError: 当出现以下情况时抛出:
|
|
- 无法获取调用者模块信息
|
|
- 无法获取模块的元数据信息
|
|
- 创建目录失败(权限不足或其他IO错误)
|
|
|
|
"""
|
|
if not plugin_name:
|
|
frame = inspect.currentframe()
|
|
module = None
|
|
if frame:
|
|
frame = frame.f_back
|
|
module = inspect.getmodule(frame)
|
|
|
|
if not module:
|
|
raise RuntimeError("无法获取调用者模块信息")
|
|
|
|
metadata = star_map.get(module.__name__, None)
|
|
|
|
if not metadata:
|
|
raise RuntimeError(f"无法获取模块 {module.__name__} 的元数据信息")
|
|
|
|
plugin_name = metadata.name
|
|
|
|
if not plugin_name:
|
|
raise ValueError("无法获取插件名称")
|
|
|
|
data_dir = Path(
|
|
os.path.join(get_astrbot_data_path(), "plugin_data", plugin_name),
|
|
)
|
|
|
|
try:
|
|
ensure_dir(data_dir)
|
|
except OSError as e:
|
|
if isinstance(e, PermissionError):
|
|
raise RuntimeError(f"无法创建目录 {data_dir}:权限不足") from e
|
|
raise RuntimeError(f"无法创建目录 {data_dir}:{e!s}") from e
|
|
|
|
return data_dir.resolve()
|