starmen1265
f41e75a1b2
- Add COMMENT_API_TEMPLATE for comment API URL - Implement crawl_comments() returning CommentsResult - Use order=hot for popular comments sorting - Only fetch first-level comments (no nested replies) - Add include_comments parameter to crawl_answer() - Export CommentAuthor, CommentData, CommentsResult in __init__.py Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
441 lines
14 KiB
Python
441 lines
14 KiB
Python
"""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,
|
|
CommentData,
|
|
CommentAuthor,
|
|
CommentsResult,
|
|
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
|
|
COMMENT_API_TEMPLATE: Template for comment API URL
|
|
"""
|
|
|
|
ANSWER_PATTERN = r"zhihu\.com/question/(\d+)/answer/(\d+)"
|
|
ARTICLE_PATTERN = r"zhuanlan\.zhihu\.com/p/(\d+)"
|
|
COMMENT_API_TEMPLATE = (
|
|
"https://www.zhihu.com/api/v4/comment_v5/"
|
|
"{content_type}/{content_id}/comments"
|
|
"?order=hot&limit=20"
|
|
)
|
|
|
|
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_comments(
|
|
self,
|
|
content_type: str,
|
|
content_id: int,
|
|
include_comments: bool = True
|
|
) -> CommentsResult:
|
|
"""Crawl hot comments for a Zhihu content.
|
|
|
|
Fetches first-level comments using the hot sorting API.
|
|
Does not expand nested replies.
|
|
|
|
Args:
|
|
content_type: Content type ("answers" or "articles")
|
|
content_id: Content ID
|
|
include_comments: Whether to fetch comments (default True)
|
|
|
|
Returns:
|
|
CommentsResult with list of CommentData
|
|
"""
|
|
if not include_comments:
|
|
return CommentsResult(
|
|
content_type=content_type,
|
|
content_id=content_id,
|
|
comments=[],
|
|
total_count=0
|
|
)
|
|
|
|
try:
|
|
self.browser = PlaywrightBrowser(
|
|
headless=self.headless,
|
|
timeout=self.timeout
|
|
)
|
|
self.browser.start(cookie=self._get_cookie())
|
|
|
|
# Build comment API URL
|
|
api_url = self.COMMENT_API_TEMPLATE.format(
|
|
content_type=content_type,
|
|
content_id=content_id
|
|
)
|
|
|
|
# Use Playwright's request API to directly call the API
|
|
response = self.browser.page.request.get(api_url)
|
|
|
|
if response.status != 200:
|
|
return CommentsResult(
|
|
content_type=content_type,
|
|
content_id=content_id,
|
|
comments=[],
|
|
total_count=0
|
|
)
|
|
|
|
data = response.json()
|
|
comments = []
|
|
|
|
for item in data.get("data", []):
|
|
author_data = item.get("author", {})
|
|
comment = CommentData(
|
|
id=item.get("id", 0),
|
|
content=item.get("content", ""),
|
|
author=CommentAuthor(
|
|
name=author_data.get("name", ""),
|
|
url_token=author_data.get("url_token")
|
|
) if author_data else None,
|
|
created_time=item.get("created_time"),
|
|
like_count=item.get("like_count", 0),
|
|
child_count=item.get("child_comment_count", 0),
|
|
is_author=item.get("is_author", False)
|
|
)
|
|
comments.append(comment)
|
|
|
|
# Get total count from paging
|
|
total_count = data.get("paging", {}).get("totals", len(comments))
|
|
|
|
return CommentsResult(
|
|
content_type=content_type,
|
|
content_id=content_id,
|
|
comments=comments,
|
|
total_count=total_count
|
|
)
|
|
|
|
except Exception:
|
|
return CommentsResult(
|
|
content_type=content_type,
|
|
content_id=content_id,
|
|
comments=[],
|
|
total_count=0
|
|
)
|
|
|
|
finally:
|
|
if self.browser:
|
|
self.browser.close()
|
|
|
|
def crawl_answer(self, url: str, include_comments: bool = False) -> CrawlResult:
|
|
"""Crawl Zhihu answer.
|
|
|
|
Args:
|
|
url: Zhihu answer URL (e.g., https://www.zhihu.com/question/123/answer/456)
|
|
include_comments: Whether to include hot comments (default False)
|
|
|
|
Returns:
|
|
CrawlResult with AnswerData if successful, error message otherwise.
|
|
If include_comments=True and successful, comments field will be populated.
|
|
"""
|
|
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
|
|
)
|
|
|
|
result = CrawlResult(
|
|
success=True,
|
|
content_type="answer",
|
|
data=result_data
|
|
)
|
|
|
|
# Fetch comments if requested
|
|
if include_comments:
|
|
# Need to close current browser first as crawl_comments creates its own
|
|
self.browser.close()
|
|
self.browser = None
|
|
|
|
comments_result = self.crawl_comments(
|
|
content_type="answers",
|
|
content_id=answer_id,
|
|
include_comments=True
|
|
)
|
|
result.comments = comments_result.comments
|
|
|
|
return result
|
|
|
|
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)
|