423 lines
18 KiB
Python
423 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Discord 适配器 (Discord Adapter)
|
||
|
||
此模块负责与 Discord API 建立连接，接收 Discord 消息，
|
||
并将其转换为本地 OneBot 数据模型，
|
||
同时提供将本地消息段发送回 Discord 的能力。
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import io
|
||
import requests
|
||
import tempfile
|
||
import subprocess
|
||
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
|
||
from core.config_loader import global_config
|
||
|
||
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`")
|
||
|
||
self.logger = ModuleLogger("DiscordAdapter")
|
||
self.token = token
|
||
self.send_channel = None
|
||
|
||
self.proxy = None
|
||
self.proxy_type = "http"
|
||
if global_config.discord.proxy:
|
||
self.proxy = global_config.discord.proxy
|
||
self.proxy_type = global_config.discord.proxy_type or "http"
|
||
|
||
proxy_url = self.proxy
|
||
if self.proxy_type.lower() in ["socks5", "socks4"]:
|
||
if not proxy_url.startswith(("socks5://", "socks4://")):
|
||
proxy_url = f"{self.proxy_type.lower()}://{proxy_url.split('://')[-1]}"
|
||
|
||
os.environ["HTTP_PROXY"] = proxy_url
|
||
os.environ["HTTPS_PROXY"] = proxy_url
|
||
self.logger.info(f"[DiscordAdapter] 代理已设置: {proxy_url} (类型: {self.proxy_type})")
|
||
|
||
intents = discord.Intents.default()
|
||
intents.message_content = True
|
||
|
||
super().__init__(intents=intents)
|
||
|
||
async def on_ready(self):
|
||
"""当 Bot 成功连接到 Discord 时触发"""
|
||
self.logger.success(f"Discord Bot 已登录: {self.user} (ID: {self.user.id})")
|
||
|
||
self.start_heartbeat_task(interval=30)
|
||
|
||
# 启动 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}")
|
||
# 记录详细的异常信息
|
||
import traceback
|
||
self.logger.error(f"异常堆栈: {traceback.format_exc()}")
|
||
|
||
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 convert_to_ogg_opus(self, audio_bytes: bytes) -> Optional[bytes]:
|
||
"""
|
||
将音频文件转换为 OGG Opus 格式，用于 Discord 语音消息
|
||
"""
|
||
try:
|
||
# 创建临时文件
|
||
with tempfile.NamedTemporaryFile(delete=False, suffix=".tmp") as temp_in:
|
||
temp_in.write(audio_bytes)
|
||
temp_in_path = temp_in.name
|
||
|
||
with tempfile.NamedTemporaryFile(delete=False, suffix=".ogg") as temp_out:
|
||
temp_out_path = temp_out.name
|
||
|
||
# 使用 ffmpeg 转换
|
||
# -c:a libopus: 使用 Opus 编码器
|
||
# -b:a 64k: 比特率 64k
|
||
# -vbr on: 开启可变比特率
|
||
# -compression_level 10: 最高压缩级别
|
||
# -frame_duration 20: 帧时长 20ms
|
||
# -application voip: 针对语音优化
|
||
cmd = [
|
||
"ffmpeg", "-y", "-i", temp_in_path,
|
||
"-c:a", "libopus", "-b:a", "64k", "-vbr", "on",
|
||
"-compression_level", "10", "-frame_duration", "20",
|
||
"-application", "voip", temp_out_path
|
||
]
|
||
|
||
process = await asyncio.create_subprocess_exec(
|
||
*cmd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE
|
||
)
|
||
stdout, stderr = await process.communicate()
|
||
|
||
if process.returncode == 0:
|
||
with open(temp_out_path, "rb") as f:
|
||
ogg_bytes = f.read()
|
||
return ogg_bytes
|
||
else:
|
||
self.logger.error(f"[DiscordAdapter] ffmpeg 转换失败: {stderr.decode('utf-8', errors='ignore')}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"[DiscordAdapter] 音频转换异常: {e}")
|
||
return None
|
||
finally:
|
||
# 清理临时文件
|
||
try:
|
||
if os.path.exists(temp_in_path):
|
||
os.remove(temp_in_path)
|
||
if os.path.exists(temp_out_path):
|
||
os.remove(temp_out_path)
|
||
except:
|
||
pass
|
||
|
||
async def handle_send_message(self, data: dict):
|
||
"""处理来自 Redis 的消息发送请求"""
|
||
try:
|
||
channel_id = data.get("channel_id")
|
||
content = data.get("content", "")
|
||
attachments = data.get("attachments", [])
|
||
embed_data = data.get("embed")
|
||
|
||
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
|
||
|
||
# 检查会话状态
|
||
if not self.is_closed():
|
||
self.logger.info(f"[DiscordAdapter] 正在发送消息到频道 {channel_id}")
|
||
else:
|
||
self.logger.error(f"[DiscordAdapter] 会话已关闭，无法发送消息到频道 {channel_id}")
|
||
# 触发重连
|
||
self.logger.warning(f"[DiscordAdapter] 会话已关闭，将触发重连")
|
||
if self.ws is not None:
|
||
# 关闭 WebSocket 连接，让 discord.py 自动重连
|
||
await self.ws.close(4000)
|
||
return
|
||
|
||
embed = None
|
||
if embed_data:
|
||
embed = discord.Embed.from_dict(embed_data)
|
||
|
||
files = []
|
||
if attachments:
|
||
proxies = None
|
||
if self.proxy:
|
||
proxies = {
|
||
"http": self.proxy,
|
||
"https": self.proxy
|
||
}
|
||
|
||
for attachment in attachments:
|
||
if isinstance(attachment, dict):
|
||
attachment_url = attachment.get("url", "")
|
||
filename = attachment.get("filename", "")
|
||
else:
|
||
attachment_url = str(attachment)
|
||
filename = ""
|
||
|
||
if attachment_url.startswith('http'):
|
||
try:
|
||
response = requests.get(attachment_url, proxies=proxies, timeout=30)
|
||
if not filename:
|
||
filename = os.path.basename(attachment_url.split('?')[0]) or "attachment"
|
||
|
||
# 检查是否是语音文件
|
||
is_voice = filename.lower().endswith(('.amr', '.silk', '.mp3', '.wav', '.ogg', '.m4a'))
|
||
|
||
if is_voice:
|
||
# 尝试转换为 OGG Opus
|
||
ogg_bytes = await self.convert_to_ogg_opus(response.content)
|
||
if ogg_bytes:
|
||
# 转换成功，作为语音消息发送
|
||
# discord.py 官方 API 目前不支持直接发送语音消息
|
||
# 我们需要使用内部的 HTTP 客户端来发送
|
||
try:
|
||
# 构造文件数据
|
||
file_data = {
|
||
"name": "voice-message.ogg",
|
||
"value": ogg_bytes,
|
||
"content_type": "audio/ogg"
|
||
}
|
||
|
||
# 构造 payload
|
||
payload = {
|
||
"flags": 8192 # IS_VOICE_MESSAGE
|
||
}
|
||
|
||
if content:
|
||
payload["content"] = content
|
||
content = "" # 清空 content，避免重复发送
|
||
|
||
if embed:
|
||
payload["embeds"] = [embed.to_dict()]
|
||
embed = None # 清空 embed，避免重复发送
|
||
|
||
# 使用内部 HTTP 客户端发送
|
||
route = discord.http.Route('POST', '/channels/{channel_id}/messages', channel_id=channel_id)
|
||
await self.http.request(
|
||
route,
|
||
form=[
|
||
{'name': 'payload_json', 'value': json.dumps(payload)},
|
||
{'name': 'files[0]', 'value': ogg_bytes, 'filename': 'voice-message.ogg', 'content_type': 'audio/ogg'}
|
||
]
|
||
)
|
||
self.logger.success(f"[DiscordAdapter] 语音消息已发送到频道 {channel_id}")
|
||
continue # 跳过后面的普通发送逻辑
|
||
except Exception as e:
|
||
self.logger.error(f"[DiscordAdapter] 发送语音消息失败: {e}，将作为普通文件发送")
|
||
files.append(discord.File(fp=io.BytesIO(ogg_bytes), filename="voice.ogg"))
|
||
else:
|
||
# 转换失败，作为普通文件发送
|
||
files.append(discord.File(fp=io.BytesIO(response.content), filename=filename))
|
||
else:
|
||
files.append(discord.File(fp=io.BytesIO(response.content), filename=filename))
|
||
except Exception as e:
|
||
self.logger.error(f"[DiscordAdapter] 下载附件失败: {attachment_url}, 错误: {e}")
|
||
|
||
if content or files or embed:
|
||
try:
|
||
await channel.send(content=content, files=files if files else None, embed=embed)
|
||
self.logger.success(f"[DiscordAdapter] 消息已发送到频道 {channel_id}")
|
||
except Exception as send_error:
|
||
self.logger.error(f"[DiscordAdapter] 发送消息失败 (channel.send): {send_error}")
|
||
# 如果发送失败，尝试检查会话状态
|
||
if self.is_closed():
|
||
self.logger.warning(f"[DiscordAdapter] 会话已关闭，将触发重连")
|
||
if self.ws is not None:
|
||
await self.ws.close(4000)
|
||
raise
|
||
else:
|
||
self.logger.debug(f"[DiscordAdapter] 没有内容需要发送到频道 {channel_id}")
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"[DiscordAdapter] 发送消息失败: {e}")
|
||
|
||
async def start_client(self, max_retries: int = -1, retry_delay: int = 5):
|
||
"""
|
||
启动 Discord 客户端
|
||
|
||
Args:
|
||
max_retries: 最大重连次数，-1 表示无限重连
|
||
retry_delay: 重连延迟（秒）
|
||
"""
|
||
if not DISCORD_AVAILABLE:
|
||
self.logger.error("无法启动 Discord 客户端：discord.py 未安装")
|
||
return
|
||
|
||
retry_count = 0
|
||
|
||
while max_retries == -1 or retry_count < max_retries:
|
||
try:
|
||
self.logger.info("正在连接 Discord...")
|
||
await self.start(self.token)
|
||
except asyncio.CancelledError:
|
||
self.logger.info("连接被取消")
|
||
break
|
||
except discord.ConnectionClosed as e:
|
||
retry_count += 1
|
||
self.logger.warning(f"Discord 连接关闭: code={e.code}, reason={e.reason}")
|
||
|
||
# 如果是正常关闭，不计入重连次数
|
||
if e.code == 1000:
|
||
self.logger.info("连接正常关闭，等待重新连接...")
|
||
continue
|
||
|
||
if max_retries != -1 and retry_count >= max_retries:
|
||
self.logger.error(f"已达到最大重连次数 ({max_retries})，停止重连")
|
||
break
|
||
|
||
self.logger.info(f"将在 {retry_delay} 秒后重连 ({retry_count}/{max_retries if max_retries != -1 else '无限'})...")
|
||
await self._cleanup_connection()
|
||
await asyncio.sleep(retry_delay)
|
||
except Exception as e:
|
||
retry_count += 1
|
||
self.logger.error(f"Discord 连接异常: {e}")
|
||
|
||
if max_retries != -1 and retry_count >= max_retries:
|
||
self.logger.error(f"已达到最大重连次数 ({max_retries})，停止重连")
|
||
break
|
||
|
||
self.logger.info(f"将在 {retry_delay} 秒后重连 ({retry_count}/{max_retries if max_retries != -1 else '无限'})...")
|
||
await self._cleanup_connection()
|
||
await asyncio.sleep(retry_delay)
|
||
|
||
self.logger.info("Discord 客户端已停止")
|
||
|
||
async def _cleanup_connection(self):
|
||
"""
|
||
清理旧的连接状态
|
||
"""
|
||
try:
|
||
# 停止心跳任务
|
||
if hasattr(self, 'heartbeat_task') and not self.heartbeat_task.done():
|
||
self.heartbeat_task.cancel()
|
||
try:
|
||
await self.heartbeat_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
except Exception as e:
|
||
self.logger.error(f"清理心跳任务时出错: {e}")
|
||
|
||
try:
|
||
# 清理 HTTP 连接
|
||
if hasattr(self, 'http') and self.http:
|
||
await self.http.close()
|
||
except Exception as e:
|
||
self.logger.error(f"清理 HTTP 连接时出错: {e}")
|
||
|
||
try:
|
||
# 清理客户端状态
|
||
self.clear()
|
||
except Exception as e:
|
||
self.logger.error(f"清理客户端状态时出错: {e}")
|
||
|
||
async def start_heartbeat(self, interval: int = 30):
|
||
"""
|
||
启动心跳机制，定期检查连接状态
|
||
|
||
Args:
|
||
interval: 心跳间隔（秒）
|
||
"""
|
||
self.logger.info(f"心跳机制已启动，间隔: {interval}秒")
|
||
|
||
while not self.is_closed():
|
||
try:
|
||
await asyncio.sleep(interval)
|
||
|
||
# 检查 WebSocket 连接状态
|
||
if self.ws is not None:
|
||
# 正确检查 WebSocket 状态
|
||
if not getattr(self.ws, 'open', False):
|
||
self.logger.warning("检测到 WebSocket 连接已关闭，触发重连...")
|
||
try:
|
||
await self.ws.close(code=4000)
|
||
except Exception as close_error:
|
||
self.logger.error(f"关闭 WebSocket 连接时出错: {close_error}")
|
||
break
|
||
|
||
self.logger.debug(f"心跳正常: {self.user}")
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"心跳检测异常: {e}")
|
||
break
|
||
|
||
def start_heartbeat_task(self, interval: int = 30):
|
||
"""
|
||
启动心跳任务（非阻塞）
|
||
|
||
Args:
|
||
interval: 心跳间隔（秒）
|
||
"""
|
||
if not hasattr(self, 'heartbeat_task') or self.heartbeat_task.done():
|
||
self.heartbeat_task = asyncio.create_task(self.start_heartbeat(interval))
|
||
self.logger.info("心跳任务已启动")
|