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

Dev
This commit is contained in:
镀铬酸钾
2026-03-15 13:49:37 +08:00
committed by GitHub
25 changed files with 2457 additions and 116 deletions

View File

@@ -114,10 +114,21 @@ async def broadcast_subscription_loop():
logger.info(f"[Broadcast] 收到跨机器人广播消息: 来源 {robot_id}")
# 获取当前机器人的实例
from core.ws import WS
if WS.instance:
await broadcast_message_to_groups(WS.instance, message_data, robot_id)
# 获取所有活跃的 Bot 实例
from core.managers.bot_manager import bot_manager
all_bots = bot_manager.get_all_bots()
if not all_bots:
logger.warning("[Broadcast] 没有活跃的 Bot 实例,无法转发广播消息")
continue
# 遍历所有 Bot 进行广播
for bot in all_bots:
# 避免重复广播:如果消息来源就是当前 Bot则跳过
if str(bot.self_id) == str(robot_id):
continue
await broadcast_message_to_groups(bot, message_data, robot_id)
except json.JSONDecodeError as e:
logger.error(f"[Broadcast] 解析广播消息失败: {e}")
@@ -178,16 +189,26 @@ async def handle_broadcast_content(event: MessageEvent):
await event.reply("捕获到的消息为空,已取消广播。")
return True
# 获取当前机器人ID使用反向WS的机器人ID
from core.ws import WS
# 获取当前机器人ID
robot_id = "unknown"
if WS.instance and hasattr(WS.instance, 'self_id'):
robot_id = str(WS.instance.self_id)
if event.bot and hasattr(event.bot, 'self_id'):
robot_id = str(event.bot.self_id)
# --- 执行本地广播 ---
# 1. 先让接收到指令的这个 Bot 进行广播
await broadcast_message_to_groups(event.bot, message_to_broadcast, robot_id)
# --- 通过 Redis 发布消息给其他机器人 ---
# 2. 获取其他所有 Bot 并进行广播(针对同一进程内的其他 Bot
from core.managers.bot_manager import bot_manager
all_bots = bot_manager.get_all_bots()
for bot in all_bots:
# 跳过已经广播过的 Bot (即当前接收指令的 Bot)
if str(bot.self_id) == robot_id:
continue
await broadcast_message_to_groups(bot, message_to_broadcast, robot_id)
# --- 通过 Redis 发布消息给其他进程的机器人 ---
try:
if redis_manager.redis:
broadcast_data = {

View File

@@ -0,0 +1,38 @@
from core.plugin import Plugin, command, on_message
from models.events.message import MessageEvent
from core.permission import Permission
# 插件元信息
__plugin_meta__ = {
"name": "类风格插件示例",
"description": "演示如何使用类风格编写插件",
"usage": "/hello - 打招呼\n/echo <msg> - 复读消息",
}
class MyPlugin(Plugin):
def __init__(self):
super().__init__()
# 可以在这里初始化一些状态
self.count = 0
@command("hello")
async def hello(self, event: MessageEvent, args: list[str]):
self.count += 1
await self.reply(event, f"Hello from class-based plugin! (Called {self.count} times)")
@command("echo", permission=Permission.USER)
async def echo(self, event: MessageEvent, args: list[str]):
if args:
await self.reply(event, " ".join(args))
else:
await self.reply(event, "请输入要复读的内容。")
@on_message()
async def handle_message(self, event: MessageEvent):
# 这是一个通用的消息处理器,会处理所有消息
# 注意:这可能会与命令冲突,通常需要过滤
if "特定关键词" in event.raw_message:
await self.reply(event, "检测到特定关键词!")
# 实例化插件以注册
plugin = MyPlugin()

703
plugins/cross_platform.py Normal file
View File

@@ -0,0 +1,703 @@
# -*- coding: utf-8 -*-
"""
跨平台消息互通插件
功能:
- Discord 频道与 QQ 群之间的消息互通
- 在消息中自动标注来源平台和子频道/群组 ID
- 支持 OneBot v11 协议和数据结构
- 支持图片、视频等媒体消息
"""
import asyncio
import json
import re
import time
from typing import Dict, List, Optional, Any
from core.managers.command_manager import matcher
from models.events.message import GroupMessageEvent, PrivateMessageEvent, MessageEvent
from models.message import MessageSegment
from core.permission import Permission
from core.utils.logger import logger
from core.managers.redis_manager import redis_manager
# --- 配置 ---
# 跨平台映射配置
# 格式: {discord_channel_id: {"qq_group_id": qq_group_id, "name": "显示名称"}}
CROSS_PLATFORM_MAP: Dict[int, Dict[str, Any]] = {
# 示例配置:
# 123456789012345678: {"qq_group_id": 123456789, "name": "主群"},
# 987654321098765432: {"qq_group_id": 987654321, "name": "测试群"},
}
# Redis 通道名称
CROSS_PLATFORM_CHANNEL = "neobot_cross_platform"
# 是否启用跨平台转发
ENABLE_CROSS_PLATFORM = True
def get_platform_info(platform: str, identifier: Any) -> str:
"""
获取平台信息字符串,用于在消息中标注来源
Args:
platform: 平台名称 ('discord''qq')
identifier: 频道 ID 或群组 ID
Returns:
格式化的平台信息字符串
"""
if platform == "discord":
channel_id = int(identifier)
if channel_id in CROSS_PLATFORM_MAP:
group_info = 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"[QQ {group_id}]"
return ""
async def format_discord_to_qq_content(
discord_username: str,
discord_discriminator: str,
content: str,
channel_id: int,
attachments: List[str] = None
) -> tuple[str, List[str]]:
"""
将 Discord 消息格式化为 QQ 消息格式
Args:
discord_username: Discord 用户名
discord_discriminator: Discord discriminator (如 #1234)
content: 消息内容
channel_id: Discord 频道 ID
attachments: 附件列表
Returns:
格式化后的消息内容和图片列表
"""
platform_info = get_platform_info("discord", channel_id)
# 构建消息头(简化版,只显示名字)
message_header = f"{platform_info}\n {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
return full_message, attachments or []
async def format_qq_to_discord_content(
qq_nickname: str,
qq_user_id: int,
group_name: str,
group_id: int,
content: str,
attachments: List[str] = None
) -> tuple[str, List[str]]:
"""
将 QQ 消息格式化为 Discord 消息格式
Args:
qq_nickname: QQ 昵称
qq_user_id: QQ 用户 ID
group_name: 群名称
group_id: QQ 群 ID
content: 消息内容
attachments: 附件列表
Returns:
格式化后的消息内容和图片列表
"""
platform_info = get_platform_info("qq", group_id)
# 构建消息头(简化版,只显示名字)
message_header = f"{platform_info} {qq_nickname}:"
# 构建消息体
message_body = content if content else ""
# 组合完整消息(移除分隔符)
if message_body:
full_message = f"{message_header} {message_body}"
else:
full_message = message_header
return full_message, attachments or []
async def send_to_discord(channel_id: int, content: str, attachments: List[str] = None):
"""
发送消息到 Discord 频道
通过 Redis 发布消息,由 Discord 适配器接收并发送
这样可以避免跨模块导入实例的问题
Args:
channel_id: Discord 频道 ID
content: 消息内容
attachments: 附件 URL 列表
"""
try:
publish_data = {
"type": "send_message",
"channel_id": channel_id,
"content": content,
"attachments": attachments or []
}
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[str] = None):
"""
发送消息到 QQ 群
Args:
group_id: QQ 群 ID
content: 消息内容
attachments: 附件 URL 列表
"""
try:
from core.managers.bot_manager import bot_manager
from models.message import MessageSegment
# 获取所有 QQ 机器人实例
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
# 发送消息(如果有附件,使用 OneBot 的图片格式)
if attachments:
# 构建完整消息:文本 + 图片
from models.message import MessageSegment
full_message = []
if content:
full_message.append(MessageSegment.text(content))
for attachment in attachments:
full_message.append(MessageSegment.image(attachment, 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[str] = None
):
"""
将 Discord 消息转发到所有映射的 QQ 群
Args:
discord_username: Discord 用户名
discord_discriminator: Discord discriminator
content: 消息内容
channel_id: Discord 频道 ID
attachments: 附件列表
"""
if channel_id not in CROSS_PLATFORM_MAP:
logger.warning(f"[CrossPlatform] 未找到 Discord 频道 {channel_id} 的映射配置")
return
group_info = 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
)
# 发送到 QQ
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[str] = None
):
"""
将 QQ 消息转发到所有映射的 Discord 频道
Args:
qq_nickname: QQ 昵称
qq_user_id: QQ 用户 ID
group_name: 群名称
group_id: QQ 群 ID
content: 消息内容
attachments: 附件列表
"""
# 查找映射的 Discord 频道
target_channels = []
for discord_channel_id, info in 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 = await format_qq_to_discord_content(
qq_nickname,
qq_user_id,
group_name,
group_id,
content,
attachments
)
# 发送到所有映射的 Discord 频道
for channel_id in target_channels:
await send_to_discord(channel_id, formatted_content, image_list)
logger.success(f"[CrossPlatform] QQ 群 {group_id} -> Discord 频道 {target_channels}")
async def publish_to_redis(platform: str, data: dict):
"""
通过 Redis 发布跨平台消息
Args:
platform: 平台名称
data: 消息数据
"""
try:
if redis_manager.redis:
publish_data = {
"platform": platform,
"data": data,
"timestamp": int(__import__('time').time())
}
await redis_manager.redis.publish(CROSS_PLATFORM_CHANNEL, json.dumps(publish_data))
logger.debug(f"[CrossPlatform] 已通过 Redis 发布消息: platform={platform}")
except Exception as e:
logger.error(f"[CrossPlatform] Redis 发布失败: {e}")
async def handle_discord_message(
username: str,
discriminator: str,
content: str,
channel_id: int,
attachments: List[str] = None
):
"""
处理 Discord 消息并转发
Args:
username: Discord 用户名
discriminator: Discord discriminator
content: 消息内容
channel_id: Discord 频道 ID
attachments: 附件列表
"""
if not ENABLE_CROSS_PLATFORM:
return
logger.info(f"[CrossPlatform] 收到 Discord 消息: {username}#{discriminator} in {channel_id}")
# 转发到映射的 QQ 群
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[str] = None
):
"""
处理 QQ 消息并转发
Args:
nickname: QQ 昵称
user_id: QQ 用户 ID
group_name: 群名称
group_id: QQ 群 ID
content: 消息内容
attachments: 附件列表
"""
if not ENABLE_CROSS_PLATFORM:
return
logger.info(f"[CrossPlatform] 收到 QQ 消息: {nickname} ({user_id}) in {group_name}({group_id})")
# 转发到映射的 Discord 频道
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 ENABLE_CROSS_PLATFORM:
return
# 检查是否是映射的群组
group_id = event.group_id
mapped_channel = None
for discord_channel_id, info in 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):
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")
if file_url:
attachments.append(str(file_url))
elif segment.type == "video":
file_url = segment.data.get("url") or segment.data.get("file")
if file_url:
attachments.append(str(file_url))
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
# 清理多余空白
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 ENABLE_CROSS_PLATFORM:
return
# 检查事件是否包含 Discord 特定信息
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 = event.raw_message or ""
attachments = []
# 从 raw_message 中提取附件 URLDiscord 附件已添加到 raw_message
import re
url_pattern = r'https?://[^\s<>"]+|www\.\S+'
raw_message_lines = content.split('\n')
content_lines = []
for line in raw_message_lines:
line = line.strip()
if re.match(url_pattern, line):
# 这是附件 URL
if line not in attachments:
attachments.append(line)
else:
# 这是普通文本内容
if line:
content_lines.append(line)
content = '\n'.join(content_lines).strip()
# 从 message 列表中提取(备用方案)
if hasattr(event, 'message') and isinstance(event.message, list):
for segment in event.message:
if isinstance(segment, MessageSegment):
if segment.type == "text":
pass # 已经在 raw_message 中
elif segment.type == "image":
file_url = segment.data.get("url") or segment.data.get("file")
if file_url and str(file_url) not in attachments:
attachments.append(str(file_url))
elif segment.type == "video":
file_url = segment.data.get("url") or segment.data.get("file")
if file_url and str(file_url) not in attachments:
attachments.append(str(file_url))
# 获取用户信息
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
)
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(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":
# 从 Discord 转发到 QQ
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":
# 从 QQ 转发到 Discord
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", [])
)
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 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] 跨平台消息订阅已停止")
async def reload_config():
"""
重新加载配置
"""
global CROSS_PLATFORM_MAP, ENABLE_CROSS_PLATFORM
try:
import os
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 = tomllib.load(f)
cross_platform_config = config.get("cross_platform", {})
ENABLE_CROSS_PLATFORM = cross_platform_config.get("enabled", True)
# 重新加载映射配置(支持两种格式)
mappings = cross_platform_config.get("mappings", {})
CROSS_PLATFORM_MAP = {}
# 格式1: [cross_platform.mappings.123456789012345678] 子表形式
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])
CROSS_PLATFORM_MAP[discord_id] = {
"qq_group_id": int(value.get("qq_group_id", 0)),
"name": value.get("name", "")
}
except (ValueError, AttributeError):
continue
# 格式2: 旧的字典形式(向后兼容)
if not CROSS_PLATFORM_MAP:
for key, value in mappings.items():
if isinstance(key, str) and key.isdigit():
CROSS_PLATFORM_MAP[int(key)] = {
"qq_group_id": int(value.get("qq_group_id", 0)),
"name": value.get("name", "")
}
logger.success(f"[CrossPlatform] 配置已重新加载: {len(CROSS_PLATFORM_MAP)} 个映射")
except Exception as e:
logger.error(f"[CrossPlatform] 重新加载配置失败: {e}")
# 插件加载时自动启动和加载配置
import asyncio
try:
asyncio.create_task(reload_config())
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}")
# 命令处理器
@matcher.command("cross_config", "跨平台配置", permission=Permission.ADMIN)
async def cross_config_command(event: MessageEvent):
"""
查看跨平台配置
"""
if not ENABLE_CROSS_PLATFORM:
await event.reply("跨平台功能已禁用")
return
config_lines = ["=== 跨平台映射配置 ==="]
if not CROSS_PLATFORM_MAP:
config_lines.append("当前没有配置任何映射")
else:
for discord_id, info in 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 reload_config()
await event.reply("跨平台配置已重载")
# 清理函数
def cleanup():
"""清理资源"""
asyncio.create_task(stop_cross_platform_subscription())

