Discord 适配器 API 参考
概述
Discord 适配器提供完整的 Discord 平台集成功能,包括消息处理、图像管理、语音连接等。所有功能基于 discord.py 库的原生 API 实现。
核心 API
DiscordAdapter 类
Discord 平台适配器主类,提供与 Discord 平台交互的完整接口。
消息相关方法
send_message(target_type, target_id, message)
发送消息到指定频道。
参数:
target_type
(str): 目标类型,通常为 "channel"target_id
(str): 目标频道 IDmessage
(MessageChain): 要发送的消息链
功能:
- 支持文本消息发送
- 支持图像文件批量发送
- 自动处理消息格式转换
示例:
# 发送纯文本消息
await adapter.send_message("channel", "123456789", MessageChain([Plain("Hello!")]))
# 发送带图像的消息
message_chain = MessageChain([
Plain("看看这张图片:"),
Image(path="path/to/image.jpg"),
Image(url="https://example.com/image.png")
])
await adapter.send_message("channel", "123456789", message_chain)
reply_message(message_source, message, quote_origin=False)
回复指定消息。
参数:
message_source
(MessageEvent): 要回复的原始消息事件message
(MessageChain): 回复内容quote_origin
(bool): 是否引用原消息
示例:
# 简单回复
await adapter.reply_message(
message_event,
MessageChain([Plain("收到你的消息了!")])
)
# 引用回复
await adapter.reply_message(
message_event,
MessageChain([Plain("这是对你消息的回复")]),
quote_origin=True
)
语音相关方法
join_voice_channel(guild_id, channel_id, user_id=None)
加入指定的语音频道。
参数:
guild_id
(int): Discord 服务器 IDchannel_id
(int): 语音频道 IDuser_id
(int, optional): 请求用户 ID,用于权限验证
返回值:
discord.VoiceClient
: 语音客户端对象
异常:
VoicePermissionError
: 权限不足VoiceNetworkError
: 网络连接问题VoiceConnectionError
: 其他连接错误
示例:
try:
voice_client = await adapter.join_voice_channel(
guild_id=123456789,
channel_id=987654321,
user_id=555666777
)
print(f"成功连接,延迟: {voice_client.latency*1000:.2f}ms")
except VoicePermissionError as e:
print(f"权限错误: {e.missing_permissions}")
leave_voice_channel(guild_id)
离开指定服务器的语音频道。
参数:
guild_id
(int): Discord 服务器 ID
返回值:
bool
: 成功断开返回True
示例:
success = await adapter.leave_voice_channel(guild_id=123456789)
if success:
print("成功离开语音频道")
get_voice_client(guild_id)
获取指定服务器的语音客户端对象。
参数:
guild_id
(int): Discord 服务器 ID
返回值:
discord.VoiceClient
: 语音客户端对象,如果存在连接None
: 如果没有活跃连接
示例:
voice_client = await adapter.get_voice_client(guild_id=123456789)
if voice_client:
# 播放音频
import discord
audio = discord.FFmpegPCMAudio("path/to/audio.mp3")
voice_client.play(audio)
is_connected_to_voice(guild_id)
检查是否连接到指定服务器的语音频道。
参数:
guild_id
(int): Discord 服务器 ID
返回值:
bool
: 已连接返回True
示例:
if await adapter.is_connected_to_voice(guild_id=123456789):
print("已连接到语音频道")
get_voice_connection_status(guild_id)
获取语音连接的详细状态信息。
参数:
guild_id
(int): Discord 服务器 ID
返回值:
dict
: 连接状态信息字典,如果存在连接None
: 如果没有连接
状态信息结构:
{
'guild_id': int,
'channel_id': int,
'channel_name': str,
'connected': bool,
'connection_time': datetime,
'last_activity': datetime,
'status': str, # VoiceConnectionStatus 枚举值
'user_count': int,
'latency': float,
'connection_health': str
}
list_active_voice_connections()
获取所有活跃的语音连接列表。
返回值:
List[dict]
: 活跃连接状态列表
示例:
connections = await adapter.list_active_voice_connections()
for conn in connections:
print(f"服务器: {conn['guild_id']}, 频道: {conn['channel_name']}")
get_voice_channel_info(guild_id, channel_id)
获取指定语音频道的详细信息。
参数:
guild_id
(int): 服务器 IDchannel_id
(int): 频道 ID
返回值:
dict
: 频道信息字典None
: 如果频道不存在
频道信息结构:
{
'channel_id': int,
'channel_name': str,
'guild_id': int,
'guild_name': str,
'user_limit': int,
'current_users': List[dict],
'user_count': int,
'bitrate': int,
'permissions': dict
}
管理方法
is_muted(group_id)
检查群组是否被静音(Discord 适配器始终返回 False)。
参数:
group_id
(int): 群组 ID
返回值:
bool
: 是否静音
register_listener(event_type, callback)
注册事件监听器。
参数:
event_type
(Type[Event]): 事件类型callback
(Callable): 回调函数
unregister_listener(event_type, callback)
注销事件监听器。
参数:
event_type
(Type[Event]): 事件类型callback
(Callable): 回调函数
run_async()
启动 Discord 适配器,建立与 Discord 的连接。
功能:
- 初始化语音管理器
- 启动连接监控
- 建立 WebSocket 连接
kill()
关闭 Discord 适配器。
返回值:
bool
: 是否成功关闭
功能:
- 断开所有语音连接
- 关闭 Discord 客户端
- 清理资源
消息转换器 API
DiscordMessageConverter 类
负责在 LangBot 和 Discord 消息格式之间进行转换。
yiri2target(message_chain)
将 LangBot MessageChain 转换为 Discord 格式。
参数:
message_chain
(MessageChain): LangBot 消息链
返回值:
Tuple[str, List[discord.File]]
: (文本内容, 文件列表)
支持的消息元素:
Plain
: 纯文本内容Image
: 图像内容(支持 base64、URL、文件路径)At
: @ 提及(会被移除)Forward
: 转发消息
图像处理特性:
- Base64 图像: 自动检测格式并解码
- URL 图像: 异步下载并转换
- 文件路径: 读取本地文件
- 格式检测: 支持 PNG、JPG、GIF、WebP
- 错误处理: 跳过无效图像
示例:
message_chain = MessageChain([
Plain("文本内容"),
Image(base64="data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg=="),
Image(url="https://example.com/image.jpg"),
Image(path="/path/to/local/image.png")
])
text, files = await DiscordMessageConverter.yiri2target(message_chain)
# text: "文本内容"
# files: [discord.File, discord.File, discord.File]
target2yiri(message)
将 Discord 消息转换为 LangBot MessageChain。
参数:
message
(discord.Message): Discord 消息对象
返回值:
MessageChain
: LangBot 消息链
转换规则:
- 文本内容 →
Plain
元素 - 附件图像 →
Image
元素 - 嵌入内容 →
Plain
元素(描述文本)
事件转换器 API
DiscordEventConverter 类
负责将 Discord 事件转换为 LangBot 平台事件。
target2yiri(message)
将 Discord 消息转换为 LangBot MessageEvent。
参数:
message
(discord.Message): Discord 消息对象
返回值:
MessageEvent
: LangBot 消息事件
事件字段:
message_chain
: 转换后的消息链sender
: 发送者信息query
: 查询上下文信息
异常类
VoiceConnectionError
基础语音连接异常类。
属性:
error_code
(str): 错误代码guild_id
(int): 相关服务器 IDtimestamp
(datetime): 错误发生时间
常见错误代码:
"GUILD_NOT_FOUND"
: 服务器不存在"CHANNEL_NOT_FOUND"
: 频道不存在"MANAGER_NOT_READY"
: 语音管理器未初始化"OPUS_NOT_LOADED"
: Opus编码器未加载"UNKNOWN_ERROR"
: 未知错误
VoicePermissionError
语音权限异常,继承自 VoiceConnectionError
。
属性:
missing_permissions
(List[str]): 缺失的权限列表user_id
(int): 相关用户 IDchannel_id
(int): 相关频道 ID
常见权限错误:
"user_not_in_channel"
: 用户不在语音频道中"member_not_found"
: 找不到用户"connect"
: Bot缺少连接权限"speak"
: Bot缺少发言权限
VoiceNetworkError
语音网络异常,继承自 VoiceConnectionError
。
属性:
retry_count
(int): 重试次数last_attempt
(datetime): 最后尝试时间
使用示例
综合消息处理
async def handle_complex_message(adapter, channel_id):
"""处理包含多种元素的复杂消息"""
# 构建包含文本和图像的消息
message_chain = MessageChain([
Plain("这是一个包含多种内容的消息:\\n"),
Plain("1. 文本内容\\n"),
Plain("2. 网络图像\\n"),
Image(url="https://example.com/image1.jpg"),
Plain("3. 本地图像\\n"),
Image(path="/path/to/local/image.png"),
Plain("4. Base64 图像\\n"),
Image(base64="data:image/png;base64,iVBORw0KGgo...")
])
# 发送消息
await adapter.send_message("channel", channel_id, message_chain)
语音播放控制
async def voice_player_example(adapter, guild_id, channel_id, audio_files):
"""语音播放器示例"""
# 连接到语音频道
try:
voice_client = await adapter.join_voice_channel(guild_id, channel_id)
except VoicePermissionError as e:
print(f"权限不足: {e.missing_permissions}")
return
# 播放音频队列
for audio_file in audio_files:
if voice_client.is_connected():
# 等待当前播放完成
while voice_client.is_playing():
await asyncio.sleep(1)
# 播放下一个音频
audio_source = discord.FFmpegPCMAudio(audio_file)
voice_client.play(audio_source)
print(f"正在播放: {audio_file}")
# 播放完成后离开频道
await adapter.leave_voice_channel(guild_id)
状态监控
async def monitor_connections(adapter):
"""监控所有语音连接状态"""
connections = await adapter.list_active_voice_connections()
for conn in connections:
print(f"服务器: {conn['guild_id']}")
print(f"频道: {conn['channel_name']}")
print(f"延迟: {conn['latency']:.2f}ms")
print(f"用户数: {conn['user_count']}")
print(f"健康度: {conn['connection_health']}")
print("---")
最佳实践
错误处理
async def safe_discord_operation():
try:
# Discord 操作
pass
except VoicePermissionError as e:
if "user_not_in_channel" in e.missing_permissions:
await send_error("请先加入语音频道")
elif "connect" in e.missing_permissions:
await send_error("机器人没有连接权限")
except discord.HTTPException as e:
if e.status == 403:
await send_error("权限不足")
elif e.status == 404:
await send_error("频道不存在")
except Exception as e:
await send_error(f"未知错误: {e}")
资源管理
async def with_voice_connection(adapter, guild_id, channel_id, operation):
"""安全的语音连接管理"""
voice_client = None
try:
voice_client = await adapter.join_voice_channel(guild_id, channel_id)
await operation(voice_client)
finally:
if voice_client:
await adapter.leave_voice_channel(guild_id)
图像处理优化
async def batch_image_processing(image_list):
"""批量图像处理"""
processed_images = []
for image_element in image_list:
try:
if image_element.url:
# 异步下载
async with aiohttp.ClientSession() as session:
async with session.get(image_element.url) as response:
image_data = await response.read()
processed_images.append(process_image_data(image_data))
elif image_element.path:
# 本地文件处理
with open(image_element.path, 'rb') as f:
image_data = f.read()
processed_images.append(process_image_data(image_data))
except Exception as e:
print(f"图像处理失败: {e}")
continue
return processed_images
注意: 本 API 文档基于实际的 discord.py 实现,涵盖了 Discord 适配器的完整功能集。图像处理功能无需额外依赖,语音功能需要安装可选依赖。