feat: 添加Discord适配器与跨平台消息互通功能
新增Discord适配器支持，实现Discord与QQ之间的消息互通 添加通用数据模型用于跨平台消息转换 扩展配置系统以支持Discord和日志配置 重构日志系统以使用配置中的日志级别 在反向WebSocket管理器中注册Bot实例 更新主程序以支持Discord客户端启动 添加测试脚本验证核心功能
This commit is contained in:
136
adapters/discord_adapter.py
Normal file
136
adapters/discord_adapter.py
Normal file
@@ -0,0 +1,136 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Discord 适配器 (Discord Adapter)
|
||||
|
||||
此模块负责与 Discord API 建立连接，接收 Discord 消息，
|
||||
并将其转换为通用数据模型 (Universal Data Models)，
|
||||
同时提供将通用消息段发送回 Discord 的能力。
|
||||
"""
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import io
|
||||
import requests
|
||||
from typing import Union, List, Optional
|
||||
|
||||
try:
|
||||
import discord
|
||||
DISCORD_AVAILABLE = True
|
||||
except ImportError:
|
||||
DISCORD_AVAILABLE = False
|
||||
|
||||
from core.utils.logger import ModuleLogger
|
||||
from .router import DiscordToOneBotConverter
|
||||
from core.managers.redis_manager import redis_manager
|
||||
|
||||
class DiscordAdapter(discord.Client if DISCORD_AVAILABLE else object):
|
||||
"""
|
||||
Discord 客户端适配器。
|
||||
继承自 discord.Client，负责处理 Discord 的底层事件。
|
||||
"""
|
||||
def __init__(self, token: str):
|
||||
if not DISCORD_AVAILABLE:
|
||||
raise ImportError("discord.py 未安装，请运行 `pip install discord.py`")
|
||||
|
||||
# 必须声明 Intents，否则无法读取消息内容
|
||||
intents = discord.Intents.default()
|
||||
intents.message_content = True
|
||||
|
||||
super().__init__(intents=intents)
|
||||
self.token = token
|
||||
self.logger = ModuleLogger("DiscordAdapter")
|
||||
self.send_channel = None
|
||||
|
||||
async def on_ready(self):
|
||||
"""当 Bot 成功连接到 Discord 时触发"""
|
||||
self.logger.success(f"Discord Bot 已登录: {self.user} (ID: {self.user.id})")
|
||||
|
||||
# 启动 Redis 订阅以处理跨平台消息
|
||||
asyncio.create_task(self.start_redis_subscription())
|
||||
|
||||
async def on_message(self, message: 'discord.Message'):
|
||||
"""当收到 Discord 消息时触发"""
|
||||
# 忽略机器人自己的消息
|
||||
if message.author.bot:
|
||||
return
|
||||
|
||||
self.logger.info(f"[Discord 消息] {message.author}: {message.content}")
|
||||
|
||||
# 1. 将 discord.Message 伪装成 OneBot 事件模型
|
||||
# 2. 触发业务逻辑
|
||||
# 将伪装后的事件丢给现有的命令管理器 (matcher)
|
||||
from core.managers.command_manager import matcher
|
||||
|
||||
# matcher.handle_event 需要 bot 实例和 event 实例
|
||||
# 我们在 create_mock_event 中已经注入了一个假的 bot 对象
|
||||
try:
|
||||
mock_event = DiscordToOneBotConverter.create_mock_event(message, self)
|
||||
await matcher.handle_event(mock_event.bot, mock_event)
|
||||
except Exception as e:
|
||||
self.logger.error(f"处理 Discord 消息时发生异常: {e}")
|
||||
|
||||
async def start_redis_subscription(self):
|
||||
"""启动 Redis 订阅以处理跨平台消息发送"""
|
||||
if redis_manager.redis is None:
|
||||
self.logger.warning("[DiscordAdapter] Redis 未初始化，跳过订阅")
|
||||
return
|
||||
|
||||
try:
|
||||
channel_name = "neobot_discord_send"
|
||||
pubsub = redis_manager.redis.pubsub()
|
||||
await pubsub.subscribe(channel_name)
|
||||
|
||||
self.logger.success(f"[DiscordAdapter] 已订阅 Redis 频道: {channel_name}")
|
||||
|
||||
async for message in pubsub.listen():
|
||||
if message["type"] == "message":
|
||||
try:
|
||||
data = json.loads(message["data"])
|
||||
if data.get("type") == "send_message":
|
||||
await self.handle_send_message(data)
|
||||
except json.JSONDecodeError as e:
|
||||
self.logger.error(f"[DiscordAdapter] 解析 Redis 消息失败: {e}")
|
||||
except Exception as e:
|
||||
self.logger.error(f"[DiscordAdapter] 处理 Redis 消息失败: {e}")
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"[DiscordAdapter] Redis 订阅异常: {e}")
|
||||
|
||||
async def handle_send_message(self, data: dict):
|
||||
"""处理来自 Redis 的消息发送请求"""
|
||||
try:
|
||||
channel_id = data.get("channel_id")
|
||||
content = data.get("content", "")
|
||||
attachments = data.get("attachments", [])
|
||||
|
||||
if channel_id is None:
|
||||
self.logger.error("[DiscordAdapter] 缺少 channel_id")
|
||||
return
|
||||
|
||||
channel = self.get_channel(channel_id)
|
||||
if channel is None:
|
||||
self.logger.error(f"[DiscordAdapter] 未找到频道: {channel_id}")
|
||||
return
|
||||
|
||||
self.logger.info(f"[DiscordAdapter] 正在发送消息到频道 {channel_id}")
|
||||
|
||||
# 发送内容和附件（合并为一条消息）
|
||||
if content or attachments:
|
||||
await channel.send(content=content, files=[discord.File(fp=io.BytesIO(requests.get(attachment_url).content), filename=os.path.basename(attachment_url)) for attachment_url in attachments if attachment_url.startswith('http')] if attachments else None)
|
||||
|
||||
self.logger.success(f"[DiscordAdapter] 消息已发送到频道 {channel_id}")
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"[DiscordAdapter] 发送消息失败: {e}")
|
||||
|
||||
async def start_client(self):
|
||||
"""启动 Discord 客户端（非阻塞方式）"""
|
||||
if not DISCORD_AVAILABLE:
|
||||
self.logger.error("无法启动 Discord 客户端：discord.py 未安装")
|
||||
return
|
||||
|
||||
try:
|
||||
self.logger.info("正在连接 Discord...")
|
||||
await self.start(self.token)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Discord 连接失败: {e}")
|
||||
406
adapters/router.py
Normal file
406
adapters/router.py
Normal file
@@ -0,0 +1,406 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
事件路由与转换器 (Event Router & Converter)
|
||||
|
||||
此模块负责在不同平台（如 Discord）和 OneBot 业务逻辑之间进行数据转换。
|
||||
核心目标是：**让现有的 OneBot 插件（如 bili.py）在不修改任何代码的情况下，能够处理 Discord 消息。**
|
||||
|
||||
实现原理：
|
||||
1. 接收 Discord 消息 (`discord.Message`)。
|
||||
2. 将其“伪装”成 OneBot 的 `GroupMessageEvent` 或 `PrivateMessageEvent`。
|
||||
3. 拦截插件调用的 `event.reply()` 方法。
|
||||
4. 将插件返回的 OneBot `MessageSegment` 转换为 Discord 格式并发送。
|
||||
"""
|
||||
import asyncio
|
||||
from typing import Union, List, Any, Optional
|
||||
|
||||
try:
|
||||
import discord
|
||||
DISCORD_AVAILABLE = True
|
||||
except ImportError:
|
||||
DISCORD_AVAILABLE = False
|
||||
|
||||
from models.events.message import GroupMessageEvent, PrivateMessageEvent
|
||||
from models.message import MessageSegment as OneBotMessageSegment
|
||||
from models.sender import Sender
|
||||
from core.utils.logger import ModuleLogger
|
||||
|
||||
logger = ModuleLogger("EventRouter")
|
||||
|
||||
class DiscordToOneBotConverter:
|
||||
"""
|
||||
将 Discord 消息转换为 OneBot 消息事件的转换器。
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def create_mock_event(discord_message: 'discord.Message', adapter: Any) -> Union[GroupMessageEvent, PrivateMessageEvent]:
|
||||
"""
|
||||
将 discord.Message 伪装成 OneBot 的 MessageEvent。
|
||||
|
||||
Args:
|
||||
discord_message: 原始的 Discord 消息对象
|
||||
adapter: DiscordAdapter 实例，用于回调发送消息
|
||||
|
||||
Returns:
|
||||
伪装后的 OneBot 事件对象
|
||||
"""
|
||||
# 1. 提取基础信息
|
||||
user_id = discord_message.author.id
|
||||
message_id = discord_message.id
|
||||
|
||||
# 处理 Discord 的 raw_message
|
||||
# 如果消息是以 @机器人 开头，Discord 的 content 会是 "<@机器人ID> /echo 1"
|
||||
# 我们需要把前面的 @ 提及去掉，否则命令匹配器 (matcher) 无法识别以 "/" 开头的命令
|
||||
raw_message = discord_message.content
|
||||
|
||||
# 添加附件信息到 raw_message
|
||||
if discord_message.attachments:
|
||||
for attachment in discord_message.attachments:
|
||||
raw_message += f"\n{attachment.url}"
|
||||
bot_mention = f"<@{adapter.user.id}>"
|
||||
if raw_message.startswith(bot_mention):
|
||||
raw_message = raw_message[len(bot_mention):].strip()
|
||||
|
||||
# 构造发送者信息
|
||||
sender = Sender(
|
||||
user_id=user_id,
|
||||
nickname=discord_message.author.display_name,
|
||||
card=getattr(discord_message.author, 'nick', ''), # 群名片
|
||||
role="member" # 简化处理，默认都是普通成员
|
||||
)
|
||||
|
||||
# 2. 判断是群聊还是私聊
|
||||
is_private = isinstance(discord_message.channel, discord.DMChannel)
|
||||
|
||||
# 构造 message 列表 (将纯文本转换为 MessageSegment)
|
||||
message_list = [OneBotMessageSegment.text(raw_message)]
|
||||
|
||||
import time
|
||||
current_time = int(time.time())
|
||||
self_id = adapter.user.id if adapter.user else 0
|
||||
|
||||
# 注入 Discord 特定信息（用于跨平台插件识别）
|
||||
discord_channel_id = discord_message.channel.id if not isinstance(discord_message.channel, discord.DMChannel) else None
|
||||
discord_username = discord_message.author.name
|
||||
discord_discriminator = f"#{discord_message.author.discriminator}" if discord_message.author.discriminator != "0" else ""
|
||||
|
||||
if is_private:
|
||||
# 构造私聊事件
|
||||
event = PrivateMessageEvent(
|
||||
time=current_time,
|
||||
self_id=self_id,
|
||||
message_type="private",
|
||||
sub_type="friend",
|
||||
message_id=message_id,
|
||||
user_id=user_id,
|
||||
raw_message=raw_message,
|
||||
message=message_list,
|
||||
sender=sender
|
||||
)
|
||||
else:
|
||||
# 构造群聊事件
|
||||
group_id = discord_message.channel.id
|
||||
event = GroupMessageEvent(
|
||||
time=current_time,
|
||||
self_id=self_id,
|
||||
message_type="group",
|
||||
sub_type="normal",
|
||||
message_id=message_id,
|
||||
user_id=user_id,
|
||||
group_id=group_id,
|
||||
raw_message=raw_message,
|
||||
message=message_list,
|
||||
sender=sender
|
||||
)
|
||||
|
||||
# 注入 Discord 特定属性（用于跨平台插件识别）
|
||||
event._is_discord_message = True
|
||||
event.discord_channel_id = discord_channel_id
|
||||
event.discord_username = discord_username
|
||||
event.discord_discriminator = discord_discriminator
|
||||
|
||||
# 3. 拦截并重写 reply 方法 (核心魔法)
|
||||
# 插件调用 event.reply() 时，实际上会执行这个闭包
|
||||
async def mock_reply(message: Union[str, OneBotMessageSegment, List[OneBotMessageSegment]], auto_escape: bool = False):
|
||||
await DiscordToOneBotConverter.send_discord_reply(discord_message, message, adapter)
|
||||
|
||||
# 覆盖实例方法
|
||||
event.reply = mock_reply
|
||||
|
||||
# 注入一个假的 bot 对象，防止插件调用 event.bot.xxx 时报错
|
||||
# 这里只提供最基础的属性，如果插件调用了复杂的 API，可能会报错
|
||||
class MockBot:
|
||||
def __init__(self):
|
||||
self.self_id = adapter.user.id if adapter.user else 0
|
||||
|
||||
async def send(self, event, message, **kwargs):
|
||||
await DiscordToOneBotConverter.send_discord_reply(discord_message, message, adapter)
|
||||
|
||||
async def send_forwarded_messages(self, target, nodes):
|
||||
"""
|
||||
模拟发送合并转发消息。
|
||||
Discord 不支持像 QQ 那样的合并转发，所以我们将其转换为普通消息发送。
|
||||
"""
|
||||
content = ""
|
||||
files = []
|
||||
|
||||
for node in nodes:
|
||||
if node.get("type") == "node":
|
||||
node_data = node.get("data", {})
|
||||
node_content = node_data.get("content", [])
|
||||
|
||||
# 提取节点中的文本和图片
|
||||
if isinstance(node_content, str):
|
||||
# 尝试解析 CQ 码
|
||||
import re
|
||||
cq_pattern = r'\[CQ:([^,]+)(?:,([^\]]+))?\]'
|
||||
matches = list(re.finditer(cq_pattern, node_content))
|
||||
|
||||
if not matches:
|
||||
content += f"{node_content}\n"
|
||||
else:
|
||||
last_end = 0
|
||||
for match in matches:
|
||||
if match.start() > last_end:
|
||||
content += node_content[last_end:match.start()]
|
||||
|
||||
cq_type = match.group(1)
|
||||
cq_params_str = match.group(2) or ""
|
||||
|
||||
params = {}
|
||||
if cq_params_str:
|
||||
for param in cq_params_str.split(','):
|
||||
if '=' in param:
|
||||
k, v = param.split('=', 1)
|
||||
params[k] = v
|
||||
|
||||
if cq_type in ("image", "video"):
|
||||
file_url = params.get("url") or params.get("file")
|
||||
if file_url:
|
||||
if str(file_url).startswith("http"):
|
||||
content += f"\n{file_url}\n"
|
||||
elif str(file_url).startswith("base64://"):
|
||||
import base64
|
||||
import io
|
||||
b64_data = str(file_url)[9:]
|
||||
if b64_data.startswith("data:image"):
|
||||
b64_data = b64_data.split(",", 1)[1]
|
||||
try:
|
||||
image_bytes = base64.b64decode(b64_data)
|
||||
files.append(discord.File(fp=io.BytesIO(image_bytes), filename="image.png"))
|
||||
except Exception as e:
|
||||
logger.error(f"解析 Base64 图片失败: {e}")
|
||||
else:
|
||||
try:
|
||||
files.append(discord.File(file_url))
|
||||
except Exception as e:
|
||||
logger.error(f"无法读取本地文件 {file_url}: {e}")
|
||||
elif cq_type == "at":
|
||||
qq_id = params.get("qq")
|
||||
if qq_id == "all":
|
||||
content += "@everyone "
|
||||
else:
|
||||
content += f"<@{qq_id}> "
|
||||
|
||||
last_end = match.end()
|
||||
|
||||
if last_end < len(node_content):
|
||||
content += node_content[last_end:]
|
||||
content += "\n"
|
||||
elif isinstance(node_content, list):
|
||||
for seg in node_content:
|
||||
if isinstance(seg, dict):
|
||||
seg_type = seg.get("type")
|
||||
seg_data = seg.get("data", {})
|
||||
|
||||
if seg_type == "text":
|
||||
content += seg_data.get("text", "")
|
||||
elif seg_type == "image" or seg_type == "video":
|
||||
file_url = seg_data.get("url") or seg_data.get("file")
|
||||
if file_url:
|
||||
if isinstance(file_url, bytes):
|
||||
import io
|
||||
try:
|
||||
files.append(discord.File(fp=io.BytesIO(file_url), filename="image.png"))
|
||||
except Exception as e:
|
||||
logger.error(f"解析 bytes 图片失败: {e}")
|
||||
elif str(file_url).startswith("http"):
|
||||
content += f"\n{file_url}\n"
|
||||
elif str(file_url).startswith("base64://") or "data:image" in str(file_url):
|
||||
import base64
|
||||
import io
|
||||
b64_data = str(file_url)
|
||||
if b64_data.startswith("base64://"):
|
||||
b64_data = b64_data[9:]
|
||||
if b64_data.startswith("data:image"):
|
||||
b64_data = b64_data.split(",", 1)[1]
|
||||
try:
|
||||
image_bytes = base64.b64decode(b64_data)
|
||||
files.append(discord.File(fp=io.BytesIO(image_bytes), filename="image.png"))
|
||||
except Exception as e:
|
||||
logger.error(f"解析 Base64 图片失败: {e}")
|
||||
else:
|
||||
try:
|
||||
files.append(discord.File(file_url))
|
||||
except Exception as e:
|
||||
logger.error(f"无法读取本地文件 {file_url}: {e}")
|
||||
content += "\n"
|
||||
|
||||
try:
|
||||
if content or files:
|
||||
await discord_message.channel.send(content=content, files=files if files else None)
|
||||
except Exception as e:
|
||||
logger.error(f"发送 Discord 合并转发消息失败: {e}")
|
||||
|
||||
event.bot = MockBot()
|
||||
|
||||
return event
|
||||
|
||||
@staticmethod
|
||||
async def send_discord_reply(
|
||||
original_message: 'discord.Message',
|
||||
message: Union[str, OneBotMessageSegment, List[OneBotMessageSegment]],
|
||||
adapter: Any
|
||||
):
|
||||
"""
|
||||
将 OneBot 的消息段转换为 Discord 格式并发送。
|
||||
|
||||
Args:
|
||||
original_message: 触发此回复的原始 Discord 消息
|
||||
message: 插件返回的 OneBot 消息内容 (字符串或 MessageSegment 列表)
|
||||
adapter: DiscordAdapter 实例
|
||||
"""
|
||||
content = ""
|
||||
files = []
|
||||
|
||||
# 统一转换为列表处理
|
||||
if not isinstance(message, list):
|
||||
message = [message]
|
||||
|
||||
import re
|
||||
|
||||
for segment in message:
|
||||
if isinstance(segment, str):
|
||||
# 尝试解析 CQ 码
|
||||
cq_pattern = r'\[CQ:([^,]+)(?:,([^\]]+))?\]'
|
||||
matches = list(re.finditer(cq_pattern, segment))
|
||||
|
||||
if not matches:
|
||||
content += segment
|
||||
continue
|
||||
|
||||
last_end = 0
|
||||
for match in matches:
|
||||
# 添加 CQ 码之前的纯文本
|
||||
if match.start() > last_end:
|
||||
content += segment[last_end:match.start()]
|
||||
|
||||
cq_type = match.group(1)
|
||||
cq_params_str = match.group(2) or ""
|
||||
|
||||
# 解析参数
|
||||
params = {}
|
||||
if cq_params_str:
|
||||
for param in cq_params_str.split(','):
|
||||
if '=' in param:
|
||||
k, v = param.split('=', 1)
|
||||
params[k] = v
|
||||
|
||||
if cq_type in ("image", "video"):
|
||||
file_url = params.get("url") or params.get("file")
|
||||
if file_url:
|
||||
if str(file_url).startswith("http"):
|
||||
content += f"\n{file_url}"
|
||||
elif str(file_url).startswith("base64://") or "data:image" in str(file_url):
|
||||
import base64
|
||||
import io
|
||||
b64_data = str(file_url)
|
||||
if b64_data.startswith("base64://"):
|
||||
b64_data = b64_data[9:]
|
||||
if b64_data.startswith("data:image"):
|
||||
b64_data = b64_data.split(",", 1)[1]
|
||||
try:
|
||||
image_bytes = base64.b64decode(b64_data)
|
||||
files.append(discord.File(fp=io.BytesIO(image_bytes), filename="image.png"))
|
||||
except Exception as e:
|
||||
logger.error(f"解析 Base64 图片失败: {e}")
|
||||
else:
|
||||
try:
|
||||
files.append(discord.File(file_url))
|
||||
except Exception as e:
|
||||
logger.error(f"无法读取本地文件 {file_url}: {e}")
|
||||
elif cq_type == "at":
|
||||
qq_id = params.get("qq")
|
||||
if qq_id == "all":
|
||||
content += "@everyone "
|
||||
else:
|
||||
content += f"<@{qq_id}> "
|
||||
|
||||
last_end = match.end()
|
||||
|
||||
# 添加最后一个 CQ 码之后的纯文本
|
||||
if last_end < len(segment):
|
||||
content += segment[last_end:]
|
||||
|
||||
elif isinstance(segment, OneBotMessageSegment):
|
||||
# 解析 OneBot 的 MessageSegment
|
||||
seg_type = segment.type
|
||||
seg_data = segment.data
|
||||
|
||||
if seg_type == "text":
|
||||
content += seg_data.get("text", "")
|
||||
elif seg_type == "image" or seg_type == "video":
|
||||
# OneBot 的图片/视频通常有 file (URL或本地路径) 或 url 字段
|
||||
file_url = seg_data.get("url") or seg_data.get("file")
|
||||
|
||||
if file_url:
|
||||
# 处理 bytes 类型
|
||||
if isinstance(file_url, bytes):
|
||||
import io
|
||||
try:
|
||||
files.append(discord.File(fp=io.BytesIO(file_url), filename="image.png"))
|
||||
except Exception as e:
|
||||
logger.error(f"解析 bytes 图片失败: {e}")
|
||||
elif str(file_url).startswith("http"):
|
||||
# 如果是网络 URL，直接拼接到文本中，Discord 会自动解析预览
|
||||
content += f"\n{file_url}"
|
||||
elif str(file_url).startswith("base64://") or "data:image" in str(file_url):
|
||||
# 处理 Base64 图片 (需要解码并作为文件上传)
|
||||
import base64
|
||||
import io
|
||||
b64_data = str(file_url)
|
||||
if b64_data.startswith("base64://"):
|
||||
b64_data = b64_data[9:]
|
||||
if b64_data.startswith("data:image"):
|
||||
b64_data = b64_data.split(",", 1)[1]
|
||||
try:
|
||||
image_bytes = base64.b64decode(b64_data)
|
||||
files.append(discord.File(fp=io.BytesIO(image_bytes), filename="image.png"))
|
||||
except Exception as e:
|
||||
logger.error(f"解析 Base64 图片失败: {e}")
|
||||
else:
|
||||
# 假设是本地文件路径
|
||||
try:
|
||||
files.append(discord.File(file_url))
|
||||
except Exception as e:
|
||||
logger.error(f"无法读取本地文件 {file_url}: {e}")
|
||||
elif seg_type == "at":
|
||||
qq_id = seg_data.get("qq")
|
||||
if qq_id == "all":
|
||||
content += "@everyone "
|
||||
else:
|
||||
# 尝试将 QQ 号映射回 Discord ID (这里简单处理，直接拼接)
|
||||
content += f"<@{qq_id}> "
|
||||
elif seg_type == "reply":
|
||||
# 忽略回复段，或者你可以尝试映射 message_id
|
||||
pass
|
||||
|
||||
# 发送消息到 Discord
|
||||
try:
|
||||
# 如果内容为空但有文件，Discord 允许发送
|
||||
if content or files:
|
||||
await original_message.channel.send(content=content, files=files if files else None)
|
||||
else:
|
||||
logger.warning("尝试发送空消息到 Discord，已拦截")
|
||||
except Exception as e:
|
||||
logger.error(f"发送 Discord 消息失败: {e}")
|
||||
101
adapters/universal_model.py
Normal file
101
adapters/universal_model.py
Normal file
@@ -0,0 +1,101 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
通用数据模型 (Universal Data Models)
|
||||
|
||||
此模块定义了平台无关的数据结构，用于在不同平台（如 OneBot, Discord）
|
||||
和业务逻辑层（如 Plugins）之间传递数据。
|
||||
"""
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional, Union, Dict, Any
|
||||
|
||||
@dataclass
|
||||
class UniversalMessageSegment:
|
||||
"""
|
||||
平台无关的通用消息段模型。
|
||||
业务逻辑层只负责生成这个对象，由底层的 Adapter 负责将其翻译成特定平台的格式。
|
||||
"""
|
||||
type: str # 消息类型：'text', 'image', 'video', 'audio', 'at', 'reply' 等
|
||||
data: Dict[str, Any] # 消息数据载荷
|
||||
|
||||
@staticmethod
|
||||
def text(text: str) -> "UniversalMessageSegment":
|
||||
return UniversalMessageSegment("text", {"text": text})
|
||||
|
||||
@staticmethod
|
||||
def image(url: Optional[str] = None, base64: Optional[str] = None, file_path: Optional[str] = None) -> "UniversalMessageSegment":
|
||||
"""
|
||||
图片消息。
|
||||
Discord 支持直接发 URL 或上传本地文件；OneBot 支持 URL、Base64 或本地路径。
|
||||
"""
|
||||
return UniversalMessageSegment("image", {"url": url, "base64": base64, "file_path": file_path})
|
||||
|
||||
@staticmethod
|
||||
def video(url: Optional[str] = None, file_path: Optional[str] = None) -> "UniversalMessageSegment":
|
||||
"""
|
||||
视频消息。
|
||||
Discord 通常直接发 URL 或作为附件上传；OneBot 支持 URL 或本地路径。
|
||||
"""
|
||||
return UniversalMessageSegment("video", {"url": url, "file_path": file_path})
|
||||
|
||||
@staticmethod
|
||||
def at(user_id: str) -> "UniversalMessageSegment":
|
||||
"""
|
||||
@某人。
|
||||
注意：为了兼容 Discord 的雪花 ID (Snowflake)，user_id 必须是字符串。
|
||||
"""
|
||||
return UniversalMessageSegment("at", {"user_id": user_id})
|
||||
|
||||
@staticmethod
|
||||
def reply(message_id: str) -> "UniversalMessageSegment":
|
||||
"""
|
||||
回复某条消息。
|
||||
"""
|
||||
return UniversalMessageSegment("reply", {"message_id": message_id})
|
||||
|
||||
@dataclass
|
||||
class UniversalUser:
|
||||
"""通用用户模型"""
|
||||
id: str # 用户唯一ID (QQ号 或 Discord Snowflake ID)
|
||||
name: str # 用户昵称/群名片
|
||||
avatar_url: str # 头像URL
|
||||
is_bot: bool # 是否是机器人
|
||||
|
||||
@dataclass
|
||||
class UniversalChannel:
|
||||
"""通用频道/群组模型"""
|
||||
id: str # 频道/群组唯一ID (QQ群号 或 Discord Channel ID)
|
||||
name: str # 频道/群组名称
|
||||
type: str # 类型：'private' (私聊), 'group' (QQ群), 'guild_text' (Discord文字频道) 等
|
||||
guild_id: Optional[str] = None # 仅 Discord 有效：服务器(Guild) ID
|
||||
|
||||
@dataclass
|
||||
class UniversalMessageEvent:
|
||||
"""
|
||||
平台无关的通用消息事件模型。
|
||||
这是传递给业务逻辑层（如 bili.py）的最终对象。
|
||||
"""
|
||||
platform: str # 来源平台标识：'onebot' 或 'discord'
|
||||
|
||||
message_id: str # 消息唯一ID (QQ消息ID 或 Discord Message ID)
|
||||
|
||||
user: UniversalUser # 发送者信息
|
||||
channel: UniversalChannel # 消息来源频道/群组信息
|
||||
|
||||
raw_message: str # 纯文本形式的消息内容（用于正则匹配、命令解析）
|
||||
|
||||
# 解析后的消息段列表（可选，如果你需要处理图文混排）
|
||||
message: List[UniversalMessageSegment] = field(default_factory=list)
|
||||
|
||||
# 原始的底层事件对象（保留引用，方便高级操作）
|
||||
# 例如：OneBot 的原始 JSON 字典，或 discord.py 的 discord.Message 对象
|
||||
raw_event: Any = field(repr=False, default=None)
|
||||
|
||||
async def reply(self, message: Union[str, UniversalMessageSegment, List[UniversalMessageSegment]]):
|
||||
"""
|
||||
统一的回复接口。
|
||||
这个方法应该是一个抽象方法或由具体的 Adapter 注入实现。
|
||||
业务逻辑层调用此方法时，不需要关心底层是调用 OneBot API 还是 Discord API。
|
||||
"""
|
||||
raise NotImplementedError("此方法应由具体的 Platform Adapter 实现")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user