diff --git a/.planning/ROADMAP.md b/.planning/ROADMAP.md
index 2ccbf75..13359d7 100644
--- a/.planning/ROADMAP.md
+++ b/.planning/ROADMAP.md
@@ -90,12 +90,12 @@ Plans:
2. 扫码失败时用户可以手动设置 Cookie 进行认证
3. Cookie 自动保存到本地文件，下次运行自动复用
4. Cookie 过期时系统自动检测并提示用户重新登录
-**Plans**: 3 plans
+**Plans**: 3 plans (all complete)
Plans:
-- [ ] 07-01-PLAN.md — 二维码扫码登录核心逻辑 (AUTH-01) + Wave 0 测试脚手架
-- [ ] 07-02-PLAN.md — Cookie 导入和持久化管理 (AUTH-02, AUTH-03)
-- [ ] 07-03-PLAN.md — Cookie 验证 + CLI login 子命令 (AUTH-04)
+- [x] 07-01-PLAN.md — 二维码扫码登录核心逻辑 (AUTH-01) + Wave 0 测试脚手架
+- [x] 07-02-PLAN.md — Cookie 导入和持久化管理 (AUTH-02, AUTH-03)
+- [x] 07-03-PLAN.md — Cookie 验证 + CLI login 子命令 (AUTH-04)
#### Phase 8: 内容爬取实现
**Goal**: 实现知乎问答、专栏/文章、评论的完整爬取功能，包含网络反爬措施
@@ -107,12 +107,12 @@ Plans:
3. 用户可以选择是否爬取内容下的评论
4. 爬取结果包含作者信息、发布时间、点赞数等完整元数据
5. 用户可以配置代理 IP 以降低被封风险
-**Plans**: TBD
+**Plans**: 3 plans in 2 waves
Plans:
-- [ ] 08-01: 问答和专栏内容爬取 (CONT-01, CONT-02, CONT-04)
-- [ ] 08-02: 评论爬取功能 (CONT-03)
-- [ ] 08-03: 网络反爬措施 (NET-01, NET-02, NET-03)
+- [ ] 08-01-PLAN.md — 问答和专栏内容爬取 (CONT-01, CONT-02, CONT-04) — Wave 1
+- [ ] 08-02-PLAN.md — 评论爬取功能 (CONT-03) — Wave 1 (parallel)
+- [ ] 08-03-PLAN.md — 网络反爬措施 (NET-01, NET-02, NET-03) — Wave 2
#### Phase 9: Pipeline 整合
**Goal**: 将新爬虫实现整合进现有 Pipeline 架构，替换原有知乎处理器
@@ -140,6 +140,6 @@ Phases execute in numeric order: 1 → 2 → 3 → 4 → 5 → 6 → 7 → 8 →
| 4. Pipeline | v1.0 | 2/2 | Complete | 2026-03-15 |
| 5. 归档与目录优化 | v1.1 | 3/3 | Complete | 2026-03-16 |
| 6. 代码整合 | v1.1 | 3/3 | Complete | 2026-03-16 |
-| 7. 登录认证体系 | v1.2 | 0/3 | Ready to execute | - |
-| 8. 内容爬取实现 | v1.2 | 0/3 | Not started | - |
+| 7. 登录认证体系 | v1.2 | 3/3 | Complete | 2026-03-17 |
+| 8. 内容爬取实现 | v1.2 | 0/3 | Ready to execute | - |
| 9. Pipeline 整合 | v1.2 | 0/1 | Not started | - |
diff --git a/.planning/phases/08-crawler/08-01-PLAN.md b/.planning/phases/08-crawler/08-01-PLAN.md
new file mode 100644
index 0000000..40f93bf
--- /dev/null
+++ b/.planning/phases/08-crawler/08-01-PLAN.md
@@ -0,0 +1,429 @@
+---
+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. 所有测试通过
+
+
+
diff --git a/.planning/phases/08-crawler/08-02-PLAN.md b/.planning/phases/08-crawler/08-02-PLAN.md
new file mode 100644
index 0000000..671ae14
--- /dev/null
+++ b/.planning/phases/08-crawler/08-02-PLAN.md
@@ -0,0 +1,439 @@
+---
+phase: 08-crawler
+plan: 02
+type: execute
+wave: 1
+depends_on: []
+files_modified:
+ - kb_workflow/crawler/zhihu_crawler.py
+ - kb_workflow/crawler/models.py
+ - tests/test_crawler/test_comment_crawler.py
+autonomous: true
+requirements: [CONT-03]
+user_setup: []
+
+must_haves:
+ truths:
+ - "用户可以选择是否爬取内容下的评论"
+ - "爬取的评论是热门评论（非全部评论）"
+ - "只爬取第一层评论，不展开嵌套回复"
+ artifacts:
+ - path: "kb_workflow/crawler/models.py"
+ provides: "CommentData 数据模型"
+ contains: "class CommentData"
+ - path: "kb_workflow/crawler/zhihu_crawler.py"
+ provides: "crawl_comments() 方法"
+ contains: "def crawl_comments"
+ - path: "tests/test_crawler/test_comment_crawler.py"
+ provides: "评论爬取测试"
+ exports: ["test_crawl_comments"]
+ key_links:
+ - from: "ZhihuCrawler.crawl_comments"
+ to: "/api/v4/comment_v5/{type}/{id}/comments"
+ via: "热门评论 API"
+ pattern: "comment_v5.*order=hot"
+---
+
+
+实现知乎评论爬取功能，通过热门评论 API 获取第一层评论，作为可选功能集成到 ZhihuCrawler。
+
+Purpose: 提供可选的评论爬取能力，丰富内容元数据
+Output: crawl_comments() 方法 + CommentData 模型 + 测试
+
+
+
+@/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
+
+
+
+
+评论 API 格式:
+```
+https://www.zhihu.com/api/v4/comment_v5/{content_type}/{content_id}/comments
+?order=hot&limit=20
+```
+
+参数说明:
+- content_type: "answers" (回答) 或 "articles" (文章)
+- content_id: 内容 ID
+- order: "hot" (热门) 或 "ts" (最新)
+- limit: 每页数量，默认 20
+
+响应结构 (预期):
+```json
+{
+ "data": [
+ {
+ "id": 12345,
+ "content": "评论内容",
+ "author": {"name": "用户名"},
+ "created_time": 1234567890,
+ "like_count": 10,
+ "child_comment_count": 2
+ }
+ ],
+ "paging": {...}
+}
+```
+
+
+
+
+
+
+ Task 1: 添加 CommentData 数据模型
+ kb_workflow/crawler/models.py
+
+在 models.py 中添加评论相关的数据模型。
+
+**添加到 kb_workflow/crawler/models.py:**
+
+```python
+class CommentAuthor(BaseModel):
+ """评论作者信息"""
+ name: str = Field(..., description="作者名称")
+ url_token: Optional[str] = Field(None, description="作者 URL token")
+
+class CommentData(BaseModel):
+ """评论数据"""
+ id: int
+ content: str = Field(..., description="评论内容")
+ author: Optional[CommentAuthor] = None
+ created_time: Optional[int] = Field(None, description="创建时间戳")
+ like_count: int = Field(default=0, description="点赞数")
+ child_count: int = Field(default=0, description="子评论数")
+ is_author: bool = Field(default=False, description="是否为作者回复")
+
+class CommentsResult(BaseModel):
+ """评论爬取结果"""
+ content_type: str # "answers" or "articles"
+ content_id: int
+ comments: List[CommentData] = Field(default_factory=list)
+ total_count: int = Field(default=0, description="总评论数（可能超过实际返回）")
+```
+
+同时更新 CrawlResult 以支持包含评论:
+```python
+class CrawlResult(BaseModel):
+ """爬取结果"""
+ success: bool
+ content_type: str # "answer" or "article"
+ data: Optional[AnswerData | ArticleData] = None
+ comments: Optional[List[CommentData]] = None # 新增
+ error: Optional[str] = None
+```
+
+
+ python -c "from kb_workflow.crawler.models import CommentData, CommentsResult; print('OK')"
+
+
+ - CommentData 模型定义完成
+ - CommentsResult 模型定义完成
+ - CrawlResult 支持 comments 字段
+
+
+
+
+ Task 2: 实现 crawl_comments 方法
+ kb_workflow/crawler/zhihu_crawler.py, tests/test_crawler/test_comment_crawler.py
+
+ - Test 1: crawl_comments("answers", 12345) 返回 CommentsResult
+ - Test 2: 评论列表包含热门评论（order=hot）
+ - Test 3: 每条评论包含 content, author, like_count
+ - Test 4: 不展开嵌套回复（只有第一层评论）
+ - Test 5: crawl_with_comments=True 参数在 crawl_answer 中触发评论爬取
+
+
+在 ZhihuCrawler 类中实现评论爬取功能。
+
+**1. 在 ZhihuCrawler 类中添加方法:**
+
+```python
+# 在 kb_workflow/crawler/zhihu_crawler.py 中添加
+
+from .models import CommentData, CommentsResult, CommentAuthor
+
+class ZhihuCrawler:
+ # ... 现有代码 ...
+
+ COMMENT_API_TEMPLATE = (
+ "https://www.zhihu.com/api/v4/comment_v5/"
+ "{content_type}/{content_id}/comments"
+ "?order=hot&limit=20"
+ )
+
+ def crawl_comments(
+ self,
+ content_type: str, # "answers" or "articles"
+ content_id: int,
+ include_comments: bool = True
+ ) -> CommentsResult:
+ """爬取热门评论
+
+ Args:
+ content_type: 内容类型 ("answers" 或 "articles")
+ content_id: 内容 ID
+ include_comments: 是否包含评论（保留参数兼容性）
+
+ Returns:
+ CommentsResult 包含评论列表
+ """
+ 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())
+
+ # 构建评论 API URL
+ api_url = self.COMMENT_API_TEMPLATE.format(
+ content_type=content_type,
+ content_id=content_id
+ )
+
+ # 使用 Playwright 的 request 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)
+
+ # 获取总数
+ 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 as e:
+ 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:
+ """爬取知乎回答（支持可选评论）
+
+ Args:
+ url: 回答 URL
+ include_comments: 是否包含热门评论
+
+ Returns:
+ CrawlResult 包含回答数据和可选评论
+ """
+ # 原有逻辑...
+ result = self._crawl_answer_original(url)
+
+ if result.success and include_comments:
+ # 爬取评论
+ comments_result = self.crawl_comments(
+ content_type="answers",
+ content_id=result.data.id,
+ include_comments=True
+ )
+ result.comments = comments_result.comments
+
+ return result
+```
+
+**2. 创建 tests/test_crawler/test_comment_crawler.py:**
+
+```python
+import pytest
+from unittest.mock import Mock, patch, MagicMock
+from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+from kb_workflow.crawler.models import CommentData, CommentsResult
+
+class TestCommentCrawler:
+
+ @pytest.fixture
+ def crawler(self):
+ return ZhihuCrawler(headless=True)
+
+ @pytest.fixture
+ def mock_comment_response(self):
+ return {
+ "data": [
+ {
+ "id": 111,
+ "content": "这是一条热门评论",
+ "author": {"name": "用户A", "url_token": "user-a"},
+ "created_time": 1234567890,
+ "like_count": 100,
+ "child_comment_count": 5,
+ "is_author": False
+ },
+ {
+ "id": 222,
+ "content": "作者回复",
+ "author": {"name": "作者", "url_token": "author"},
+ "created_time": 1234567900,
+ "like_count": 50,
+ "child_comment_count": 0,
+ "is_author": True
+ }
+ ],
+ "paging": {"totals": 100}
+ }
+
+ def test_crawl_comments_returns_comments_result(self, crawler, mock_comment_response):
+ """crawl_comments 返回 CommentsResult"""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
+ mock_browser = MockBrowser.return_value
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+ mock_browser.page.request.get.return_value = mock_response
+
+ result = crawler.crawl_comments("answers", 12345)
+
+ assert isinstance(result, CommentsResult)
+ assert result.content_type == "answers"
+ assert result.content_id == 12345
+
+ def test_comments_are_hot_order(self, crawler, mock_comment_response):
+ """评论使用热门排序"""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
+ mock_browser = MockBrowser.return_value
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+ mock_browser.page.request.get.return_value = mock_response
+
+ crawler.crawl_comments("answers", 12345)
+
+ # 验证 API URL 包含 order=hot
+ call_args = mock_browser.page.request.get.call_args
+ assert "order=hot" in call_args[0][0]
+
+ def test_comment_has_required_fields(self, crawler, mock_comment_response):
+ """每条评论包含必要字段"""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
+ mock_browser = MockBrowser.return_value
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+ mock_browser.page.request.get.return_value = mock_response
+
+ result = crawler.crawl_comments("answers", 12345)
+
+ assert len(result.comments) == 2
+ comment = result.comments[0]
+ assert comment.content == "这是一条热门评论"
+ assert comment.author.name == "用户A"
+ assert comment.like_count == 100
+
+ def test_no_nested_replies(self, crawler, mock_comment_response):
+ """不展开嵌套回复"""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as MockBrowser:
+ mock_browser = MockBrowser.return_value
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+ mock_browser.page.request.get.return_value = mock_response
+
+ result = crawler.crawl_comments("answers", 12345)
+
+ # 只返回第一层评论，不包含子评论
+ assert len(result.comments) == 2
+ # child_count 记录子评论数但不展开
+ assert result.comments[0].child_count == 5
+
+ def test_include_comments_parameter(self, crawler):
+ """include_comments=False 不爬取评论"""
+ result = crawler.crawl_comments("answers", 12345, include_comments=False)
+
+ assert result.comments == []
+ assert result.total_count == 0
+```
+
+
+ pytest tests/test_crawler/test_comment_crawler.py -x --tb=short
+
+
+ - crawl_comments() 方法返回 CommentsResult
+ - 评论使用 order=hot 排序
+ - 只返回第一层评论
+ - include_comments 参数控制是否爬取
+ - 所有测试通过
+
+
+
+
+
+
+**自动化验证:**
+```bash
+pytest tests/test_crawler/test_comment_crawler.py -x -v
+```
+
+**集成验证:**
+```python
+from kb_workflow.crawler import ZhihuCrawler
+crawler = ZhihuCrawler()
+result = crawler.crawl_answer("https://www.zhihu.com/question/123/answer/456", include_comments=True)
+assert result.comments is not None
+```
+
+
+
+1. CommentData 模型定义完成
+2. crawl_comments() 方法实现完成
+3. 评论使用热门排序 (order=hot)
+4. 只爬取第一层评论，不展开嵌套
+5. include_comments 参数控制评论爬取
+6. 所有测试通过
+
+
+
diff --git a/.planning/phases/08-crawler/08-03-PLAN.md b/.planning/phases/08-crawler/08-03-PLAN.md
new file mode 100644
index 0000000..008e914
--- /dev/null
+++ b/.planning/phases/08-crawler/08-03-PLAN.md
@@ -0,0 +1,710 @@
+---
+phase: 08-crawler
+plan: 03
+type: execute
+wave: 2
+depends_on: [08-01]
+files_modified:
+ - kb_workflow/crawler/config.py
+ - kb_workflow/crawler/zhihu_crawler.py
+ - tests/test_crawler/test_proxy_config.py
+ - tests/test_crawler/test_rate_limiter.py
+ - tests/test_crawler/test_api_intercept.py
+autonomous: true
+requirements: [NET-01, NET-02, NET-03]
+user_setup:
+ - service: proxy
+ why: "防止 IP 被知乎封禁"
+ env_vars:
+ - name: HTTP_PROXY
+ source: "代理服务地址，如 http://127.0.0.1:7890"
+ - name: HTTPS_PROXY
+ source: "代理服务地址，如 http://127.0.0.1:7890"
+ dashboard_config: []
+
+must_haves:
+ truths:
+ - "用户可以通过环境变量配置代理 IP"
+ - "不配置代理时，爬虫拒绝运行并提示错误"
+ - "每次请求间有随机延迟（1-3秒）"
+ - "请求失败时有限重试（3次），间隔递增"
+ - "x-zse-96 签名由浏览器自动处理，无需手动逆向"
+ artifacts:
+ - path: "kb_workflow/crawler/config.py"
+ provides: "CrawlerConfig 配置类，代理和延迟设置"
+ contains: "class CrawlerConfig"
+ - path: "kb_workflow/crawler/zhihu_crawler.py"
+ provides: "代理应用和重试逻辑"
+ contains: "def _get_proxy_config"
+ - path: "tests/test_crawler/test_proxy_config.py"
+ provides: "代理配置测试"
+ exports: ["test_proxy_required"]
+ - path: "tests/test_crawler/test_rate_limiter.py"
+ provides: "频率控制测试"
+ exports: ["test_random_delay"]
+ key_links:
+ - from: "CrawlerConfig"
+ to: "HTTP_PROXY/HTTPS_PROXY"
+ via: "环境变量读取"
+ pattern: "os\\.environ\\.get.*PROXY"
+ - from: "ZhihuCrawler"
+ to: "browser.new_context(proxy=...)"
+ via: "Playwright 代理配置"
+ pattern: "new_context.*proxy"
+---
+
+
+实现网络反爬措施：强制代理配置、请求频率控制、有限重试机制，以及利用浏览器自动处理 x-zse-96 签名。
+
+Purpose: 降低被封风险，提高爬取稳定性
+Output: CrawlerConfig + 代理/延迟/重试逻辑 + 测试
+
+
+
+@/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/08-crawler/08-01-PLAN.md
+
+
+
+
+From kb_workflow/crawler/zhihu_crawler.py (Plan 01):
+```python
+class ZhihuCrawler:
+ def __init__(self, headless: bool = True, timeout: int = 30)
+ def _get_cookie(self) -> str
+ def crawl_answer(self, url: str) -> CrawlResult
+ def crawl_article(self, url: str) -> CrawlResult
+```
+
+代理配置模式 (from RESEARCH.md):
+```python
+import os
+from urllib.parse import urlparse
+
+def get_proxy_config():
+ proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
+ if not proxy_url:
+ return None
+
+ parsed = urlparse(proxy_url)
+ config = {"server": f"{parsed.scheme}://{parsed.hostname}:{parsed.port}"}
+
+ if parsed.username and parsed.password:
+ config["username"] = parsed.username
+ config["password"] = parsed.password
+
+ return config
+```
+
+重试模式 (from RESEARCH.md):
+```python
+from tenacity import Retrying, stop_after_attempt, wait_exponential
+
+for attempt in Retrying(
+ stop=stop_after_attempt(3),
+ wait=wait_exponential(multiplier=1, min=2, max=10),
+ reraise=True
+):
+ with attempt:
+ # 爬取逻辑
+```
+
+
+
+
+
+
+ Task 1: 创建 CrawlerConfig 配置类
+ kb_workflow/crawler/config.py, tests/test_crawler/test_proxy_config.py
+
+创建爬虫配置类，处理代理和延迟设置。
+
+**1. 创建 kb_workflow/crawler/config.py:**
+
+```python
+"""Crawler configuration
+
+Handles proxy settings, rate limiting, and retry configuration.
+"""
+
+import os
+from urllib.parse import urlparse
+from typing import Optional, Dict, Any
+
+from pydantic import BaseModel, Field
+
+
+class ProxyConfig(BaseModel):
+ """代理配置"""
+ server: str = Field(..., description="代理服务器地址")
+ username: Optional[str] = Field(None, description="代理用户名")
+ password: Optional[str] = Field(None, description="代理密码")
+
+ def to_playwright_format(self) -> Dict[str, Any]:
+ """转换为 Playwright 代理格式"""
+ config = {"server": self.server}
+ if self.username:
+ config["username"] = self.username
+ if self.password:
+ config["password"] = self.password
+ return config
+
+
+class CrawlerConfig(BaseModel):
+ """爬虫配置"""
+
+ # 代理设置
+ proxy: Optional[ProxyConfig] = Field(None, description="代理配置")
+ proxy_required: bool = Field(default=True, description="是否强制要求代理")
+
+ # 请求频率
+ min_delay: float = Field(default=1.0, description="最小延迟（秒）")
+ max_delay: float = Field(default=3.0, description="最大延迟（秒）")
+
+ # 重试设置
+ max_retries: int = Field(default=3, description="最大重试次数")
+ retry_min_wait: float = Field(default=2.0, description="重试最小等待（秒）")
+ retry_max_wait: float = Field(default=10.0, description="重试最大等待（秒）")
+
+ @classmethod
+ def from_env(cls) -> "CrawlerConfig":
+ """从环境变量加载配置"""
+ proxy = cls._parse_proxy_env()
+ return cls(proxy=proxy)
+
+ @staticmethod
+ def _parse_proxy_env() -> Optional[ProxyConfig]:
+ """解析代理环境变量"""
+ proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
+
+ if not proxy_url:
+ return None
+
+ parsed = urlparse(proxy_url)
+ server = f"{parsed.scheme}://{parsed.hostname}"
+
+ if parsed.port:
+ server += f":{parsed.port}"
+
+ return ProxyConfig(
+ server=server,
+ username=parsed.username,
+ password=parsed.password
+ )
+
+ def validate_proxy(self) -> None:
+ """验证代理配置
+
+ Raises:
+ ValueError: 如果 proxy_required=True 但未配置代理
+ """
+ if self.proxy_required and not self.proxy:
+ raise ValueError(
+ "代理配置缺失！请设置环境变量 HTTP_PROXY 或 HTTPS_PROXY。\n"
+ "示例: export HTTPS_PROXY=http://127.0.0.1:7890"
+ )
+
+
+def get_crawler_config() -> CrawlerConfig:
+ """获取爬虫配置单例"""
+ return CrawlerConfig.from_env()
+```
+
+**2. 创建 tests/test_crawler/test_proxy_config.py:**
+
+```python
+import os
+import pytest
+from unittest.mock import patch
+
+from kb_workflow.crawler.config import (
+ CrawlerConfig,
+ ProxyConfig,
+ get_crawler_config
+)
+
+
+class TestProxyConfig:
+
+ def test_parse_proxy_from_https_env(self):
+ """从 HTTPS_PROXY 解析代理"""
+ with patch.dict(os.environ, {"HTTPS_PROXY": "http://127.0.0.1:7890"}):
+ config = CrawlerConfig.from_env()
+ assert config.proxy is not None
+ assert config.proxy.server == "http://127.0.0.1:7890"
+
+ def test_parse_proxy_from_http_env(self):
+ """从 HTTP_PROXY 解析代理（HTTPS_PROXY 优先）"""
+ with patch.dict(os.environ, {
+ "HTTP_PROXY": "http://127.0.0.1:8080",
+ "HTTPS_PROXY": "http://127.0.0.1:7890"
+ }, clear=False):
+ config = CrawlerConfig.from_env()
+ assert config.proxy.server == "http://127.0.0.1:7890"
+
+ def test_parse_proxy_with_auth(self):
+ """解析带认证的代理"""
+ with patch.dict(os.environ, {
+ "HTTPS_PROXY": "http://user:pass@proxy.example.com:8080"
+ }):
+ config = CrawlerConfig.from_env()
+ assert config.proxy.username == "user"
+ assert config.proxy.password == "pass"
+
+ def test_no_proxy_configured(self):
+ """未配置代理"""
+ with patch.dict(os.environ, {}, clear=True):
+ # 清除所有代理相关环境变量
+ os.environ.pop("HTTP_PROXY", None)
+ os.environ.pop("HTTPS_PROXY", None)
+ config = CrawlerConfig.from_env()
+ assert config.proxy is None
+
+ def test_proxy_required_validation_fails(self):
+ """强制代理但未配置时抛出异常"""
+ config = CrawlerConfig(proxy=None, proxy_required=True)
+ with pytest.raises(ValueError, match="代理配置缺失"):
+ config.validate_proxy()
+
+ def test_proxy_required_validation_passes(self):
+ """配置代理后验证通过"""
+ config = CrawlerConfig(
+ proxy=ProxyConfig(server="http://127.0.0.1:7890"),
+ proxy_required=True
+ )
+ config.validate_proxy() # 不应抛出异常
+
+ def test_proxy_not_required(self):
+ """不强制代理时验证通过"""
+ config = CrawlerConfig(proxy=None, proxy_required=False)
+ config.validate_proxy() # 不应抛出异常
+
+ def test_to_playwright_format(self):
+ """转换为 Playwright 格式"""
+ proxy = ProxyConfig(
+ server="http://127.0.0.1:7890",
+ username="user",
+ password="pass"
+ )
+ result = proxy.to_playwright_format()
+ assert result == {
+ "server": "http://127.0.0.1:7890",
+ "username": "user",
+ "password": "pass"
+ }
+```
+
+
+ pytest tests/test_crawler/test_proxy_config.py -x --tb=short
+
+
+ - CrawlerConfig 类定义完成
+ - 从 HTTP_PROXY/HTTPS_PROXY 读取代理
+ - proxy_required=True 时未配置代理抛出异常
+ - 所有测试通过
+
+
+
+
+ Task 2: 整合代理和延迟到 ZhihuCrawler
+ kb_workflow/crawler/zhihu_crawler.py, tests/test_crawler/test_rate_limiter.py, tests/test_crawler/test_api_intercept.py
+
+ - Test 1: 未配置代理且 proxy_required=True 时抛出异常
+ - Test 2: 请求间有随机延迟 (1-3 秒)
+ - Test 3: 失败请求重试 3 次，间隔递增
+ - Test 4: API 拦截正确处理 x-zse-96 签名（由浏览器自动处理）
+
+
+更新 ZhihuCrawler 类，整合代理配置、随机延迟和重试机制。
+
+**1. 更新 kb_workflow/crawler/zhihu_crawler.py:**
+
+```python
+import random
+import time
+from typing import Optional, Dict, Any
+from playwright.sync_api import Route, Response
+from tenacity import Retrying, stop_after_attempt, wait_exponential, RetryError
+
+from kb_workflow.metadata.zhihu_extractor import PlaywrightBrowser
+from kb_workflow.pipeline.config import get_settings
+from .models import AnswerData, ArticleData, ContentMetadata, AuthorInfo, CrawlResult
+from .config import CrawlerConfig, get_crawler_config
+
+
+class ZhihuCrawler:
+ """知乎内容爬虫
+
+ 支持:
+ - 强制代理配置
+ - 随机请求延迟
+ - 有限重试
+ - 自动处理 x-zse-96 签名
+ """
+
+ # ... URL patterns 等现有代码 ...
+
+ def __init__(
+ self,
+ headless: bool = True,
+ timeout: int = 30,
+ config: Optional[CrawlerConfig] = None
+ ):
+ self.headless = headless
+ self.timeout = timeout
+ self.config = config or get_crawler_config()
+ 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 _apply_random_delay(self) -> None:
+ """应用随机延迟"""
+ delay = random.uniform(self.config.min_delay, self.config.max_delay)
+ time.sleep(delay)
+
+ def _start_browser_with_proxy(self, cookie: str) -> None:
+ """启动浏览器并应用代理配置"""
+ # 验证代理
+ self.config.validate_proxy()
+
+ from playwright.sync_api import sync_playwright
+
+ self.playwright = sync_playwright().start()
+ browser_args = {"headless": self.headless}
+
+ # 应用代理
+ context_args = {
+ 'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
+ }
+
+ if self.config.proxy:
+ context_args["proxy"] = self.config.proxy.to_playwright_format()
+
+ # 阻止 Service Worker 干扰 API 拦截
+ context_args["service_workers"] = "block"
+
+ self.browser = self.playwright.chromium.launch(**browser_args)
+ self.context = self.browser.new_context(**context_args)
+
+ # 添加 Cookie
+ if cookie:
+ self._add_cookies(cookie)
+
+ self.page = self.context.new_page()
+ self.page.set_default_timeout(self.timeout * 1000)
+
+ def _add_cookies(self, cookie_str: str) -> None:
+ """添加 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_browser(self) -> None:
+ """关闭浏览器"""
+ if hasattr(self, 'page') and self.page:
+ self.page.close()
+ if hasattr(self, 'context') and self.context:
+ self.context.close()
+ if hasattr(self, 'browser') and self.browser:
+ self.browser.close()
+ if hasattr(self, 'playwright'):
+ self.playwright.stop()
+
+ def _crawl_with_retry(self, crawl_func, *args, **kwargs):
+ """带重试的爬取"""
+ try:
+ for attempt in Retrying(
+ stop=stop_after_attempt(self.config.max_retries),
+ wait=wait_exponential(
+ multiplier=1,
+ min=self.config.retry_min_wait,
+ max=self.config.retry_max_wait
+ ),
+ reraise=True
+ ):
+ with attempt:
+ # 随机延迟
+ self._apply_random_delay()
+ return crawl_func(*args, **kwargs)
+ except RetryError as e:
+ return CrawlResult(
+ success=False,
+ error=f"重试 {self.config.max_retries} 次后仍失败: {str(e)}"
+ )
+
+ def crawl_answer(self, url: str, include_comments: bool = False) -> CrawlResult:
+ """爬取知乎回答（带重试）"""
+ return self._crawl_with_retry(self._crawl_answer_impl, url, include_comments)
+
+ def _crawl_answer_impl(self, url: str, include_comments: bool = False) -> 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._start_browser_with_proxy(self._get_cookie())
+ self.interceptor = ZhihuApiInterceptor()
+
+ # 设置路由拦截 - x-zse-96 签名由浏览器自动处理
+ self.page.route("**/api/v4/**", self.interceptor.handle_route)
+
+ # 访问页面触发 API 请求
+ self.page.goto(url, wait_until="networkidle")
+
+ # 等待数据捕获
+ 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)
+ )
+
+ result = CrawlResult(success=True, content_type="answer", data=result_data)
+
+ # 可选：爬取评论
+ if include_comments:
+ comments_result = self.crawl_comments(
+ content_type="answers",
+ content_id=int(answer_id)
+ )
+ result.comments = comments_result.comments
+
+ return result
+
+ except Exception as e:
+ return CrawlResult(success=False, error=str(e))
+ finally:
+ self._close_browser()
+
+ def crawl_article(self, url: str, include_comments: bool = False) -> CrawlResult:
+ """爬取知乎专栏文章（带重试）"""
+ return self._crawl_with_retry(self._crawl_article_impl, url, include_comments)
+
+ def _crawl_article_impl(self, url: str, include_comments: bool = False) -> CrawlResult:
+ """爬取知乎专栏文章实现"""
+ # ... 类似 crawl_answer 的实现 ...
+ pass
+
+ # crawl_comments 方法保持不变
+ # ...
+```
+
+**2. 创建 tests/test_crawler/test_rate_limiter.py:**
+
+```python
+import pytest
+import time
+from unittest.mock import Mock, patch, MagicMock
+
+from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+from kb_workflow.crawler.config import CrawlerConfig
+
+
+class TestRateLimiter:
+
+ def test_random_delay_applied(self):
+ """请求间有随机延迟"""
+ config = CrawlerConfig(
+ min_delay=0.1,
+ max_delay=0.2,
+ proxy_required=False
+ )
+ crawler = ZhihuCrawler(config=config)
+
+ start = time.time()
+ crawler._apply_random_delay()
+ elapsed = time.time() - start
+
+ assert 0.1 <= elapsed <= 0.25 # 允许少量误差
+
+ def test_delay_is_random(self):
+ """延迟是随机的"""
+ config = CrawlerConfig(
+ min_delay=0.05,
+ max_delay=0.15,
+ proxy_required=False
+ )
+ crawler = ZhihuCrawler(config=config)
+
+ delays = []
+ for _ in range(10):
+ start = time.time()
+ crawler._apply_random_delay()
+ delays.append(time.time() - start)
+
+ # 延迟应该有变化
+ assert max(delays) - min(delays) > 0.01
+
+ def test_exponential_backoff_retry(self):
+ """重试间隔递增"""
+ config = CrawlerConfig(
+ max_retries=3,
+ retry_min_wait=0.1,
+ retry_max_wait=1.0,
+ proxy_required=False
+ )
+ crawler = ZhihuCrawler(config=config)
+
+ call_times = []
+
+ def failing_func():
+ call_times.append(time.time())
+ raise Exception("Simulated failure")
+
+ result = crawler._crawl_with_retry(failing_func)
+
+ assert result.success is False
+ assert "重试 3 次后仍失败" in result.error
+ assert len(call_times) == 3 # 初始 + 2 次重试
+```
+
+**3. 创建 tests/test_crawler/test_api_intercept.py:**
+
+```python
+import pytest
+from unittest.mock import Mock, MagicMock
+
+from kb_workflow.crawler.zhihu_crawler import ZhihuApiInterceptor
+
+
+class TestApiIntercept:
+
+ def test_intercept_answer_api(self):
+ """拦截答案 API"""
+ interceptor = ZhihuApiInterceptor()
+
+ # 模拟 route 对象
+ route = Mock()
+ route.request.url = "https://www.zhihu.com/api/v4/answers/12345"
+
+ mock_response = Mock()
+ mock_response.json.return_value = {"id": 12345, "content": "test"}
+ route.fetch.return_value = mock_response
+
+ interceptor.handle_route(route)
+
+ assert "answer" in interceptor.captured_data
+ assert interceptor.captured_data["answer"]["id"] == 12345
+
+ def test_intercept_article_api(self):
+ """拦截文章 API"""
+ interceptor = ZhihuApiInterceptor()
+
+ route = Mock()
+ route.request.url = "https://www.zhihu.com/api/v4/articles/67890"
+
+ mock_response = Mock()
+ mock_response.json.return_value = {"id": 67890, "title": "test article"}
+ route.fetch.return_value = mock_response
+
+ interceptor.handle_route(route)
+
+ assert "article" in interceptor.captured_data
+
+ def test_non_api_requests_continue(self):
+ """非 API 请求继续正常加载"""
+ interceptor = ZhihuApiInterceptor()
+
+ route = Mock()
+ route.request.url = "https://www.zhihu.com/static/some.js"
+
+ interceptor.handle_route(route)
+
+ route.continue_.assert_called_once()
+ route.fulfill.assert_not_called()
+
+ def test_x_zse_96_auto_handled(self):
+ """x-zse-96 签名由浏览器自动处理（通过 route.fetch）"""
+ interceptor = ZhihuApiInterceptor()
+
+ route = Mock()
+ route.request.url = "https://www.zhihu.com/api/v4/answers/123"
+
+ interceptor.handle_route(route)
+
+ # fetch() 会自动带上浏览器生成的签名
+ route.fetch.assert_called_once()
+ # fulfill() 返回响应给页面
+ route.fulfill.assert_called_once()
+```
+
+
+ pytest tests/test_crawler/test_rate_limiter.py tests/test_crawler/test_api_intercept.py -x --tb=short
+
+
+ - 代理配置整合到 ZhihuCrawler
+ - 随机延迟 1-3 秒生效
+ - 失败重试 3 次，间隔递增
+ - x-zse-96 签名由浏览器自动处理
+ - 所有测试通过
+
+
+
+
+
+
+**自动化验证:**
+```bash
+pytest tests/test_crawler/ -x -v
+```
+
+**代理配置验证:**
+```bash
+# 不配置代理时应报错
+unset HTTP_PROXY HTTPS_PROXY
+python -c "from kb_workflow.crawler import ZhihuCrawler; ZhihuCrawler().crawl_answer('https://www.zhihu.com/question/1/answer/1')"
+# 应输出: ValueError: 代理配置缺失！
+```
+
+
+
+1. 代理配置从 HTTP_PROXY/HTTPS_PROXY 读取
+2. proxy_required=True 时未配置代理抛出异常
+3. 请求间有随机延迟 (1-3 秒)
+4. 失败请求重试 3 次，间隔递增
+5. x-zse-96 签名由浏览器自动处理
+6. 所有测试通过
+
+
+