diff --git a/.gitignore b/.gitignore index e9506861e..9d95aee4f 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/plugins/bilibli/__init__.py b/plugins/bilibli/__init__.py new file mode 100644 index 000000000..ca649acac --- /dev/null +++ b/plugins/bilibli/__init__.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Bilibili 插件包 +提供B站视频观看体验功能,像真实用户一样浏览和评价视频 +""" + +# 插件会通过 @register_plugin 装饰器自动注册,这里不需要额外的导入 diff --git a/plugins/bilibli/_manifest.json b/plugins/bilibli/_manifest.json new file mode 100644 index 000000000..1e5942bc7 --- /dev/null +++ b/plugins/bilibli/_manifest.json @@ -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编码", + "自动处理重定向链接", + "详细的解析状态反馈" + ] + } +} \ No newline at end of file diff --git a/plugins/bilibli/bilibli_base.py b/plugins/bilibli/bilibli_base.py new file mode 100644 index 000000000..66dc2697e --- /dev/null +++ b/plugins/bilibli/bilibli_base.py @@ -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 \ No newline at end of file diff --git a/plugins/bilibli/plugin.py b/plugins/bilibli/plugin.py new file mode 100644 index 000000000..fec3af74c --- /dev/null +++ b/plugins/bilibli/plugin.py @@ -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) + ]