feat: Improve kb-bilibili-content video download

- Simplify download logic using playurl API
- Support video download without Cookie (FLV format)
- Improve subtitle extraction
- Better error handling and progress display
- Remove unnecessary dependencies
This commit is contained in:
OpenClaw Agent
2026-02-10 13:40:03 +08:00
parent a7ab478dbc
commit 51347f671b
+128 -378
View File
@@ -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)