自动同步: 2026-03-18 02:10
This commit is contained in:
@@ -0,0 +1,12 @@
|
||||
"""KB Workflow Metadata Module
|
||||
|
||||
提供 B站和知乎的元数据提取功能
|
||||
"""
|
||||
|
||||
from .bilibili_extractor import BilibiliExtractor
|
||||
from .zhihu_extractor import ZhihuPlaywrightExtractor
|
||||
|
||||
__all__ = [
|
||||
"BilibiliExtractor",
|
||||
"ZhihuPlaywrightExtractor",
|
||||
]
|
||||
@@ -0,0 +1,666 @@
|
||||
#!/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))
|
||||
@@ -0,0 +1,228 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
B站元数据提取 CLI 工具
|
||||
|
||||
支持命令行调用，输出 JSON 格式数据
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any
|
||||
|
||||
from bilibili_extractor import BilibiliExtractor
|
||||
|
||||
|
||||
def format_number(num: int) -> str:
|
||||
"""格式化数字"""
|
||||
if num >= 100000000:
|
||||
return f"{num / 100000000:.1f}亿"
|
||||
elif num >= 10000:
|
||||
return f"{num / 10000:.1f}万"
|
||||
else:
|
||||
return f"{num:,}"
|
||||
|
||||
|
||||
def extract_video(extractor: BilibiliExtractor, bvid: str, output_file: str = None) -> Dict[str, Any]:
|
||||
"""提取视频元数据"""
|
||||
result = extractor.extract_video_metadata(bvid)
|
||||
|
||||
# 如果需要输出到文件
|
||||
if output_file:
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def extract_article(extractor: BilibiliExtractor, cvid: str, output_file: str = None) -> Dict[str, Any]:
|
||||
"""提取专栏元数据"""
|
||||
result = extractor.extract_article_metadata(cvid)
|
||||
|
||||
if output_file:
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def extract_up(extractor: BilibiliExtractor, mid: str, output_file: str = None) -> Dict[str, Any]:
|
||||
"""提取UP主信息（包含订阅跟踪所需信息）"""
|
||||
result = asyncio.run(extractor._extract_up_async(mid))
|
||||
|
||||
if output_file:
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def extract_series(extractor: BilibiliExtractor, mid: str, series_id: str, output_file: str = None) -> Dict[str, Any]:
|
||||
"""提取系列合集元数据"""
|
||||
result = asyncio.run(extractor._extract_series_async(mid, series_id))
|
||||
|
||||
if output_file:
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def extract_readlist(extractor: BilibiliExtractor, rl_id: str, output_file: str = None) -> Dict[str, Any]:
|
||||
"""提取专栏合集元数据"""
|
||||
result = asyncio.run(extractor._extract_readlist_async(rl_id))
|
||||
|
||||
if output_file:
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def extract_favlist(extractor: BilibiliExtractor, fid: str, output_file: str = None) -> Dict[str, Any]:
|
||||
"""提取收藏夹元数据"""
|
||||
result = asyncio.run(extractor._extract_favlist_async(fid))
|
||||
|
||||
if output_file:
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
parser = argparse.ArgumentParser(
|
||||
description='B站元数据提取工具',
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
示例:
|
||||
# 提取视频元数据
|
||||
python bilibili_metadata.py --id "BV1xx411c7mD" --type video
|
||||
|
||||
# 提取专栏元数据
|
||||
python bilibili_metadata.py --id "cv43780186" --type article
|
||||
|
||||
# 提取UP主信息
|
||||
python bilibili_metadata.py --id "7770724" --type up
|
||||
|
||||
# 提取系列合集信息
|
||||
python bilibili_metadata.py --id "123456" --type series --mid "7770724"
|
||||
|
||||
# 提取专栏合集信息
|
||||
python bilibili_metadata.py --id "rl123456" --type readlist
|
||||
|
||||
# 提取收藏夹信息
|
||||
python bilibili_metadata.py --id "123456" --type favlist
|
||||
|
||||
# 输出到文件
|
||||
python bilibili_metadata.py --id "BV1xx411c7mD" --type video --output result.json
|
||||
"""
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--id',
|
||||
required=True,
|
||||
help='B站资源ID（BV号、CV号或UP主ID）'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--type',
|
||||
required=True,
|
||||
choices=['video', 'article', 'up', 'series', 'readlist', 'favlist'],
|
||||
help='资源类型'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--mid',
|
||||
help='UP主ID (用于series类型)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--output',
|
||||
'-o',
|
||||
help='输出文件路径（JSON格式）'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--headless',
|
||||
action='store_true',
|
||||
default=True,
|
||||
help='使用无头模式（浏览器，默认开启，使用API）'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--timeout',
|
||||
type=int,
|
||||
default=30,
|
||||
help='超时时间（秒，默认30）'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--pretty',
|
||||
action='store_true',
|
||||
default=True,
|
||||
help='格式化输出（默认开启）'
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# 创建提取器
|
||||
extractor = BilibiliExtractor(headless=args.headless, timeout=args.timeout)
|
||||
|
||||
try:
|
||||
# 根据类型提取数据
|
||||
if args.type == 'video':
|
||||
result = extract_video(extractor, args.id, args.output)
|
||||
elif args.type == 'article':
|
||||
result = extract_article(extractor, args.id, args.output)
|
||||
elif args.type == 'up':
|
||||
result = extract_up(extractor, args.id, args.output)
|
||||
elif args.type == 'series':
|
||||
if not args.mid:
|
||||
print(json.dumps({
|
||||
"success": False,
|
||||
"error": "series类型需要提供 --mid 参数"
|
||||
}, ensure_ascii=False, indent=2))
|
||||
sys.exit(1)
|
||||
result = extract_series(extractor, args.mid, args.id, args.output)
|
||||
elif args.type == 'readlist':
|
||||
result = extract_readlist(extractor, args.id, args.output)
|
||||
elif args.type == 'favlist':
|
||||
result = extract_favlist(extractor, args.id, args.output)
|
||||
else:
|
||||
print(json.dumps({
|
||||
"success": False,
|
||||
"error": f"不支持的类型 {args.type}"
|
||||
}, ensure_ascii=False, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
# 生成文件名
|
||||
platform = "bilibili"
|
||||
filename = f"{platform}_{args.id}.meta.json"
|
||||
filepath = args.output or filename
|
||||
|
||||
# 统一输出格式
|
||||
output = {
|
||||
"success": True,
|
||||
"filename": filename,
|
||||
"filepath": str(Path(filepath).resolve()) if filepath else None,
|
||||
"message": f"{args.type}元数据提取成功"
|
||||
}
|
||||
|
||||
print(json.dumps(output, ensure_ascii=False, indent=2))
|
||||
|
||||
# 如果指定了输出路径，则写入文件
|
||||
if args.output:
|
||||
with open(args.output, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
except Exception as e:
|
||||
print(json.dumps({
|
||||
"success": False,
|
||||
"error": str(e),
|
||||
"detail": type(e).__name__
|
||||
}, ensure_ascii=False, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,70 @@
|
||||
import re
|
||||
from bs4 import BeautifulSoup
|
||||
from markdownify import MarkdownConverter as _MarkdownConverter
|
||||
|
||||
class MarkdownConverter:
|
||||
"""HTML 转 Markdown 转换器，保留完整格式
|
||||
|
||||
处理知乎等平台的图片懒加载机制：
|
||||
- 知乎图片的 src 属性可能是 SVG 占位符 (data:image/svg+xml;...)
|
||||
- 真实图片 URL 存储在 data-actualsrc 或 data-original 属性中
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.converter = _MarkdownConverter(
|
||||
heading_style="ATX",
|
||||
bullets="*,-",
|
||||
code_blocks=True,
|
||||
tables=True,
|
||||
strip=["script", "style", "nav", "footer", "aside"]
|
||||
)
|
||||
|
||||
def convert(self, html: str) -> str:
|
||||
"""转换 HTML 为 Markdown
|
||||
|
||||
Args:
|
||||
html: 输入的 HTML 字符串
|
||||
|
||||
Returns:
|
||||
转换后的 Markdown 字符串
|
||||
"""
|
||||
# 预处理：处理懒加载图片
|
||||
html = self._fix_lazy_load_images(html)
|
||||
return self.converter.convert(html)
|
||||
|
||||
def _fix_lazy_load_images(self, html: str) -> str:
|
||||
"""修复懒加载图片的 src 属性
|
||||
|
||||
将 data-actualsrc 或 data-original 属性的值替换到 src 属性中，
|
||||
以便 markdownify 能够正确提取真实图片 URL。
|
||||
|
||||
同时移除 <noscript> 标签，因为知乎等平台在 noscript 中包含备用图片，
|
||||
会导致修复后图片重复出现。
|
||||
|
||||
Args:
|
||||
html: 输入的 HTML 字符串
|
||||
|
||||
Returns:
|
||||
处理后的 HTML 字符串
|
||||
"""
|
||||
soup = BeautifulSoup(html, 'html.parser')
|
||||
|
||||
# 先移除 noscript 标签，避免图片重复
|
||||
for noscript in soup.find_all('noscript'):
|
||||
noscript.decompose()
|
||||
|
||||
for img in soup.find_all('img'):
|
||||
src = img.get('src', '')
|
||||
|
||||
# 如果 src 不是占位符，保持原样
|
||||
if not src or not src.startswith('data:'):
|
||||
continue
|
||||
|
||||
# 尝试获取真实图片 URL
|
||||
# 优先 data-actualsrc，其次 data-original
|
||||
real_src = img.get('data-actualsrc') or img.get('data-original')
|
||||
|
||||
if real_src:
|
||||
img['src'] = real_src
|
||||
|
||||
return str(soup)
|
||||
@@ -0,0 +1,236 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
知乎文章爬取脚本 - 基于 MediaCrawler
|
||||
用于爬取知乎专栏文章并导出到知识库格式
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import subprocess
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
# 配置
|
||||
MEDIACRAWLER_DIR = Path.home() / "work" / "MediaCrawler"
|
||||
DEFAULT_OUTPUT_DIR = Path.home() / "work" / "knowledge-base" / "data" / "zhihu"
|
||||
|
||||
|
||||
def load_env():
|
||||
"""加载环境变量"""
|
||||
env_file = Path.home() / ".kb_env"
|
||||
if env_file.exists():
|
||||
with open(env_file) as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, val = line.split("=", 1)
|
||||
os.environ[key.strip()] = val.strip()
|
||||
|
||||
|
||||
def get_zhihu_cookie():
|
||||
"""获取知乎 Cookie"""
|
||||
return os.environ.get("ZHIHU_COOKIE", "")
|
||||
|
||||
|
||||
def crawl_zhihu_article(article_url: str, output_dir: Path) -> dict:
|
||||
"""
|
||||
使用 MediaCrawler 爬取知乎文章
|
||||
|
||||
Args:
|
||||
article_url: 知乎文章链接
|
||||
output_dir: 输出目录
|
||||
|
||||
Returns:
|
||||
dict: 爬取结果
|
||||
"""
|
||||
# 清理旧数据
|
||||
data_dir = MEDIACRAWLER_DIR / "data" / "zhihu" / "json"
|
||||
data_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 备份现有数据
|
||||
backup_dir = data_dir / "backup"
|
||||
backup_dir.mkdir(exist_ok=True)
|
||||
for f in data_dir.glob("detail_contents_*.json"):
|
||||
shutil.move(str(f), str(backup_dir / f.name))
|
||||
|
||||
# 配置 MediaCrawler
|
||||
config_file = MEDIACRAWLER_DIR / "config" / "zhihu_config.py"
|
||||
|
||||
# 读取配置文件并修改
|
||||
with open(config_file) as f:
|
||||
config_content = f.read()
|
||||
|
||||
# 强制更新配置中的 ID 列表（每次都更新，确保爬取正确的文章）
|
||||
article_id = article_url.split("/")[-1]
|
||||
|
||||
# 使用正则表达式替换整个 ZHIHU_SPECIFIED_ID_LIST 部分
|
||||
pattern = r'(ZHIHU_SPECIFIED_ID_LIST\s*=\s*\[)([\s\S]*?)(\])'
|
||||
replacement = f'\\1\n "{article_url}", # article\n\\3'
|
||||
|
||||
new_config = re.sub(pattern, replacement, config_content)
|
||||
|
||||
# 如果正则替换失败（没有匹配到），使用简单替换
|
||||
if new_config == config_content:
|
||||
# 直接替换整个配置项
|
||||
if 'ZHIHU_SPECIFIED_ID_LIST' in config_content:
|
||||
# 找到配置项的范围并替换
|
||||
lines = config_content.split('\n')
|
||||
new_lines = []
|
||||
in_list = False
|
||||
for line in lines:
|
||||
if 'ZHIHU_SPECIFIED_ID_LIST' in line and '=' in line:
|
||||
new_lines.append(f'ZHIHU_SPECIFIED_ID_LIST = ["{article_url}", # article')
|
||||
in_list = True
|
||||
elif in_list and line.strip().startswith(']'):
|
||||
new_lines.append(line)
|
||||
in_list = False
|
||||
elif not in_list:
|
||||
new_lines.append(line)
|
||||
new_config = '\n'.join(new_lines)
|
||||
|
||||
with open(config_file, "w") as f:
|
||||
f.write(new_config)
|
||||
|
||||
# 运行 MediaCrawler
|
||||
env = os.environ.copy()
|
||||
cookie = get_zhihu_cookie()
|
||||
if cookie:
|
||||
env["ZHIHU_COOKIE"] = cookie
|
||||
|
||||
cmd = [
|
||||
"python3", "main.py",
|
||||
"--platform", "zhihu",
|
||||
"--lt", "cookie",
|
||||
"--type", "detail"
|
||||
]
|
||||
|
||||
print(f"运行命令: {' '.join(cmd)}")
|
||||
print(f"工作目录: {MEDIACRAWLER_DIR}")
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
cwd=MEDIACRAWLER_DIR,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120
|
||||
)
|
||||
|
||||
print(f"返回码: {result.returncode}")
|
||||
if result.returncode != 0:
|
||||
print(f"stderr: {result.stderr[:500]}")
|
||||
|
||||
# 查找最新的输出文件
|
||||
output_files = sorted(data_dir.glob("detail_contents_*.json"), reverse=True)
|
||||
|
||||
if output_files:
|
||||
latest_file = output_files[0]
|
||||
with open(latest_file) as f:
|
||||
data = json.load(f)
|
||||
|
||||
if data:
|
||||
item = data[0]
|
||||
return {
|
||||
"success": True,
|
||||
"title": item.get("title", ""),
|
||||
"content": item.get("content_text", ""),
|
||||
"url": item.get("content_url", ""),
|
||||
"author": item.get("user_nickname", ""),
|
||||
"created_time": item.get("created_time", 0),
|
||||
"voteup": item.get("voteup_count", 0),
|
||||
"comments": item.get("comment_count", 0),
|
||||
"content_type": item.get("content_type", "article"),
|
||||
}
|
||||
|
||||
return {"success": False, "error": "未获取到数据"}
|
||||
|
||||
|
||||
def save_to_markdown(data: dict, output_path: Path) -> str:
|
||||
"""保存为 Markdown 格式"""
|
||||
created = datetime.fromtimestamp(data["created_time"]).strftime("%Y-%m-%d %H:%M:%S") if data.get("created_time") else ""
|
||||
|
||||
content = f"""---
|
||||
title: "{data['title']}"
|
||||
author: "{data['author']}"
|
||||
url: "{data['url']}"
|
||||
created: "{created}"
|
||||
voteup: {data.get('voteup', 0)}
|
||||
comments: {data.get('comments', 0)}
|
||||
platform: zhihu
|
||||
type: article
|
||||
---
|
||||
|
||||
# {data['title']}
|
||||
|
||||
**作者**: {data['author']}
|
||||
**来源**: [知乎专栏]({data['url']})
|
||||
**发布时间**: {created}
|
||||
**点赞**: {data.get('voteup', 0)}
|
||||
**评论**: {data.get('comments', 0)}
|
||||
|
||||
---
|
||||
|
||||
{data['content']}
|
||||
"""
|
||||
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
|
||||
return str(output_path)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="知乎文章爬取工具 (基于 MediaCrawler)")
|
||||
parser.add_argument("--url", "-u", required=True, help="知乎文章链接")
|
||||
parser.add_argument("--output", "-o", default=str(DEFAULT_OUTPUT_DIR), help="输出目录")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# 加载环境变量
|
||||
load_env()
|
||||
|
||||
# 确保输出目录存在
|
||||
output_dir = Path(args.output)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
print(f"目标文章: {args.url}")
|
||||
print(f"输出目录: {output_dir}")
|
||||
|
||||
# 爬取文章
|
||||
print("\n开始爬取...")
|
||||
result = crawl_zhihu_article(args.url, output_dir)
|
||||
|
||||
if not result.get("success"):
|
||||
print(f"爬取失败: {result.get('error', '未知错误')}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"\n爬取成功!")
|
||||
print(f"标题: {result['title']}")
|
||||
print(f"内容长度: {len(result['content'])} 字符")
|
||||
|
||||
# 生成文件名
|
||||
safe_title = "".join(c for c in result['title'] if c.isalnum() or c in " -_")[:50]
|
||||
filename = f"zhihu_article_{safe_title}.md"
|
||||
output_path = output_dir / filename
|
||||
|
||||
# 保存为 Markdown
|
||||
save_to_markdown(result, output_path)
|
||||
print(f"\n已保存到: {output_path}")
|
||||
|
||||
# 输出 JSON 格式
|
||||
print("\n--- JSON 输出 ---")
|
||||
print(json.dumps({
|
||||
"success": True,
|
||||
"type": "article",
|
||||
"title": result["title"],
|
||||
"filepath": str(output_path),
|
||||
"content_length": len(result["content"]),
|
||||
"message": f"爬取成功: {result['title']}"
|
||||
}, ensure_ascii=False, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,470 @@
|
||||
"""
|
||||
知乎 Playwright 内容提取器
|
||||
使用 Playwright 浏览器自动化工具抓取知乎内容
|
||||
"""
|
||||
|
||||
import re
|
||||
import os
|
||||
from typing import Dict, Optional, Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from .markdown_converter import MarkdownConverter
|
||||
|
||||
|
||||
class PlaywrightBrowser:
|
||||
"""Playwright 浏览器管理器"""
|
||||
|
||||
def __init__(self, headless: bool = True, timeout: int = 30):
|
||||
self.headless = headless
|
||||
self.timeout = timeout * 1000
|
||||
self.browser = None
|
||||
self.context = None
|
||||
self.page = None
|
||||
|
||||
def start(self, cookie: str = None):
|
||||
"""启动浏览器"""
|
||||
from playwright.sync_api import sync_playwright
|
||||
self.playwright = sync_playwright().start()
|
||||
self.browser = self.playwright.chromium.launch(headless=self.headless)
|
||||
|
||||
# 创建浏览器上下文
|
||||
context_options = {
|
||||
'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'
|
||||
}
|
||||
|
||||
self.context = self.browser.new_context(**context_options)
|
||||
|
||||
# 如果提供了 Cookie，设置 Cookie
|
||||
if cookie:
|
||||
self._add_cookies(cookie)
|
||||
|
||||
self.page = self.context.new_page()
|
||||
self.page.set_default_timeout(self.timeout)
|
||||
|
||||
def _add_cookies(self, cookie_str: str):
|
||||
"""添加 Cookie 到浏览器上下文"""
|
||||
cookies = []
|
||||
for item in cookie_str.split(';'):
|
||||
item = item.strip()
|
||||
if '=' in item:
|
||||
name, value = item.split('=', 1)
|
||||
cookies.append({
|
||||
'name': name.strip(),
|
||||
'value': value.strip(),
|
||||
'domain': '.zhihu.com',
|
||||
'path': '/',
|
||||
'secure': True,
|
||||
'httpOnly': False,
|
||||
'sameSite': 'Lax'
|
||||
})
|
||||
self.context.add_cookies(cookies)
|
||||
|
||||
def close(self):
|
||||
"""关闭浏览器"""
|
||||
if self.page:
|
||||
self.page.close()
|
||||
if self.context:
|
||||
self.context.close()
|
||||
if self.browser:
|
||||
self.browser.close()
|
||||
if hasattr(self, 'playwright'):
|
||||
self.playwright.stop()
|
||||
|
||||
def goto(self, url: str, wait_until: str = "networkidle"):
|
||||
"""导航到指定URL"""
|
||||
return self.page.goto(url, wait_until=wait_until)
|
||||
|
||||
def wait_for_selector(self, selector: str, timeout: int = None):
|
||||
"""等待选择器出现"""
|
||||
timeout_ms = timeout * 1000 if timeout else self.timeout
|
||||
return self.page.wait_for_selector(selector, timeout=timeout_ms)
|
||||
|
||||
def query_selector(self, selector: str):
|
||||
"""查询选择器"""
|
||||
return self.page.query_selector(selector)
|
||||
|
||||
def query_selector_all(self, selector: str):
|
||||
"""查询所有匹配的选择器"""
|
||||
return self.page.query_selector_all(selector)
|
||||
|
||||
def evaluate(self, script: str):
|
||||
"""执行 JavaScript 代码"""
|
||||
return self.page.evaluate(script)
|
||||
|
||||
|
||||
class ZhihuPlaywrightExtractor:
|
||||
"""知乎内容提取器 - 使用 Playwright"""
|
||||
|
||||
URL_PATTERNS = {
|
||||
'zvideo': r'zhihu\.com/zvideo/(\d+)',
|
||||
'zhuanlan': r'zhuanlan\.zhihu\.com/p/(\d+)',
|
||||
'answer': r'zhihu\.com/question/\d+/answer/(\d+)',
|
||||
'question': r'zhihu\.com/question/(\d+)',
|
||||
}
|
||||
|
||||
def __init__(self, headless: bool = True, timeout: int = 30, cookie: str = None):
|
||||
"""
|
||||
初始化提取器
|
||||
|
||||
Args:
|
||||
headless: 是否使用无头浏览器
|
||||
timeout: 超时时间（秒）
|
||||
cookie: 知乎 Cookie 字符串（可选）
|
||||
"""
|
||||
self.cookie = cookie
|
||||
self.browser = PlaywrightBrowser(headless=headless, timeout=timeout)
|
||||
self.markdown_converter = MarkdownConverter()
|
||||
|
||||
def identify_url_type(self, url: str) -> Optional[str]:
|
||||
"""
|
||||
识别 URL 类型
|
||||
|
||||
Returns:
|
||||
'answer', 'article', 'question' 或 None
|
||||
"""
|
||||
for url_type, pattern in self.URL_PATTERNS.items():
|
||||
if re.search(pattern, url):
|
||||
return url_type
|
||||
return None
|
||||
|
||||
def extract_id(self, url: str) -> Optional[str]:
|
||||
"""
|
||||
从 URL 中提取内容 ID
|
||||
|
||||
Returns:
|
||||
内容 ID 或 None
|
||||
"""
|
||||
url_type = self.identify_url_type(url)
|
||||
if url_type:
|
||||
pattern = self.URL_PATTERNS[url_type]
|
||||
match = re.search(pattern, url)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return None
|
||||
|
||||
def _extract_text_content(self, element) -> str:
|
||||
"""提取元素的文本内容"""
|
||||
if element is None:
|
||||
return ''
|
||||
return element.inner_text() if hasattr(element, 'inner_text') else ''
|
||||
|
||||
def _extract_html_content(self, element) -> str:
|
||||
"""提取元素的 HTML 内容"""
|
||||
if element is None:
|
||||
return ''
|
||||
return element.inner_html() if hasattr(element, 'inner_html') else ''
|
||||
|
||||
def _extract_zvideo_content(self) -> Dict[str, Any]:
|
||||
"""提取知乎视频内容"""
|
||||
result = {
|
||||
'type': 'zvideo',
|
||||
'title': '',
|
||||
'author': '',
|
||||
'content': '',
|
||||
'content_markdown': '',
|
||||
'publish_time': '',
|
||||
'vote_up': 0,
|
||||
}
|
||||
|
||||
# 打印页面标题用于调试
|
||||
page_title = self.browser.page.title()
|
||||
print(f"页面标题: {page_title}")
|
||||
|
||||
# 提取视频标题
|
||||
title_elem = self.browser.query_selector('.VideoPlaylistInfo-title')
|
||||
if title_elem:
|
||||
result['title'] = self._extract_text_content(title_elem)
|
||||
|
||||
# 备用：尝试其他选择器
|
||||
if not result['title']:
|
||||
title_elem = self.browser.query_selector('h1')
|
||||
if title_elem:
|
||||
result['title'] = self._extract_text_content(title_elem)
|
||||
|
||||
# 提取作者信息
|
||||
author_elem = self.browser.query_selector('.AuthorInfo-name')
|
||||
if author_elem:
|
||||
result['author'] = self._extract_text_content(author_elem)
|
||||
|
||||
# 提取视频描述
|
||||
desc_elem = self.browser.query_selector('.Video-desc')
|
||||
if desc_elem:
|
||||
html_content = self._extract_html_content(desc_elem)
|
||||
result['content'] = html_content
|
||||
result['content_markdown'] = self.markdown_converter.convert(html_content)
|
||||
|
||||
# 提取发布时间
|
||||
time_elem = self.browser.query_selector('.Video-meta time')
|
||||
if time_elem:
|
||||
result['publish_time'] = time_elem.get_attribute('datetime') or self._extract_text_content(time_elem)
|
||||
|
||||
# 提取点赞数
|
||||
vote_elem = self.browser.query_selector('.VideoLikeButton-count')
|
||||
if vote_elem:
|
||||
vote_text = self._extract_text_content(vote_elem)
|
||||
try:
|
||||
result['vote_up'] = int(vote_text.replace(',', ''))
|
||||
except (ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
return result
|
||||
|
||||
def _extract_zhuanlan_content(self) -> Dict[str, Any]:
|
||||
"""提取知乎专栏文章内容"""
|
||||
result = {
|
||||
'type': 'zhuanlan',
|
||||
'title': '',
|
||||
'author': '',
|
||||
'content': '',
|
||||
'content_markdown': '',
|
||||
'publish_time': '',
|
||||
'vote_up': 0,
|
||||
}
|
||||
|
||||
# 打印页面标题用于调试
|
||||
page_title = self.browser.page.title()
|
||||
print(f"页面标题: {page_title}")
|
||||
|
||||
# 提取文章标题
|
||||
title_elem = self.browser.query_selector('.Post-Title')
|
||||
if title_elem:
|
||||
result['title'] = self._extract_text_content(title_elem)
|
||||
|
||||
# 备用
|
||||
if not result['title']:
|
||||
title_elem = self.browser.query_selector('h1')
|
||||
if title_elem:
|
||||
result['title'] = self._extract_text_content(title_elem)
|
||||
|
||||
# 提取作者信息
|
||||
author_elem = self.browser.query_selector('.AuthorInfo-name')
|
||||
if author_elem:
|
||||
result['author'] = self._extract_text_content(author_elem)
|
||||
|
||||
# 提取文章内容
|
||||
content_elem = self.browser.query_selector('.Post-RichText')
|
||||
if content_elem:
|
||||
html_content = self._extract_html_content(content_elem)
|
||||
result['content'] = html_content
|
||||
result['content_markdown'] = self.markdown_converter.convert(html_content)
|
||||
|
||||
# 提取发布时间
|
||||
time_elem = self.browser.query_selector('.ContentItem-time')
|
||||
if time_elem:
|
||||
result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
|
||||
|
||||
# 提取点赞数
|
||||
vote_elem = self.browser.query_selector('.VoteButton--up .VoteButton-label')
|
||||
if vote_elem:
|
||||
vote_text = self._extract_text_content(vote_elem)
|
||||
try:
|
||||
result['vote_up'] = int(vote_text.replace(',', ''))
|
||||
except (ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
return result
|
||||
|
||||
def _extract_answer_content(self) -> Dict[str, Any]:
|
||||
"""提取知乎回答内容"""
|
||||
result = {
|
||||
'type': 'answer',
|
||||
'title': '', # 问题标题
|
||||
'question_id': '',
|
||||
'answer_id': '',
|
||||
'author': '',
|
||||
'content': '',
|
||||
'content_markdown': '',
|
||||
'publish_time': '',
|
||||
'vote_up': 0,
|
||||
}
|
||||
|
||||
# 打印页面标题用于调试
|
||||
page_title = self.browser.page.title()
|
||||
print(f"页面标题: {page_title}")
|
||||
|
||||
# 提取问题标题
|
||||
question_title_elem = self.browser.query_selector('.QuestionHeader-title')
|
||||
if not question_title_elem:
|
||||
question_title_elem = self.browser.query_selector('.QuestionHeader-content h1')
|
||||
if question_title_elem:
|
||||
result['title'] = self._extract_text_content(question_title_elem)
|
||||
|
||||
# 提取回答内容
|
||||
content_elem = self.browser.query_selector('.RichText')
|
||||
if not content_elem:
|
||||
content_elem = self.browser.query_selector('.AnswerItem-richText')
|
||||
if not content_elem:
|
||||
content_elem = self.browser.query_selector('[class*="RichText"]')
|
||||
if content_elem:
|
||||
html_content = self._extract_html_content(content_elem)
|
||||
result['content'] = html_content
|
||||
result['content_markdown'] = self.markdown_converter.convert(html_content)
|
||||
|
||||
# 提取作者信息
|
||||
author_elem = self.browser.query_selector('.AuthorInfo-name')
|
||||
if author_elem:
|
||||
result['author'] = self._extract_text_content(author_elem)
|
||||
|
||||
# 提取发布时间
|
||||
time_elem = self.browser.query_selector('.AnswerItem-time')
|
||||
if not time_elem:
|
||||
time_elem = self.browser.query_selector('.ContentItem-time')
|
||||
if time_elem:
|
||||
result['publish_time'] = time_elem.get_attribute('data-tooltip') or self._extract_text_content(time_elem)
|
||||
|
||||
# 提取点赞数
|
||||
vote_elem = self.browser.query_selector('.VoteButton--up .VoteButton-label')
|
||||
if not vote_elem:
|
||||
vote_elem = self.browser.query_selector('.VoteButton .Count')
|
||||
if vote_elem:
|
||||
vote_text = self._extract_text_content(vote_elem)
|
||||
try:
|
||||
result['vote_up'] = int(vote_text.replace(',', ''))
|
||||
except (ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
return result
|
||||
|
||||
def _extract_question_content(self) -> Dict[str, Any]:
|
||||
"""提取知乎问题内容（包含所有回答）"""
|
||||
result = {
|
||||
'type': 'question',
|
||||
'title': '',
|
||||
'question_id': '',
|
||||
'author': '',
|
||||
'content': '',
|
||||
'content_markdown': '',
|
||||
'publish_time': '',
|
||||
'answer_count': 0,
|
||||
}
|
||||
|
||||
# 打印页面标题用于调试
|
||||
page_title = self.browser.page.title()
|
||||
print(f"页面标题: {page_title}")
|
||||
|
||||
# 提取问题标题
|
||||
title_elem = self.browser.query_selector('.QuestionHeader-title')
|
||||
if not title_elem:
|
||||
title_elem = self.browser.query_selector('.QuestionHeader-content h1')
|
||||
if title_elem:
|
||||
result['title'] = self._extract_text_content(title_elem)
|
||||
|
||||
# 提取问题描述
|
||||
desc_elem = self.browser.query_selector('.QuestionHeader-detail')
|
||||
if not desc_elem:
|
||||
desc_elem = self.browser.query_selector('.QuestionRichText')
|
||||
if desc_elem:
|
||||
html_content = self._extract_html_content(desc_elem)
|
||||
result['content'] = html_content
|
||||
result['content_markdown'] = self.markdown_converter.convert(html_content)
|
||||
|
||||
# 提取回答数量
|
||||
answer_count_elem = self.browser.query_selector('.List-headerText span')
|
||||
if answer_count_elem:
|
||||
count_text = self._extract_text_content(answer_count_elem)
|
||||
try:
|
||||
# 提取数字
|
||||
import re
|
||||
match = re.search(r'(\d+)', count_text)
|
||||
if match:
|
||||
result['answer_count'] = int(match.group(1))
|
||||
except (ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
return result
|
||||
|
||||
def extract(self, url: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
从 URL 提取知乎内容
|
||||
|
||||
Args:
|
||||
url: 知乎 URL
|
||||
|
||||
Returns:
|
||||
包含提取内容的字典，结构如下:
|
||||
{
|
||||
'type': 'zvideo' | 'zhuanlan',
|
||||
'title': str,
|
||||
'author': str,
|
||||
'content': str (HTML),
|
||||
'content_markdown': str (Markdown),
|
||||
'publish_time': str,
|
||||
'vote_up': int,
|
||||
}
|
||||
如果提取失败返回 None
|
||||
"""
|
||||
try:
|
||||
# 识别 URL 类型
|
||||
url_type = self.identify_url_type(url)
|
||||
if not url_type:
|
||||
print(f"无法识别 URL 类型: {url}")
|
||||
return None
|
||||
|
||||
print(f"URL 类型: {url_type}")
|
||||
print(f"启动浏览器...")
|
||||
|
||||
# 启动浏览器（传入 Cookie）
|
||||
self.browser.start(cookie=self.cookie)
|
||||
|
||||
print(f"访问页面: {url}")
|
||||
# 访问页面
|
||||
self.browser.goto(url, wait_until="domcontentloaded")
|
||||
|
||||
print("等待页面加载...")
|
||||
import time
|
||||
time.sleep(3) # 等待 JavaScript 执行
|
||||
|
||||
# 根据类型提取内容
|
||||
if url_type == 'zvideo':
|
||||
result = self._extract_zvideo_content()
|
||||
elif url_type == 'zhuanlan':
|
||||
result = self._extract_zhuanlan_content()
|
||||
elif url_type == 'answer':
|
||||
result = self._extract_answer_content()
|
||||
elif url_type == 'question':
|
||||
result = self._extract_question_content()
|
||||
else:
|
||||
print(f"不支持的类型: {url_type}")
|
||||
return None
|
||||
|
||||
# 添加额外信息
|
||||
result['url'] = url
|
||||
result['content_id'] = self.extract_id(url)
|
||||
|
||||
print(f"提取成功! 标题: {result.get('title', '')[:50]}...")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
print(f"提取知乎内容时出错: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
finally:
|
||||
# 关闭浏览器
|
||||
self.browser.close()
|
||||
|
||||
def __enter__(self):
|
||||
"""上下文管理器入口"""
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""上下文管理器出口"""
|
||||
self.browser.close()
|
||||
return False
|
||||
|
||||
|
||||
# 便捷函数
|
||||
def extract_zhihu_content(url: str, headless: bool = True, timeout: int = 30) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
便捷函数：提取知乎内容
|
||||
|
||||
Args:
|
||||
url: 知乎 URL
|
||||
headless: 是否使用无头浏览器
|
||||
timeout: 超时时间（秒）
|
||||
|
||||
Returns:
|
||||
提取的内容字典
|
||||
"""
|
||||
with ZhihuPlaywrightExtractor(headless=headless, timeout=timeout) as extractor:
|
||||
return extractor.extract(url)
|
||||
@@ -0,0 +1,201 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
知乎元数据获取工具
|
||||
|
||||
支持以下内容类型：
|
||||
- zvideo: 知乎视频 (https://www.zhihu.com/zvideo/xxx)
|
||||
- zhuanlan: 知乎专栏文章 (https://zhuanlan.zhihu.com/p/xxx)
|
||||
- answer: 知乎回答 (https://www.zhihu.com/question/xxx/answer/yyy)
|
||||
- question: 知乎问题 (https://www.zhihu.com/question/xxx)
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
from typing import Dict, Optional, Any
|
||||
|
||||
from .zhihu_extractor import ZhihuPlaywrightExtractor
|
||||
|
||||
|
||||
def extract_zvideo(video_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
|
||||
"""
|
||||
获取知乎视频元数据
|
||||
|
||||
Args:
|
||||
video_id: 视频 ID (数字字符串)
|
||||
headless: 是否使用无头浏览器
|
||||
cookie: 知乎 Cookie（可选）
|
||||
|
||||
Returns:
|
||||
视频元数据字典
|
||||
"""
|
||||
url = f"https://www.zhihu.com/zvideo/{video_id}"
|
||||
return _extract_metadata(url, "zvideo", headless, cookie)
|
||||
|
||||
|
||||
def extract_zhuanlan(article_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
|
||||
"""
|
||||
获取知乎专栏文章元数据
|
||||
|
||||
Args:
|
||||
article_id: 文章 ID (数字字符串)
|
||||
headless: 是否使用无头浏览器
|
||||
cookie: 知乎 Cookie（可选）
|
||||
|
||||
Returns:
|
||||
文章元数据字典
|
||||
"""
|
||||
url = f"https://zhuanlan.zhihu.com/p/{article_id}"
|
||||
return _extract_metadata(url, "zhuanlan", headless, cookie)
|
||||
|
||||
|
||||
def extract_answer(answer_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
|
||||
"""
|
||||
获取知乎回答元数据
|
||||
|
||||
Args:
|
||||
answer_id: 回答 ID (数字字符串)
|
||||
headless: 是否使用无头浏览器
|
||||
cookie: 知乎 Cookie（可选）
|
||||
|
||||
Returns:
|
||||
回答元数据字典
|
||||
"""
|
||||
# 回答需要构建包含问题ID的URL，但answer_id本身足以识别
|
||||
# 使用一个占位符问题ID，因为提取器会从URL中提取answer_id
|
||||
url = f"https://www.zhihu.com/question/0/answer/{answer_id}"
|
||||
return _extract_metadata(url, "answer", headless, cookie)
|
||||
|
||||
|
||||
def extract_question(question_id: str, headless: bool = True, cookie: str = None) -> Dict[str, Any]:
|
||||
"""
|
||||
获取知乎问题元数据
|
||||
|
||||
Args:
|
||||
question_id: 问题 ID (数字字符串)
|
||||
headless: 是否使用无头浏览器
|
||||
cookie: 知乎 Cookie（可选）
|
||||
|
||||
Returns:
|
||||
问题元数据字典
|
||||
"""
|
||||
url = f"https://www.zhihu.com/question/{question_id}"
|
||||
return _extract_metadata(url, "question", headless, cookie)
|
||||
|
||||
|
||||
def _extract_metadata(url: str, content_type: str, headless: bool, cookie: str = None) -> Dict[str, Any]:
|
||||
"""
|
||||
内部函数：提取元数据
|
||||
|
||||
Args:
|
||||
url: 知乎 URL
|
||||
content_type: 内容类型 (zvideo/zhuanlan)
|
||||
headless: 是否使用无头浏览器
|
||||
cookie: 知乎 Cookie（可选）
|
||||
|
||||
Returns:
|
||||
元数据字典
|
||||
"""
|
||||
extractor = ZhihuPlaywrightExtractor(headless=headless, cookie=cookie)
|
||||
result = extractor.extract(url)
|
||||
|
||||
if result:
|
||||
# 标准化输出格式
|
||||
metadata = {
|
||||
"type": content_type,
|
||||
"id": result.get("content_id", ""),
|
||||
"url": result.get("url", url),
|
||||
"title": result.get("title", ""),
|
||||
"author": result.get("author", ""),
|
||||
"publish_time": result.get("publish_time", ""),
|
||||
"vote_up": result.get("vote_up", 0),
|
||||
"content_markdown": result.get("content_markdown", ""),
|
||||
"raw_html": result.get("content", ""),
|
||||
}
|
||||
return metadata
|
||||
else:
|
||||
return {
|
||||
"type": content_type,
|
||||
"error": "Failed to extract content",
|
||||
"url": url
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
"""命令行入口"""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="知乎元数据获取工具",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
支持的类型:
|
||||
zvideo - 知乎视频 (https://www.zhihu.com/zvideo/123456)
|
||||
zhuanlan - 知乎专栏文章 (https://zhuanlan.zhihu.com/p/123456)
|
||||
answer - 知乎回答 (https://www.zhihu.com/question/xxx/answer/yyy)
|
||||
question - 知乎问题 (https://www.zhihu.com/question/xxx)
|
||||
|
||||
示例:
|
||||
python zhihu_metadata.py --id "123456" --type zvideo
|
||||
python zhihu_metadata.py --id "123456" --type zhuanlan
|
||||
python zhihu_metadata.py --id "123456" --type answer
|
||||
python zhihu_metadata.py --id "123456" --type question
|
||||
"""
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--id", "-i",
|
||||
required=True,
|
||||
help="知乎内容 ID (视频ID/文章ID/回答ID/问题ID)"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--type", "-t",
|
||||
required=True,
|
||||
choices=["zvideo", "zhuanlan", "answer", "question"],
|
||||
help="内容类型: zvideo=视频, zhuanlan=专栏文章, answer=回答, question=问题"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--output", "-o",
|
||||
default=None,
|
||||
help="输出文件路径 (默认: stdout)"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--headless",
|
||||
action="store_true",
|
||||
default=True,
|
||||
help="使用无头浏览器模式 (默认: True)"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# 获取知乎 Cookie（优先使用环境变量，其次使用参数）
|
||||
cookie = os.environ.get('ZHIHU_COOKIE')
|
||||
|
||||
# 提取元数据（传入 Cookie）
|
||||
if args.type == "zvideo":
|
||||
result = extract_zvideo(args.id, headless=args.headless, cookie=cookie)
|
||||
elif args.type == "zhuanlan":
|
||||
result = extract_zhuanlan(args.id, headless=args.headless, cookie=cookie)
|
||||
elif args.type == "answer":
|
||||
result = extract_answer(args.id, headless=args.headless, cookie=cookie)
|
||||
elif args.type == "question":
|
||||
result = extract_question(args.id, headless=args.headless, cookie=cookie)
|
||||
else:
|
||||
print(f"不支持的类型: {args.type}")
|
||||
sys.exit(1)
|
||||
|
||||
# 输出结果
|
||||
json_output = json.dumps(result, ensure_ascii=False, indent=2)
|
||||
|
||||
if args.output:
|
||||
with open(args.output, 'w', encoding='utf-8') as f:
|
||||
f.write(json_output)
|
||||
print(f"结果已保存到: {args.output}")
|
||||
else:
|
||||
print(json_output)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user