252 lines
10 KiB
Python
252 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Bilibili Content Downloader - B站内容获取工具
|
||
|
||
功能:
|
||
- download_video(bvid, quality=480) - 下载视频
|
||
- extract_subtitle(bvid) - 提取字幕
|
||
- extract_article(cvid) - 获取专栏全文（使用Playwright）
|
||
- 命令行参数支持
|
||
- JSON 输出
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
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
|
||
|
||
|
||
class BilibiliContent:
|
||
"""B站内容获取器"""
|
||
|
||
# 完整的浏览器请求头
|
||
BROWSER_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',
|
||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
|
||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||
'Connection': 'keep-alive',
|
||
'Upgrade-Insecure-Requests': '1',
|
||
'Sec-Fetch-Dest': 'document',
|
||
'Sec-Fetch-Mode': 'navigate',
|
||
'Sec-Fetch-Site': 'none',
|
||
'Sec-Fetch-User': '?1',
|
||
'Cache-Control': 'max-age=0',
|
||
}
|
||
|
||
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]:
|
||
"""获取请求头（包含Cookie）"""
|
||
headers = self.BROWSER_HEADERS.copy()
|
||
if self.cookie:
|
||
headers['Cookie'] = self.cookie
|
||
return headers
|
||
|
||
async def download_video(self, bvid: str, quality: int = 32) -> Dict[str, Any]:
|
||
"""下载B站视频"""
|
||
result = {"success": False, "bvid": bvid, "message": "", "filepath": None, "metadata": {}}
|
||
|
||
try:
|
||
# API 请求使用自动解压
|
||
async with aiohttp.ClientSession(headers=self._get_headers()) as api_session:
|
||
# 获取视频信息
|
||
print(f"获取视频信息: {bvid}")
|
||
async with api_session.get("https://api.bilibili.com/x/web-interface/view", params={'bvid': bvid}) as resp:
|
||
data = await resp.json()
|
||
|
||
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 api_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"] = "未找到视频流"
|
||
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"
|
||
|
||
# 视频下载使用禁用自动解压的 session
|
||
async with aiohttp.ClientSession(headers=self._get_headers(), auto_decompress=False) as download_session:
|
||
async with download_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]:
|
||
"""提取字幕"""
|
||
result = {"success": False, "bvid": bvid, "subtitles": [], "message": ""}
|
||
|
||
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()
|
||
|
||
pages = data.get('data', [])
|
||
if not pages:
|
||
result["message"] = "未找到字幕"
|
||
return result
|
||
|
||
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]:
|
||
"""获取专栏全文（使用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', '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')
|
||
|
||
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)
|
||
else:
|
||
result = await downloader.extract_article(args.id.replace('cv', ''))
|
||
|
||
# 统一输出格式
|
||
if result.get("success"):
|
||
output = {
|
||
"success": True,
|
||
"type": args.type,
|
||
"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)
|