#!/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', 'Accept-Encoding': 'gzip, deflate, br', '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: 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]: """获取专栏全文（使用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') parser.add_argument('--json', action='store_true', help='JSON输出') 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 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)