diff --git a/skills/kb-bilibili-content/bilibili_content.py b/skills/kb-bilibili-content/bilibili_content.py index 7d81341..5898310 100644 --- a/skills/kb-bilibili-content/bilibili_content.py +++ b/skills/kb-bilibili-content/bilibili_content.py @@ -17,423 +17,178 @@ import json import os import re import sys +import aiohttp from datetime import datetime from pathlib import Path from typing import Optional, Dict, Any, List -import aiohttp - class BilibiliContent: """B站内容获取器""" - def __init__(self, output_dir: str = "./output"): + def __init__(self, output_dir: str = "./output", cookie: str = None): self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) + self.cookie = cookie def _get_headers(self) -> Dict[str, str]: - """获取请求头""" - return { + 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', - 'Accept': 'application/json, text/plain, */*', - 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', + 'Accept': 'application/json', + 'Accept-Language': 'zh-CN,zh;q=0.9', } + if self.cookie: + headers['Cookie'] = self.cookie + return headers - async def _make_request(self, session, url: str, params: Dict = None) -> Optional[Dict]: - """发送API请求""" + async def download_video(self, bvid: str, quality: int = 32) -> Dict[str, Any]: + """下载B站视频""" + result = {"success": False, "bvid": bvid, "message": "", "filepath": None, "metadata": {}} + try: - import aiohttp - async with session.get(url, params=params, headers=self._get_headers(), timeout=aiohttp.ClientTimeout(total=30)) as response: - if response.status == 200: - return await response.json() - return None - except Exception as e: - print(f"请求错误: {e}") - return None - - async def download_video(self, bvid: str, quality: int = 480) -> Dict[str, Any]: - """ - 下载B站视频 - - Args: - bvid: B站视频BV号 (如: BV1xx411c7mD) - quality: 视频质量 (默认480) - - 127: 8K - - 126:杜比 - - 125:HDR - - 120:4K - - 116:1080P60 - - 112:1080P+ - - 80:1080P - - 74:720P60 - - 64:720P - - 32:480P - - 16:360P - - Returns: - Dict: 包含状态、消息和文件路径的结果 - """ - result = { - "success": False, - "bvid": bvid, - "message": "", - "filepath": None, - "metadata": {} - } - - async with aiohttp.ClientSession() as session: - # 1. 获取视频信息 - print(f"正在获取视频信息: {bvid}") - api_url = "https://api.bilibili.com/x/web-interface/view" - data = await self._make_request(session, api_url, {'bvid': bvid}) - - if not data or data.get('code') != 0: - result["message"] = f"获取视频信息失败: {data.get('message', '未知错误')}" - return result - - video_data = data.get('data', {}) - title = video_data.get('title', '未知标题') - aid = video_data.get('aid') - cid = video_data.get('cid') - duration = video_data.get('duration', 0) - author = video_data.get('owner', {}).get('name', '未知作者') - - result["metadata"] = { - "title": title, - "author": author, - "aid": aid, - "cid": cid, - "duration": duration, - "bvid": bvid - } - - print(f"视频标题: {title}") - print(f"作者: {author}") - print(f"时长: {duration}秒") - - # 2. 获取视频流地址 - print(f"\n正在获取视频流...") - - # 选择合适的质量 - quality_map = {127: 127, 126: 126, 125: 125, 120: 120, 116: 116, 112: 112, 80: 80, 74: 74, 64: 64, 32: 32, 16: 16} - target_quality = quality_map.get(quality, 32) # 默认480P - - playurl_api = "https://api.bilibili.com/x/player/playurl" - params = { - 'bvid': bvid, - 'cid': cid, - 'qn': target_quality, - 'otype': 'json', - 'fnver': 0, - 'fnval': 16, # 支持DASH格式 - } - - playurl_data = await self._make_request(session, playurl_api, params) - - if not playurl_data or playurl_data.get('code') != 0: - result["message"] = f"获取视频流失败: {playurl_data.get('message', '未知错误')}" - return result - - play_data = playurl_data.get('data', {}) - - # 获取视频和音频流 - dash_data = play_data.get('dash') - if dash_data: - video_list = dash_data.get('video', []) - audio_list = dash_data.get('audio', []) + async with aiohttp.ClientSession(headers=self._get_headers()) as session: + # 获取视频信息 + print(f"获取视频信息: {bvid}") + async with session.get("https://api.bilibili.com/x/web-interface/view", params={'bvid': bvid}) as resp: + data = await resp.json() - if video_list: - # 选择质量最高的视频流 - video_stream = video_list[0] - video_url = video_stream.get('base_url') or video_stream.get('baseUrl') - video_size = video_list[0].get('size', 0) - - # 获取音频流 - audio_stream = audio_list[0] if audio_list else None - audio_url = audio_stream.get('base_url') or audio_stream.get('baseUrl') if audio_stream else None - - print(f"找到视频流: {video_url[:100]}...") - - # 3. 下载视频流 - print(f"\n正在下载视频...") - - # 下载视频 - async with session.get(video_url, headers=self._get_headers()) as resp: - if resp.status == 200: - video_content = await resp.read() - print(f"视频下载完成: {len(video_content) / 1024 / 1024:.2f}MB") - else: - result["message"] = f"视频下载失败: HTTP {resp.status}" - return result - - # 下载音频 - audio_content = None - if audio_url: - print(f"正在下载音频...") - async with session.get(audio_url, headers=self._get_headers()) as resp: - if resp.status == 200: - audio_content = await resp.read() - print(f"音频下载完成: {len(audio_content) / 1024 / 1024:.2f}MB") - - # 4. 合并音视频 - print(f"\n正在合并音视频...") - - safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:50] - safe_title = re.sub(r'[<>:"/\\|?*]', '', safe_title) - - output_path = self.output_dir / f"{safe_title}_{bvid}" - - # 保存文件 - if video_content: - video_path = output_path.with_suffix('.mp4') - with open(video_path, 'wb') as f: - f.write(video_content) - - result["filepath"] = str(video_path) - result["success"] = True - result["message"] = f"视频已保存: {video_path}" - print(f"✅ 下载完成: {video_path}") - else: + if data.get('code') != 0: + result["message"] = f"获取失败: {data.get('message')}" + return result + + video_data = data.get('data', {}) + title = video_data.get('title', '未知标题') + cid = video_data.get('cid') + + result["metadata"] = {"title": title, "cid": cid, "bvid": bvid} + print(f"标题: {title}") + + # 获取播放地址 + print(f"获取视频流...") + async with session.get("https://api.bilibili.com/x/player/playurl", + params={'bvid': bvid, 'cid': cid, 'qn': quality}) as resp: + play_data = await resp.json() + + if play_data.get('code') != 0: + result["message"] = f"获取失败: {play_data.get('message')}" + return result + + durl = play_data.get('data', {}).get('durl', []) + if not durl: result["message"] = "未找到视频流" - else: - # 回退到FLV格式 - durl = play_data.get('durl', []) - if durl: - video_url = durl[0].get('url') - print(f"使用FLV格式: {video_url[:100]}...") - - async with session.get(video_url, headers=self._get_headers()) as resp: - if resp.status == 200: - video_content = await resp.read() - - safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:50] - safe_title = re.sub(r'[<>:"/\\|?*]', '', safe_title) - - output_path = self.output_dir / f"{safe_title}_{bvid}.flv" - with open(output_path, 'wb') as f: - f.write(video_content) - - result["filepath"] = str(output_path) - result["success"] = True - result["message"] = f"视频已保存: {output_path}" - print(f"✅ 下载完成: {output_path}") - else: - result["message"] = f"视频下载失败: HTTP {resp.status}" - else: - result["message"] = "未找到可用的视频流" + return result + + video_url = durl[0].get('url') + print(f"下载视频...") + + # 下载 + safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:30] + output_path = self.output_dir / f"{safe_title}_{bvid}.flv" + + async with session.get(video_url) as resp: + with open(output_path, 'wb') as f: + async for chunk in resp.content.iter_chunked(81920): + f.write(chunk) + print(f"\r下载中...", end="", flush=True) + + print(f"\n✅ 已保存: {output_path}") + result["success"] = True + result["filepath"] = str(output_path) + result["message"] = f"下载完成" + + except Exception as e: + result["message"] = f"错误: {str(e)}" return result async def extract_subtitle(self, bvid: str) -> Dict[str, Any]: - """ - 提取B站视频字幕 + """提取字幕""" + result = {"success": False, "bvid": bvid, "subtitles": [], "message": ""} - Args: - bvid: B站视频BV号 - - Returns: - Dict: 包含字幕内容的JSON结果 - """ - result = { - "success": False, - "bvid": bvid, - "subtitles": [], - "message": "" - } - - async with aiohttp.ClientSession() as session: - # 获取视频信息 - api_url = "https://api.bilibili.com/x/web-interface/view" - data = await self._make_request(session, api_url, {'bvid': bvid}) - - if not data or data.get('code') != 0: - result["message"] = f"获取视频信息失败: {data.get('message', '未知错误')}" - return result - - cid = data.get('data', {}).get('cid') - - # 获取字幕列表 - subtitle_api = "https://api.bilibili.com/x/player/pagelist" - page_data = await self._make_request(session, subtitle_api, {'bvid': bvid}) - - # 获取字幕详情 - subtitle_list = [] - if page_data and page_data.get('code') == 0: - cid = page_data.get('data', [{}])[0].get('cid', cid) + try: + async with aiohttp.ClientSession(headers=self._get_headers()) as session: + async with session.get("https://api.bilibili.com/x/player/pagelist", params={'bvid': bvid}) as resp: + data = await resp.json() - subtitle_url = f"https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}" - subtitle_data = await self._make_request(session, subtitle_url) + pages = data.get('data', []) + if not pages: + result["message"] = "未找到字幕" + return result - if subtitle_data and subtitle_data.get('code') == 0: - subtitles = subtitle_data.get('data', {}).get('subtitle', {}).get('list', []) - - for sub in subtitles: - sub_title = sub.get('lan_doc', '未知语言') - sub_url = sub.get('subtitle_url') - - if sub_url: - # 下载字幕文件 - async with session.get(sub_url) as resp: - if resp.status == 200: - sub_content = await resp.json() - subtitle_list.append({ - "language": sub.get('lan', 'unknown'), - "title": sub_title, - "content": sub_content - }) - - result["subtitles"] = subtitle_list - result["success"] = len(subtitle_list) > 0 - result["message"] = f"找到 {len(subtitle_list)} 个字幕" if subtitle_list else "未找到字幕" - - # 保存字幕到文件 - if subtitle_list: - output_path = self.output_dir / f"{bvid}_subtitles.json" - with open(output_path, 'w', encoding='utf-8') as f: - json.dump(result, f, ensure_ascii=False, indent=2) - result["filepath"] = str(output_path) - print(f"✅ 字幕已保存: {output_path}") + cid = pages[0].get('cid', '') + + async with session.get(f"https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}") as resp: + sub_data = await resp.json() + + subtitles = sub_data.get('data', {}).get('subtitle', {}).get('list', []) + + if not subtitles: + result["message"] = "未找到字幕" + return result + + result["success"] = True + result["subtitles"] = [{"lan": s.get('lan'), "title": s.get('lan_doc')} for s in subtitles] + result["message"] = f"找到 {len(subtitles)} 个字幕" + print(result["message"]) + + except Exception as e: + result["message"] = f"错误: {str(e)}" return result async def extract_article(self, cvid: str) -> Dict[str, Any]: - """ - 获取B站专栏全文 + """获取专栏全文""" + result = {"success": False, "cvid": cvid, "title": "", "content": "", "message": ""} - Args: - cvid: B站专栏CV号 + try: + async with aiohttp.ClientSession(headers=self._get_headers()) as session: + async with session.get("https://api.bilibili.com/x/article/viewinfo", params={'id': cvid}) as resp: + data = await resp.json() + + if data.get('code') != 0: + result["message"] = f"获取失败: {data.get('message')}" + return result + + article = data.get('data', {}) + result["title"] = article.get('title', '') + result["content"] = article.get('content', '') + result["success"] = bool(result["title"] and result["content"]) + result["message"] = f"获取: {result['title']}" if result["success"] else "获取失败" + + if result["success"]: + print(f"✅ {result['title']}") - Returns: - Dict: 包含专栏内容的JSON结果 - """ - result = { - "success": False, - "cvid": cvid, - "title": "", - "content": "", - "message": "" - } - - async with aiohttp.ClientSession() as session: - # 获取专栏信息 - api_url = "https://api.bilibili.com/x/article/viewinfo" - data = await self._make_request(session, api_url, {'id': cvid}) - - if not data or data.get('code') != 0: - result["message"] = f"获取专栏信息失败: {data.get('message', '未知错误')}" - return result - - article_data = data.get('data', {}) - title = article_data.get('title', '') - content = article_data.get('content', '') - - # 获取作者信息 - author = article_data.get('author', {}).get('name', '未知作者') - publish_time = article_data.get('publish_time', 0) - view = article_data.get('stats', {}).get('view', 0) - like = article_data.get('stats', {}).get('like', 0) - - result["title"] = title - result["content"] = content - result["author"] = author - result["publish_time"] = datetime.fromtimestamp(publish_time).strftime('%Y-%m-%d %H:%M:%S') if publish_time else '' - result["stats"] = {"view": view, "like": like} - result["success"] = bool(title and content) - result["message"] = f"成功获取专栏: {title}" if result["success"] else "获取专栏内容失败" - - # 保存到文件 - if result["success"]: - safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:30] - safe_title = re.sub(r'[<>:"/\\|?*]', '', safe_title) - - # 保存JSON - json_path = self.output_dir / f"{safe_title}_{cvid}.json" - with open(json_path, 'w', encoding='utf-8') as f: - json.dump(result, f, ensure_ascii=False, indent=2) - - # 保存Markdown - md_content = f"""# {title} - -> **来源**: [B站专栏](https://www.bilibili.com/read/cv{cvid}) -> **作者**: {author} -> **发布时间**: {result["publish_time"]} -> **阅读**: {view} | **点赞**: {like} - ---- - -{content} -""" - md_path = self.output_dir / f"{safe_title}_{cvid}.md" - with open(md_path, 'w', encoding='utf-8') as f: - f.write(md_content) - - result["json_path"] = str(json_path) - result["md_path"] = str(md_path) - print(f"✅ 专栏已保存: {md_path}") + except Exception as e: + result["message"] = f"错误: {str(e)}" return result async def main(): - """命令行入口""" - parser = argparse.ArgumentParser( - description='Bilibili Content Downloader - B站内容获取工具', - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -示例: - %(prog)s --id "BV1xx411c7mD" --type video # 下载视频 - %(prog)s --id "BV1xx411c7mD" --type video --quality 720 # 下载720P视频 - %(prog)s --id "BV1xx411c7mD" --type subtitle # 提取字幕 - %(prog)s --id "cv43780186" --type article # 获取专栏 - """ - ) - - parser.add_argument('--id', '-i', required=True, help='B站内容ID (BV号或CV号)') - parser.add_argument('--type', '-t', required=True, choices=['video', 'subtitle', 'article'], - help='内容类型: video=视频, subtitle=字幕, article=专栏') - parser.add_argument('--output', '-o', default='./output', help='输出目录 (默认: ./output)') - parser.add_argument('--quality', '-q', type=int, default=480, - help='视频质量 (默认: 480, 可选: 360, 480, 720, 1080)') - parser.add_argument('--json', action='store_true', help='JSON格式输出') + parser = argparse.ArgumentParser(description='Bilibili Content Downloader') + parser.add_argument('--id', '-i', required=True, help='B站内容ID') + parser.add_argument('--type', '-t', required=True, choices=['video', 'subtitle', 'article'], help='内容类型') + parser.add_argument('--output', '-o', default='./output', help='输出目录') + parser.add_argument('--quality', '-q', type=int, default=32, help='视频质量: 16=360P, 32=480P, 64=720P, 80=1080P') + parser.add_argument('--json', action='store_true', help='JSON输出') args = parser.parse_args() - - # 确定内容类型 - is_article = args.id.startswith('cv') - - if args.type == 'article' or is_article: - content_type = 'article' - content_id = args.id.replace('cv', '') - elif args.type == 'subtitle': - content_type = 'subtitle' - content_id = args.id - else: - content_type = 'video' - content_id = args.id - - # 创建获取器 downloader = BilibiliContent(output_dir=args.output) - # 执行操作 - if content_type == 'video': - result = await downloader.download_video(content_id, quality=args.quality) - elif content_type == 'subtitle': - result = await downloader.extract_subtitle(content_id) + if args.type == 'video': + result = await downloader.download_video(args.id, quality=args.quality) + elif args.type == 'subtitle': + result = await downloader.extract_subtitle(args.id) else: - result = await downloader.extract_article(content_id) + result = await downloader.extract_article(args.id.replace('cv', '')) - # 输出结果 if args.json: print(json.dumps(result, ensure_ascii=False, indent=2)) else: - if result["success"]: - print(f"\n✅ {result['message']}") - else: - print(f"\n❌ {result['message']}") - - if result.get("filepath"): - print(f"文件路径: {result['filepath']}") + print(f"\n{result['message']}") + if result.get('filepath'): + print(f"文件: {result['filepath']}") return result @@ -443,10 +198,5 @@ if __name__ == '__main__': result = asyncio.run(main()) sys.exit(0 if result.get("success") else 1) except KeyboardInterrupt: - print("\n操作已取消") - sys.exit(1) - except Exception as e: - print(f"\n错误: {e}") - import traceback - traceback.print_exc() + print("\n已取消") sys.exit(1)