OpenClaw Agent
51347f671b
- 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
203 lines
7.9 KiB
Python
203 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Bilibili Content Downloader - B站内容获取工具
|
|
|
|
功能:
|
|
- download_video(bvid, quality=480) - 下载低分辨率视频
|
|
- extract_subtitle(bvid) - 提取字幕
|
|
- extract_article(cvid) - 获取专栏全文
|
|
- 命令行参数支持
|
|
- 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站内容获取器"""
|
|
|
|
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]:
|
|
headers = {
|
|
'Referer': 'https://www.bilibili.com/',
|
|
'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 download_video(self, bvid: str, quality: int = 32) -> Dict[str, Any]:
|
|
"""下载B站视频"""
|
|
result = {"success": False, "bvid": bvid, "message": "", "filepath": None, "metadata": {}}
|
|
|
|
try:
|
|
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 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"] = "未找到视频流"
|
|
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]:
|
|
"""提取字幕"""
|
|
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]:
|
|
"""获取专栏全文"""
|
|
result = {"success": False, "cvid": cvid, "title": "", "content": "", "message": ""}
|
|
|
|
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']}")
|
|
|
|
except Exception as e:
|
|
result["message"] = f"错误: {str(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')
|
|
parser.add_argument('--json', action='store_true', help='JSON输出')
|
|
|
|
args = parser.parse_args()
|
|
downloader = BilibiliContent(output_dir=args.output)
|
|
|
|
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 args.json:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
else:
|
|
print(f"\n{result['message']}")
|
|
if result.get('filepath'):
|
|
print(f"文件: {result['filepath']}")
|
|
|
|
return result
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
result = asyncio.run(main())
|
|
sys.exit(0 if result.get("success") else 1)
|
|
except KeyboardInterrupt:
|
|
print("\n已取消")
|
|
sys.exit(1)
|