gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Bilibili video analysis plugin

Browse Source 
Introduced a new 'bilibli' plugin for Bilibili video parsing and AI-based content analysis. Added plugin manifest, base module for video info retrieval and analysis, and plugin integration with tool registration. Updated .gitignore to include the new plugin directory.
This commit is contained in:
雅诺狐
2025-08-30 00:26:44 +08:00
parent 18a739dc2b
commit cb93b38162
5 changed files with 624 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -325,7 +325,7 @@ run_pet.bat
!/plugins/set_emoji_like
!/plugins/permission_example
!/plugins/hello_world_plugin
!/plugins/take_picture_plugin
!/plugins/bilibli
!/plugins/napcat_adapter_plugin
!/plugins/echo_example

8
plugins/bilibli/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Bilibili 插件包
提供B站视频观看体验功能，像真实用户一样浏览和评价视频
"""
# 插件会通过 @register_plugin 装饰器自动注册，这里不需要额外的导入

43
plugins/bilibli/_manifest.json Normal file
View File

@@ -0,0 +1,43 @@
{
"manifest_version": 1,
"name": "哔哩哔哩视频解析插件 (Bilibili Video Parser)",
"version": "1.0.0",
"description": "解析哔哩哔哩视频链接，获取视频的base64编码数据（无音频版本），支持BV号、AV号和短链接",
"author": {
"name": "雅诺狐"
},
"license": "GPL-v3.0-or-later",
"host_application": {
"min_version": "0.8.0"
},
"keywords": [
"bilibili",
"video",
"parser",
"tool",
"媒体处理"
],
"categories": [
"Media",
"Tools"
],
"default_locale": "zh-CN",
"plugin_info": {
"is_built_in": false,
"plugin_type": "tool",
"components": [
{
"type": "tool",
"name": "bilibili_video_parser",
"description": "解析哔哩哔哩视频链接，获取视频的base64编码数据（无音频版本）"
}
],
"features": [
"支持BV号、AV号视频链接解析",
"支持b23.tv短链接解析",
"返回无音频的视频base64编码",
"自动处理重定向链接",
"详细的解析状态反馈"
]
}
}

349
plugins/bilibli/bilibli_base.py Normal file
View File

@@ -0,0 +1,349 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Bilibili 工具基础模块
提供 B 站视频信息获取和视频分析功能
"""
import re
import aiohttp
import asyncio
import tempfile
import os
from typing import Optional, Dict, Any
from src.common.logger import get_logger
from src.chat.utils.utils_video import get_video_analyzer
logger = get_logger("bilibili_tool")
class BilibiliVideoAnalyzer:
"""哔哩哔哩视频分析器，集成视频下载和AI分析功能"""
def __init__(self):
self.video_analyzer = get_video_analyzer()
self.headers = {
'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',
'Referer': 'https://www.bilibili.com/',
}
def extract_bilibili_url(self, text: str) -> Optional[str]:
"""从文本中提取哔哩哔哩视频链接"""
# 哔哩哔哩短链接模式
short_pattern = re.compile(r'https?://b23\.tv/[\w]+', re.IGNORECASE)
# 哔哩哔哩完整链接模式
full_pattern = re.compile(r'https?://(?:www\.)?bilibili\.com/video/(?:BV[\w]+|av\d+)', re.IGNORECASE)
# 先匹配短链接
short_match = short_pattern.search(text)
if short_match:
return short_match.group(0)
# 再匹配完整链接
full_match = full_pattern.search(text)
if full_match:
return full_match.group(0)
return None
async def get_video_info(self, url: str) -> Optional[Dict[str, Any]]:
"""获取哔哩哔哩视频基本信息"""
try:
logger.info(f"🔍 解析视频URL: {url}")
# 如果是短链接，先解析为完整链接
if 'b23.tv' in url:
logger.info("🔗 检测到短链接，正在解析...")
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(url, headers=self.headers, allow_redirects=True) as response:
url = str(response.url)
logger.info(f"✅ 短链接解析完成: {url}")
# 提取BV号或AV号
bv_match = re.search(r'BV([\w]+)', url)
av_match = re.search(r'av(\d+)', url)
if bv_match:
bvid = f"BV{bv_match.group(1)}"
api_url = f"https://api.bilibili.com/x/web-interface/view?bvid={bvid}"
logger.info(f"📺 提取到BV号: {bvid}")
elif av_match:
aid = av_match.group(1)
api_url = f"https://api.bilibili.com/x/web-interface/view?aid={aid}"
logger.info(f"📺 提取到AV号: av{aid}")
else:
logger.error("❌ 无法从URL中提取视频ID")
return None
# 获取视频信息
logger.info("📡 正在获取视频信息...")
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(api_url, headers=self.headers) as response:
if response.status != 200:
logger.error(f"❌ API请求失败，状态码: {response.status}")
return None
data = await response.json()
if data.get('code') != 0:
error_msg = data.get('message', '未知错误')
logger.error(f"❌ B站API返回错误: {error_msg} (code: {data.get('code')})")
return None
video_data = data['data']
# 验证必要字段
if not video_data.get('title'):
logger.error("❌ 视频数据不完整，缺少标题")
return None
result = {
'title': video_data.get('title', ''),
'desc': video_data.get('desc', ''),
'duration': video_data.get('duration', 0),
'view': video_data.get('stat', {}).get('view', 0),
'like': video_data.get('stat', {}).get('like', 0),
'coin': video_data.get('stat', {}).get('coin', 0),
'favorite': video_data.get('stat', {}).get('favorite', 0),
'share': video_data.get('stat', {}).get('share', 0),
'owner': video_data.get('owner', {}).get('name', ''),
'pubdate': video_data.get('pubdate', 0),
'aid': video_data.get('aid'),
'bvid': video_data.get('bvid'),
'cid': video_data.get('cid') or (video_data.get('pages', [{}])[0].get('cid') if video_data.get('pages') else None)
}
logger.info(f"✅ 视频信息获取成功: {result['title']}")
return result
except asyncio.TimeoutError:
logger.error("❌ 获取视频信息超时")
return None
except aiohttp.ClientError as e:
logger.error(f"❌ 网络请求失败: {e}")
return None
except Exception as e:
logger.error(f"❌ 获取哔哩哔哩视频信息时发生未知错误: {e}")
logger.exception("详细错误信息:")
return None
async def get_video_stream_url(self, aid: int, cid: int) -> Optional[str]:
"""获取视频流URL"""
try:
logger.info(f"🎥 获取视频流URL: aid={aid}, cid={cid}")
# 构建播放信息API请求
api_url = f"https://api.bilibili.com/x/player/playurl?avid={aid}&cid={cid}&qn=80&type=&otype=json&fourk=1&fnver=0&fnval=4048&session="
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(api_url, headers=self.headers) as response:
if response.status != 200:
logger.error(f"❌ 播放信息API请求失败，状态码: {response.status}")
return None
data = await response.json()
if data.get('code') != 0:
error_msg = data.get('message', '未知错误')
logger.error(f"❌ 获取播放信息失败: {error_msg} (code: {data.get('code')})")
return None
play_data = data['data']
# 尝试获取DASH格式的视频流
if 'dash' in play_data and play_data['dash'].get('video'):
videos = play_data['dash']['video']
logger.info(f"🎬 找到 {len(videos)} 个DASH视频流")
# 选择最高质量的视频流
video_stream = max(videos, key=lambda x: x.get('bandwidth', 0))
stream_url = video_stream.get('baseUrl') or video_stream.get('base_url')
if stream_url:
logger.info(f"✅ 获取到DASH视频流URL (带宽: {video_stream.get('bandwidth', 0)})")
return stream_url
# 降级到FLV格式
if 'durl' in play_data and play_data['durl']:
logger.info("📹 使用FLV格式视频流")
stream_url = play_data['durl'][0].get('url')
if stream_url:
logger.info("✅ 获取到FLV视频流URL")
return stream_url
logger.error("❌ 未找到可用的视频流")
return None
except asyncio.TimeoutError:
logger.error("❌ 获取视频流URL超时")
return None
except aiohttp.ClientError as e:
logger.error(f"❌ 网络请求失败: {e}")
return None
except Exception as e:
logger.error(f"❌ 获取视频流URL时发生未知错误: {e}")
logger.exception("详细错误信息:")
return None
async def download_video_bytes(self, stream_url: str, max_size_mb: int = 100) -> Optional[bytes]:
"""下载视频字节数据
Args:
stream_url: 视频流URL
max_size_mb: 最大下载大小限制（MB），默认100MB
Returns:
视频字节数据或None
"""
try:
logger.info(f"📥 开始下载视频: {stream_url[:50]}...")
# 设置超时和大小限制
timeout = aiohttp.ClientTimeout(total=300, connect=30) # 5分钟总超时，30秒连接超时
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(stream_url, headers=self.headers) as response:
if response.status != 200:
logger.error(f"❌ 下载失败，HTTP状态码: {response.status}")
return None
# 检查内容长度
content_length = response.headers.get('content-length')
if content_length:
size_mb = int(content_length) / 1024 / 1024
if size_mb > max_size_mb:
logger.error(f"❌ 视频文件过大: {size_mb:.1f}MB > {max_size_mb}MB")
return None
logger.info(f"📊 预计下载大小: {size_mb:.1f}MB")
# 分块下载并监控大小
video_bytes = bytearray()
downloaded_mb = 0
async for chunk in response.content.iter_chunked(8192): # 8KB块
video_bytes.extend(chunk)
downloaded_mb = len(video_bytes) / 1024 / 1024
# 检查大小限制
if downloaded_mb > max_size_mb:
logger.error(f"❌ 下载中止，文件过大: {downloaded_mb:.1f}MB > {max_size_mb}MB")
return None
final_size_mb = len(video_bytes) / 1024 / 1024
logger.info(f"✅ 视频下载完成，实际大小: {final_size_mb:.2f}MB")
return bytes(video_bytes)
except asyncio.TimeoutError:
logger.error("❌ 下载超时")
return None
except aiohttp.ClientError as e:
logger.error(f"❌ 网络请求失败: {e}")
return None
except Exception as e:
logger.error(f"❌ 下载视频时发生未知错误: {e}")
logger.exception("详细错误信息:")
return None
async def analyze_bilibili_video(self, url: str, prompt: str = None) -> Dict[str, Any]:
"""分析哔哩哔哩视频并返回详细信息和AI分析结果"""
try:
logger.info(f"🎬 开始分析哔哩哔哩视频: {url}")
# 1. 获取视频基本信息
video_info = await self.get_video_info(url)
if not video_info:
logger.error("❌ 无法获取视频基本信息")
return {"error": "无法获取视频信息"}
logger.info(f"📺 视频标题: {video_info['title']}")
logger.info(f"👤 UP主: {video_info['owner']}")
logger.info(f"⏱️ 时长: {video_info['duration']}秒")
# 2. 获取视频流URL
stream_url = await self.get_video_stream_url(video_info['aid'], video_info['cid'])
if not stream_url:
logger.warning("⚠️ 无法获取视频流，仅返回基本信息")
return {
"video_info": video_info,
"error": "无法获取视频流，仅返回基本信息"
}
# 3. 下载视频
video_bytes = await self.download_video_bytes(stream_url)
if not video_bytes:
logger.warning("⚠️ 视频下载失败，仅返回基本信息")
return {
"video_info": video_info,
"error": "视频下载失败，仅返回基本信息"
}
# 4. 构建增强的元数据信息
enhanced_metadata = {
"title": video_info['title'],
"uploader": video_info['owner'],
"duration": video_info['duration'],
"view_count": video_info['view'],
"like_count": video_info['like'],
"description": video_info['desc'],
"bvid": video_info['bvid'],
"aid": video_info['aid'],
"file_size": len(video_bytes),
"source": "bilibili"
}
# 5. 使用新的视频分析API，传递完整的元数据
logger.info("🤖 开始AI视频分析...")
analysis_result = await self.video_analyzer.analyze_video_from_bytes(
video_bytes=video_bytes,
filename=f"{video_info['title']}.mp4",
prompt=prompt # 使用新API的prompt参数而不是user_question
)
# 6. 检查分析结果
if not analysis_result or not analysis_result.get('summary'):
logger.error("❌ 视频分析失败或返回空结果")
return {
"video_info": video_info,
"error": "视频分析失败，仅返回基本信息"
}
# 7. 格式化返回结果
duration_str = f"{video_info['duration'] // 60}分{video_info['duration'] % 60}秒"
result = {
"video_info": {
"标题": video_info['title'],
"UP主": video_info['owner'],
"时长": duration_str,
"播放量": f"{video_info['view']:,}",
"点赞": f"{video_info['like']:,}",
"投币": f"{video_info['coin']:,}",
"收藏": f"{video_info['favorite']:,}",
"转发": f"{video_info['share']:,}",
"简介": video_info['desc'][:200] + "..." if len(video_info['desc']) > 200 else video_info['desc']
},
"ai_analysis": analysis_result.get('summary', ''),
"success": True,
"metadata": enhanced_metadata # 添加元数据信息
}
logger.info("✅ 哔哩哔哩视频分析完成")
return result
except Exception as e:
error_msg = f"分析哔哩哔哩视频时发生异常: {str(e)}"
logger.error(f"❌ {error_msg}")
logger.exception("详细错误信息:") # 记录完整的异常堆栈
return {"error": f"分析失败: {str(e)}"}
# 全局实例
_bilibili_analyzer = None
def get_bilibili_analyzer() -> BilibiliVideoAnalyzer:
"""获取哔哩哔哩视频分析器实例（单例模式）"""
global _bilibili_analyzer
if _bilibili_analyzer is None:
_bilibili_analyzer = BilibiliVideoAnalyzer()
return _bilibili_analyzer

