K2CrO4/NeoBot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #71 from Fairy-Oracle-Sanctuary/dev

Browse Source 
Dev
This commit is contained in:
镀铬酸钾
2026-03-21 13:47:21 +08:00
committed by GitHub
parent f916807dfb f753f4de87
commit 7988943ccf
24 changed files with 1563 additions and 2150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1015
plugins/cross_platform.py
View File

File diff suppressed because it is too large Load Diff

24
plugins/discord-cross/__init__.py Normal file
View File

@@ -0,0 +1,24 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件入口
"""
import asyncio
from core.utils.logger import logger
from .config import config
from .subscription import start_cross_platform_subscription, stop_cross_platform_subscription
from .handlers import *
# 插件加载时自动启动和加载配置
try:
asyncio.create_task(config.reload())
except Exception as e:
logger.error(f"[CrossPlatform] 重新加载配置失败: {e}")
try:
asyncio.create_task(start_cross_platform_subscription())
except Exception as e:
logger.error(f"[CrossPlatform] 启动订阅失败: {e}")
def cleanup():
"""清理资源"""
asyncio.create_task(stop_cross_platform_subscription())

61
plugins/discord-cross/config.py Normal file
View File

@@ -0,0 +1,61 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件配置模块
"""
import os
from typing import Dict, Any
from core.utils.logger import logger
class CrossPlatformConfig:
def __init__(self):
self.CROSS_PLATFORM_MAP: Dict[int, Dict[str, Any]] = {}
self.CROSS_PLATFORM_CHANNEL = "neobot_cross_platform"
self.ENABLE_CROSS_PLATFORM = True
# DeepSeek API 配置
self.DEEPSEEK_API_KEY = "sk-Cn4BeHyTHDPRKuDadLy6dUnjSSHxrz5wQa54ZFAdQovXguLD"
self.DEEPSEEK_API_URL = "https://api.gptgod.online/v1/chat/completions"
self.DEEPSEEK_MODEL = "gemini-3-flash-preview"
# 是否启用翻译功能
self.ENABLE_TRANSLATION = True
async def reload(self):
"""重新加载配置"""
try:
config_path = os.path.join(os.path.dirname(__file__), "..", "..", "config.toml")
if os.path.exists(config_path):
try:
import tomllib
except ImportError:
import tomli as tomllib
with open(config_path, "rb") as f:
config_data = tomllib.load(f)
cross_platform_config = config_data.get("cross_platform", {})
self.ENABLE_CROSS_PLATFORM = cross_platform_config.get("enabled", True)
# 重新加载映射配置
mappings = cross_platform_config.get("mappings", {})
self.CROSS_PLATFORM_MAP.clear()
if isinstance(mappings, dict) and mappings:
for key, value in mappings.items():
if isinstance(value, dict) and "qq_group_id" in value:
try:
discord_id = int(key) if str(key).isdigit() else int(str(key).split('.')[-1])
self.CROSS_PLATFORM_MAP[discord_id] = {
"qq_group_id": int(value.get("qq_group_id", 0)),
"name": value.get("name", "")
}
except (ValueError, AttributeError):
continue
logger.success(f"[CrossPlatform] 配置已重新加载: {len(self.CROSS_PLATFORM_MAP)} 个映射")
except Exception as e:
logger.error(f"[CrossPlatform] 重新加载配置失败: {e}")
config = CrossPlatformConfig()

229
plugins/discord-cross/handlers.py Normal file
View File