View File

@@ -6,7 +6,6 @@ Echo 与交互插件
from core.managers.command_manager import matcher
from core.bot import Bot
from models.events.message import MessageEvent
from core.permission import Permission
__plugin_meta__ = {
"name": "echo",
@@ -14,7 +13,7 @@ __plugin_meta__ = {
"usage": "/echo [内容] - 复读内容\n/赞我 - 让机器人给你点赞",
}
@matcher.command("echo", permission=Permission.ADMIN)
@matcher.command("echo")
async def handle_echo(bot: Bot, event: MessageEvent, args: list[str]):
"""
处理 echo 指令,原样回复用户输入的内容

View File

@@ -0,0 +1,41 @@
from core.plugin import SimplePlugin
from models.events.message import MessageEvent
# 插件元信息
__plugin_meta__ = {
"name": "极简插件示例",
"description": "演示面向新手的极简插件写法",
"usage": "/ping - 测试\n/add <a> <b> - 加法\n/greet <name> - 问候",
}
class MySimplePlugin(SimplePlugin):
async def ping(self, event: MessageEvent):
"""
发送 /ping 即可调用
"""
return "Pong! (来自极简插件)"
async def greet(self, event: MessageEvent, name: str):
"""
发送 /greet Neo 即可调用
"""
return f"你好, {name}!"
async def add(self, event: MessageEvent, a: int, b: int):
"""
发送 /add 10 20 即可调用
自动处理类型转换
"""
return f"{a} + {b} = {a + b}"
async def echo_all(self, event: MessageEvent, msg: str):
"""
只有一个参数时,会自动捕获所有剩余文本
发送 /echo_all 这是一个 测试 消息
msg 将会是 "这是一个 测试 消息"
"""
return f"复读: {msg}"
# 实例化插件以生效
plugin = MySimplePlugin()

View File

@@ -1,5 +1,9 @@
# -*- coding: utf-8 -*-
import re
import os
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, Dict, Any, List, Union
from urllib.parse import urlparse, parse_qs
@@ -13,9 +17,25 @@ from bilibili_api.exceptions import ResponseCodeException
from core.config_loader import global_config
from core.services.local_file_server import download_to_local
try:
import aiohttp
AIOHTTP_AVAILABLE = True
except ImportError:
AIOHTTP_AVAILABLE = False
logger.warning("[B站解析器] aiohttp 未安装,音视频合并功能将不可用")
# bilibili_api-python 可用性标志
BILI_API_AVAILABLE = True
# ffmpeg 可用性标志
FFMPEG_AVAILABLE = False
try:
subprocess.run(['ffmpeg', '-version'], capture_output=True, check=True)
FFMPEG_AVAILABLE = True
logger.success("[B站解析器] ffmpeg 已安装,支持合并音视频")
except (subprocess.CalledProcessError, FileNotFoundError):
logger.warning("[B站解析器] ffmpeg 未安装,视频可能没有声音。建议安装 ffmpeg 以获得完整音视频体验")
# 显式指定使用 aiohttp避免与其他库冲突
try:
select_client("aiohttp")
@@ -273,20 +293,51 @@ class BiliParser(BaseParser):
if not cid:
return None
# 获取下载链接数据
download_url_data = await v.get_download_url(cid=cid)
# 获取下载链接数据,使用 html5=True 获取网页格式(通常包含合并的音视频)
download_url_data = await v.get_download_url(cid=cid, html5=True)
# 使用 VideoDownloadURLDataDetecter 解析数据
detecter = video.VideoDownloadURLDataDetecter(data=download_url_data)
# 尝试获取 MP4 格式的合并流(包含音视频)
streams = detecter.detect_best_streams()
# 如果没有获取到流,尝试其他格式
if not streams:
logger.warning(f"[{self.name}] 无法获取 html5 格式,尝试获取其他格式...")
download_url_data = await v.get_download_url(cid=cid, html5=False)
detecter = video.VideoDownloadURLDataDetecter(data=download_url_data)
streams = detecter.detect_best_streams()
if streams:
# 获取视频直链
video_direct_url = streams[0].url
# 检查是否是分离的 m4s 流(可能没有声音)
is_m4s_stream = '.m4s' in video_direct_url
if is_m4s_stream:
logger.warning(f"[{self.name}] 检测到分离的 m4s 流B站 API 返回的 m4s 流通常是分离的视频和音频,需要客户端合并才能有声音")
logger.info(f"[{self.name}] 建议: 使用支持合并 m4s 流的下载工具(如 ffmpeg合并视频和音频")
logger.info(f"[{self.name}] 获取到视频直链,开始下载到本地...")
# 使用本地文件服务器下载
local_url = await download_to_local(video_direct_url, timeout=120)
# B站下载需要 Referer 和 User-Agent
headers = {
"Referer": "https://www.bilibili.com",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
# 调试:打印 download_url_data 结构
logger.debug(f"[{self.name}] download_url_data 类型: {type(download_url_data)}")
if isinstance(download_url_data, dict):
logger.debug(f"[{self.name}] download_url_data keys: {list(download_url_data.keys())}")
# 如果是 m4s 流且 ffmpeg 可用,先保存 download_url_data 供合并使用
if is_m4s_stream and FFMPEG_AVAILABLE and AIOHTTP_AVAILABLE:
local_url = await self._download_and_merge_m4s(video_direct_url, headers, bvid, download_url_data)
else:
# 使用本地文件服务器下载
local_url = await download_to_local(video_direct_url, timeout=120, headers=headers)
if local_url:
logger.success(f"[{self.name}] 视频已下载到本地: {local_url}")
@@ -300,6 +351,199 @@ class BiliParser(BaseParser):
return None
async def _download_and_merge_m4s(self, video_url: str, headers: Dict[str, str], bvid: str, download_url_data: Dict) -> Optional[str]:
"""
下载并合并 m4s 视频和音频流
Args:
video_url (str): 视频流 URL
headers (Dict[str, str]): 请求头
bvid (str): BV号
download_url_data (Dict): 下载 URL 数据
Returns:
Optional[str]: 合并后的本地视频 URL如果失败则返回None
"""
if not FFMPEG_AVAILABLE:
logger.warning("[B站解析器] ffmpeg 不可用,无法合并音视频")
return None
if not AIOHTTP_AVAILABLE:
logger.warning("[B站解析器] aiohttp 不可用,无法合并音视频")
return None
try:
logger.info(f"[{self.name}] 开始下载并合并 m4s 音视频...")
# 创建共享的 ClientSession 用于下载
async with aiohttp.ClientSession() as session:
# 下载视频流
video_file = tempfile.NamedTemporaryFile(suffix='.m4s', delete=False)
video_file.close()
async with session.get(video_url, headers=headers, timeout=60) as response:
if response.status != 200:
logger.error(f"[{self.name}] 下载视频流失败: HTTP {response.status}")
return None
with open(video_file.name, 'wb') as f:
while True:
chunk = await response.content.read(8192)
if not chunk:
break
f.write(chunk)
logger.info(f"[{self.name}] 视频流下载完成: {video_file.name}")
# 从 download_url_data 中提取音频 URL
# B站的 dash 格式包含视频和音频流
audio_url = None
if isinstance(download_url_data, dict):
# 尝试 dash 格式(推荐)
if 'dash' in download_url_data and isinstance(download_url_data['dash'], dict):
dash = download_url_data['dash']
if 'audio' in dash and isinstance(dash['audio'], list) and len(dash['audio']) > 0:
# 获取第一个音频流
audio_item = dash['audio'][0]
audio_url = audio_item.get('baseUrl') or audio_item.get('url') or audio_item.get('backupUrl')
logger.debug(f"[{self.name}] 从 dash.audio 提取音频 URL: {audio_url is not None}")
elif 'audio' in dash and isinstance(dash['audio'], dict):
audio_url = dash['audio'].get('baseUrl') or dash['audio'].get('url')
logger.debug(f"[{self.name}] 从 dash.audio (dict) 提取音频 URL: {audio_url is not None}")
# 尝试 durl 格式(非分段流)
elif 'durl' in download_url_data:
if isinstance(download_url_data['durl'], list) and len(download_url_data['durl']) > 0:
main_url = download_url_data['durl'][0].get('url') or download_url_data['durl'][0].get('baseUrl')
if main_url:
video_url = main_url
logger.debug(f"[{self.name}] 使用 durl 主 URL: {video_url}")
if not audio_url and not video_url.startswith('http'):
logger.warning(f"[{self.name}] 无法从 download_url_data 中提取音频 URL")
logger.debug(f"[{self.name}] download_url_data 结构: {download_url_data}")
os.unlink(video_file.name)
return None
# 下载音频流
audio_file = tempfile.NamedTemporaryFile(suffix='.m4s', delete=False)
audio_file.close()
async with session.get(audio_url, headers=headers, timeout=60) as response:
if response.status != 200:
logger.error(f"[{self.name}] 下载音频流失败: HTTP {response.status}")
os.unlink(video_file.name)
return None
with open(audio_file.name, 'wb') as f:
while True:
chunk = await response.content.read(8192)
if not chunk:
break
f.write(chunk)
logger.info(f"[{self.name}] 音频流下载完成: {audio_file.name}")
# 使用 ffmpeg 合并视频和音频
merged_file = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
merged_file.close()
# ffmpeg命令使用ffmpeg -i多次输入然后合并
# 先转换视频流(移除音频),然后添加音频流
ffmpeg_cmd = [
'ffmpeg', '-y', '-i', video_file.name, '-i', audio_file.name,
'-c:v', 'libx264', '-c:a', 'aac',
'-shortest', merged_file.name
]
logger.debug(f"[{self.name}] ffmpeg命令: {' '.join(ffmpeg_cmd)}")
result = subprocess.run(ffmpeg_cmd, capture_output=True, text=True)
# 详细记录ffmpeg输出
if result.stdout:
logger.debug(f"[{self.name}] ffmpeg stdout: {result.stdout}")
if result.stderr:
logger.debug(f"[{self.name}] ffmpeg stderr: {result.stderr}")
if result.returncode != 0:
logger.error(f"[{self.name}] ffmpeg 合并失败: {result.stderr}")
os.unlink(video_file.name)
os.unlink(audio_file.name)
return None
# 验证输出文件
merged_size = os.path.getsize(merged_file.name)
logger.debug(f"[{self.name}] 合并文件大小: {merged_size} bytes")
if merged_size == 0:
logger.error(f"[{self.name}] ffmpeg生成了空文件命令可能有问题")
logger.error(f"[{self.name}] ffmpeg命令: {' '.join(ffmpeg_cmd)}")
if result.stderr:
logger.error(f"[{self.name}] ffmpeg错误输出: {result.stderr}")
os.unlink(video_file.name)
os.unlink(audio_file.name)
return None
logger.info(f"[{self.name}] 音视频合并成功: {merged_file.name} ({merged_size} bytes)")
# 上传合并后的文件到本地文件服务器
from core.services.local_file_server import get_local_file_server
server = get_local_file_server()
if server:
try:
file_id = server._generate_file_id(f'file://{merged_file.name}')
dest_path = server.download_dir / file_id
# 获取合并文件大小
merged_size = os.path.getsize(merged_file.name)
logger.debug(f"[{self.name}] 合并文件大小: {merged_size} bytes")
if merged_size == 0:
logger.error(f"[{self.name}] 合并文件为空ffmpeg可能失败了")
merged_url = None
else:
# 复制本地文件到服务器目录
import shutil
shutil.copy2(merged_file.name, dest_path)
server.file_map[file_id] = dest_path
# 验证复制后的文件
if dest_path.exists():
dest_size = dest_path.stat().st_size
logger.debug(f"[{self.name}] 复制后文件大小: {dest_size} bytes")
if dest_size == merged_size:
merged_url = f"http://127.0.0.1:{server.port}/download?id={file_id}"
logger.success(f"[{self.name}] 合并后的视频已上传到本地服务器: {merged_url}")
else:
logger.error(f"[{self.name}] 文件大小不匹配: 原始 {merged_size} vs 复制 {dest_size}")
merged_url = None
else:
logger.error(f"[{self.name}] 文件复制失败: {dest_path} 不存在")
merged_url = None
except Exception as e:
logger.error(f"[{self.name}] 上传合并文件失败: {e}")
merged_url = None
else:
merged_url = None
# 清理临时文件
try:
os.unlink(video_file.name)
os.unlink(audio_file.name)
os.unlink(merged_file.name)
except Exception as e:
logger.warning(f"[{self.name}] 清理临时文件失败: {e}")
if merged_url:
logger.success(f"[{self.name}] 合并后的视频已上传到本地服务器: {merged_url}")
return merged_url
except Exception as e:
logger.error(f"[{self.name}] 合并音视频失败: {e}")
return None
async def format_response(self, event: MessageEvent, data: Dict[str, Any]) -> List[Any]:
"""
格式化B站视频响应消息
@@ -314,8 +558,8 @@ class BiliParser(BaseParser):
# 检查视频时长
video_message: Union[str, MessageSegment]
direct_url = None
if data['duration'] > 1200: # 20分钟 = 1200秒
video_message = "视频时长超过20分钟,不进行解析。"
if data['duration'] > 7200: # 2小时 = 7200秒
video_message = "视频时长超过2小时,不进行解析。"
else:
# 构建完整的B站视频URL
video_url = f"https://www.bilibili.com/video/{data.get('bvid', '')}"