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

feat: 添加多线程架构支持并优化性能

Browse Source 
实现线程管理器以支持高并发场景，添加GIL-free模式提升Python 3.14下的多线程性能
新增B站API集成和本地文件服务器功能，改进镜像插件支持GIF处理
更新文档说明多线程架构和GIL-free模式的使用方法
This commit is contained in:
K2cr2O1
2026-03-01 16:01:51 +08:00
parent 734c112ee4
commit ff4a4d92a5
20 changed files with 2071 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
plugins/broadcast.py
View File

@@ -4,18 +4,25 @@
功能：
- 仅限管理员在私聊中调用。
- 通过回复一条消息并发送指令，将该消息转发给机器人所在的所有群聊。
- 此插件不写入 __plugin_meta__，保持隐藏。
- 支持跨机器人广播：当任意机器人接收到广播消息时，会通过 Redis 发布消息，
所有其他机器人订阅后也会转发给它们各自的群聊。
- 使用通用消息格式，不使用合并转发（聊天记录）格式。
"""
import asyncio
import json
from core.managers.command_manager import matcher
from models.events.message import MessageEvent, PrivateMessageEvent
from core.permission import Permission
from core.utils.logger import logger
from core.managers.redis_manager import redis_manager
# --- 会话状态管理 ---
# 结构: {user_id: asyncio.TimerHandle}
broadcast_sessions: dict[int, asyncio.TimerHandle] = {}
# 广播消息订阅任务
_broadcast_subscription_task = None
def cleanup_session(user_id: int):
"""
清理超时的广播会话。
@@ -24,6 +31,103 @@ def cleanup_session(user_id: int):
del broadcast_sessions[user_id]
logger.info(f"[Broadcast] 会话 {user_id} 已超时，自动取消。")
async def broadcast_message_to_groups(bot, message, source_robot_id: str = "unknown"):
"""
将消息广播到所有群聊
Args:
bot: 机器人实例
message: 要发送的消息
source_robot_id: 消息来源机器人ID（用于日志）
"""
try:
group_list = await bot.get_group_list()
if not group_list:
logger.warning(f"[Broadcast] 机器人 {source_robot_id} 目前没有加入任何群聊")
return
success_count, failed_count = 0, 0
total_groups = len(group_list)
for group in group_list:
try:
await bot.send_group_msg(group.group_id, message)
success_count += 1
except Exception as e:
failed_count += 1
logger.error(f"[Broadcast] 机器人 {source_robot_id} 发送至群聊 {group.group_id} 失败: {e}")
logger.success(f"[Broadcast] 机器人 {source_robot_id} 广播完成: {total_groups} 个群聊, 成功 {success_count}, 失败 {failed_count}")
except Exception as e:
logger.error(f"[Broadcast] 机器人 {source_robot_id} 获取群聊列表失败: {e}")
async def start_broadcast_subscription():
"""
启动 Redis 广播消息订阅
"""
global _broadcast_subscription_task
if _broadcast_subscription_task is None:
_broadcast_subscription_task = asyncio.create_task(broadcast_subscription_loop())
logger.success("[Broadcast] Redis 广播订阅已启动")
async def stop_broadcast_subscription():
"""
停止 Redis 广播消息订阅
"""
global _broadcast_subscription_task
if _broadcast_subscription_task:
_broadcast_subscription_task.cancel()
try:
await _broadcast_subscription_task
except asyncio.CancelledError:
pass
_broadcast_subscription_task = None
logger.info("[Broadcast] Redis 广播订阅已停止")
async def broadcast_subscription_loop():
"""
Redis 广播消息订阅循环
"""
if redis_manager.redis is None:
logger.warning("[Broadcast] Redis 未初始化，无法启动广播订阅")
return
try:
pubsub = redis_manager.redis.pubsub()
await pubsub.subscribe("neobot_broadcast")
logger.success("[Broadcast] 已订阅 Redis 广播频道")
async for message in pubsub.listen():
if message["type"] == "message":
try:
data = json.loads(message["data"])
robot_id = data.get("robot_id", "unknown")
message_data = data.get("message")
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)
except json.JSONDecodeError as e:
logger.error(f"[Broadcast] 解析广播消息失败: {e}")
except Exception as e:
logger.error(f"[Broadcast] 处理广播消息失败: {e}")
except Exception as e:
logger.error(f"[Broadcast] 广播订阅循环异常: {e}")
@matcher.command("broadcast", "广播", permission=Permission.ADMIN)
async def broadcast_start(event: MessageEvent):
"""
@@ -49,12 +153,15 @@ async def broadcast_start(event: MessageEvent):
user_id
)
broadcast_sessions[user_id] = timeout_handler
# 确保广播订阅已启动
await start_broadcast_subscription()
@matcher.on_message()
async def handle_broadcast_content(event: MessageEvent):
"""
通用消息处理器，用于捕获广播模式下的消息输入。
将捕获到的消息打包成一个新的合并转发消息并广播。
将捕获到的消息直接发送给机器人所在的所有群聊，并通过 Redis 发布给其他机器人。
"""
# 仅处理私聊消息，且用户在广播会话中
if not isinstance(event, PrivateMessageEvent) or event.user_id not in broadcast_sessions:
@@ -71,46 +178,27 @@ async def handle_broadcast_content(event: MessageEvent):
await event.reply("捕获到的消息为空，已取消广播。")
return True
# --- 执行广播逻辑 ---
bot = event.bot
try:
group_list = await bot.get_group_list()
if not group_list:
await event.reply("机器人目前没有加入任何群聊。")
return True
except Exception as e:
logger.error(f"[Broadcast] 获取群聊列表失败: {e}")
await event.reply(f"获取群聊列表时发生错误: {e}")
return True
success_count, failed_count = 0, 0
total_groups = len(group_list)
await event.reply(f"已收到广播内容，准备打包并向 {total_groups} 个群聊广播...")
# --- 将管理员发送的消息打包成一个单节点的合并转发消息 ---
try:
nodes_to_send = [
bot.build_forward_node(
user_id=event.user_id,
nickname=event.sender.nickname if event.sender else "未知用户",
message=message_to_broadcast
)
]
except Exception as e:
logger.error(f"[Broadcast] 构建转发节点失败: {e}")
await event.reply(f"构建转发消息节点时发生错误: {e}")
return True
# --- 向所有群聊发送打包好的合并转发消息 ---
for group in group_list:
try:
await bot.send_group_forward_msg(group.group_id, nodes_to_send)
success_count += 1
except Exception as e:
failed_count += 1
logger.error(f"[Broadcast] 转发至群聊 {group.group_id} 失败: {e}")
# 获取当前机器人ID（使用反向WS的机器人ID）
from core.ws import WS
robot_id = "unknown"
if WS.instance and hasattr(WS.instance, 'self_id'):
robot_id = str(WS.instance.self_id)
report = f"广播完成。\n总群聊: {total_groups}\n成功: {success_count}\n失败: {failed_count}"
await event.reply(report)
# --- 执行本地广播 ---
await broadcast_message_to_groups(event.bot, message_to_broadcast, robot_id)
# --- 通过 Redis 发布消息给其他机器人 ---
try:
if redis_manager.redis:
broadcast_data = {
"robot_id": robot_id,
"message": message_to_broadcast
}
await redis_manager.redis.publish("neobot_broadcast", json.dumps(broadcast_data))
logger.success(f"[Broadcast] 已通过 Redis 发布广播消息: 来源 {robot_id}")
except Exception as e:
logger.error(f"[Broadcast] 发布 Redis 消息失败: {e}")
await event.reply("广播已完成！")
return True # 消费事件，防止其他处理器响应

109
plugins/mirror_avatar.py
View File

@@ -2,12 +2,12 @@
镜像头像插件
提供 /镜像 指令，将@的用户头像或用户发送的图片处理成轴对称图形。
支持普通图片和 GIF 动画。
"""
from core.managers.command_manager import matcher
from core.bot import Bot
from models.events.message import MessageEvent
from core.permission import Permission
from PIL import Image
from PIL import Image, ImageSequence
import io
import aiohttp
import base64
@@ -16,7 +16,7 @@ import asyncio
__plugin_meta__ = {
"name": "mirror_avatar",
"description": "将用户头像或图片处理成轴对称图形",
"usage": "/镜像 @人 - 将@的用户头像处理成轴对称图形\n/镜像 - 等待用户发送图片进行镜像处理",
"usage": "/镜像 @人 - 将@的用户头像处理成轴对称图形\n/镜像 gif - 将@的用户头像处理成轴对称GIF动画\n/镜像 - 等待用户发送图片进行镜像处理",
}
# 存储等待图片的用户信息
@@ -71,7 +71,6 @@ def process_avatar(image_bytes: bytes) -> bytes:
# 分割图片为左右两部分
left_half = img.crop((0, 0, mid_x, height))
right_half = img.crop((mid_x, 0, width, height))
# 翻转左侧部分到右侧
left_half_flipped = left_half.transpose(Image.FLIP_LEFT_RIGHT)
@@ -90,6 +89,75 @@ def process_avatar(image_bytes: bytes) -> bytes:
return output.read()
def process_gif_avatar(gif_bytes: bytes) -> bytes:
"""
处理GIF动画为轴对称图形
:param gif_bytes: 原始GIF字节
:return: 处理后的GIF字节
"""
# 打开GIF
gif = Image.open(io.BytesIO(gif_bytes))
# 检查是否为动画GIF
if not getattr(gif, "is_animated", False):
# 如果不是动画，当作普通图片处理
return process_avatar(gif_bytes)
# 获取GIF的所有帧
frames = []
durations = []
disposal_methods = []
for frame in ImageSequence.Iterator(gif):
# 如果是P模式（调色板模式），需要特殊处理
if frame.mode == 'P':
# 转换为RGB进行处理
frame_rgb = frame.convert('RGB')
else:
frame_rgb = frame.convert('RGB')
# 获取图片尺寸
width, height = frame_rgb.size
# 计算对称轴位置（中间）
mid_x = width // 2
# 分割图片为左右两部分
left_half = frame_rgb.crop((0, 0, mid_x, height))
# 翻转左侧部分到右侧
left_half_flipped = left_half.transpose(Image.FLIP_LEFT_RIGHT)
# 创建新图片
new_frame = Image.new('RGB', (width, height))
# 粘贴左侧原始部分和右侧翻转部分
new_frame.paste(left_half, (0, 0))
new_frame.paste(left_half_flipped, (mid_x, 0))
frames.append(new_frame)
durations.append(frame.info.get('duration', 100))
disposal_methods.append(frame.info.get('disposal', 0))
# 保存处理后的GIF
output = io.BytesIO()
if frames:
# 使用save_all保存多帧GIF
frames[0].save(
output,
format='GIF',
save_all=True,
append_images=frames[1:],
duration=durations,
loop=0,
optimize=False,
disposal=disposal_methods
)
output.seek(0)
return output.read()
async def wait_for_image(bot: Bot, event: MessageEvent):
"""
等待用户发送图片
@@ -98,8 +166,6 @@ async def wait_for_image(bot: Bot, event: MessageEvent):
:param event: 消息事件对象
"""
user_id = event.user_id
chat_id = event.group_id if hasattr(event, 'group_id') else event.user_id
is_group = hasattr(event, 'group_id')
# 设置超时时间
timeout = 30
@@ -138,11 +204,19 @@ async def handle_image_message(bot: Bot, event: MessageEvent):
# 查找消息中的图片
images = []
is_gif = False
for segment in event.message:
if segment.type == "image" and segment.data.get("url"):
images.append(segment.data["url"])
if segment.type == "image":
url = segment.data.get("url", "")
# 检查是否为GIF图片
if ".gif" in url.lower() or segment.data.get("sub_type", 0) == 1:
is_gif = True
if url:
images.append((url, is_gif))
if not images:
del waiting_for_image[user_id]
await event.reply("未找到图片，请重新发送")
return
# 取消等待任务
@@ -150,13 +224,16 @@ async def handle_image_message(bot: Bot, event: MessageEvent):
try:
# 获取第一张图片
image_url = images[0]
image_url, is_gif = images[0]
# 下载图片
image_bytes = await get_image_from_url(image_url)
# 处理图片
processed_image = process_avatar(image_bytes)
if is_gif:
processed_image = process_gif_avatar(image_bytes)
else:
processed_image = process_avatar(image_bytes)
# 检查是否可以发送图片
can_send = await bot.can_send_image()
@@ -189,6 +266,11 @@ async def handle_mirror(bot: Bot, event: MessageEvent, args: list[str]):
if segment.type == "at" and segment.data.get("qq"):
at_users.append(int(segment.data["qq"]))
# 检查是否为GIF模式
is_gif_mode = False
if args and args[0] == "gif":
is_gif_mode = True
if at_users:
# 获取第一个@的用户
user_id = at_users[0]
@@ -198,7 +280,10 @@ async def handle_mirror(bot: Bot, event: MessageEvent, args: list[str]):
avatar_bytes = await get_avatar(user_id)
# 处理头像
processed_avatar = process_avatar(avatar_bytes)
if is_gif_mode:
processed_avatar = process_gif_avatar(avatar_bytes)
else:
processed_avatar = process_avatar(avatar_bytes)
# 检查是否可以发送图片
can_send = await bot.can_send_image()
@@ -218,4 +303,4 @@ async def handle_mirror(bot: Bot, event: MessageEvent, args: list[str]):
else:
# 没有@用户，等待用户发送图片
# 启动等待任务
asyncio.create_task(wait_for_image(bot, event))
asyncio.create_task(wait_for_image(bot, event))

