feat(bilibili): 使用 yt-dlp 替换原有下载方案
- 重写 download_video 使用 yt-dlp + Netscape cookie 文件 - 支持 BILIBILI_COOKIE 环境变量 - 更新默认质量为 720P - 更新 SKILL.md 文档 环境变量已在 ~/.bashrc 中配置
This commit is contained in:
@@ -4,21 +4,34 @@
|
||||
|
||||
## Features
|
||||
|
||||
- Video download (multiple quality options)
|
||||
- Video download using **yt-dlp** (recommended)
|
||||
- Subtitle extraction
|
||||
- Article content extraction
|
||||
- Low quality for OCR (480p default)
|
||||
- Multiple quality options (360P/480P/720P/1080P)
|
||||
- Supports `BILIBILI_COOKIE` environment variable
|
||||
|
||||
## Environment Variables
|
||||
|
||||
| Variable | Required | Description |
|
||||
|----------|----------|-------------|
|
||||
| `BILIBILI_COOKIE` | No | B站登录 Cookie (for member-only videos) |
|
||||
|
||||
## Commands
|
||||
|
||||
```bash
|
||||
# Download video (传入 --output 指定完整路径)
|
||||
python bilibili_content.py --id "BV1xx411c7mD" --type video --quality 480 --output "/path/to/downloads/BV1xx.mp4"
|
||||
# Download video (720P default)
|
||||
python bilibili_content.py --id "BV1xx411c7mD" --type video --output "/path/to/downloads/"
|
||||
|
||||
# Extract subtitle (传入 --output 指定完整路径)
|
||||
python bilibili_content.py --id "BV1xx411c7mD" --type subtitle --output "/path/to/subtitles/BV1xx.srt"
|
||||
# Download video (1080P)
|
||||
python bilibili_content.py --id "BV1xx411c7mD" --type video --quality 80 --output "/path/to/downloads/"
|
||||
|
||||
# Get article content (传入 --output 指定完整路径)
|
||||
# Download video (480P)
|
||||
python bilibili_content.py --id "BV1xx411c7mD" --type video --quality 32 --output "/path/to/downloads/"
|
||||
|
||||
# Extract subtitle
|
||||
python bilibili_content.py --id "BV1xx411c7mD" --type subtitle --output "/path/to/subtitles/"
|
||||
|
||||
# Get article content
|
||||
python bilibili_content.py --id "cv43780186" --type article --output "/path/to/article.md"
|
||||
```
|
||||
|
||||
@@ -28,15 +41,51 @@ python bilibili_content.py --id "cv43780186" --type article --output "/path/to/a
|
||||
{
|
||||
"success": true,
|
||||
"type": "video",
|
||||
"filename": "BV1xx.mp4",
|
||||
"filepath": "/path/to/downloads/BV1xx.mp4",
|
||||
"message": "下载完成"
|
||||
"filename": "video_title.mp4",
|
||||
"filepath": "/path/to/downloads/video_title.mp4",
|
||||
"message": "下载完成: video_title.mp4"
|
||||
}
|
||||
```
|
||||
|
||||
## Cookie Setup
|
||||
|
||||
### Get Cookie from Browser
|
||||
1. Login to bilibili.com
|
||||
2. Press F12 → Network → Refresh page
|
||||
3. Copy Cookie from request headers
|
||||
|
||||
### Set Environment Variable
|
||||
```bash
|
||||
# Add to ~/.bashrc for persistence
|
||||
export BILIBILI_COOKIE="SESSDATA=xxx; bili_jct=xxx; DedeUserID=xxx; DedeUserID__ckMd5=xxx"
|
||||
|
||||
# Or set temporarily
|
||||
export BILIBILI_COOKIE="your_cookie_here"
|
||||
```
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
cp -r kb-bilibili-content/ ~/skills/
|
||||
pip install aiohttp
|
||||
pip install aiohttp playwright
|
||||
playwright install chromium
|
||||
```
|
||||
|
||||
## How It Works
|
||||
|
||||
Video download uses **yt-dlp** with Netscape-format cookie file:
|
||||
1. Parses `BILIBILI_COOKIE` environment variable
|
||||
2. Creates temporary Netscape-format cookie file
|
||||
3. Runs `yt-dlp --cookies <file> -f <format> -o <output>`
|
||||
4. Cleans up temporary file
|
||||
|
||||
## Quality Options
|
||||
|
||||
| Quality | ID | Resolution | Notes |
|
||||
|---------|-----|------------|-------|
|
||||
| 16 | 360P | 640x360 | Fastest |
|
||||
| 32 | 480P | 852x480 | Default for low bandwidth |
|
||||
| 64 | 720P | 1280x720 | **Recommended** |
|
||||
| 80 | 1080P | 1920x1080 | Best quality |
|
||||
|
||||
Default quality: **64 (720P)**
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
Bilibili Content Downloader - B站内容获取工具
|
||||
|
||||
功能:
|
||||
- download_video(bvid, quality=480) - 下载视频
|
||||
- download_video(bvid, quality=480) - 下载视频(使用 yt-dlp)
|
||||
- extract_subtitle(bvid) - 提取字幕
|
||||
- extract_article(cvid) - 获取专栏全文(使用Playwright)
|
||||
- 命令行参数支持
|
||||
@@ -16,8 +16,9 @@ import asyncio
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import aiohttp
|
||||
import tempfile
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any, List
|
||||
@@ -25,155 +26,203 @@ 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站视频"""
|
||||
|
||||
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:
|
||||
# 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}")
|
||||
# 先获取视频信息
|
||||
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(output_path)
|
||||
result["message"] = f"下载完成"
|
||||
|
||||
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]:
|
||||
"""提取字幕"""
|
||||
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:
|
||||
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"] = "未找到字幕"
|
||||
return result
|
||||
|
||||
cid = pages[0].get('cid', '')
|
||||
|
||||
|
||||
pages = data.get('data', [])
|
||||
if not pages:
|
||||
result["message"] = "未找到分P"
|
||||
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"] = "未找到字幕"
|
||||
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"])
|
||||
|
||||
|
||||
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:
|
||||
@@ -185,18 +234,18 @@ class BilibiliContent:
|
||||
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
|
||||
|
||||
|
||||
@@ -205,20 +254,20 @@ async def main():
|
||||
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('--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)
|
||||
else:
|
||||
result = await downloader.extract_article(args.id.replace('cv', ''))
|
||||
|
||||
|
||||
# 统一输出格式
|
||||
if result.get("success"):
|
||||
output = {
|
||||
@@ -234,7 +283,7 @@ async def main():
|
||||
"error": result.get('message', '处理失败'),
|
||||
"detail": f"ID: {args.id}, 类型: {args.type}"
|
||||
}
|
||||
|
||||
|
||||
print(json.dumps(output, ensure_ascii=False, indent=2))
|
||||
return result
|
||||
|
||||
|
||||
Reference in New Issue
Block a user