@@ -0,0 +1,229 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件事件处理器模块
"""
import os
import html
from typing import List, Any
from core.managers.command_manager import matcher
from models.events.message import GroupMessageEvent, MessageEvent
from models.message import MessageSegment
from core.permission import Permission
from core.utils.logger import logger
from .config import config
from .parser import parse_forward_nodes
from .sender import forward_discord_to_qq, forward_qq_to_discord
async def handle_discord_message(
username: str,
discriminator: str,
content: str,
channel_id: int,
attachments: List[dict] = None,
embed: dict = None
):
"""处理 Discord 消息并转发"""
if not config.ENABLE_CROSS_PLATFORM:
return
logger.info(f"[CrossPlatform] 收到 Discord 消息: {username}#{discriminator} in {channel_id}")
await forward_discord_to_qq(username, discriminator, content, channel_id, attachments)
async def handle_qq_message(
nickname: str,
user_id: int,
group_name: str,
group_id: int,
content: str,
attachments: List[dict] = None
):
"""处理 QQ 消息并转发"""
if not config.ENABLE_CROSS_PLATFORM:
return
logger.info(f"[CrossPlatform] 收到 QQ 消息: {nickname} ({user_id}) in {group_name}({group_id})")
await forward_qq_to_discord(nickname, user_id, group_name, group_id, content, attachments)
@matcher.on_message()
async def handle_qq_group_message(event: GroupMessageEvent):
"""处理 QQ 群消息，转发到 Discord"""
if not config.ENABLE_CROSS_PLATFORM:
return
group_id = event.group_id
mapped_channel = None
for discord_channel_id, info in config.CROSS_PLATFORM_MAP.items():
if info["qq_group_id"] == group_id:
mapped_channel = discord_channel_id
break
if mapped_channel is None:
return
content = ""
attachments = []
if isinstance(event.message, list):
has_forward_node = any(isinstance(seg, MessageSegment) and seg.type == "node" for seg in event.message)
if has_forward_node:
forward_nodes = [seg for seg in event.message if isinstance(seg, MessageSegment) and seg.type == "node"]
forward_nodes_dict = [{"type": seg.type, "data": seg.data} for seg in forward_nodes]
content, attachments = await parse_forward_nodes(forward_nodes_dict)
else:
for segment in event.message:
if isinstance(segment, MessageSegment):
if segment.type == "text":
content += segment.data.get("text", "")
elif segment.type == "image":
file_url = segment.data.get("url") or segment.data.get("file")
file_name = segment.data.get("filename")
if file_url:
file_url = html.unescape(str(file_url))
if not file_name:
file_name = os.path.basename(file_url.split('?')[0]) or f"image_{len(attachments)}.jpg"
attachments.append({"type": "image", "url": file_url, "filename": file_name})
elif segment.type == "video":
file_url = segment.data.get("url") or segment.data.get("file")
file_name = segment.data.get("filename")
if file_url:
file_url = html.unescape(str(file_url))
if not file_name:
file_name = os.path.basename(file_url.split('?')[0]) or f"video_{len(attachments)}.mp4"
attachments.append({"type": "video", "url": file_url, "filename": file_name})
elif segment.type == "record":
file_url = segment.data.get("url") or segment.data.get("file")
file_name = segment.data.get("filename")
if file_url:
file_url = html.unescape(str(file_url))
if not file_name:
file_name = os.path.basename(file_url.split('?')[0]) or f"record_{len(attachments)}.amr"
attachments.append({"type": "record", "url": file_url, "filename": file_name})
content += f"\n[语音: {file_name}]\n"
elif segment.type == "at":
qq_id = segment.data.get("qq")
if qq_id and qq_id != "all":
content += f"@{qq_id} "
elif qq_id == "all":
content += "@所有人 "
elif isinstance(segment, str):
content += segment
elif isinstance(event.message, str):
content = event.message
import re
local_file_pattern = r'(http://[\w\.-]+:\d+/download\?id=file_[a-zA-Z0-9_]+)'
matches = re.finditer(local_file_pattern, content)
for match in matches:
file_url = match.group(1)
file_name = f"video_{len(attachments)}.mp4"
attachments.append({"type": "video", "url": file_url, "filename": file_name})
content = content.strip()
group_name = ""
try:
group_info = await event.bot.get_group_info(event.group_id)
group_name = group_info.get("group_name", "")
except Exception:
group_name = f"群{group_id}"
await handle_qq_message(
nickname=event.sender.nickname or event.sender.card or str(event.user_id),
user_id=event.user_id,
group_name=group_name,
group_id=group_id,
content=content,
attachments=attachments
)
@matcher.on_message()
async def handle_discord_message_event(event: Any):
"""处理 Discord 消息事件（通过适配器注入）"""
if not config.ENABLE_CROSS_PLATFORM:
return
if not hasattr(event, '_is_discord_message'):
return
discord_channel_id = getattr(event, 'discord_channel_id', None)
if discord_channel_id is None:
return
content = ""
attachments = []
if hasattr(event, 'message') and isinstance(event.message, list):
for segment in event.message:
if isinstance(segment, MessageSegment):
if segment.type == "text":
content += segment.data.get("text", "")
elif segment.type == "image":
file_url = segment.data.get("url") or segment.data.get("file")
file_name = segment.data.get("filename")
if file_url:
file_name = file_name or os.path.basename(str(file_url).split('?')[0]) or "image"
attachment_item = {"type": "image", "url": str(file_url), "filename": file_name}
if attachment_item not in attachments:
attachments.append(attachment_item)
elif segment.type == "video":
file_url = segment.data.get("url") or segment.data.get("file")
file_name = segment.data.get("filename")
if file_url:
file_name = file_name or os.path.basename(str(file_url).split('?')[0]) or "video"
attachment_item = {"type": "video", "url": str(file_url), "filename": file_name}
if attachment_item not in attachments:
attachments.append(attachment_item)
elif segment.type == "record":
file_url = segment.data.get("url") or segment.data.get("file")
file_name = segment.data.get("filename")
if file_url:
file_name = file_name or os.path.basename(str(file_url).split('?')[0]) or "record"
attachment_item = {"type": "record", "url": str(file_url), "filename": file_name}
if attachment_item not in attachments:
attachments.append(attachment_item)
else:
content = event.raw_message or ""
content = content.strip()
discord_username = getattr(event, 'discord_username', 'Unknown')
discord_discriminator = getattr(event, 'discord_discriminator', '')
await handle_discord_message(
username=discord_username,
discriminator=discord_discriminator,
content=content,
channel_id=discord_channel_id,
attachments=attachments,
embed=None
)
@matcher.command("cross_config", "跨平台配置", permission=Permission.ADMIN)
async def cross_config_command(event: MessageEvent):
"""查看跨平台配置"""
if not config.ENABLE_CROSS_PLATFORM:
await event.reply("跨平台功能已禁用")
return
config_lines = ["=== 跨平台映射配置 ==="]
if not config.CROSS_PLATFORM_MAP:
config_lines.append("当前没有配置任何映射")
else:
for discord_id, info in config.CROSS_PLATFORM_MAP.items():
discord_channel = f"Discord: {discord_id}"
qq_group = f"QQ: {info['qq_group_id']}"
name = info.get("name", "")
if name:
config_lines.append(f"• {discord_channel} ↔ {qq_group} ({name})")
else:
config_lines.append(f"• {discord_channel} ↔ {qq_group}")
await event.reply("\n".join(config_lines))
@matcher.command("cross_reload", "跨平台重载", permission=Permission.ADMIN)
async def cross_reload_command(event: MessageEvent):
"""重新加载跨平台配置"""
await config.reload()
await event.reply("跨平台配置已重载")

364
plugins/discord-cross/parser.py Normal file
View File

@@ -0,0 +1,364 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件解析器模块
"""
import os
import json
from typing import Dict, List, Any
from models.message import MessageSegment
from .config import config
async def parse_forward_nodes(nodes: List[Dict[str, Any]]) -> tuple[str, List[dict]]:
"""解析 OneBot 合并转发消息节点"""
content_parts = []
attachments = []
for node in nodes:
if not isinstance(node, dict):
continue
node_data = node.get("data", {})
node_content = node_data.get("content", "")
sender_name = node_data.get("name", node_data.get("uin", "Unknown"))
if isinstance(node_content, str):
if "[object Object]" in node_content:
content = f"[合并转发消息: {sender_name}]"
content_parts.append(f"**{sender_name}**:\n{content}")
elif '[CQ:' in node_content:
content = parse_cq_code(node_content, attachments)
content_parts.append(f"**{sender_name}**:\n{content}")
else:
content = node_content
content_parts.append(f"**{sender_name}**:\n{content}")
elif isinstance(node_content, list):
content = parse_message_segments(node_content, attachments)
content_parts.append(f"**{sender_name}**:\n{content}")
full_content = "\n\n".join(content_parts) if content_parts else ""
return full_content, attachments
def parse_cq_code(cq_code: str, attachments: List[dict]) -> str:
"""解析 CQ 码字符串"""
import re
cq_pattern = r'\[CQ:([^,]+)(?:,([^\]]+))?\]'
matches = list(re.finditer(cq_pattern, cq_code))
if not matches:
return cq_code
result = []
last_end = 0
for match in matches:
if match.start() > last_end:
result.append(cq_code[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 == "text":
result.append(params.get("text", ""))
elif cq_type == "image":
file_url = params.get("url") or params.get("file")
if file_url:
file_name = params.get("file", "")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "image"
attachments.append({"type": "image", "url": str(file_url), "filename": file_name})
result.append(f"\n[图片: {file_name}]\n")
elif cq_type == "video":
file_url = params.get("url") or params.get("file")
if file_url:
file_name = params.get("file", "")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "video"
attachments.append({"type": "video", "url": str(file_url), "filename": file_name})
result.append(f"\n[视频: {file_name}]\n")
elif cq_type == "record":
file_url = params.get("url") or params.get("file")
if file_url:
file_name = params.get("file", "")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "record"
attachments.append({"type": "record", "url": str(file_url), "filename": file_name})
result.append(f"\n[语音: {file_name}]\n")
elif cq_type == "at":
qq_id = params.get("qq")
if qq_id == "all":
result.append("@所有人 ")
else:
result.append(f"@{qq_id} ")
elif cq_type == "face":
face_id = params.get("id", "")
result.append(f"[表情:{face_id}] ")
elif cq_type == "reply":
reply_id = params.get("id", "")
result.append(f"[回复:{reply_id}] ")
elif cq_type == "file":
file_url = params.get("file", "")
if file_url:
file_name = os.path.basename(str(file_url).split('?')[0]) or "file"
attachments.append({"type": "file", "url": str(file_url), "filename": file_name})
result.append(f"\n[文件: {file_name}]\n")
last_end = match.end()
if last_end < len(cq_code):
result.append(cq_code[last_end:])
return "".join(result)
def parse_message_segments(segments: List[Any], attachments: List[dict]) -> str:
"""解析 MessageSegment 列表"""
result = []
for seg in segments:
if isinstance(seg, str):
result.append(seg)
elif isinstance(seg, MessageSegment):
seg_type = seg.type
seg_data = seg.data
if seg_type == "text":
result.append(seg_data.get("text", ""))
elif seg_type == "image":
file_url = seg_data.get("url") or seg_data.get("file")
if file_url:
file_name = seg_data.get("filename")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "image"
attachments.append({"type": "image", "url": str(file_url), "filename": file_name})
result.append(f"\n[图片: {file_name}]\n")
elif seg_type == "video":
file_url = seg_data.get("url") or seg_data.get("file")
if file_url:
file_name = seg_data.get("filename")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "video"
attachments.append({"type": "video", "url": str(file_url), "filename": file_name})
result.append(f"\n[视频: {file_name}]\n")
elif seg_type == "record":
file_url = seg_data.get("url") or seg_data.get("file")
if file_url:
file_name = seg_data.get("filename")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "record"
attachments.append({"type": "record", "url": str(file_url), "filename": file_name})
result.append(f"\n[语音: {file_name}]\n")
elif seg_type == "at":
qq_id = seg_data.get("qq")
if qq_id == "all":
result.append("@所有人 ")
else:
result.append(f"@{qq_id} ")
elif seg_type == "face":
face_id = seg_data.get("id", "")
result.append(f"[表情:{face_id}] ")
elif seg_type == "reply":
reply_id = seg_data.get("id", "")
result.append(f"[回复:{reply_id}] ")
elif seg_type == "file":
file_url = seg_data.get("file", "")
if file_url:
file_name = os.path.basename(str(file_url).split('?')[0]) or "file"
attachments.append({"type": "file", "url": str(file_url), "filename": file_name})
result.append(f"\n[文件: {file_name}]\n")
elif seg_type == "json":
json_data = seg_data.get("data", "")
try:
parsed = json.loads(json_data)
if isinstance(parsed, dict):
result.append(f"\n[JSON数据: {json_data[:100]}...]\n")
except:
result.append(f"\n[JSON数据]\n")
elif seg_type == "xml":
result.append(f"\n[XML数据]\n")
elif isinstance(seg, dict):
seg_type = seg.get("type")
seg_data = seg.get("data", {})
if seg_type == "text":
result.append(seg_data.get("text", ""))
elif seg_type == "image":
file_url = seg_data.get("url") or seg_data.get("file")
if file_url:
file_name = seg_data.get("filename")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "image"
attachments.append({"type": "image", "url": str(file_url), "filename": file_name})
result.append(f"\n[图片: {file_name}]\n")
elif seg_type == "video":
file_url = seg_data.get("url") or seg_data.get("file")
if file_url:
file_name = seg_data.get("filename")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "video"
attachments.append({"type": "video", "url": str(file_url), "filename": file_name})
result.append(f"\n[视频: {file_name}]\n")
elif seg_type == "record":
file_url = seg_data.get("url") or seg_data.get("file")
if file_url:
file_name = seg_data.get("filename")
if not file_name:
file_name = os.path.basename(str(file_url).split('?')[0]) or "record"
attachments.append({"type": "record", "url": str(file_url), "filename": file_name})
result.append(f"\n[语音: {file_name}]\n")
elif seg_type == "at":
qq_id = seg_data.get("qq")
if qq_id == "all":
result.append("@所有人 ")
else:
result.append(f"@{qq_id} ")
return "".join(result)
def get_platform_info(platform: str, identifier: Any) -> str:
"""获取平台信息字符串"""
if platform == "discord":
channel_id = int(identifier)
if channel_id in config.CROSS_PLATFORM_MAP:
group_info = config.CROSS_PLATFORM_MAP[channel_id]
group_name = group_info.get("name", f"群组 {group_info['qq_group_id']}")
return f"[Discord {group_name}]"
return f"[Discord]"
elif platform == "qq":
group_id = int(identifier)
return f"[PAW qq]"
return ""
async def format_discord_to_qq_content(
discord_username: str,
discord_discriminator: str,
content: str,
channel_id: int,
attachments: List[dict] = None
) -> tuple[str, List[dict]]:
"""将 Discord 消息格式化为 QQ 消息格式"""
platform_info = get_platform_info("discord", channel_id)
message_header = f"{discord_username}:"
message_body = content.strip() if content else ""
if message_body:
full_message = f"{message_header}\n{message_body}"
else:
full_message = message_header
processed_attachments = []
if attachments:
for att in attachments:
if isinstance(att, dict):
url = att.get("url", "")
filename = att.get("filename", "").lower()
att_type = att.get("type", "")
if att_type == "image" or filename.endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp')):
processed_attachments.append({"type": "image", "url": url})
elif att_type == "record" or filename.endswith(('.amr', '.silk', '.mp3', '.wav', '.ogg', '.m4a')):
processed_attachments.append({"type": "record", "url": url})
elif att_type == "video" or filename.endswith(('.mp4', '.avi', '.mkv', '.mov', '.flv', '.wmv')):
processed_attachments.append({"type": "video", "url": url})
else:
url = str(att)
if url.lower().endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp')):
processed_attachments.append({"type": "image", "url": url})
elif url.lower().endswith(('.amr', '.silk', '.mp3', '.wav', '.ogg', '.m4a')):
processed_attachments.append({"type": "record", "url": url})
elif url.lower().endswith(('.mp4', '.avi', '.mkv', '.mov', '.flv', '.wmv')):
processed_attachments.append({"type": "video", "url": url})
return full_message, processed_attachments
async def format_qq_to_discord_content(
qq_nickname: str,
qq_user_id: int,
group_name: str,
group_id: int,
content: str,
attachments: List[dict] = None
) -> tuple[str, List[dict], dict]:
"""将 QQ 消息格式化为 Discord 消息格式（Embed 卡片）"""
platform_info = get_platform_info("qq", group_id)
embed = {
"type": "rich",
"color": 0x5865F2,
"author": {
"name": f"{qq_nickname}",
"icon_url": f"https://q1.qlogo.cn/g?b=qq&nk={qq_user_id}&s=640"
},
"description": content if content else "",
"timestamp": None,
"footer": {
"text": f"来自 QQ"
}
}
if attachments:
image_urls = []
voice_urls = []
video_urls = []
other_urls = []
filtered_attachments = []
for att in attachments:
url = att.get("url", "")
filename = att.get("filename", "").lower()
att_type = att.get("type", "")
if att_type == "image" or filename.endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp')):
image_urls.append(url)
if len(image_urls) > 1:
filtered_attachments.append(att)
elif att_type == "record" or filename.endswith(('.amr', '.silk', '.mp3', '.wav', '.ogg', '.m4a')):
voice_urls.append(url)
other_urls.append(url)
filtered_attachments.append(att)
elif att_type == "video" or filename.endswith(('.mp4', '.avi', '.mkv', '.mov', '.flv', '.wmv')):
video_urls.append(url)
other_urls.append(url)
filtered_attachments.append(att)
else:
other_urls.append(url)
filtered_attachments.append(att)
attachments = filtered_attachments
embed["description"] = content if content else ""
if image_urls:
embed["image"] = {"url": image_urls[0]}
if voice_urls:
voice_filenames = [att.get("filename", "voice") for att in attachments if att.get("url") in voice_urls]
voice_list = "\n".join([f"🎤 {fname}" for fname in voice_filenames[:5]])
embed["description"] += f"\n\n**语音消息:**\n{voice_list}"
if len(voice_urls) > 5:
embed["description"] += f"\n...还有 {len(voice_urls) - 5} 条语音"
if video_urls:
video_filenames = [att.get("filename", "video") for att in attachments if att.get("url") in video_urls]
video_list = "\n".join([f"🎬 {fname}" for fname in video_filenames[:5]])
embed["description"] += f"\n\n**视频文件:**\n{video_list}"
if len(video_urls) > 5:
embed["description"] += f"\n...还有 {len(video_urls) - 5} 个视频"
non_media_other_urls = [u for u in other_urls if u not in voice_urls and u not in video_urls]
if non_media_other_urls:
file_filenames = [att.get("filename", "file") for att in attachments if att.get("url") in non_media_other_urls]
file_list = "\n".join([f"📄 {fname}" for fname in file_filenames[:5]])
embed["description"] += f"\n\n**附加文件:**\n{file_list}"
if len(non_media_other_urls) > 5:
embed["description"] += f"\n...还有 {len(non_media_other_urls) - 5} 个文件"
return "", attachments or [], embed

165
plugins/discord-cross/sender.py Normal file
View File

@@ -0,0 +1,165 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件发送器模块
"""
import json
from typing import List
from core.utils.logger import logger
from core.managers.redis_manager import redis_manager
from .config import config
from .translator import translate_with_deepseek
from .parser import format_discord_to_qq_content, format_qq_to_discord_content
async def send_to_discord(channel_id: int, content: str, attachments: List[dict] = None, embed: dict = None):
"""发送消息到 Discord 频道"""
try:
publish_data = {
"type": "send_message",
"channel_id": channel_id,
"content": content,
"attachments": attachments or [],
"embed": embed
}
await redis_manager.redis.publish("neobot_discord_send", json.dumps(publish_data))
logger.info(f"[CrossPlatform] 消息已发布到 Redis 供 Discord 适配器发送: {channel_id}")
except Exception as e:
logger.error(f"[CrossPlatform] 发送消息到 Discord 失败: {e}")
async def send_to_qq(group_id: int, content: str, attachments: List[dict] = None):
"""发送消息到 QQ 群"""
try:
from core.managers.bot_manager import bot_manager
from models.message import MessageSegment
all_bots = bot_manager.get_all_bots()
if not all_bots:
logger.error(f"[CrossPlatform] 没有可用的 QQ 机器人实例")
return
logger.debug(f"[CrossPlatform] 找到 {len(all_bots)} 个 QQ 机器人实例")
for bot in all_bots:
try:
message = content
if attachments:
full_message = []
if content:
full_message.append(MessageSegment.text(content))
for attachment in attachments:
if isinstance(attachment, dict):
att_type = attachment.get("type", "image")
attachment_url = attachment.get("url", "")
if att_type == "image":
full_message.append(MessageSegment.image(attachment_url, cache=True, proxy=True, timeout=30))
elif att_type == "record":
full_message.append(MessageSegment.record(attachment_url, cache=True, proxy=True, timeout=30))
elif att_type == "video":
full_message.append(MessageSegment.video(attachment_url))
else:
attachment_url = str(attachment)
if attachment_url.lower().endswith(('.mp4', '.avi', '.mkv', '.mov', '.flv', '.wmv')):
full_message.append(MessageSegment.video(attachment_url))
elif attachment_url.lower().endswith(('.amr', '.silk', '.mp3', '.wav', '.ogg', '.m4a')):
full_message.append(MessageSegment.record(attachment_url, cache=True, proxy=True, timeout=30))
else:
full_message.append(MessageSegment.image(attachment_url, cache=True, proxy=True, timeout=30))
logger.debug(f"[CrossPlatform] 准备发送消息到 QQ 群 {group_id}: {full_message}")
await bot.send_group_msg(group_id, full_message)
logger.info(f"[CrossPlatform] 消息已发送到 QQ 群 {group_id}")
else:
await bot.send_group_msg(group_id, message)
logger.info(f"[CrossPlatform] 消息已发送到 QQ 群 {group_id}")
break
except Exception as e:
logger.error(f"[CrossPlatform] 发送消息到 QQ 群 {group_id} 失败: {e}")
except Exception as e:
logger.error(f"[CrossPlatform] 发送消息到 QQ 失败: {e}")
async def forward_discord_to_qq(
discord_username: str,
discord_discriminator: str,
content: str,
channel_id: int,
attachments: List[dict] = None
):
"""将 Discord 消息转发到所有映射的 QQ 群"""
if channel_id not in config.CROSS_PLATFORM_MAP:
logger.warning(f"[CrossPlatform] 未找到 Discord 频道 {channel_id} 的映射配置")
return
group_info = config.CROSS_PLATFORM_MAP[channel_id]
target_qq_group = group_info["qq_group_id"]
formatted_content, image_list = await format_discord_to_qq_content(
discord_username,
discord_discriminator,
content,
channel_id,
attachments
)
if formatted_content:
translated_content = await translate_with_deepseek(formatted_content, "zh-CN", channel_id, "en2zh")
if translated_content != formatted_content:
formatted_content = f"{formatted_content}\n\n━━━━━ 翻译 ━━━━━\n{translated_content}"
await send_to_qq(target_qq_group, formatted_content, image_list)
logger.success(f"[CrossPlatform] Discord 频道 {channel_id} -> QQ 群 {target_qq_group}")
async def forward_qq_to_discord(
qq_nickname: str,
qq_user_id: int,
group_name: str,
group_id: int,
content: str,
attachments: List[dict] = None
):
"""将 QQ 消息转发到所有映射的 Discord 频道"""
target_channels = []
for discord_channel_id, info in config.CROSS_PLATFORM_MAP.items():
if info["qq_group_id"] == group_id:
target_channels.append(discord_channel_id)
if not target_channels:
logger.warning(f"[CrossPlatform] 未找到 QQ 群 {group_id} 的映射配置")
return
formatted_content, image_list, embed = await format_qq_to_discord_content(
qq_nickname,
qq_user_id,
group_name,
group_id,
content,
attachments
)
if embed and embed.get("description"):
original_text = embed["description"]
translated_text = await translate_with_deepseek(original_text, "en", group_id, "zh2en")
if translated_text != original_text:
embed["description"] = f"{original_text}\n\n**Translation:**\n{translated_text}"
for channel_id in target_channels:
await send_to_discord(channel_id, formatted_content, image_list, embed)
logger.success(f"[CrossPlatform] QQ 群 {group_id} -> Discord 频道 {target_channels}")
async def publish_to_redis(platform: str, data: dict):
"""通过 Redis 发布跨平台消息"""
try:
if redis_manager.redis:
publish_data = {
"platform": platform,
"data": data,
"timestamp": int(__import__('time').time())
}
await redis_manager.redis.publish(config.CROSS_PLATFORM_CHANNEL, json.dumps(publish_data))
logger.debug(f"[CrossPlatform] 已通过 Redis 发布消息: platform={platform}")
except Exception as e:
logger.error(f"[CrossPlatform] Redis 发布失败: {e}")

81
plugins/discord-cross/subscription.py Normal file
View File

@@ -0,0 +1,81 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件订阅模块
"""
import json
import asyncio
from core.utils.logger import logger
from core.managers.redis_manager import redis_manager
from .config import config
from .sender import forward_discord_to_qq, forward_qq_to_discord
async def cross_platform_subscription_loop():
"""Redis 跨平台消息订阅循环"""
if redis_manager.redis is None:
logger.warning("[CrossPlatform] Redis 未初始化，无法启动订阅")
return
try:
pubsub = redis_manager.redis.pubsub()
await pubsub.subscribe(config.CROSS_PLATFORM_CHANNEL)
logger.success("[CrossPlatform] 已订阅 Redis 跨平台频道")
async for message in pubsub.listen():
if message["type"] == "message":
try:
data = json.loads(message["data"])
platform = data.get("platform", "")
message_data = data.get("data", {})
logger.info(f"[CrossPlatform] 收到跨平台消息: {platform}")
if platform == "discord":
await forward_discord_to_qq(
discord_username=message_data.get("username", "Unknown"),
discord_discriminator=message_data.get("discriminator", ""),
content=message_data.get("content", ""),
channel_id=message_data.get("channel_id", 0),
attachments=message_data.get("attachments", [])
)
elif platform == "qq":
await forward_qq_to_discord(
qq_nickname=message_data.get("nickname", "Unknown"),
qq_user_id=message_data.get("user_id", 0),
group_name=message_data.get("group_name", ""),
group_id=message_data.get("group_id", 0),
content=message_data.get("content", ""),
attachments=message_data.get("attachments", []),
embed=message_data.get("embed")
)
except json.JSONDecodeError as e:
logger.error(f"[CrossPlatform] 解析消息失败: {e}")
except Exception as e:
logger.error(f"[CrossPlatform] 处理跨平台消息失败: {e}")
except Exception as e:
logger.error(f"[CrossPlatform] 订阅循环异常: {e}")
_subscription_task = None
async def start_cross_platform_subscription():
"""启动跨平台消息订阅"""
global _subscription_task
if _subscription_task is None and config.ENABLE_CROSS_PLATFORM:
_subscription_task = asyncio.create_task(cross_platform_subscription_loop())
logger.success("[CrossPlatform] 跨平台消息订阅已启动")
async def stop_cross_platform_subscription():
"""停止跨平台消息订阅"""
global _subscription_task
if _subscription_task:
_subscription_task.cancel()
try:
await _subscription_task
except asyncio.CancelledError:
pass
_subscription_task = None
logger.info("[CrossPlatform] 跨平台消息订阅已停止")

152
plugins/discord-cross/translator.py Normal file
View File

@@ -0,0 +1,152 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件翻译模块
"""
from typing import Dict, List
from core.utils.logger import logger
from .config import config
# 翻译上下文缓存（每个通道15条消息）
TRANSLATION_CONTEXT_CACHE: Dict[str, List[Dict[str, str]]] = {}
MAX_CONTEXT_MESSAGES = 15
def get_translation_context(channel_id: int, direction: str) -> List[Dict[str, str]]:
"""获取翻译上下文缓存"""
cache_key = f"{channel_id}_{direction}"
return TRANSLATION_CONTEXT_CACHE.get(cache_key, [])
def add_translation_context(channel_id: int, direction: str, original: str, translated: str):
"""添加翻译到上下文缓存"""
cache_key = f"{channel_id}_{direction}"
if cache_key not in TRANSLATION_CONTEXT_CACHE:
TRANSLATION_CONTEXT_CACHE[cache_key] = []
TRANSLATION_CONTEXT_CACHE[cache_key].append({
"original": original,
"translated": translated
})
if len(TRANSLATION_CONTEXT_CACHE[cache_key]) > MAX_CONTEXT_MESSAGES:
TRANSLATION_CONTEXT_CACHE[cache_key] = TRANSLATION_CONTEXT_CACHE[cache_key][-MAX_CONTEXT_MESSAGES:]
async def translate_with_deepseek(
text: str,
target_lang: str = "zh-CN",
channel_id: int = 0,
direction: str = "en2zh"
) -> str:
"""使用 DeepSeek API 翻译文本"""
if not config.ENABLE_TRANSLATION or not text.strip():
return text
if config.DEEPSEEK_API_KEY == "your-deepseek-api-key-here":
logger.warning("[CrossPlatform] DeepSeek API 密钥未配置，跳过翻译")
return text
lang_name = "中文" if target_lang == "zh-CN" else "英文"
messages = []
context_ref = ""
if channel_id > 0:
context = get_translation_context(channel_id, direction)
if context:
context_ref = "\n\n参考之前的翻译：\n"
for i, ctx in enumerate(context[-5:], 1):
context_ref += f"{i}. 原文: {ctx['original'][:100]}\n 译文: {ctx['translated'][:100]}\n"
system_prompt = f"""你是一个专业的翻译助手。请将以下文本翻译成{lang_name}。
只返回翻译后的文本，不要添加任何解释、注释或其他内容。避免翻译出仇视言论以及违反中国大陆相关法律法规的内容。如果有，请在翻译后有敏感的词语中把文本替换成井号（#）
保持原文的语气和格式。如果文本已经是目标语言，直接返回原文。{context_ref}"""
messages.append({"role": "user", "content": text})
try:
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key=config.DEEPSEEK_API_KEY,
base_url=config.DEEPSEEK_API_URL.replace("/chat/completions", "")
)
response = await client.chat.completions.create(
model=config.DEEPSEEK_MODEL,
messages=[{"role": "system", "content": system_prompt}] + messages,
temperature=0.3,
max_tokens=4000
)
translated_text = response.choices[0].message.content
if translated_text:
translated_text = translated_text.strip()
logger.info(f"[CrossPlatform] 翻译成功: {text[:50]}... -> {translated_text[:50]}...")
if channel_id > 0:
add_translation_context(channel_id, direction, text, translated_text)
return translated_text
else:
logger.warning("[CrossPlatform] DeepSeek 返回空翻译结果")
return text
except ImportError:
logger.warning("[CrossPlatform] openai 库未安装，尝试使用同步请求")
return await translate_with_deepseek_sync(text, target_lang, channel_id, direction)
except Exception as e:
logger.error(f"[CrossPlatform] 翻译失败: {e}")
return text
async def translate_with_deepseek_sync(
text: str,
target_lang: str = "zh-CN",
channel_id: int = 0,
direction: str = "en2zh"
) -> str:
"""使用同步请求的 DeepSeek 翻译（备用方案）"""
if not config.ENABLE_TRANSLATION or not text.strip():
return text
if config.DEEPSEEK_API_KEY == "your-deepseek-api-key-here":
return text
lang_name = "中文" if target_lang == "zh-CN" else "英文"
context_ref = ""
if channel_id > 0:
context = get_translation_context(channel_id, direction)
if context:
context_ref = "\n\n参考之前的翻译：\n"
for i, ctx in enumerate(context[-5:], 1):
context_ref += f"{i}. 原文: {ctx['original'][:100]}\n 译文: {ctx['translated'][:100]}\n"
system_prompt = f"""你是一个专业的翻译助手。请将以下文本翻译成{lang_name}。
只返回翻译后的文本，不要添加任何解释、注释或其他内容。避免翻译出仇视言论以及违反中国大陆相关法律法规的内容。如果有，请在翻译后有敏感的词语中把文本替换成井号（#）
保持原文的语气和格式。如果文本已经是目标语言，直接返回原文。{context_ref}"""
messages = [{"role": "user", "content": text}]
try:
from openai import OpenAI
client = OpenAI(
api_key=config.DEEPSEEK_API_KEY,
base_url=config.DEEPSEEK_API_URL.replace("/chat/completions", "")
)
response = client.chat.completions.create(
model=config.DEEPSEEK_MODEL,
messages=[{"role": "system", "content": system_prompt}] + messages,
temperature=0.3,
max_tokens=4000
)
translated_text = response.choices[0].message.content
if translated_text:
translated_text = translated_text.strip()
if channel_id > 0:
add_translation_context(channel_id, direction, text, translated_text)
return translated_text
return text
except Exception as e:
logger.error(f"[CrossPlatform] 同步翻译失败: {e}")
return text

0
plugins/osu!_plugin/__init__.py Normal file
View File

11
plugins/osu!_plugin/test.py Normal file
View File

@@ -0,0 +1,11 @@
from ossapi import Ossapi
# 初始化客户端（替换为自己的client_id和client_secret）
api = Ossapi("49746", "3sLQQC92twXgETwkJwixZWs5Chvhpo1HHQbYklLN")
# 根据用户名查询用户信息
print(api.user("[PAW]K2CRO4"))
# 根据用户ID查询osu模式下的用户信息
print(api.user(12092800, mode="osu").username)
# 查询指定谱面的ID
print(api.beatmap(221777).id)