starmen1265
d2c262973e
- zhihu_extractor.py: - 添加 URL_PATTERNS 支持 answer 和 question - 添加 _extract_answer_content() 方法提取知乎回答 - 添加 _extract_question_content() 方法提取知乎问题 - zhihu_metadata.py: - 添加 extract_answer() 函数 - 添加 extract_question() 函数 - 更新命令行参数支持 answer/question 类型 - 更新帮助文档
451 lines
15 KiB
Python
451 lines
15 KiB
Python
"""
|
|
知乎 Playwright 内容提取器
|
|
使用 Playwright 浏览器自动化工具抓取知乎内容
|
|
"""
|
|
|
|
import re
|
|
import sys
|
|
import os
|
|
from typing import Dict, Optional, Any
|
|
from urllib.parse import urlparse
|
|
|
|
# 添加当前目录到 Python 路径以支持绝对导入
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
if current_dir not in sys.path:
|
|
sys.path.insert(0, current_dir)
|
|
|
|
try:
|
|
from .markdown_converter import MarkdownConverter
|
|
except ImportError:
|
|
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):
|
|
"""启动浏览器"""
|
|
from playwright.sync_api import sync_playwright
|
|
self.playwright = sync_playwright().start()
|
|
self.browser = self.playwright.chromium.launch(headless=self.headless)
|
|
self.context = self.browser.new_context(
|
|
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.page = self.context.new_page()
|
|
self.page.set_default_timeout(self.timeout)
|
|
|
|
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):
|
|
"""
|
|
初始化提取器
|
|
|
|
Args:
|
|
headless: 是否使用无头浏览器
|
|
timeout: 超时时间（秒）
|
|
"""
|
|
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"启动浏览器...")
|
|
|
|
# 启动浏览器
|
|
self.browser.start()
|
|
|
|
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)
|