#!/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]: """提取字幕""" result = {"success": False, "bvid": bvid, "subtitles": [], "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" 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"]) 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=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 = { "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)