420 lines
15 KiB
Python
420 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Bilibili Content Downloader - B站内容获取工具
|
||
|
||
功能:
|
||
- download_video(bvid, quality=480) - 下载视频（使用 yt-dlp）
|
||
- extract_subtitle(bvid) - 提取字幕
|
||
- extract_article(cvid) - 获取专栏全文（使用Playwright）
|
||
- 命令行参数支持
|
||
- JSON 输出
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
|
||
class BilibiliContent:
|
||
"""B站内容获取器"""
|
||
|
||
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 _create_cookie_file(self) -> str:
|
||
"""创建 Netscape 格式的 Cookie 文件"""
|
||
if not self.cookie:
|
||
return None
|
||
|
||
# 解析 Cookie 字符串
|
||
cookies = {}
|
||
for item in self.cookie.split(';'):
|
||
item = item.strip()
|
||
if '=' in item:
|
||
key, value = item.split('=', 1)
|
||
cookies[key.strip()] = value.strip()
|
||
|
||
# 创建 Netscape 格式的 Cookie 文件
|
||
cookie_content = """# Netscape HTTP Cookie File
|
||
# Generated by bilibili_content.py
|
||
|
||
"""
|
||
|
||
for name, value in cookies.items():
|
||
cookie_content += f".bilibili.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n"
|
||
|
||
# 写入临时文件
|
||
fd, path = tempfile.mkstemp(suffix='.txt')
|
||
with os.fdopen(fd, 'w') as f:
|
||
f.write(cookie_content)
|
||
|
||
return path
|
||
|
||
async def download_video(self, bvid: str, quality: int = 64) -> Dict[str, Any]:
|
||
"""下载B站视频（使用 yt-dlp）"""
|
||
result = {"success": False, "bvid": bvid, "message": "", "filepath": None, "metadata": {}}
|
||
|
||
try:
|
||
# 先获取视频信息
|
||
print(f"获取视频信息: {bvid}")
|
||
|
||
cookie_file = self._create_cookie_file()
|
||
url = f"https://www.bilibili.com/video/{bvid}"
|
||
|
||
try:
|
||
# 获取视频标题（用于文件名）
|
||
if cookie_file:
|
||
cmd = ["yt-dlp", "--cookies", cookie_file, "--get-title", url]
|
||
else:
|
||
cmd = ["yt-dlp", "--get-title", url]
|
||
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE
|
||
)
|
||
stdout, stderr = await proc.communicate()
|
||
|
||
title = stdout.decode('utf-8').strip()
|
||
if not title or 'error' in title.lower():
|
||
# 尝试备用方法
|
||
title = f"video_{bvid}"
|
||
except Exception as e:
|
||
title = f"video_{bvid}"
|
||
|
||
# 构建 yt-dlp 命令
|
||
safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_', '丨')).strip()[:50]
|
||
|
||
# 选择视频格式
|
||
if quality == 80:
|
||
# 1080P
|
||
format_str = "30112+30280"
|
||
elif quality == 64:
|
||
# 720P
|
||
format_str = "30064+30280"
|
||
elif quality == 32:
|
||
# 480P
|
||
format_str = "30032+30280"
|
||
else:
|
||
# 360P
|
||
format_str = "30016+30280"
|
||
|
||
print(f"下载视频: {title}")
|
||
print(f"格式: {format_str}")
|
||
|
||
output_template = str(self.output_dir / f"{safe_title}.%(ext)s")
|
||
|
||
cmd = [
|
||
"yt-dlp",
|
||
"-f", format_str,
|
||
"-o", output_template,
|
||
"--no-download-archive",
|
||
"-q",
|
||
url
|
||
]
|
||
|
||
if cookie_file:
|
||
cmd.extend(["--cookies", cookie_file])
|
||
|
||
print(f"执行命令: {' '.join(cmd[:4])}...")
|
||
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE
|
||
)
|
||
stdout, stderr = await proc.communicate()
|
||
|
||
# 查找下载的文件
|
||
downloaded_files = list(self.output_dir.glob(f"{safe_title}.*"))
|
||
if downloaded_files:
|
||
video_file = downloaded_files[0]
|
||
result["success"] = True
|
||
result["filepath"] = str(video_file)
|
||
result["message"] = f"下载完成: {video_file.name}"
|
||
result["metadata"] = {"title": title, "bvid": bvid}
|
||
print(f"✅ {result['message']}")
|
||
else:
|
||
result["message"] = "未找到下载的文件"
|
||
print(f"❌ {result['message']}")
|
||
|
||
# 清理临时 Cookie 文件
|
||
if cookie_file and os.path.exists(cookie_file):
|
||
os.remove(cookie_file)
|
||
|
||
except Exception as e:
|
||
result["message"] = f"错误: {str(e)}"
|
||
print(f"❌ {result['message']}")
|
||
|
||
return result
|
||
|
||
async def extract_subtitle(self, bvid: str) -> Dict[str, Any]:
|
||
"""
|
||
提取字幕
|
||
|
||
注意：没有找到官方字幕是正常的！很多视频没有自动生成的字幕。
|
||
返回 success=True 即使没有字幕，让工作流继续执行后续步骤（如 Whisper）。
|
||
"""
|
||
result = {"success": False, "bvid": bvid, "subtitles": [], "has_subtitle": False, "message": ""}
|
||
|
||
try:
|
||
import aiohttp
|
||
|
||
api_url = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}"
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(api_url) as resp:
|
||
data = await resp.json()
|
||
|
||
pages = data.get('data', [])
|
||
if not pages:
|
||
result["message"] = "未找到分P（这是正常的）"
|
||
# 仍然返回 success=True，让工作流继续
|
||
result["success"] = True
|
||
return result
|
||
|
||
cid = pages[0].get('cid', '')
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
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"] = "没有官方字幕（这是正常的），将使用 Whisper 进行语音识别"
|
||
# 仍然返回 success=True，让工作流继续
|
||
result["success"] = True
|
||
return result
|
||
|
||
result["success"] = True
|
||
result["has_subtitle"] = 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"获取字幕出错（这是正常的），将使用 Whisper: {str(e)}"
|
||
# 出错时也返回 success=True，让工作流继续
|
||
result["success"] = True
|
||
|
||
return result
|
||
|
||
async def download_subtitle(self, bvid: str) -> Dict[str, Any]:
|
||
"""
|
||
下载字幕并转换为 SRT 格式
|
||
|
||
如果没有官方字幕，返回 success=True 但 has_subtitle=False，
|
||
让工作流继续使用 Whisper 进行语音识别。
|
||
"""
|
||
result = {"success": False, "bvid": bvid, "filepath": None, "has_subtitle": False, "message": ""}
|
||
|
||
try:
|
||
import aiohttp
|
||
|
||
# 获取 CID
|
||
api_url = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}"
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(api_url) as resp:
|
||
data = await resp.json()
|
||
|
||
pages = data.get('data', [])
|
||
if not pages:
|
||
result["message"] = "未找到分P，将使用 Whisper 进行语音识别"
|
||
result["success"] = True
|
||
return result
|
||
|
||
cid = pages[0].get('cid', '')
|
||
|
||
# 获取字幕列表
|
||
async with aiohttp.ClientSession() as session:
|
||
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"] = "没有官方字幕，将使用 Whisper 进行语音识别"
|
||
result["success"] = True
|
||
return result
|
||
|
||
# 选择中文字幕，如果没有则选第一个
|
||
zh_sub = None
|
||
for s in subtitles:
|
||
if s.get('lan') in ('ai-zh', 'zh-CN', 'zh'):
|
||
zh_sub = s
|
||
break
|
||
if not zh_sub:
|
||
zh_sub = subtitles[0]
|
||
|
||
# 下载字幕文件
|
||
sub_url = zh_sub.get('subtitle_url', '')
|
||
if not sub_url:
|
||
result["message"] = "字幕URL为空，将使用 Whisper"
|
||
result["success"] = True
|
||
return result
|
||
|
||
if sub_url.startswith('//'):
|
||
sub_url = 'https:' + sub_url
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(sub_url) as resp:
|
||
sub_content = await resp.json()
|
||
|
||
# 转换为 SRT 格式
|
||
body = sub_content.get('body', [])
|
||
if not body:
|
||
result["message"] = "字幕内容为空，将使用 Whisper"
|
||
result["success"] = True
|
||
return result
|
||
|
||
def format_srt_time(t: float) -> str:
|
||
"""将秒数转换为 SRT 时间格式"""
|
||
h = int(t // 3600)
|
||
m = int((t % 3600) // 60)
|
||
s = int(t % 60)
|
||
ms = int((t - int(t)) * 1000)
|
||
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
|
||
|
||
srt_lines = []
|
||
for i, line in enumerate(body, 1):
|
||
start = line.get('from', 0)
|
||
end = line.get('to', 0)
|
||
content = line.get('content', '')
|
||
srt_lines.extend([str(i), f"{format_srt_time(start)} --> {format_srt_time(end)}", content, ""])
|
||
|
||
srt_content = "\n".join(srt_lines)
|
||
|
||
# 保存 SRT 文件
|
||
output_path = self.output_dir / f"{bvid}.srt"
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
f.write(srt_content)
|
||
|
||
result["success"] = True
|
||
result["has_subtitle"] = True
|
||
result["filepath"] = str(output_path)
|
||
result["message"] = f"✅ 官方字幕已下载: {output_path.name} ({len(body)}条)"
|
||
print(result["message"])
|
||
|
||
except Exception as e:
|
||
result["message"] = f"下载字幕出错: {str(e)}，将使用 Whisper"
|
||
result["success"] = True # 仍然返回成功，让工作流继续
|
||
|
||
return result
|
||
|
||
async def extract_article(self, cvid: str) -> Dict[str, Any]:
|
||
"""获取专栏全文（使用Playwright异步API）"""
|
||
result = {"success": False, "cvid": cvid, "title": "", "content": "", "message": ""}
|
||
|
||
try:
|
||
from playwright.async_api import async_playwright
|
||
|
||
cvid_clean = cvid.replace('cv', '')
|
||
url = f"https://www.bilibili.com/read/cv{cvid_clean}"
|
||
print(f"获取专栏: {url}")
|
||
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(headless=True)
|
||
page = await browser.new_page()
|
||
|
||
await page.goto(url, wait_until="networkidle")
|
||
await page.wait_for_timeout(3000)
|
||
|
||
# 提取标题
|
||
title_elem = await page.query_selector('h1')
|
||
if title_elem:
|
||
result["title"] = (await title_elem.inner_text()).strip()
|
||
print(f"标题: {result['title'][:50]}...")
|
||
|
||
# 提取正文内容
|
||
content_elem = await page.query_selector('.read-article-holder')
|
||
if content_elem:
|
||
html_content = await content_elem.inner_html()
|
||
text_content = await content_elem.inner_text()
|
||
result["content"] = text_content.strip()
|
||
result["html_content"] = html_content.strip()
|
||
result["success"] = bool(result["title"] and result["content"])
|
||
print(f"内容长度: {len(result['content'])} 字符")
|
||
else:
|
||
result["message"] = "未找到专栏内容"
|
||
|
||
await browser.close()
|
||
|
||
if result["success"]:
|
||
result["message"] = f"✅ 获取成功: {result['title'][:30]}..."
|
||
else:
|
||
result["message"] = f"获取失败: {result.get('message', '未知错误')}"
|
||
|
||
except Exception as e:
|
||
result["message"] = f"错误: {str(e)}"
|
||
print(f"错误: {e}")
|
||
|
||
return result
|
||
|
||
|
||
async def main():
|
||
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', 'subtitle-download', 'article'], help='内容类型')
|
||
parser.add_argument('--output', '-o', default='./output', help='输出目录')
|
||
parser.add_argument('--quality', '-q', type=int, default=64, help='视频质量: 16=360P, 32=480P, 64=720P, 80=1080P')
|
||
|
||
args = parser.parse_args()
|
||
|
||
cookie = os.environ.get('BILIBILI_COOKIE')
|
||
downloader = BilibiliContent(output_dir=args.output, cookie=cookie)
|
||
|
||
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)
|
||
elif args.type == 'subtitle-download':
|
||
# 下载并转换字幕为 SRT 格式
|
||
result = await downloader.download_subtitle(args.id)
|
||
else:
|
||
result = await downloader.extract_article(args.id.replace('cv', ''))
|
||
|
||
# 统一输出格式
|
||
if result.get("success"):
|
||
output = {
|
||
"success": True,
|
||
"type": args.type,
|
||
"has_subtitle": result.get("has_subtitle", False),
|
||
"filename": Path(result.get('filepath', '')).name if result.get('filepath') else None,
|
||
"filepath": str(Path(result.get('filepath', '')).resolve()) if result.get('filepath') else None,
|
||
"message": result.get('message', '处理成功')
|
||
}
|
||
else:
|
||
output = {
|
||
"success": False,
|
||
"error": result.get('message', '处理失败'),
|
||
"detail": f"ID: {args.id}, 类型: {args.type}"
|
||
}
|
||
|
||
print(json.dumps(output, ensure_ascii=False, indent=2))
|
||
return result
|
||
|
||
|
||
if __name__ == '__main__':
|
||
try:
|
||
result = asyncio.run(main())
|
||
sys.exit(0 if result.get("success") else 1)
|
||
except KeyboardInterrupt:
|
||
print(json.dumps({"success": False, "error": "用户取消"}, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|