223
plugins/bilibli/plugin.py Normal file
View File

@@ -0,0 +1,223 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Bilibili 视频观看体验工具
支持哔哩哔哩视频链接解析和AI视频内容分析
"""
from typing import Dict, Any, List, Tuple, Type
from src.plugin_system import BaseTool, ToolParamType, BasePlugin, register_plugin, ComponentInfo, ConfigField
from .bilibli_base import get_bilibili_analyzer
from src.common.logger import get_logger
logger = get_logger("bilibili_tool")
class BilibiliTool(BaseTool):
"""哔哩哔哩视频观看体验工具 - 像真实用户一样观看和评价用户分享的哔哩哔哩视频"""
name = "bilibili_video_watcher"
description = "观看用户分享的哔哩哔哩视频，以真实用户视角给出观看感受和评价"
available_for_llm = True
parameters = [
("url", ToolParamType.STRING, "用户分享给我的哔哩哔哩视频链接，我会认真观看这个视频并给出真实的观看感受", True, None),
("interest_focus", ToolParamType.STRING, "你特别感兴趣的方面（如：搞笑内容、学习资料、美食、游戏、音乐等），我会重点关注这些内容", False, None)
]
def __init__(self, plugin_config: dict = None):
super().__init__(plugin_config)
self.analyzer = get_bilibili_analyzer()
async def execute(self, function_args: Dict[str, Any]) -> Dict[str, Any]:
"""执行哔哩哔哩视频观看体验"""
try:
url = function_args.get("url", "").strip()
interest_focus = function_args.get("interest_focus", "").strip() or None
if not url:
return {
"name": self.name,
"content": "🤔 你想让我看哪个视频呢？给我个链接吧！"
}
logger.info(f"开始'观看'哔哩哔哩视频: {url}")
# 验证是否为哔哩哔哩链接
extracted_url = self.analyzer.extract_bilibili_url(url)
if not extracted_url:
return {
"name": self.name,
"content": "🤨 这好像不是哔哩哔哩的链接诶，我只会看哔哩哔哩的视频哦~ 给我一个bilibili.com或b23.tv的链接吧！"
}
# 构建个性化的观看提示词
watch_prompt = self._build_watch_prompt(interest_focus)
# 执行视频分析
result = await self.analyzer.analyze_bilibili_video(extracted_url, watch_prompt)
if result.get("error"):
return {
"name": self.name,
"content": f"😔 唉，这个视频我看不了... {result['error']}\n可能是网络问题或者视频有限制"
}
# 格式化输出结果
video_info = result.get("video_info", {})
ai_analysis = result.get("ai_analysis", "")
# 构建个性化的观看体验报告
content = self._format_watch_experience(video_info, ai_analysis, interest_focus)
logger.info("✅ 哔哩哔哩视频观看体验完成")
return {
"name": self.name,
"content": content.strip()
}
except Exception as e:
error_msg = f"😅 看视频的时候出了点问题: {str(e)}"
logger.error(error_msg)
return {
"name": self.name,
"content": error_msg
}
def _build_watch_prompt(self, interest_focus: str = None) -> str:
"""构建个性化的观看提示词"""
base_prompt = """请以一个真实哔哩哔哩用户的视角来观看用户分享给我的这个视频。用户特意分享了这个视频给我，我需要认真观看并给出真实的反馈。
我希望你能：
1. 像普通用户一样对这个分享的视频内容做出真实的反应和评价
2. 根据你的"个人喜好"判断这个用户分享的视频是否有趣、是否值得观看
3. 给出主观的观看感受，包括：喜欢什么、不喜欢什么、有什么感想
4. 用轻松、自然的语气表达，就像在和分享视频的朋友聊天
5. 可以表达个人偏好，比如"我比较喜欢..."、"这种类型不太符合我的口味"等
7. 对用户的分享表示感谢，体现出这是用户主动分享给我的内容"""
if interest_focus:
base_prompt += f"\n\n特别关注点：我对 {interest_focus} 相关的内容比较感兴趣，请重点评价这方面的内容。"
return base_prompt
def _format_watch_experience(self, video_info: Dict, ai_analysis: str, interest_focus: str = None) -> str:
"""格式化观看体验报告"""
# 根据播放量生成热度评价
view_count = video_info.get('播放量', '0').replace(',', '')
if view_count.isdigit():
views = int(view_count)
if views > 1000000:
popularity = "🔥 超火爆"
elif views > 100000:
popularity = "🔥 很热门"
elif views > 10000:
popularity = "👍 还不错"
else:
popularity = "🆕 比较新"
else:
popularity = "🤷‍♀️ 数据不明"
# 生成时长评价
duration = video_info.get('时长', '')
if '分' in duration:
time_comment = self._get_duration_comment(duration)
else:
time_comment = ""
content = f"""🎬 **谢谢你分享的这个哔哩哔哩视频！我认真看了一下~**
📺 **视频速览**
• 标题：{video_info.get('标题', '未知')}
• UP主：{video_info.get('UP主', '未知')}
• 时长：{duration} {time_comment}
• 热度：{popularity} ({video_info.get('播放量', '0')}播放)
• 互动：👍{video_info.get('点赞', '0')} 🪙{video_info.get('投币', '0')} ⭐{video_info.get('收藏', '0')}
📝 **UP主说了什么**
{video_info.get('简介', '这个UP主很懒，什么都没写...')[:150]}{'...' if len(video_info.get('简介', '')) > 150 else ''}
🤔 **我的观看感受**
{ai_analysis}
"""
if interest_focus:
content += f"\n💭 **关于你感兴趣的'{interest_focus}'**\n我特别注意了这方面的内容，感觉{self._get_focus_comment()}~"
return content
def _get_duration_comment(self, duration: str) -> str:
"""根据时长生成评价"""
if '分' in duration:
try:
minutes = int(duration.split('分')[0])
if minutes < 3:
return "(短小精悍)"
elif minutes < 10:
return "(时长刚好)"
elif minutes < 30:
return "(有点长，适合闲时观看)"
else:
return "(超长视频，需要耐心)"
except:
return ""
return ""
def _get_focus_comment(self) -> str:
"""生成关注点评价"""
import random
comments = [
"挺符合你的兴趣的",
"内容还算不错",
"可能会让你感兴趣",
"值得一看",
"可能不太符合你的口味",
"内容比较一般"
]
return random.choice(comments)
@register_plugin
class BilibiliPlugin(BasePlugin):
"""哔哩哔哩视频观看体验插件 - 处理用户分享的视频内容"""
# 插件基本信息
plugin_name: str = "bilibili_video_watcher"
enable_plugin: bool = True
dependencies: List[str] = []
python_dependencies: List[str] = []
config_file_name: str = "config.toml"
# 配置节描述
config_section_descriptions = {
"plugin": "插件基本信息",
"bilibili": "哔哩哔哩视频观看配置",
"tool": "工具配置"
}
# 配置Schema定义
config_schema: dict = {
"plugin": {
"name": ConfigField(type=str, default="bilibili_video_watcher", description="插件名称"),
"version": ConfigField(type=str, default="2.0.0", description="插件版本"),
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
"config_version": ConfigField(type=str, default="2.0.0", description="配置文件版本"),
},
"bilibili": {
"timeout": ConfigField(type=int, default=300, description="观看超时时间（秒）"),
"verbose_logging": ConfigField(type=bool, default=True, description="是否启用详细日志"),
"max_retries": ConfigField(type=int, default=3, description="最大重试次数"),
},
"tool": {
"available_for_llm": ConfigField(type=bool, default=True, description="是否对LLM可用"),
"name": ConfigField(type=str, default="bilibili_video_watcher", description="工具名称"),
"description": ConfigField(type=str, default="观看用户分享的哔哩哔哩视频并给出真实观看体验", description="工具描述"),
}
}
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
"""返回插件包含的工具组件"""
return [
(BilibiliTool.get_tool_info(), BilibiliTool)
]