starmen1265
2b305b0eb9
- Remove unnecessary sys.path.insert for current directory - Use pure relative import for markdown_converter - Package imports work correctly without path manipulation Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
471 lines
16 KiB
Python
471 lines
16 KiB
Python
"""
|
||
知乎 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)
|