starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(08-01): implement ZhihuCrawler with API interception

Browse Source 
- Add ZhihuCrawler class with crawl_answer() and crawl_article() methods
- Implement ZhihuApiInterceptor for capturing /api/v4/ responses
- Add _parse_author(), _parse_metadata(), _parse_answer_data(), _parse_article_data() helpers
- Support DOM parsing fallback for zhuanlan articles
- Export ZhihuCrawler and ZhihuApiInterceptor from __init__.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
starmen1265
2026-03-17 21:10:10 +08:00
parent 912773fc65
commit aa96bc963c
2 changed files with 325 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kb_workflow/crawler/__init__.py
+3
View File
@@ -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",
]
kb_workflow/crawler/zhihu_crawler.py
+322
View File
@@ -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)