""" 知乎 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)