352
plugins/web_parser/parsers/bili.py
View File

@@ -1,20 +1,31 @@
# -*- coding: utf-8 -*-
import re
import orjson
import aiohttp
from typing import Optional, Dict, Any, List, Union
from bs4 import BeautifulSoup
from urllib.parse import urlparse, parse_qs
from core.utils.logger import logger
from models import MessageEvent, MessageSegment
from ..base import BaseParser
from ..utils import format_duration
from cachetools import TTLCache
from bilibili_api import video, select_client, Credential
from bilibili_api.exceptions import ResponseCodeException
from core.config_loader import global_config
from core.services.local_file_server import download_to_local
# bilibili_api-python 可用性标志
BILI_API_AVAILABLE = True
# 显式指定使用 aiohttp，避免与其他库冲突
try:
select_client("aiohttp")
except Exception as e:
logger.warning(f"设置 bilibili_api 客户端失败: {e}")
class BiliParser(BaseParser):
"""
B站视频解析器
B站视频解析器（使用 bilibili-api-python 库）
"""
def __init__(self):
@@ -22,9 +33,24 @@ class BiliParser(BaseParser):
self.name = "B站解析器"
self.url_pattern = re.compile(r"https?://(?:www\.)?(bilibili\.com/video/\w+|b23\.tv/[a-zA-Z0-9]+)")
self.nickname = "B站视频解析"
# 消息去重缓存
self.processed_messages: TTLCache[int, bool] = TTLCache(maxsize=100, ttl=10)
def _get_credential(self) -> Optional[Credential]:
"""获取 B 站登录凭证"""
try:
bili_config = global_config.bilibili
if bili_config.sessdata and bili_config.bili_jct and bili_config.buvid3:
return Credential(
sessdata=bili_config.sessdata,
bili_jct=bili_config.bili_jct,
buvid3=bili_config.buvid3,
dedeuserid=bili_config.dedeuserid
)
except Exception:
pass
return None
async def parse(self, url: str) -> Optional[Dict[str, Any]]:
"""
解析B站视频信息
@@ -35,111 +61,172 @@ class BiliParser(BaseParser):
Returns:
Optional[Dict[str, Any]]: 视频信息字典，如果失败则返回None
"""
# 提取 BV 号
bvid = self.extract_bvid(url)
if not bvid:
logger.error(f"[{self.name}] 无法从 URL 提取 BV 号: {url}")
return None
try:
# 清理URL
if BILI_API_AVAILABLE:
# 使用 bilibili-api-python 库
credential = self._get_credential()
v = video.Video(bvid=bvid, credential=credential)
info = await v.get_info()
# 处理封面 URL
cover_url = info.get('pic', '')
if cover_url:
cover_url = cover_url.split('@')[0]
if cover_url.startswith('//'):
cover_url = 'https:' + cover_url
# 处理 UP 主头像
owner = info.get('owner', {})
owner_name = owner.get('name', '未知UP主')
owner_face = owner.get('face', '')
if owner_face:
if owner_face.startswith('//'):
owner_face = 'https:' + owner_face
owner_face = owner_face.split('@')[0]
# 处理统计信息
stat = info.get('stat', {})
return {
"title": info.get('title', '未知标题'),
"bvid": bvid,
"aid": info.get('aid', 0),
"duration": info.get('duration', 0),
"cover_url": cover_url,
"play": stat.get('view', 0),
"like": stat.get('like', 0),
"coin": stat.get('coin', 0),
"favorite": stat.get('favorite', 0),
"share": stat.get('share', 0),
"danmaku": stat.get('danmaku', 0),
"owner_name": owner_name,
"owner_avatar": owner_face,
"followers": info.get('owner', {}).get('fans', 0),
"description": info.get('desc', ''),
"pubdate": info.get('pubdate', 0),
}
else:
# 备用方案：直接解析页面
return await self._parse_fallback(url, bvid)
except ResponseCodeException as e:
logger.error(f"[{self.name}] API 返回错误: {e.code} - {e.msg}")
except Exception as e:
logger.error(f"[{self.name}] 解析视频信息失败: {e}")
if BILI_API_AVAILABLE:
logger.info(f"[{self.name}] 尝试备用解析方案")
return await self._parse_fallback(url, bvid)
return None
async def _parse_fallback(self, url: str, bvid: str) -> Optional[Dict[str, Any]]:
"""
备用解析方案（不使用 bilibili-api-python）
Args:
url (str): B站视频URL
bvid (str): BV号
Returns:
Optional[Dict[str, Any]]: 视频信息字典
"""
try:
session = self.get_session()
clean_url = url.split('?')[0]
if '#/' in clean_url:
clean_url = clean_url.split('#/')[0]
session = self.get_session()
async with session.get(clean_url, headers=self.HEADERS, timeout=aiohttp.ClientTimeout(total=5)) as response:
async with session.get(clean_url, headers=self.HEADERS, timeout=5) as response:
response.raise_for_status()
text = await response.text()
soup = BeautifulSoup(text, 'html.parser')
# 尝试多种方式获取视频数据
# 方式1: 尝试获取 __INITIAL_STATE__
script_tag = soup.find('script', text=re.compile('window.__INITIAL_STATE__'))
if not script_tag or not script_tag.string:
# 方式2: 尝试获取 __PLAYINFO__
script_tag = soup.find('script', text=re.compile('window.__PLAYINFO__'))
if not script_tag or not script_tag.string:
# 方式3: 尝试获取页面标题和其他信息
title_tag = soup.find('title')
if title_tag:
title = title_tag.get_text().strip()
# 提取BV号
bv_match = re.search(r'(BV\w{10})', clean_url)
bvid = bv_match.group(1) if bv_match else '未知BV号'
return {
"title": title.replace('_哔哩哔哩_bilibili', '').strip(),
"bvid": bvid,
"duration": 0,
"cover_url": '',
"play": 0,
"like": 0,
"coin": 0,
"favorite": 0,
"share": 0,
"owner_name": '未知UP主',
"owner_avatar": '',
"followers": 0,
}
return None
# 原始解析逻辑
match = re.search(r'window\.__INITIAL_STATE__\s*=\s*(\{[^}]*\});', script_tag.string)
if not match:
# 尝试另一种正则表达式
match = re.search(r'window\.__INITIAL_STATE__\s*=\s*(\{.*?\});', script_tag.string, re.DOTALL)
if not match:
return None
# 提取标题
import re
title_match = re.search(r'<h1[^>]*>([^<]+)</h1>', text)
title = title_match.group(1).strip() if title_match else '未知标题'
# 提取播放量等信息
play_match = re.search(r'"view":(\d+)', text)
play = int(play_match.group(1)) if play_match else 0
like_match = re.search(r'"like":(\d+)', text)
like = int(like_match.group(1)) if like_match else 0
coin_match = re.search(r'"coin":(\d+)', text)
coin = int(coin_match.group(1)) if coin_match else 0
favorite_match = re.search(r'"favorite":(\d+)', text)
favorite = int(favorite_match.group(1)) if favorite_match else 0
share_match = re.search(r'"share":(\d+)', text)
share = int(share_match.group(1)) if share_match else 0
# 提取 UP 主信息
owner_match = re.search(r'"name":"([^"]+)"', text)
owner_name = owner_match.group(1) if owner_match else '未知UP主'
face_match = re.search(r'"face":"([^"]+)"', text)
owner_face = face_match.group(1) if face_match else ''
if owner_face:
if owner_face.startswith('//'):
owner_face = 'https:' + owner_face
owner_face = owner_face.split('@')[0]
return {
"title": title,
"bvid": bvid,
"aid": 0,
"duration": 0,
"cover_url": '',
"play": play,
"like": like,
"coin": coin,
"favorite": favorite,
"share": share,
"danmaku": 0,
"owner_name": owner_name,
"owner_avatar": owner_face,
"followers": 0,
"description": '',
"pubdate": 0,
}
json_str = match.group(1)
# 清理JSON字符串中的潜在问题字符
json_str = json_str.strip().rstrip(';')
try:
data = orjson.loads(json_str)
except ValueError:
# 如果直接解析失败，尝试清理JSON字符串
# 移除可能的注释或无效字符
cleaned_json = re.sub(r',\s*[}]', '}', json_str) # 移除末尾多余的逗号
cleaned_json = re.sub(r'/\*.*?\*/', '', cleaned_json) # 移除注释
cleaned_json = re.sub(r'//.*', '', cleaned_json) # 移除行注释
data = orjson.loads(cleaned_json)
video_data = data.get('videoData', {})
up_data = data.get('upData', {})
stat = video_data.get('stat', {})
owner = video_data.get('owner', {})
cover_url = video_data.get('pic', '')
if cover_url:
cover_url = cover_url.split('@')[0]
if cover_url.startswith('//'):
cover_url = 'https:' + cover_url
owner_avatar = owner.get('face', '')
if owner_avatar:
if owner_avatar.startswith('//'):
owner_avatar = 'https:' + owner_avatar
owner_avatar = owner_avatar.split('@')[0]
return {
"title": video_data.get('title', '未知标题'),
"bvid": video_data.get('bvid', '未知BV号'),
"duration": video_data.get('duration', 0),
"cover_url": cover_url,
"play": stat.get('view', 0),
"like": stat.get('like', 0),
"coin": stat.get('coin', 0),
"favorite": stat.get('favorite', 0),
"share": stat.get('share', 0),
"owner_name": owner.get('name', '未知UP主'),
"owner_avatar": owner_avatar,
"followers": up_data.get('fans', 0),
}
except (aiohttp.ClientError, KeyError, AttributeError, ValueError) as e:
logger.error(f"[{self.name}] 解析视频信息失败: {e}")
logger.debug(f"失败的URL: {url}")
except Exception as e:
logger.error(f"[{self.name}] 解析视频信息时发生未知错误: {e}")
logger.debug(f"失败的URL: {url}")
logger.error(f"[{self.name}] 备用解析方案失败: {e}")
return None
def extract_bvid(self, url: str) -> Optional[str]:
"""
从 URL 中提取 BV 号
Args:
url (str): B站视频URL
Returns:
Optional[str]: BV号，如果失败则返回None
"""
# 方式1: 直接从 URL 中提取
bvid_match = re.search(r'/video/(BV\w+)', url)
if bvid_match:
return bvid_match.group(1)
# 方式2: 从短链接跳转后提取
if 'b23.tv' in url:
try:
session = self.get_session()
# 简单处理，不实际跳转，直接尝试提取
bvid_match = re.search(r'BV\w{10}', url)
if bvid_match:
return bvid_match.group(0)
except Exception:
pass
return None
@@ -155,34 +242,62 @@ class BiliParser(BaseParser):
"""
try:
session = self.get_session()
async with session.head(short_url, headers=self.HEADERS, allow_redirects=False, timeout=aiohttp.ClientTimeout(total=5)) as response:
async with session.head(short_url, headers=self.HEADERS, allow_redirects=False, timeout=5) as response:
if response.status == 302:
return response.headers.get('Location')
except Exception as e:
logger.error(f"[{self.name}] 获取真实URL失败: {e}")
return None
async def get_direct_video_url(self, video_url: str) -> Optional[str]:
async def get_direct_video_url(self, video_url: str, bvid: str) -> Optional[str]:
"""
调用第三方API解析B站视频直链
获取B站视频直链（通过本地文件服务器下载）
Args:
video_url (str): B站视频的完整URL
bvid (str): BV号
Returns:
Optional[str]: 视频直链URL，如果失败则返回None
Optional[str]: 本地视频 URL，如果失败则返回None
"""
api_url = f"https://api.mir6.com/api/bzjiexi?url={video_url}&type=json"
if not BILI_API_AVAILABLE:
return None
try:
async with aiohttp.ClientSession() as session:
async with session.get(api_url, headers=self.HEADERS, timeout=aiohttp.ClientTimeout(total=10)) as response:
response.raise_for_status()
# 使用 content_type=None 来忽略 Content-Type 检查
data = await response.json(content_type=None)
if data.get("code") == 200 and data.get("data"):
return data["data"][0].get("video_url")
except (aiohttp.ClientError, ValueError, KeyError, IndexError) as e:
logger.error(f"[{self.name}] 调用第三方API解析视频失败: {e}")
credential = self._get_credential()
v = video.Video(bvid=bvid, credential=credential)
# 先获取视频信息以获取 cid
info = await v.get_info()
cid = info.get('cid', 0)
if not cid:
return None
# 获取下载链接数据
download_url_data = await v.get_download_url(cid=cid)
# 使用 VideoDownloadURLDataDetecter 解析数据
detecter = video.VideoDownloadURLDataDetecter(data=download_url_data)
streams = detecter.detect_best_streams()
if streams:
# 获取视频直链
video_direct_url = streams[0].url
logger.info(f"[{self.name}] 获取到视频直链，开始下载到本地...")
# 使用本地文件服务器下载
local_url = await download_to_local(video_direct_url, timeout=120)
if local_url:
logger.success(f"[{self.name}] 视频已下载到本地: {local_url}")
return local_url
else:
logger.error(f"[{self.name}] 下载到本地失败")
return None
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]:
@@ -204,7 +319,8 @@ class BiliParser(BaseParser):
else:
# 构建完整的B站视频URL
video_url = f"https://www.bilibili.com/video/{data.get('bvid', '')}"
direct_url = await self.get_direct_video_url(video_url)
bvid = data.get('bvid', '')
direct_url = await self.get_direct_video_url(video_url, bvid)
if direct_url:
video_message = MessageSegment.video(direct_url)
else:
@@ -226,6 +342,7 @@ class BiliParser(BaseParser):
f" 投币: {self.format_count(data['coin'])}\n"
f" 收藏: {self.format_count(data['favorite'])}\n"
f" 转发: {self.format_count(data['share'])}\n"
f" 弹幕: {self.format_count(data.get('danmaku', 0))}\n"
)
image_message_segment = [
@@ -264,5 +381,4 @@ class BiliParser(BaseParser):
Returns:
bool: 是否应该处理
"""
# 检查是否是B站相关域名，包括短链接
return bool(self.url_pattern.search(url))