#!/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]: """ 提取字幕 注意：没有找到官方字幕是正常的！很多视频没有自动生成的字幕。 返回 success=True 即使没有字幕，让工作流继续执行后续步骤（如 Whisper）。 """ result = {"success": False, "bvid": bvid, "subtitles": [], "has_subtitle": False, "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（这是正常的）" # 仍然返回 success=True，让工作流继续 result["success"] = True 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"] = "没有官方字幕（这是正常的），将使用 Whisper 进行语音识别" # 仍然返回 success=True，让工作流继续 result["success"] = True return result result["success"] = True result["has_subtitle"] = 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"获取字幕出错（这是正常的），将使用 Whisper: {str(e)}" # 出错时也返回 success=True，让工作流继续 result["success"] = True return result async def download_subtitle(self, bvid: str) -> Dict[str, Any]: """ 下载字幕并转换为 SRT 格式 如果没有官方字幕，返回 success=True 但 has_subtitle=False， 让工作流继续使用 Whisper 进行语音识别。 """ result = {"success": False, "bvid": bvid, "filepath": None, "has_subtitle": False, "message": ""} try: import aiohttp # 获取 CID 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，将使用 Whisper 进行语音识别" result["success"] = True 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"] = "没有官方字幕，将使用 Whisper 进行语音识别" result["success"] = True return result # 选择中文字幕，如果没有则选第一个 zh_sub = None for s in subtitles: if s.get('lan') in ('ai-zh', 'zh-CN', 'zh'): zh_sub = s break if not zh_sub: zh_sub = subtitles[0] # 下载字幕文件 sub_url = zh_sub.get('subtitle_url', '') if not sub_url: result["message"] = "字幕URL为空，将使用 Whisper" result["success"] = True return result if sub_url.startswith('//'): sub_url = 'https:' + sub_url async with aiohttp.ClientSession() as session: async with session.get(sub_url) as resp: sub_content = await resp.json() # 转换为 SRT 格式 body = sub_content.get('body', []) if not body: result["message"] = "字幕内容为空，将使用 Whisper" result["success"] = True return result def format_srt_time(t: float) -> str: """将秒数转换为 SRT 时间格式""" h = int(t // 3600) m = int((t % 3600) // 60) s = int(t % 60) ms = int((t - int(t)) * 1000) return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" srt_lines = [] for i, line in enumerate(body, 1): start = line.get('from', 0) end = line.get('to', 0) content = line.get('content', '') srt_lines.extend([str(i), f"{format_srt_time(start)} --> {format_srt_time(end)}", content, ""]) srt_content = "\n".join(srt_lines) # 保存 SRT 文件 output_path = self.output_dir / f"{bvid}.srt" with open(output_path, 'w', encoding='utf-8') as f: f.write(srt_content) result["success"] = True result["has_subtitle"] = True result["filepath"] = str(output_path) result["message"] = f"✅ 官方字幕已下载: {output_path.name} ({len(body)}条)" print(result["message"]) except Exception as e: result["message"] = f"下载字幕出错: {str(e)}，将使用 Whisper" result["success"] = True # 仍然返回成功，让工作流继续 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', 'subtitle-download', '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) elif args.type == 'subtitle-download': # 下载并转换字幕为 SRT 格式 result = await downloader.download_subtitle(args.id) else: result = await downloader.extract_article(args.id.replace('cv', '')) # 统一输出格式 if result.get("success"): output = { "success": True, "type": args.type, "has_subtitle": result.get("has_subtitle", False), "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)