#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ B站元数据提取器 独立版本，不依赖 obsidian-kb 结构 支持视频、专栏、UP主信息提取 """ import asyncio import json import logging import re from datetime import datetime from typing import Dict, Any, Optional from urllib.parse import urlparse logger = logging.getLogger(__name__) # 默认请求头 DEFAULT_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', 'Referer': 'https://www.bilibili.com/', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', } try: from playwright.async_api import async_playwright PLAYWRIGHT_AVAILABLE = True except ImportError: PLAYWRIGHT_AVAILABLE = False try: from undetected_playwright import stealth_async STEALTH_AVAILABLE = True except ImportError: STEALTH_AVAILABLE = False try: import httpx HTTPX_AVAILABLE = True except ImportError: HTTPX_AVAILABLE = False class BilibiliExtractor: """ B站元数据提取器 支持： - 视频信息提取（标题、作者、统计数据、字幕等） - 专栏文章提取 - UP主信息提取 """ # 正则模式 BV_PATTERN = re.compile(r'BV[a-zA-Z0-9]{10}') CV_PATTERN = re.compile(r'cv(\d+)') URL_PATTERN = re.compile(r'https?://www\.bilibili\.com/video/(BV[a-zA-Z0-9]{10})') ARTICLE_URL_PATTERN = re.compile(r'https?://www\.bilibili\.com/read/(cv\d+)') UP_URL_PATTERN = re.compile(r'https?://space\.bilibili\.com/(\d+)') def __init__(self, headless: bool = True, timeout: int = 30): """ 初始化提取器 Args: headless: 是否使用无头模式（浏览器） timeout: 超时时间（秒） """ self.headless = headless self.timeout = timeout * 1000 # 转换为毫秒 def extract_video_metadata(self, bvid: str) -> Dict[str, Any]: """ 提取视频元数据 Args: bvid: B站视频BV号（如 BV1xx411c7mD） Returns: Dict containing metadata, stats, description, subtitle """ # 验证并解析BV号 bvid = self._resolve_bvid(bvid) if not bvid: raise ValueError(f"无效的BV号: {bvid}") # 使用异步方法提取 return asyncio.run(self._extract_video_async(bvid)) def extract_article_metadata(self, cvid: str) -> Dict[str, Any]: """ 提取专栏元数据 Args: cvid: B站专栏CV号（如 cv43780186） Returns: Dict containing article metadata """ # 解析CV号 cvid = self._resolve_cvid(cvid) if not cvid: raise ValueError(f"无效的CV号: {cvid}") return asyncio.run(self._extract_article_async(cvid)) def extract_up_metadata(self, mid: str) -> Dict[str, Any]: """ 提取UP主信息 Args: mid: B站用户ID（如 7770724） Returns: Dict containing UP主信息 """ # 解析MID mid = self._resolve_mid(mid) if not mid: raise ValueError(f"无效的MID: {mid}") return asyncio.run(self._extract_up_async(mid)) def _resolve_bvid(self, source_id: str) -> Optional[str]: """解析并返回BV号""" source = source_id.strip() # 直接BV号 if self.BV_PATTERN.fullmatch(source): return source # 完整URL match = self.URL_PATTERN.match(source) if match: return match.group(1) return None def _resolve_cvid(self, source_id: str) -> Optional[str]: """解析并返回CV号""" source = source_id.strip() # 直接CV号（带cv前缀） if source.startswith('cv'): match = self.CV_PATTERN.match(source) if match: return source # 纯数字 if source.isdigit(): return f"cv{source}" # 完整URL match = self.ARTICLE_URL_PATTERN.match(source) if match: return match.group(1) return None def _resolve_mid(self, source_id: str) -> Optional[str]: """解析并返回MID""" source = source_id.strip() # 纯数字 if source.isdigit(): return source # 完整URL match = self.UP_URL_PATTERN.match(source) if match: return match.group(1) return None async def _extract_video_async(self, bvid: str) -> Dict[str, Any]: """异步提取视频信息""" url = f"https://www.bilibili.com/video/{bvid}" result = { 'metadata': {}, 'stats': {}, 'subtitle': '', 'description': '' } if not PLAYWRIGHT_AVAILABLE: # 使用 HTTP API return await self._extract_video_by_api(bvid) async with async_playwright() as playwright: try: browser = await playwright.chromium.launch( headless=self.headless, args=[ '--no-sandbox', '--disable-setuid-sandbox', '--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage', ] ) context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, 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', locale='zh-CN', ) if STEALTH_AVAILABLE: await stealth_async(context) page = await context.new_page() await page.goto(url, wait_until='domcontentloaded', timeout=self.timeout) await asyncio.sleep(2) content = await page.content() video_data = await self._try_extract_video_methods(page, content, bvid) if video_data: result['metadata'] = { 'title': video_data.get('title', ''), 'author': video_data.get('author', ''), 'bvid': video_data.get('bvid', ''), 'aid': video_data.get('aid', ''), 'mid': video_data.get('mid', ''), 'pubtime': video_data.get('pubtime', ''), 'duration': video_data.get('duration', ''), 'url': url, 'tags': video_data.get('tags', ''), } result['stats'] = { 'view': video_data.get('view', 0), 'danmaku': video_data.get('danmaku', 0), 'reply': video_data.get('reply', 0), 'favorite': video_data.get('favorite', 0), 'coin': video_data.get('coin', 0), 'share': video_data.get('share', 0), 'like': video_data.get('like', 0), } result['description'] = video_data.get('desc', '') await browser.close() except Exception as e: logger.error(f"视频提取错误: {e}") # 回退到 API return await self._extract_video_by_api(bvid) return result async def _extract_video_by_api(self, bvid: str) -> Dict[str, Any]: """使用B站API提取视频信息""" result = { 'metadata': {}, 'stats': {}, 'subtitle': '', 'description': '' } try: async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: # 获取视频信息 response = await client.get('https://api.bilibili.com/x/web-interface/view', params={'bvid': bvid}) if response.status_code == 200: data = response.json() if data.get('code') == 0: video_data = data.get('data', {}) result['metadata'] = { 'title': video_data.get('title', ''), 'author': video_data.get('owner', {}).get('name', ''), 'bvid': video_data.get('bvid', ''), 'aid': video_data.get('aid', ''), 'mid': video_data.get('owner', {}).get('mid', ''), 'pubtime': datetime.fromtimestamp(video_data.get('pubdate', 0)).strftime('%Y-%m-%d %H:%M:%S') if video_data.get('pubdate') else '', 'duration': self._format_duration(video_data.get('duration', 0)), 'url': f"https://www.bilibili.com/video/{bvid}", 'tags': ', '.join([t.get('tag_name', '') for t in video_data.get('tag', [])]), } stat = video_data.get('stat', {}) result['stats'] = { 'view': stat.get('view', 0), 'danmaku': stat.get('danmaku', 0), 'reply': stat.get('reply', 0), 'favorite': stat.get('favorite', 0), 'coin': stat.get('coin', 0), 'share': stat.get('share', 0), 'like': stat.get('like', 0), } result['description'] = video_data.get('desc', '') except Exception as e: logger.error(f"API提取错误: {e}") return result async def _try_extract_video_methods(self, page, content: str, bvid: str) -> Dict[str, Any]: """尝试多种方法提取视频信息""" video_data = {} # 方法1: 从 __INITIAL_STATE__ try: initial_state = await page.evaluate('() => typeof window.__INITIAL_STATE__ !== "undefined" ? window.__INITIAL_STATE__ : null') if initial_state: video_data = self._parse_video_state(initial_state) if video_data and video_data.get('title'): return video_data except: pass # 方法2: 使用 API（更可靠） return await self._extract_video_by_api(bvid) def _parse_video_state(self, data: Dict) -> Dict[str, Any]: """解析 __INITIAL_STATE__ 数据""" video_data = {} try: video = data.get('videoData', {}) video_data['title'] = video.get('title', '') video_data['bvid'] = video.get('bvid', '') video_data['aid'] = video.get('aid', '') owner = video.get('owner', {}) video_data['author'] = owner.get('name', '') video_data['mid'] = owner.get('mid', '') stat = video.get('stat', {}) video_data['view'] = stat.get('view', 0) video_data['danmaku'] = stat.get('danmaku', 0) video_data['reply'] = stat.get('reply', 0) video_data['favorite'] = stat.get('favorite', 0) video_data['coin'] = stat.get('coin', 0) video_data['share'] = stat.get('share', 0) video_data['like'] = stat.get('like', 0) pubdate = video.get('pubdate', 0) video_data['pubtime'] = datetime.fromtimestamp(pubdate).strftime('%Y-%m-%d %H:%M:%S') if pubdate else '' video_data['duration'] = self._format_duration(video.get('duration', 0)) video_data['desc'] = video.get('desc', '') video_data['tags'] = ', '.join([t.get('tag_name', '') for t in video.get('tag', [])]) except: pass return video_data async def _extract_article_async(self, cvid: str) -> Dict[str, Any]: """异步提取专栏信息""" result = {} if not HTTPX_AVAILABLE: return {'error': 'httpx 未安装'} try: async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: # 获取专栏信息 response = await client.get('https://api.bilibili.com/x/article/viewinfo', params={'id': cvid[2:]}) if response.status_code == 200: data = response.json() if data.get('code') == 0: article = data.get('data', {}) result = { 'cvid': cvid, 'title': article.get('title', ''), 'author': article.get('author_name', ''), 'mid': article.get('author_mid', ''), 'publish_time': datetime.fromtimestamp(article.get('publish_time', 0)).strftime('%Y-%m-%d %H:%M:%S') if article.get('publish_time') else '', 'view_count': article.get('stats', {}).get('view', 0), 'like_count': article.get('stats', {}).get('like', 0), 'coin_count': article.get('stats', {}).get('coin', 0), 'share_count': article.get('stats', {}).get('share', 0), 'reply_count': article.get('stats', {}).get('reply', 0), 'url': f"https://www.bilibili.com/read/{cvid}", 'summary': article.get('summary', ''), 'word_count': article.get('words', 0), } except Exception as e: logger.error(f"专栏提取错误: {e}") result = {'error': str(e)} return result async def _extract_up_async(self, mid: str) -> Dict[str, Any]: """ 异步提取UP主信息（支持订阅跟踪） Args: mid: UP主ID Returns: Dict containing UP主完整信息 """ result = { 'type': 'up', 'mid': mid, 'subscribed': False, 'basic_info': {}, 'subscription_info': {} } if not HTTPX_AVAILABLE: return {'error': 'httpx 未安装'} try: async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: # 获取UP主基本信息 response = await client.get('https://api.bilibili.com/x/space/acc/info', params={'mid': mid}) if response.status_code == 200: data = response.json() if data.get('code') == 0: up = data.get('data', {}) result['basic_info'] = { 'name': up.get('name', ''), 'sex': up.get('sex', ''), 'face': up.get('face', ''), 'sign': up.get('sign', ''), 'level': up.get('level', 0), 'vip_type': up.get('vip', {}).get('type', 0), 'vip_status': up.get('vip', {}).get('status', 0), 'following_count': up.get('following', 0), 'follower_count': up.get('follower', 0), 'url': f"https://space.bilibili.com/{mid}", } # 获取UP主视频统计 stats_response = await client.get('https://api.bilibili.com/x/space/navnum', params={'mid': mid}) if stats_response.status_code == 200: stats_data = stats_response.json() if stats_data.get('code') == 0: result['basic_info']['video_count'] = stats_data.get('data', {}).get('video', 0) # 获取UP主专栏数量 article_response = await client.get('https://api.bilibili.com/x/space/article', params={'mid': mid, 'pn': 1, 'ps': 1}) if article_response.status_code == 200: article_data = article_response.json() if article_data.get('code') == 0: result['basic_info']['article_count'] = article_data.get('data', {}).get('count', 0) # 获取UP主最近视频（用于跟踪更新） video_response = await client.get('https://api.bilibili.com/x/space/arc/search', params={'mid': mid, 'pn': 1, 'ps': 1, 'order': 'pubtime'}) if video_response.status_code == 200: video_data = video_response.json() if video_data.get('code') == 0: videos = video_data.get('data', {}).get('list', {}).get('vlist', []) if videos: latest_video = videos[0] result['subscription_info'] = { 'last_video_id': latest_video.get('bvid', ''), 'last_video_title': latest_video.get('title', ''), 'last_video_time': datetime.fromtimestamp(latest_video.get('created', 0)).strftime('%Y-%m-%d %H:%M:%S') if latest_video.get('created') else '', } # 获取UP主最近专栏（用于跟踪更新） article_response = await client.get('https://api.bilibili.com/x/space/article', params={'mid': mid, 'pn': 1, 'ps': 1, 'sort': 'publish_time'}) if article_response.status_code == 200: article_data = article_response.json() if article_data.get('code') == 0: articles = article_data.get('data', {}).get('articles', []) if articles: latest_article = articles[0] if 'last_article' not in result['subscription_info']: result['subscription_info']['last_article'] = {} result['subscription_info']['last_article'] = { 'id': latest_article.get('id', 0), 'title': latest_article.get('title', ''), 'time': datetime.fromtimestamp(latest_article.get('publish_time', 0)).strftime('%Y-%m-%d %H:%M:%S') if latest_article.get('publish_time') else '', } except Exception as e: logger.error(f"UP主提取错误: {e}") result = {'error': str(e)} return result def _format_duration(self, seconds: int) -> str: """格式化时长（秒 -> HH:MM:SS）""" if not seconds: return '' hours = seconds // 3600 minutes = (seconds % 3600) // 60 secs = seconds % 60 if hours > 0: return f"{hours:02d}:{minutes:02d}:{secs:02d}" else: return f"{minutes:02d}:{secs:02d}" async def _extract_series_async(self, mid: str, series_id: str) -> Dict[str, Any]: """ 异步提取系列合集信息（不包含列表内容） Args: mid: UP主ID series_id: 系列ID Returns: Dict containing series basic info """ result = { 'type': 'series', 'series_id': series_id, 'mid': mid, 'subscribed': False, 'basic_info': {} } if not HTTPX_AVAILABLE: return {'error': 'httpx 未安装'} try: async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: # 获取系列信息 response = await client.get('https://api.bilibili.com/x/space/series', params={'mid': mid, 'series_id': series_id, 'pn': 1, 'ps': 1}) if response.status_code == 200: data = response.json() if data.get('code') == 0: series_info = data.get('data', {}).get('list', {}) result['basic_info'] = { 'title': series_info.get('title', ''), 'description': series_info.get('desc', ''), 'cover': series_info.get('cover', ''), 'mid': mid, 'creator_name': '', 'total_count': series_info.get('cnt', 0), 'last_updated': '', 'update_count': 0 } # 获取创作者信息 up_info = await client.get('https://api.bilibili.com/x/space/acc/info', params={'mid': mid}) if up_info.status_code == 200: up_data = up_info.json() if up_data.get('code') == 0: result['basic_info']['creator_name'] = up_data.get('data', {}).get('name', '') except Exception as e: logger.error(f"系列合集提取错误: {e}") result = {'error': str(e)} return result async def _extract_readlist_async(self, rl_id: str) -> Dict[str, Any]: """ 异步提取专栏合集信息（不包含列表内容） Args: rl_id: 专栏合集ID (rl开头) Returns: Dict containing readlist basic info """ result = { 'type': 'readlist', 'readlist_id': rl_id, 'subscribed': False, 'basic_info': {} } if not HTTPX_AVAILABLE: return {'error': 'httpx 未安装'} try: async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: # 获取专栏合集信息 response = await client.get('https://api.bilibili.com/x/article/readlist', params={'id': rl_id.replace('rl', '')}) if response.status_code == 200: data = response.json() if data.get('code') == 0: list_info = data.get('data', {}) result['basic_info'] = { 'title': list_info.get('name', ''), 'description': list_info.get('description', ''), 'cover': list_info.get('cover', ''), 'mid': list_info.get('mid', ''), 'creator_name': '', 'total_count': list_info.get('articles_count', 0), 'last_updated': '', 'is_public': list_info.get('public', True) } # 获取创作者信息 mid = list_info.get('mid', '') if mid: up_info = await client.get('https://api.bilibili.com/x/space/acc/info', params={'mid': mid}) if up_info.status_code == 200: up_data = up_info.json() if up_data.get('code') == 0: result['basic_info']['creator_name'] = up_data.get('data', {}).get('name', '') except Exception as e: logger.error(f"专栏合集提取错误: {e}") result = {'error': str(e)} return result async def _extract_favlist_async(self, fid: str) -> Dict[str, Any]: """ 异步提取收藏夹信息（不包含列表内容） Args: fid: 收藏夹ID Returns: Dict containing favlist basic info """ result = { 'type': 'favlist', 'favlist_id': fid, 'subscribed': False, 'basic_info': {} } if not HTTPX_AVAILABLE: return {'error': 'httpx 未安装'} try: async with httpx.AsyncClient(timeout=30.0, headers=DEFAULT_HEADERS) as client: # 获取收藏夹信息 response = await client.get('https://api.bilibili.com/x/space/favlist', params={'fid': fid, 'pn': 1, 'ps': 1}) if response.status_code == 200: data = response.json() if data.get('code') == 0: fav_info = data.get('data', {}) result['basic_info'] = { 'title': fav_info.get('title', ''), 'description': fav_info.get('intro', ''), 'cover': fav_info.get('cover', ''), 'mid': fav_info.get('mid', ''), 'creator_name': '', 'total_count': fav_info.get('media_count', 0), 'is_public': fav_info.get('public', True), 'like_count': fav_info.get('like', 0) } # 获取创作者信息 mid = fav_info.get('mid', '') if mid: up_info = await client.get('https://api.bilibili.com/x/space/acc/info', params={'mid': mid}) if up_info.status_code == 200: up_data = up_info.json() if up_data.get('code') == 0: result['basic_info']['creator_name'] = up_data.get('data', {}).get('name', '') except Exception as e: logger.error(f"收藏夹提取错误: {e}") result = {'error': str(e)} return result # 测试代码 if __name__ == '__main__': extractor = BilibiliExtractor() print("=" * 60) print("B站元数据提取器测试") print("=" * 60) # 测试视频 print("\n测试视频提取: BV1MnFNz5EDL") result = extractor.extract_video_metadata('BV1MnFNz5EDL') print(json.dumps(result, ensure_ascii=False, indent=2))