diff --git a/kb_workflow/crawler/__init__.py b/kb_workflow/crawler/__init__.py index 4b5a571..cc4b69a 100644 --- a/kb_workflow/crawler/__init__.py +++ b/kb_workflow/crawler/__init__.py @@ -10,6 +10,7 @@ from .models import ( ArticleData, CrawlResult, ) +from .zhihu_crawler import ZhihuCrawler, ZhihuApiInterceptor __all__ = [ "AuthorInfo", @@ -17,4 +18,6 @@ __all__ = [ "AnswerData", "ArticleData", "CrawlResult", + "ZhihuCrawler", + "ZhihuApiInterceptor", ] diff --git a/kb_workflow/crawler/zhihu_crawler.py b/kb_workflow/crawler/zhihu_crawler.py new file mode 100644 index 0000000..f7c9643 --- /dev/null +++ b/kb_workflow/crawler/zhihu_crawler.py @@ -0,0 +1,322 @@ +"""Zhihu Content Crawler + +Implements content crawling using Playwright API interception. +Captures structured JSON data from Zhihu API responses. +""" + +import re +import time +from typing import Optional, Dict, Any + +from playwright.sync_api import Route, Response + +from kb_workflow.metadata.zhihu_extractor import PlaywrightBrowser +from kb_workflow.pipeline.config import get_settings +from .models import ( + AnswerData, + ArticleData, + ContentMetadata, + AuthorInfo, + CrawlResult, +) + + +class ZhihuApiInterceptor: + """API Interceptor for capturing Zhihu API responses. + + Intercepts network requests and captures JSON responses from + Zhihu API endpoints (/api/v4/answers/, /api/v4/articles/, etc). + + Attributes: + captured_data: Dictionary storing captured API response data + """ + + def __init__(self): + self.captured_data: Dict[str, Any] = {} + + def handle_route(self, route: Route) -> None: + """Handle intercepted route/request. + + Args: + route: Playwright Route object for the intercepted request + """ + url = route.request.url + + # Intercept answer API + if "/api/v4/answers/" in url: + response = route.fetch() + self.captured_data["answer"] = response.json() + route.fulfill(response=response) + return + + # Intercept article API (zhuanlan) + if "/api/v4/articles/" in url or "zhuanlan" in url: + response = route.fetch() + self.captured_data["article"] = response.json() + route.fulfill(response=response) + return + + route.continue_() + + +class ZhihuCrawler: + """Zhihu Content Crawler. + + Crawls Zhihu answers and articles using Playwright API interception. + Reuses Cookie management from Phase 7 auth module. + + Attributes: + ANSWER_PATTERN: Regex pattern for answer URLs + ARTICLE_PATTERN: Regex pattern for zhuanlan article URLs + """ + + ANSWER_PATTERN = r"zhihu\.com/question/(\d+)/answer/(\d+)" + ARTICLE_PATTERN = r"zhuanlan\.zhihu\.com/p/(\d+)" + + def __init__(self, headless: bool = True, timeout: int = 30): + """Initialize ZhihuCrawler. + + Args: + headless: Whether to run browser in headless mode + timeout: Request timeout in seconds + """ + self.headless = headless + self.timeout = timeout + self.browser: Optional[PlaywrightBrowser] = None + self.interceptor: Optional[ZhihuApiInterceptor] = None + + def _get_cookie(self) -> str: + """Get Cookie from configuration. + + Returns: + Cookie string from settings + """ + settings = get_settings() + return settings.zhihu_cookie.get_secret_value() + + def _parse_author(self, data: Optional[Dict]) -> Optional[AuthorInfo]: + """Parse author information from API response. + + Args: + data: Author data dictionary from API response + + Returns: + AuthorInfo object or None if data is empty/None + """ + if data is None: + return None + + # Empty dict should return AuthorInfo with empty name + return AuthorInfo( + name=data.get("name", ""), + url_token=data.get("url_token"), + headline=data.get("headline") + ) + + def _parse_metadata(self, data: Dict) -> ContentMetadata: + """Parse content metadata from API response. + + Args: + data: API response data dictionary + + Returns: + ContentMetadata object + """ + return ContentMetadata( + author=self._parse_author(data.get("author")), + voteup_count=data.get("voteup_count", 0), + comment_count=data.get("comment_count", 0), + created_time=data.get("created_time"), + updated_time=data.get("updated_time") + ) + + def _parse_answer_data( + self, + data: Dict, + answer_id: int, + question_id: int + ) -> AnswerData: + """Parse answer data from API response. + + Args: + data: API response data dictionary + answer_id: Answer ID + question_id: Question ID + + Returns: + AnswerData object + """ + question = data.get("question", {}) + return AnswerData( + id=answer_id, + question_id=question_id, + question_title=question.get("title", ""), + content=data.get("content", ""), + excerpt=data.get("excerpt", ""), + metadata=self._parse_metadata(data) + ) + + def _parse_article_data(self, data: Dict, article_id: int) -> ArticleData: + """Parse article data from API response. + + Args: + data: API response data dictionary + article_id: Article ID + + Returns: + ArticleData object + """ + return ArticleData( + id=article_id, + title=data.get("title", ""), + content=data.get("content", ""), + excerpt=data.get("excerpt", ""), + metadata=self._parse_metadata(data) + ) + + def crawl_answer(self, url: str) -> CrawlResult: + """Crawl Zhihu answer. + + Args: + url: Zhihu answer URL (e.g., https://www.zhihu.com/question/123/answer/456) + + Returns: + CrawlResult with AnswerData if successful, error message otherwise + """ + match = re.search(self.ANSWER_PATTERN, url) + if not match: + return CrawlResult(success=False, error="Invalid answer URL format") + + question_id, answer_id = int(match.group(1)), int(match.group(2)) + + try: + self.browser = PlaywrightBrowser( + headless=self.headless, + timeout=self.timeout + ) + self.browser.start(cookie=self._get_cookie()) + self.interceptor = ZhihuApiInterceptor() + + # Set up route interception + self.browser.page.route("**/api/v4/**", self.interceptor.handle_route) + + # Visit page to trigger API requests + self.browser.goto(url, wait_until="networkidle") + + # Wait for data capture + time.sleep(2) + + if "answer" not in self.interceptor.captured_data: + return CrawlResult( + success=False, + error="Failed to capture answer data from API" + ) + + answer_data = self.interceptor.captured_data["answer"] + result_data = self._parse_answer_data( + answer_data, + answer_id=answer_id, + question_id=question_id + ) + + return CrawlResult( + success=True, + content_type="answer", + data=result_data + ) + + except Exception as e: + return CrawlResult(success=False, error=str(e)) + + finally: + if self.browser: + self.browser.close() + + def crawl_article(self, url: str) -> CrawlResult: + """Crawl Zhihu zhuanlan article. + + Args: + url: Zhihu zhuanlan URL (e.g., https://zhuanlan.zhihu.com/p/123) + + Returns: + CrawlResult with ArticleData if successful, error message otherwise + """ + match = re.search(self.ARTICLE_PATTERN, url) + if not match: + return CrawlResult(success=False, error="Invalid article URL format") + + article_id = int(match.group(1)) + + try: + self.browser = PlaywrightBrowser( + headless=self.headless, + timeout=self.timeout + ) + self.browser.start(cookie=self._get_cookie()) + self.interceptor = ZhihuApiInterceptor() + + self.browser.page.route("**/api/v4/**", self.interceptor.handle_route) + + self.browser.goto(url, wait_until="networkidle") + + time.sleep(2) + + if "article" not in self.interceptor.captured_data: + # Fallback: DOM parsing + return self._crawl_article_from_dom(url, article_id) + + article_data = self.interceptor.captured_data["article"] + result_data = self._parse_article_data(article_data, article_id) + + return CrawlResult( + success=True, + content_type="article", + data=result_data + ) + + except Exception as e: + return CrawlResult(success=False, error=str(e)) + + finally: + if self.browser: + self.browser.close() + + def _crawl_article_from_dom(self, url: str, article_id: int) -> CrawlResult: + """Crawl article using DOM parsing as fallback. + + Used when API interception fails for zhuanlan articles. + + Args: + url: Article URL + article_id: Article ID + + Returns: + CrawlResult with ArticleData from DOM extraction + """ + from kb_workflow.metadata.zhihu_extractor import ZhihuPlaywrightExtractor + + extractor = ZhihuPlaywrightExtractor( + headless=self.headless, + timeout=self.timeout, + cookie=self._get_cookie() + ) + result = extractor.extract(url) + + if not result: + return CrawlResult( + success=False, + error="Failed to extract article from DOM" + ) + + result_data = ArticleData( + id=article_id, + title=result.get("title", ""), + content=result.get("content", ""), + excerpt=result.get("content_markdown", "")[:200], + metadata=ContentMetadata( + author=AuthorInfo(name=result.get("author", "")), + voteup_count=result.get("vote_up", 0) + ) + ) + + return CrawlResult(success=True, content_type="article", data=result_data)