starmen1265
e2edce2e29
- Rename kb-workflow to kb_workflow for Python package compatibility - Move bilibili_extractor.py and bilibili_metadata.py to kb_workflow/metadata/ - Create metadata/__init__.py with BilibiliExporter export Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
667 lines
26 KiB
Python
667 lines
26 KiB
Python
#!/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))
|