---
phase: 08-crawler
plan: 01
type: execute
wave: 1
depends_on: []
files_modified:
- kb_workflow/crawler/__init__.py
- kb_workflow/crawler/models.py
- kb_workflow/crawler/zhihu_crawler.py
- tests/test_crawler/__init__.py
- tests/test_crawler/conftest.py
- tests/test_crawler/test_answer_crawler.py
- tests/test_crawler/test_article_crawler.py
- tests/test_crawler/test_metadata_extraction.py
autonomous: true
requirements: [CONT-01, CONT-02, CONT-04]
user_setup: []
must_haves:
truths:
- "用户可以爬取知乎问答，获取问题标题、描述、回答内容"
- "用户可以爬取知乎专栏/文章，获取标题和正文"
- "爬取结果包含作者信息、发布时间、点赞数等元数据"
artifacts:
- path: "kb_workflow/crawler/zhihu_crawler.py"
provides: "ZhihuCrawler 类，实现 API 拦截和内容提取"
min_lines: 100
- path: "kb_workflow/crawler/models.py"
provides: "Pydantic 数据模型定义"
contains: "class.*AnswerData"
- path: "tests/test_crawler/test_answer_crawler.py"
provides: "问答爬取测试"
exports: ["test_crawl_answer"]
- path: "tests/test_crawler/test_article_crawler.py"
provides: "专栏爬取测试"
exports: ["test_crawl_article"]
key_links:
- from: "kb_workflow/crawler/zhihu_crawler.py"
to: "kb_workflow/auth/zhihu_login.py"
via: "Cookie 加载"
pattern: "from kb_workflow.auth"
- from: "kb_workflow/crawler/zhihu_crawler.py"
to: "Playwright page.route"
via: "API 拦截"
pattern: "page\\.route"
---
实现知乎问答和专栏/文章的内容爬取功能，使用 Playwright 拦截 API 请求获取结构化 JSON 数据。
Purpose: 为 Pipeline 提供可靠的知乎内容爬取能力，复用 Phase 7 的 Cookie 管理
Output: ZhihuCrawler 类 + 数据模型 + 测试
@/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md
@/home/starmen1265/.claude/get-shit-done/templates/summary.md
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/08-crawler/08-CONTEXT.md
@.planning/phases/08-crawler/08-RESEARCH.md
@.planning/phases/07-login-auth/07-03-SUMMARY.md
From kb_workflow/auth/zhihu_login.py:
```python
class ZhihuLoginResult(BaseModel):
success: bool
cookie_str: str
qrcode_path: Optional[Path]
error_message: Optional[str]
class CookieValidationResult(BaseModel):
valid: bool
username: Optional[str]
user_id: Optional[str]
reason: Optional[str]
```
From kb_workflow/metadata/zhihu_extractor.py:
```python
class PlaywrightBrowser:
def start(self, cookie: str = None)
def goto(self, url: str, wait_until: str = "networkidle")
def close(self)
```
From kb_workflow/pipeline/config.py:
```python
class KBSettings(BaseSettings):
zhihu_cookie: SecretStr
intermediate_path: str
```
API 拦截目标 (from RESEARCH.md):
- 问答: `/api/v4/answers/{id}` → JSON 响应
- 专栏: zhuanlan 页面 API 或 DOM 解析
- 评论: `/api/v4/comment_v5/{type}/{id}/comments?order=hot`
Task 1: 创建数据模型和测试脚手架
kb_workflow/crawler/__init__.py, kb_workflow/crawler/models.py, tests/test_crawler/__init__.py, tests/test_crawler/conftest.py
创建 ZhihuCrawler 模块的数据模型和测试基础设施。
**1. 创建 kb_workflow/crawler/models.py:**
```python
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
class AuthorInfo(BaseModel):
"""作者信息"""
name: str = Field(..., description="作者名称")
url_token: Optional[str] = Field(None, description="作者 URL token")
headline: Optional[str] = Field(None, description="作者签名")
class ContentMetadata(BaseModel):
"""内容元数据"""
author: Optional[AuthorInfo] = None
voteup_count: int = Field(default=0, description="点赞数")
comment_count: int = Field(default=0, description="评论数")
created_time: Optional[int] = Field(None, description="创建时间戳")
updated_time: Optional[int] = Field(None, description="更新时间戳")
class AnswerData(BaseModel):
"""知乎回答数据"""
id: int
question_id: int
question_title: str
content: str # HTML
excerpt: str # 纯文本摘要
metadata: ContentMetadata
class ArticleData(BaseModel):
"""知乎专栏文章数据"""
id: int
title: str
content: str # HTML
excerpt: str
metadata: ContentMetadata
class CrawlResult(BaseModel):
"""爬取结果"""
success: bool
content_type: str # "answer" or "article"
data: Optional[AnswerData | ArticleData] = None
error: Optional[str] = None
```
**2. 创建 tests/test_crawler/conftest.py:**
- mock_page: Mock Playwright page 对象
- mock_browser: Mock PlaywrightBrowser 实例
- sample_answer_api_response: 示例问答 API 响应
- sample_article_api_response: 示例专栏 API 响应
**3. 创建 kb_workflow/crawler/__init__.py:** 导出主要类
pytest tests/test_crawler/ -x --tb=short -k "conftest or init"
Verify models.py can be imported: python -c "from kb_workflow.crawler.models import AnswerData; print('OK')"
- models.py 包含 AnswerData, ArticleData, ContentMetadata, AuthorInfo 类
- conftest.py 包含 mock_page, mock_browser fixtures
- 所有模块可正常导入
Task 2: 实现 ZhihuCrawler 核心类
kb_workflow/crawler/zhihu_crawler.py, tests/test_crawler/test_answer_crawler.py, tests/test_crawler/test_article_crawler.py
- Test 1: crawl_answer() 返回 AnswerData，包含问题标题、回答内容、元数据
- Test 2: crawl_article() 返回 ArticleData，包含标题、正文、元数据
- Test 3: 无效 URL 返回 CrawlResult(success=False, error=...)
- Test 4: API 拦截正确捕获 /api/v4/answers/ 响应
实现 ZhihuCrawler 类，使用 Playwright route 拦截 API 请求。
**1. 创建 kb_workflow/crawler/zhihu_crawler.py:**
```python
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 拦截器，捕获知乎 API 响应"""
def __init__(self):
self.captured_data: Dict[str, Any] = {}
def handle_route(self, route: Route):
"""处理拦截的请求"""
url = route.request.url
# 拦截答案 API
if "/api/v4/answers/" in url:
response = route.fetch()
self.captured_data["answer"] = response.json()
route.fulfill(response=response)
return
# 拦截文章 API (专栏)
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:
"""知乎内容爬虫"""
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):
self.headless = headless
self.timeout = timeout
self.browser: Optional[PlaywrightBrowser] = None
self.interceptor: Optional[ZhihuApiInterceptor] = None
def _get_cookie(self) -> str:
"""获取 Cookie"""
settings = get_settings()
return settings.zhihu_cookie.get_secret_value()
def _parse_author(self, data: Dict) -> Optional[AuthorInfo]:
"""解析作者信息"""
author = data.get("author", {})
if not author:
return None
return AuthorInfo(
name=author.get("name", ""),
url_token=author.get("url_token"),
headline=author.get("headline")
)
def _parse_metadata(self, data: Dict) -> ContentMetadata:
"""解析元数据"""
return ContentMetadata(
author=self._parse_author(data),
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 crawl_answer(self, url: str) -> CrawlResult:
"""爬取知乎回答"""
import re
match = re.search(self.ANSWER_PATTERN, url)
if not match:
return CrawlResult(success=False, error="Invalid answer URL")
question_id, answer_id = match.groups()
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)
# 访问页面触发 API 请求
self.browser.goto(url, wait_until="networkidle")
# 等待数据捕获
import time
time.sleep(2)
if "answer" not in self.interceptor.captured_data:
return CrawlResult(success=False, error="Failed to capture answer data")
answer_data = self.interceptor.captured_data["answer"]
# 解析数据
result_data = AnswerData(
id=int(answer_id),
question_id=int(question_id),
question_title=answer_data.get("question", {}).get("title", ""),
content=answer_data.get("content", ""),
excerpt=answer_data.get("excerpt", ""),
metadata=self._parse_metadata(answer_data)
)
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:
"""爬取知乎专栏文章"""
import re
match = re.search(self.ARTICLE_PATTERN, url)
if not match:
return CrawlResult(success=False, error="Invalid article URL")
article_id = match.groups()[0]
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")
import time
time.sleep(2)
if "article" not in self.interceptor.captured_data:
# 备选方案：DOM 解析
return self._crawl_article_from_dom(url, article_id)
article_data = self.interceptor.captured_data["article"]
result_data = ArticleData(
id=int(article_id),
title=article_data.get("title", ""),
content=article_data.get("content", ""),
excerpt=article_data.get("excerpt", ""),
metadata=self._parse_metadata(article_data)
)
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: str) -> CrawlResult:
"""从 DOM 解析专栏文章（备选方案）"""
# 复用现有的 ZhihuPlaywrightExtractor 逻辑
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=int(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)
```
**2. 创建测试文件:**
- tests/test_crawler/test_answer_crawler.py: 测试 crawl_answer()
- tests/test_crawler/test_article_crawler.py: 测试 crawl_article()
- tests/test_crawler/test_metadata_extraction.py: 测试元数据提取
pytest tests/test_crawler/test_answer_crawler.py tests/test_crawler/test_article_crawler.py tests/test_crawler/test_metadata_extraction.py -x --tb=short
- ZhihuCrawler.crawl_answer() 返回 CrawlResult 包含 AnswerData
- ZhihuCrawler.crawl_article() 返回 CrawlResult 包含 ArticleData
- 元数据正确提取作者、点赞数、评论数
- 所有测试通过
**Wave 0 检查:**
- [ ] tests/test_crawler/__init__.py 存在
- [ ] tests/test_crawler/conftest.py 包含 fixtures
- [ ] tests/test_crawler/test_answer_crawler.py 测试文件存在
- [ ] tests/test_crawler/test_article_crawler.py 测试文件存在
- [ ] tests/test_crawler/test_metadata_extraction.py 测试文件存在
**自动化验证:**
```bash
pytest tests/test_crawler/ -x -v
```
1. ZhihuCrawler 类可正常导入
2. crawl_answer() 能爬取知乎问答并返回结构化数据
3. crawl_article() 能爬取知乎专栏并返回结构化数据
4. 数据包含作者、点赞数、评论数等完整元数据
5. 所有测试通过