` with a real Zhihu URL
+ - Expected: Content is downloaded with comments, metadata saved correctly
+ - Why human: Requires live Zhihu connectivity and valid cookies
+
+2. **QR Code Login Flow**
+ - Test: Trigger login by deleting cookie file and processing a Zhihu URL
+ - Expected: Browser opens with QR code, scanning completes the process
+ - Why human: Requires visual interaction and mobile device for scanning
+
+3. **Proxy Configuration**
+ - Test: Configure proxy and verify requests go through proxy
+ - Expected: All Zhihu requests use configured proxy
+ - Why human: Requires proxy server setup and network monitoring
+
+### Gaps Summary
+
+No critical gaps found. Phase goal achieved.
+
+**Minor observation:** One test in `tests/test_processors/test_download.py` failed due to assertion not matching the updated API signature (source_url=None parameter). This is a test maintenance issue, not a functionality gap. The ZhihuAdapter correctly passes source_url to fetch_metadata as designed.
+
+---
+
+_Verified: 2026-03-17T16:30:00Z_
+_Verifier: Claude (gsd-verifier)_
diff --git a/.planning/state.md b/.planning/state.md
new file mode 100644
index 0000000..bcd506a
--- /dev/null
+++ b/.planning/state.md
@@ -0,0 +1,86 @@
+---
+gsd_state_version: 1.0
+milestone: v1.2
+milestone_name: Zhihu Crawler
+status: planning
+stopped_at: Phase 9 plans created
+last_updated: "2026-03-17T15:00:00Z"
+last_activity: 2026-03-17 - Phase 09 plans created
+
+progress:
+ total_phases: 9
+ completed_phases: 8
+ total_plans: 26
+ completed_plans: 24
+ percent: 100
+---
+
+# Project State
+
+## Project Reference
+
+See: .planning/PROJECT.md (updated 2026-03-17)
+
+**Core value:** 可靠、可维护的处理管道 - 每个步骤独立、可验证、可恢复
+**Current focus:** v1.2 Zhihu Crawler - Phase 9 planning
+
+ progress
+
+## Current Position
+
+Phase: 9 of 9 (Pipeline Integration - 整合)
+Plan: 0 of 02
+Status: Phase 9 plans created, ready to execution
+Last activity: 2026-03-17 - Phase 09 plans created
+
+Progress: [██████████] 100% (v1.2 milestone - Phase 7 complete)
+
+## Performance Metrics
+**Velocity:**
+- Total plans completed: 3 (v1.2)
+- Average duration: 11 min
+- Total execution time: 32 min
+| Plan | Duration | Tasks | Files | Date |
+|------|----------|-------|-------|------|
+| 07-01 | 15min | 2 | 5 | 2026-03-17 |
+| 07-02 | 5min | 2 | 4 | 2026-03-17 |
+| 07-03 | 12min | 3 | 4 | 2026-03-17 |
+| Phase 08 P01 | 3min | 2 tasks | 8 files |
+| Phase 08-crawler P02 | 5min | 2 tasks | 3 files |
+| Phase 08 P03 | 15min | 2 tasks | 5 files |
+## Accumulated Context
+### Decisions
+Key decisions from v1.0 (preserved):
+- 管道模型 - BaseProcessor + Pipeline 架构
+- 步骤级状态 - WorkflowStateManager
+- Schema 验证 - Pydantic v2
+- 顺序执行 - 简单性和可调试性
+- CLI subparsers - 可扩展命令结构
+v1.1 decisions:
+- 中间目录统一到 `~/work/intermediate/analysis/`
+- 归档只保留 md 文件,不归档 json 文件
+- 最终目录按平台分组,不再按时间创建子目录
+v1.2 decisions:
+- 知乎登录策略: 二维码扫码 + 手动 Cookie 降级
+- 参考 MediaCrawler 实现,自实现而非直接依赖
+- 使用 playwright 作为浏览器自动化框架
+- 整合进现有 Pipeline 架构
+- Cookie 文件权限设置为 600 保证安全性
+- 保存 Cookie 后清除 get_settings 缓存确保配置刷新
+- 使用 tenacity Retrying 对象而非 @retry 装饰器支持可配置超时
+- 多个 canvas 选择器降级提高 QR 码检测鲁棒性
+- login CLI 命令支持三种模式: 扫码登录、导入 Cookie、 状态检查
+- [Phase 08-02]: use Playwright request API for direct comment API calls instead of route interception
+- [Phase ?]: - [Phase 08-02]: only fetch first-level comments (no nested replies)
+- [Phase ?]: - [Phase 08-02]: use hot sorting (order=hot) for popular comments
+- [Phase 09-01]: 修改 ZhihuAdapter 内部使用 ZhihuCrawler, 保持接口不变
+- [Phase 09-01]: Cookie 过期自动触发扫码登录
+- [Phase 09-01]: 评论追加到正文末尾 (Markdown 格式)
+- [Phase 09-01]: 直接报错不降级到旧版 ZhihuPlaywrightExtractor
+### Blockers/Concerns
+None.
+## Session Continuity
+Last session: 2026-03-17T15:00:00Z
+Stopped at: Phase 9 plans created
+resume file: .planning/phases/09-pipeline-integration/09-CONTEXT.md
+**Next step:** Execute `/gsd:execute-phase 09`
diff --git a/kb_workflow/auth/__init__.py b/kb_workflow/auth/__init__.py
new file mode 100644
index 0000000..0fb159f
--- /dev/null
+++ b/kb_workflow/auth/__init__.py
@@ -0,0 +1,8 @@
+"""Auth module for KB Workflow
+
+Provides authentication and cookie management functionality.
+"""
+
+from .zhihu_login import ZhihuLogin, ZhihuLoginResult
+
+__all__ = ["ZhihuLogin", "ZhihuLoginResult"]
diff --git a/kb_workflow/auth/cookie_manager.py b/kb_workflow/auth/cookie_manager.py
new file mode 100644
index 0000000..b9927bf
--- /dev/null
+++ b/kb_workflow/auth/cookie_manager.py
@@ -0,0 +1,120 @@
+"""Cookie 管理模块
+
+处理 Cookie 的导入、转换和持久化。
+"""
+
+import json
+from pathlib import Path
+from typing import List, Dict, Any
+
+
+def parse_json_cookies(json_path: str) -> str:
+ """从 JSON 文件解析 Cookie 为字符串格式
+
+ 支持 Chrome Cookie Editor 导出的 JSON 格式:
+ [{"name": "z_c0", "value": "xxx", "domain": ".zhihu.com"}, ...]
+
+ Args:
+ json_path: JSON 文件路径
+
+ Returns:
+ Cookie 字符串，格式: "name1=value1; name2=value2"
+
+ Raises:
+ ValueError: JSON 格式无效或文件不存在
+ """
+ path = Path(json_path)
+ if not path.exists():
+ raise ValueError(f"Cookie file not found: {json_path}")
+
+ try:
+ with open(path, 'r', encoding='utf-8') as f:
+ cookies = json.load(f)
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Invalid JSON format: {e}")
+
+ if not isinstance(cookies, list):
+ raise ValueError("Expected JSON array")
+
+ # 过滤知乎域名，转换为 name=value 格式
+ zhihu_cookies = []
+ for c in cookies:
+ if not isinstance(c, dict):
+ continue
+ domain = c.get('domain', '')
+ if 'zhihu.com' in domain:
+ name = c.get('name', '')
+ value = c.get('value', '')
+ if name and value:
+ zhihu_cookies.append(f"{name}={value}")
+
+ return '; '.join(zhihu_cookies)
+
+
+class CookieManager:
+ """Cookie 持久化管理器
+
+ 管理 Cookie 在 ~/.kb_env 文件中的存储和读取。
+ """
+
+ ENV_FILE_NAME = ".kb_env"
+ COOKIE_KEY = "ZHIHU_COOKIE"
+
+ def __init__(self, env_path: Path = None):
+ """初始化管理器
+
+ Args:
+ env_path: 环境文件路径，默认 ~/.kb_env
+ """
+ self.env_path = env_path or (Path.home() / self.ENV_FILE_NAME)
+
+ def save_cookie(self, cookie_str: str) -> None:
+ """保存 Cookie 到环境文件
+
+ 格式: ZHIHU_COOKIE="name1=value1; name2=value2"
+
+ Args:
+ cookie_str: Cookie 字符串
+ """
+ # 读取现有内容
+ lines = []
+ if self.env_path.exists():
+ lines = self.env_path.read_text(encoding='utf-8').splitlines()
+
+ # 更新或添加 ZHIHU_COOKIE 行
+ cookie_line = f'{self.COOKIE_KEY}="{cookie_str}"'
+ found = False
+ for i, line in enumerate(lines):
+ if line.startswith(f"{self.COOKIE_KEY}="):
+ lines[i] = cookie_line
+ found = True
+ break
+ if not found:
+ lines.append(cookie_line)
+
+ # 写回文件，设置权限 600
+ self.env_path.write_text('\n'.join(lines) + '\n', encoding='utf-8')
+ self.env_path.chmod(0o600)
+
+ # 清除配置缓存
+ from kb_workflow.pipeline.config import get_settings
+ get_settings.cache_clear()
+
+ def load_cookie(self) -> str:
+ """从环境文件加载 Cookie
+
+ Returns:
+ Cookie 字符串，不存在时返回空字符串
+ """
+ if not self.env_path.exists():
+ return ""
+
+ content = self.env_path.read_text(encoding='utf-8')
+ for line in content.splitlines():
+ if line.startswith(f"{self.COOKIE_KEY}="):
+ # 提取引号内的值
+ value = line[len(f"{self.COOKIE_KEY}="):]
+ if value.startswith('"') and value.endswith('"'):
+ value = value[1:-1]
+ return value
+ return ""
diff --git a/kb_workflow/auth/zhihu_login.py b/kb_workflow/auth/zhihu_login.py
new file mode 100644
index 0000000..45d2e52
--- /dev/null
+++ b/kb_workflow/auth/zhihu_login.py
@@ -0,0 +1,276 @@
+"""Zhihu Login via QR Code
+
+Implements QR code login flow for Zhihu:
+1. Navigate to login page
+2. Capture QR code from canvas element
+3. Poll for login cookie (z_c0)
+4. Extract cookie string on success
+
+Reference: MediaCrawler implementation pattern
+"""
+
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, List
+
+from pydantic import BaseModel, Field
+from tenacity import Retrying, stop_after_delay, wait_fixed, RetryError
+
+from kb_workflow.metadata.zhihu_extractor import PlaywrightBrowser
+
+
+class ZhihuLoginResult(BaseModel):
+ """Result of QR code login attempt.
+
+ Attributes:
+ success: Whether login was successful
+ cookie_str: Cookie string if login succeeded, empty otherwise
+ qrcode_path: Path to saved QR code image
+ error_message: Error message if login failed
+ """
+ success: bool = Field(..., description="Whether login was successful")
+ cookie_str: str = Field(default="", description="Cookie string")
+ qrcode_path: Optional[Path] = Field(None, description="Path to saved QR code image")
+ error_message: Optional[str] = Field(None, description="Error message if failed")
+
+
+class CookieValidationResult(BaseModel):
+ """Cookie validation result.
+
+ Attributes:
+ valid: Whether the cookie is valid
+ username: User's display name if valid
+ user_id: User's ID if valid
+ reason: Reason for invalid cookie
+ """
+ valid: bool = Field(..., description="Whether cookie is valid")
+ username: Optional[str] = Field(None, description="User display name")
+ user_id: Optional[str] = Field(None, description="User ID")
+ reason: Optional[str] = Field(None, description="Reason if invalid")
+
+
+class ZhihuLogin:
+ """Zhihu QR Code Login Handler.
+
+ Handles the QR code login flow for Zhihu:
+ - Navigate to login page
+ - Capture QR code screenshot
+ - Poll for login cookie
+ - Extract cookie string
+
+ Attributes:
+ LOGIN_URL: Zhihu login page URL
+ CANVAS_SELECTORS: List of selectors to find QR code canvas
+ """
+
+ LOGIN_URL = "https://www.zhihu.com/signin"
+ CANVAS_SELECTORS = [
+ "canvas.SignIn-qrcode",
+ "canvas[class*='qrcode']",
+ "canvas"
+ ]
+ VALIDATION_URL = "https://www.zhihu.com/api/v4/me"
+
+ def __init__(
+ self,
+ intermediate_path: str,
+ headless: bool = True,
+ timeout: int = 60
+ ):
+ """Initialize ZhihuLogin.
+
+ Args:
+ intermediate_path: Path to intermediate directory for QR code storage
+ headless: Whether to run browser in headless mode
+ timeout: Login timeout in seconds
+ """
+ self.intermediate_path = Path(intermediate_path)
+ self.headless = headless
+ self.timeout = timeout
+ self.browser: Optional[PlaywrightBrowser] = None
+
+ def _ensure_intermediate_dir(self) -> None:
+ """Ensure intermediate directory exists."""
+ self.intermediate_path.mkdir(parents=True, exist_ok=True)
+
+ def _find_qrcode_canvas(self):
+ """Find QR code canvas element.
+
+ Tries multiple selectors to find the QR code canvas.
+
+ Returns:
+ Canvas element handle or None
+ """
+ for selector in self.CANVAS_SELECTORS:
+ try:
+ canvas = self.browser.wait_for_selector(selector, timeout=5)
+ if canvas:
+ return canvas
+ except Exception:
+ continue
+ return None
+
+ def _save_qrcode_image(self, screenshot: bytes) -> Path:
+ """Save QR code screenshot to intermediate directory.
+
+ Args:
+ screenshot: PNG bytes from canvas screenshot
+
+ Returns:
+ Path to saved QR code image
+ """
+ self._ensure_intermediate_dir()
+
+ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+ filename = f"zhihu_qrcode_{timestamp}.png"
+ filepath = self.intermediate_path / filename
+
+ filepath.write_bytes(screenshot)
+ return filepath
+
+ def check_login_cookie(self) -> bool:
+ """Check if login cookie (z_c0) exists.
+
+ Returns:
+ True if z_c0 cookie exists, False otherwise
+ """
+ cookies = self.browser.context.cookies()
+ for cookie in cookies:
+ if cookie.get("name") == "z_c0":
+ return True
+ return False
+
+ def get_cookies_str(self) -> str:
+ """Get cookie string from browser context.
+
+ Returns:
+ Cookie string in format "name1=value1; name2=value2"
+ """
+ cookies = self.browser.context.cookies()
+ cookie_pairs = [
+ f"{cookie['name']}={cookie['value']}"
+ for cookie in cookies
+ ]
+ return "; ".join(cookie_pairs)
+
+ def _wait_for_login(self) -> None:
+ """Wait for user to scan QR code and login.
+
+ Uses tenacity Retrying to poll for login cookie with configurable timeout.
+
+ Raises:
+ RetryError: If login timeout exceeded
+ """
+ for attempt in Retrying(
+ stop=stop_after_delay(self.timeout),
+ wait=wait_fixed(1),
+ reraise=True
+ ):
+ with attempt:
+ if not self.check_login_cookie():
+ raise Exception("Not logged in yet")
+
+ def login_by_qrcode(self) -> ZhihuLoginResult:
+ """Execute QR code login flow.
+
+ Main entry point for QR code login:
+ 1. Start browser
+ 2. Navigate to login page
+ 3. Find and capture QR code
+ 4. Wait for user to scan
+ 5. Extract cookies on success
+
+ Returns:
+ ZhihuLoginResult with success status and cookies
+ """
+ qrcode_path = None
+
+ try:
+ # Start browser
+ self.browser = PlaywrightBrowser(headless=self.headless, timeout=30)
+ self.browser.start()
+
+ # Navigate to login page
+ self.browser.goto(self.LOGIN_URL, wait_until="networkidle")
+
+ # Find QR code canvas
+ canvas = self._find_qrcode_canvas()
+ if not canvas:
+ return ZhihuLoginResult(
+ success=False,
+ error_message="Could not find QR code canvas on login page"
+ )
+
+ # Capture screenshot
+ screenshot = canvas.screenshot()
+ qrcode_path = self._save_qrcode_image(screenshot)
+
+ # Wait for login
+ try:
+ self._wait_for_login()
+ except RetryError:
+ return ZhihuLoginResult(
+ success=False,
+ qrcode_path=qrcode_path,
+ error_message=f"Login timeout after {self.timeout} seconds"
+ )
+
+ # Extract cookies
+ cookie_str = self.get_cookies_str()
+
+ return ZhihuLoginResult(
+ success=True,
+ cookie_str=cookie_str,
+ qrcode_path=qrcode_path
+ )
+
+ except Exception as e:
+ return ZhihuLoginResult(
+ success=False,
+ qrcode_path=qrcode_path,
+ error_message=str(e)
+ )
+
+ finally:
+ if self.browser:
+ self.browser.close()
+
+ async def validate_cookie(self, cookie_str: str = None) -> CookieValidationResult:
+ """Validate cookie validity.
+
+ Checks cookie by requesting Zhihu user info API.
+
+ Args:
+ cookie_str: Cookie string to validate. If None, uses already loaded cookie.
+
+ Returns:
+ CookieValidationResult with validation status and user info
+ """
+ # If cookie_str provided, need to restart browser with new cookie
+ if cookie_str:
+ if self.browser.browser:
+ self.browser.close()
+ self.browser.start(cookie=cookie_str)
+ elif not self.browser.page:
+ return CookieValidationResult(valid=False, reason="No cookie provided")
+
+ try:
+ # Use Playwright's request API
+ response = await self.browser.page.request.get(self.VALIDATION_URL)
+
+ if response.status == 200:
+ user_info = await response.json()
+ return CookieValidationResult(
+ valid=True,
+ username=user_info.get("name", ""),
+ user_id=str(user_info.get("id", ""))
+ )
+ elif response.status == 401:
+ return CookieValidationResult(valid=False, reason="Cookie expired")
+ else:
+ return CookieValidationResult(
+ valid=False,
+ reason=f"Validation failed with status {response.status}"
+ )
+ except Exception as e:
+ return CookieValidationResult(valid=False, reason=str(e))
diff --git a/kb_workflow/crawler/__init__.py b/kb_workflow/crawler/__init__.py
new file mode 100644
index 0000000..8389733
--- /dev/null
+++ b/kb_workflow/crawler/__init__.py
@@ -0,0 +1,39 @@
+"""Zhihu Crawler Module
+
+Provides content crawling capabilities for Zhihu using Playwright API interception.
+"""
+
+from .models import (
+ AuthorInfo,
+ ContentMetadata,
+ AnswerData,
+ ArticleData,
+ CommentAuthor,
+ CommentData,
+ CommentsResult,
+ CrawlResult,
+)
+from .config import (
+ ProxyConfig,
+ CrawlerConfig,
+ get_crawler_config,
+ reset_crawler_config,
+)
+from .zhihu_crawler import ZhihuCrawler, ZhihuApiInterceptor
+
+__all__ = [
+ "AuthorInfo",
+ "ContentMetadata",
+ "AnswerData",
+ "ArticleData",
+ "CommentAuthor",
+ "CommentData",
+ "CommentsResult",
+ "CrawlResult",
+ "ProxyConfig",
+ "CrawlerConfig",
+ "get_crawler_config",
+ "reset_crawler_config",
+ "ZhihuCrawler",
+ "ZhihuApiInterceptor",
+]
diff --git a/kb_workflow/crawler/config.py b/kb_workflow/crawler/config.py
new file mode 100644
index 0000000..cdb0630
--- /dev/null
+++ b/kb_workflow/crawler/config.py
@@ -0,0 +1,141 @@
+"""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):
+ """Proxy configuration for crawler requests.
+
+ Attributes:
+ server: Proxy server address (e.g., http://127.0.0.1:7890)
+ username: Optional proxy username for authentication
+ password: Optional proxy password for authentication
+ """
+
+ server: str = Field(..., description="Proxy server address")
+ username: Optional[str] = Field(None, description="Proxy username")
+ password: Optional[str] = Field(None, description="Proxy password")
+
+ def to_playwright_format(self) -> Dict[str, Any]:
+ """Convert to Playwright proxy format.
+
+ Returns:
+ Dictionary compatible with browser.new_context(proxy=...)
+ """
+ config = {"server": self.server}
+ if self.username:
+ config["username"] = self.username
+ if self.password:
+ config["password"] = self.password
+ return config
+
+
+class CrawlerConfig(BaseModel):
+ """Crawler configuration.
+
+ Handles proxy settings, rate limiting, and retry behavior.
+
+ Attributes:
+ proxy: Optional proxy configuration
+ proxy_required: Whether proxy is mandatory (raises error if not configured)
+ min_delay: Minimum delay between requests (seconds)
+ max_delay: Maximum delay between requests (seconds)
+ max_retries: Maximum number of retry attempts
+ retry_min_wait: Minimum wait time between retries (seconds)
+ retry_max_wait: Maximum wait time between retries (seconds)
+ """
+
+ # Proxy settings
+ proxy: Optional[ProxyConfig] = Field(None, description="Proxy configuration")
+ proxy_required: bool = Field(default=True, description="Whether proxy is mandatory")
+
+ # Rate limiting
+ min_delay: float = Field(default=1.0, description="Minimum delay between requests (seconds)")
+ max_delay: float = Field(default=3.0, description="Maximum delay between requests (seconds)")
+
+ # Retry settings
+ max_retries: int = Field(default=3, description="Maximum retry attempts")
+ retry_min_wait: float = Field(default=2.0, description="Minimum retry wait (seconds)")
+ retry_max_wait: float = Field(default=10.0, description="Maximum retry wait (seconds)")
+
+ @classmethod
+ def from_env(cls) -> "CrawlerConfig":
+ """Load configuration from environment variables.
+
+ Reads HTTP_PROXY or HTTPS_PROXY for proxy settings.
+
+ Returns:
+ CrawlerConfig instance with proxy settings from environment
+ """
+ proxy = cls._parse_proxy_env()
+ return cls(proxy=proxy)
+
+ @staticmethod
+ def _parse_proxy_env() -> Optional[ProxyConfig]:
+ """Parse proxy configuration from environment variables.
+
+ Checks HTTPS_PROXY first, then HTTP_PROXY.
+
+ Returns:
+ ProxyConfig if proxy URL found, None otherwise
+ """
+ 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:
+ """Validate proxy configuration.
+
+ Raises:
+ ValueError: If proxy_required=True but no proxy is configured
+ """
+ if self.proxy_required and not self.proxy:
+ raise ValueError(
+ "Proxy configuration missing! Please set HTTP_PROXY or HTTPS_PROXY environment variable.\n"
+ "Example: export HTTPS_PROXY=http://127.0.0.1:7890"
+ )
+
+
+# Module-level cached config
+_crawler_config: Optional[CrawlerConfig] = None
+
+
+def get_crawler_config() -> CrawlerConfig:
+ """Get crawler configuration singleton.
+
+ Returns:
+ CrawlerConfig instance loaded from environment variables
+ """
+ global _crawler_config
+ if _crawler_config is None:
+ _crawler_config = CrawlerConfig.from_env()
+ return _crawler_config
+
+
+def reset_crawler_config() -> None:
+ """Reset the cached crawler configuration.
+
+ Useful when environment variables change during runtime.
+ """
+ global _crawler_config
+ _crawler_config = None
diff --git a/kb_workflow/crawler/models.py b/kb_workflow/crawler/models.py
new file mode 100644
index 0000000..00f52ca
--- /dev/null
+++ b/kb_workflow/crawler/models.py
@@ -0,0 +1,152 @@
+"""Zhihu Crawler Data Models
+
+Pydantic models for structured data from Zhihu API responses.
+"""
+
+from typing import Optional, Union, List
+from pydantic import BaseModel, Field
+
+
+class AuthorInfo(BaseModel):
+ """Author information from Zhihu content.
+
+ Attributes:
+ name: Author display name
+ url_token: Author URL token (e.g., 'zhang-san-12')
+ headline: Author bio/headline
+ """
+ name: str = Field(..., description="Author display name")
+ url_token: Optional[str] = Field(None, description="Author URL token")
+ headline: Optional[str] = Field(None, description="Author bio/headline")
+
+
+class ContentMetadata(BaseModel):
+ """Metadata for Zhihu content (answers, articles, etc).
+
+ Attributes:
+ author: Author information
+ voteup_count: Number of upvotes/likes
+ comment_count: Number of comments
+ created_time: Unix timestamp of creation
+ updated_time: Unix timestamp of last update
+ """
+ author: Optional[AuthorInfo] = Field(None, description="Author info")
+ voteup_count: int = Field(default=0, description="Upvote count")
+ comment_count: int = Field(default=0, description="Comment count")
+ created_time: Optional[int] = Field(None, description="Creation timestamp")
+ updated_time: Optional[int] = Field(None, description="Update timestamp")
+
+
+class AnswerData(BaseModel):
+ """Data model for Zhihu answer.
+
+ Attributes:
+ id: Answer ID
+ question_id: Question ID this answer belongs to
+ question_title: Title of the question
+ content: Answer content in HTML format
+ excerpt: Plain text excerpt/summary
+ metadata: Additional metadata (author, votes, etc.)
+ """
+ id: int = Field(..., description="Answer ID")
+ question_id: int = Field(..., description="Question ID")
+ question_title: str = Field(..., description="Question title")
+ content: str = Field(default="", description="Answer content (HTML)")
+ excerpt: str = Field(default="", description="Plain text excerpt")
+ metadata: ContentMetadata = Field(
+ default_factory=ContentMetadata,
+ description="Content metadata"
+ )
+
+
+class ArticleData(BaseModel):
+ """Data model for Zhihu zhuanlan article.
+
+ Attributes:
+ id: Article ID
+ title: Article title
+ content: Article content in HTML format
+ excerpt: Plain text excerpt/summary
+ metadata: Additional metadata (author, votes, etc.)
+ """
+ id: int = Field(..., description="Article ID")
+ title: str = Field(..., description="Article title")
+ content: str = Field(default="", description="Article content (HTML)")
+ excerpt: str = Field(default="", description="Plain text excerpt")
+ metadata: ContentMetadata = Field(
+ default_factory=ContentMetadata,
+ description="Content metadata"
+ )
+
+
+class CommentAuthor(BaseModel):
+ """Comment author information.
+
+ Attributes:
+ name: Author display name
+ url_token: Author URL token (e.g., 'zhang-san-12')
+ """
+ name: str = Field(..., description="Author display name")
+ url_token: Optional[str] = Field(None, description="Author URL token")
+
+
+class CommentData(BaseModel):
+ """Data model for Zhihu comment.
+
+ Attributes:
+ id: Comment ID
+ content: Comment content in plain text
+ author: Comment author information
+ created_time: Unix timestamp of creation
+ like_count: Number of likes/upvotes
+ child_count: Number of child/reply comments
+ is_author: Whether the commenter is the content author
+ """
+ id: int = Field(..., description="Comment ID")
+ content: str = Field(default="", description="Comment content")
+ author: Optional[CommentAuthor] = Field(None, description="Comment author")
+ created_time: Optional[int] = Field(None, description="Creation timestamp")
+ like_count: int = Field(default=0, description="Like/upvote count")
+ child_count: int = Field(default=0, description="Number of child comments")
+ is_author: bool = Field(default=False, description="Whether commenter is content author")
+
+
+class CommentsResult(BaseModel):
+ """Result of comment crawling operation.
+
+ Attributes:
+ content_type: Type of content ("answers" or "articles")
+ content_id: ID of the content
+ comments: List of comments
+ total_count: Total number of comments (may exceed returned count)
+ """
+ content_type: str = Field(..., description="Content type: answers or articles")
+ content_id: int = Field(..., description="Content ID")
+ comments: List[CommentData] = Field(
+ default_factory=list,
+ description="List of comments"
+ )
+ total_count: int = Field(default=0, description="Total comment count")
+
+
+class CrawlResult(BaseModel):
+ """Result of a crawl operation.
+
+ Attributes:
+ success: Whether the crawl was successful
+ content_type: Type of content ("answer" or "article")
+ data: Extracted content data if successful
+ comments: Optional list of comments (when include_comments=True)
+ error: Error message if failed
+ """
+ success: bool = Field(..., description="Whether crawl succeeded")
+ content_type: str = Field(default="", description="Content type: answer or article")
+ data: Optional[Union[AnswerData, ArticleData]] = Field(
+ None,
+ description="Extracted content data"
+ )
+ comments: Optional[List[CommentData]] = Field(
+ None,
+ description="Comments if include_comments was True"
+ )
+ error: Optional[str] = Field(None, description="Error message if failed")
diff --git a/kb_workflow/crawler/zhihu_crawler.py b/kb_workflow/crawler/zhihu_crawler.py
new file mode 100644
index 0000000..b424da4
--- /dev/null
+++ b/kb_workflow/crawler/zhihu_crawler.py
@@ -0,0 +1,590 @@
+"""Zhihu Content Crawler
+
+Implements content crawling using Playwright API interception.
+Captures structured JSON data from Zhihu API responses.
+
+Supports:
+- Mandatory proxy configuration
+- Random request delay
+- Retry with exponential backoff
+- Automatic x-zse-96 signature handling via browser
+"""
+
+import random
+import re
+import time
+from typing import Optional, Dict, Any
+
+from playwright.sync_api import Route, Response, sync_playwright
+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,
+ CommentData,
+ CommentAuthor,
+ CommentsResult,
+ CrawlResult,
+)
+from .config import CrawlerConfig, get_crawler_config
+
+
+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).
+ The x-zse-96 signature is automatically handled by the browser via route.fetch().
+
+ 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.
+
+ The route.fetch() call automatically includes browser-generated
+ x-zse-96 signature headers, No manual signature handling needed.
+
+ 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.
+ Includes anti-scraping measures:
+ - Mandatory proxy configuration (configurable via proxy_required)
+ - Random delay between requests (1-3 seconds by default)
+ - Retry with exponential backoff (3 attempts by default)
+ - Automatic x-zse-96 signature handling via browser
+
+ 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,
+ config: Optional[CrawlerConfig] = None
+ ):
+ """Initialize ZhihuCrawler.
+
+ Args:
+ headless: Whether to run browser in headless mode
+ timeout: Request timeout in seconds
+ config: Optional CrawlerConfig instance (uses environment if not provided)
+ """
+ self.headless = headless
+ self.timeout = timeout
+ self.config = config or get_crawler_config()
+ self.browser: Optional[PlaywrightBrowser] = None
+ self.interceptor: Optional[ZhihuApiInterceptor] = None
+ # Internal browser management for proxy support
+ self._playwright = None
+ self._browser = None
+ self._context = None
+ self._page = 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 _apply_random_delay(self) -> None:
+ """Apply random delay between requests.
+
+ Uses min_delay and max_delay from configuration.
+ """
+ delay = random.uniform(self.config.min_delay, self.config.max_delay)
+ time.sleep(delay)
+
+ def _start_browser_with_proxy(self, cookie: str) -> None:
+ """Start browser with proxy configuration.
+
+ Validates proxy if required and applies to browser context.
+
+ Args:
+ cookie: Cookie string for authentication
+
+ Raises:
+ ValueError: If proxy_required=True but no proxy is configured
+ """
+ # Validate proxy configuration
+ self.config.validate_proxy()
+
+ self._playwright = sync_playwright().start()
+ self._browser = self._playwright.chromium.launch(headless=self.headless)
+
+ # Build context options
+ context_options = {
+ 'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
+ }
+
+ # Apply proxy if configured
+ if self.config.proxy:
+ context_options["proxy"] = self.config.proxy.to_playwright_format()
+
+ # Block Service Worker to prevent interference with API interception
+ context_options["service_workers"] = "block"
+
+ self._context = self._browser.new_context(**context_options)
+
+ # Add cookies if provided
+ 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:
+ """Add cookies to browser context.
+
+ Args:
+ cookie_str: Cookie string in "name=value; name2=value2" format
+ """
+ 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:
+ """Close browser and cleanup resources."""
+ if self._page:
+ self._page.close()
+ if self._context:
+ self._context.close()
+ if self._browser:
+ self._browser.close()
+ if self._playwright:
+ self._playwright.stop()
+
+ def _crawl_with_retry(self, crawl_func, *args, **kwargs) -> CrawlResult:
+ """Execute crawl with retry logic.
+
+ Applies random delay before each attempt and retries on failure
+ with exponential backoff.
+
+ Args:
+ crawl_func: Crawl function to execute
+ *args: Positional arguments for crawl_func
+ **kwargs: Keyword arguments for crawl_func
+
+ Returns:
+ CrawlResult from crawl_func or error result after retries exhausted
+ """
+ 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:
+ # Apply random delay before each attempt
+ self._apply_random_delay()
+ return crawl_func(*args, **kwargs)
+ except RetryError as e:
+ return CrawlResult(
+ success=False,
+ error=f"Failed after {self.config.max_retries} retries: {str(e)}"
+ )
+ except Exception as e:
+ return CrawlResult(success=False, error=str(e))
+
+ 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._start_browser_with_proxy(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._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:
+ self._close_browser()
+
+ def crawl_answer(self, url: str, include_comments: bool = False) -> CrawlResult:
+ """Crawl Zhihu answer with retry support.
+
+ 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.
+ """
+ return self._crawl_with_retry(
+ self._crawl_answer_impl, url, include_comments
+ )
+
+ def _crawl_answer_impl(self, url: str, include_comments: bool = False) -> CrawlResult:
+ """Implementation of crawl_answer without retry wrapper.
+
+ Args:
+ url: Zhihu answer URL
+ include_comments: Whether to include hot comments
+
+ Returns:
+ CrawlResult
+ """
+ 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._start_browser_with_proxy(self._get_cookie())
+ self.interceptor = ZhihuApiInterceptor()
+
+ # Set up route interception - x-zse-96 handled by browser
+ self._page.route("**/api/v4/**", self.interceptor.handle_route)
+
+ # Visit page to trigger API requests
+ self._page.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:
+ 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:
+ self._close_browser()
+
+ def crawl_article(self, url: str) -> CrawlResult:
+ """Crawl Zhihu zhuanlan article with retry support.
+
+ Args:
+ url: Zhihu zhuanlan URL (e.g., https://zhuanlan.zhihu.com/p/123)
+
+ Returns:
+ CrawlResult with ArticleData if successful, error message otherwise
+ """
+ return self._crawl_with_retry(self._crawl_article_impl, url)
+
+ def _crawl_article_impl(self, url: str) -> CrawlResult:
+ """Implementation of crawl_article without retry wrapper.
+
+ Args:
+ url: Zhihu zhuanlan URL
+
+ Returns:
+ CrawlResult
+ """
+ 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._start_browser_with_proxy(self._get_cookie())
+ self.interceptor = ZhihuApiInterceptor()
+
+ self._page.route("**/api/v4/**", self.interceptor.handle_route)
+
+ self._page.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:
+ self._close_browser()
+
+ 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)
diff --git a/kb_workflow/kb_workflow.py b/kb_workflow/kb_workflow.py
index c5fda0f..c89108c 100644
--- a/kb_workflow/kb_workflow.py
+++ b/kb_workflow/kb_workflow.py
@@ -11,6 +11,9 @@ KB Workflow - Knowledge Base Workflow Orchestrator
使用方式:
python kb_workflow.py process "https://www.bilibili.com/video/BVxxx"
python kb_workflow.py process "https://..." -o /path/to/output
+ python kb_workflow.py login # QR code login
+ python kb_workflow.py login --import-cookie cookies.json
+ python kb_workflow.py login --status
"""
import argparse
@@ -102,6 +105,68 @@ def process_command(
}
+def login_command(
+ import_cookie: Optional[str] = None,
+ status: bool = False
+) -> Dict[str, Any]:
+ """Execute login command
+
+ Args:
+ import_cookie: Path to JSON cookie file
+ status: Check current cookie status
+
+ Returns:
+ Dictionary with login result
+ """
+ try:
+ from kb_workflow.auth import ZhihuLogin
+ from kb_workflow.auth.cookie_manager import CookieManager, parse_json_cookies
+ from kb_workflow.pipeline.config import get_settings
+
+ settings = get_settings()
+ manager = CookieManager()
+
+ # Check Cookie status
+ if status:
+ current_cookie = settings.zhihu_cookie.get_secret_value()
+ if not current_cookie:
+ return {"valid": False, "reason": "No cookie found in ~/.kb_env"}
+
+ async def check_status():
+ login = ZhihuLogin(intermediate_path=settings.intermediate_path)
+ try:
+ login.browser = login.browser or type('obj', (object,), {
+ 'browser': None,
+ 'page': None,
+ 'start': lambda cookie=None: None,
+ 'close': lambda: None
+ })()
+ login.browser.start(cookie=current_cookie)
+ return await login.validate_cookie(current_cookie)
+ finally:
+ if login.browser and hasattr(login.browser, 'close'):
+ login.browser.close()
+
+ result = asyncio.run(check_status())
+ return result.model_dump()
+
+ # Import Cookie
+ if import_cookie:
+ cookie_str = parse_json_cookies(import_cookie)
+ manager.save_cookie(cookie_str)
+ return {"success": True, "message": f"Cookie imported from {import_cookie}"}
+
+ # QR code login (synchronous method)
+ login = ZhihuLogin(intermediate_path=settings.intermediate_path)
+ result = login.login_by_qrcode()
+ if result.success and result.cookie_str:
+ manager.save_cookie(result.cookie_str)
+ return result.model_dump()
+
+ except Exception as e:
+ return {"success": False, "error": str(e)}
+
+
def main():
"""CLI入口 - 用于命令行调用"""
parser = argparse.ArgumentParser(
@@ -117,6 +182,15 @@ def main():
# 强制重新执行（忽略已完成状态）
python kb_workflow.py process "https://..." --no-skip
+
+ # 知乎登录 (二维码扫码)
+ python kb_workflow.py login
+
+ # 导入 Cookie
+ python kb_workflow.py login --import-cookie cookies.json
+
+ # 检查 Cookie 状态
+ python kb_workflow.py login --status
'''
)
@@ -132,6 +206,13 @@ def main():
process_parser.add_argument('--no-skip', action='store_false', dest='skip_completed',
help='Re-run all steps')
+ # Login subcommand
+ login_parser = subparsers.add_parser('login', help='Zhihu login authentication')
+ login_parser.add_argument('--import-cookie', '-i', type=str,
+ help='Import cookie from JSON file')
+ login_parser.add_argument('--status', action='store_true',
+ help='Check current cookie status')
+
args = parser.parse_args()
# Handle process subcommand
@@ -142,6 +223,13 @@ def main():
skip_completed=args.skip_completed
)
print(json.dumps(result, ensure_ascii=False, indent=2))
+ # Handle login subcommand
+ elif args.command == 'login':
+ result = login_command(
+ import_cookie=args.import_cookie,
+ status=args.status
+ )
+ print(json.dumps(result, ensure_ascii=False, indent=2))
else:
parser.print_help()
diff --git a/kb_workflow/pipeline/platforms/zhihu.py b/kb_workflow/pipeline/platforms/zhihu.py
index 013644a..26c5f45 100644
--- a/kb_workflow/pipeline/platforms/zhihu.py
+++ b/kb_workflow/pipeline/platforms/zhihu.py
@@ -3,26 +3,44 @@
Provides content extraction for Zhihu platform:
- Answer extraction from question/answer URLs
- Zhuanlan (article) extraction from zhuanlan URLs
-- Cookie-based authentication
-- Login page detection for expired cookies
+- Cookie-based authentication with auto-login
+- Comment crawling and formatting
+- Idempotent execution
+
+Phase 09 - Pipeline Integration:
+- Uses ZhihuCrawler instead of ZhihuPlaywrightExtractor
+- Comments included by default and appended to content
+- Cookie expiration triggers QR code login
"""
-from typing import Dict, Any, Optional
+from typing import Dict, Any, Optional, List
from datetime import datetime
+from pathlib import Path
import re
import asyncio
+import html2text
from kb_workflow.pipeline.platforms.base import PlatformAdapter
from kb_workflow.pipeline.errors import AuthError
from kb_workflow.pipeline.config import get_settings
-from kb_workflow.metadata.zhihu_extractor import ZhihuPlaywrightExtractor
+from kb_workflow.pipeline.idempotency import IdempotencyChecker
+from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+from kb_workflow.crawler.models import (
+ CrawlResult,
+ AnswerData,
+ ArticleData,
+ CommentData,
+)
+from kb_workflow.auth.zhihu_login import ZhihuLogin, ZhihuLoginResult
class ZhihuAdapter(PlatformAdapter):
"""Zhihu platform adapter for answer and zhuanlan content extraction.
- Uses ZhihuPlaywrightExtractor (from kb-zhihu-metadata) for content extraction.
- Requires KB_ZHIHU_COOKIE environment variable for authentication.
+ Uses ZhihuCrawler for content extraction with:
+ - Comment crawling enabled by default
+ - Cookie expiration auto-detection and QR code login
+ - Idempotent execution via IdempotencyChecker
Example:
adapter = ZhihuAdapter()
@@ -40,17 +58,30 @@ class ZhihuAdapter(PlatformAdapter):
ZHUANLAN_PATTERN = re.compile(r'zhuanlan\.zhihu\.com/p/([\w\d]+)')
QUESTION_PATTERN = re.compile(r'zhihu\.com/question/(\d+)(?!/answer)')
- def __init__(self, headless: bool = True, timeout: int = 30):
+ def __init__(
+ self,
+ headless: bool = True,
+ timeout: int = 30,
+ output_dir: Optional[Path] = None,
+ idempotency_checker: Optional[IdempotencyChecker] = None
+ ):
"""Initialize ZhihuAdapter.
Args:
headless: Run browser in headless mode (default: True)
timeout: Page load timeout in seconds (default: 30)
+ output_dir: Directory for output files (for idempotency checks)
+ idempotency_checker: Optional IdempotencyChecker instance
"""
self.headless = headless
self.timeout = timeout
self._cookie: Optional[str] = None
- self._extractor = None
+ self._crawler: Optional[ZhihuCrawler] = None
+ self._idempotency_checker = idempotency_checker or IdempotencyChecker()
+ self._output_dir = output_dir
+ self._html_to_text = html2text.HTML2Text()
+ self._html_to_text.ignore_links = False
+ self._html_to_text.body_width = 0
@property
def cookie(self) -> str:
@@ -80,66 +111,264 @@ class ZhihuAdapter(PlatformAdapter):
return "zhuanlan"
return "unknown"
- def _get_extractor(self):
- """Get or create ZhihuPlaywrightExtractor instance.
+ def _get_crawler(self) -> ZhihuCrawler:
+ """Get or create ZhihuCrawler instance.
Returns:
- ZhihuPlaywrightExtractor instance
+ ZhihuCrawler instance
"""
- if self._extractor is None:
- self._extractor = ZhihuPlaywrightExtractor(
+ if self._crawler is None:
+ self._crawler = ZhihuCrawler(
headless=self.headless,
- timeout=self.timeout,
- cookie=self.cookie
+ timeout=self.timeout
)
- return self._extractor
+ return self._crawler
- def _extract_with_playwright(self, url: str) -> Optional[Dict[str, Any]]:
- """Extract content using Playwright-based extractor.
+ def _crawl_with_crawler(
+ self,
+ url: str,
+ content_type: str,
+ include_comments: bool = True
+ ) -> CrawlResult:
+ """Crawl content using ZhihuCrawler.
Args:
- url: Full URL to extract content from
+ url: Full URL to crawl
+ content_type: Type of content ('answer' or 'zhuanlan')
+ include_comments: Whether to include comments (default: True)
Returns:
- Extraction result dictionary or None if extraction failed
+ CrawlResult from the crawler
"""
- extractor = self._get_extractor()
- return extractor.extract(url)
+ crawler = self._get_crawler()
- def _detect_login_page(self, result: Dict[str, Any]) -> bool:
+ if content_type == "answer":
+ return crawler.crawl_answer(url, include_comments=include_comments)
+ elif content_type == "zhuanlan":
+ # For articles, we need to crawl comments separately
+ result = crawler.crawl_article(url)
+ if result.success and include_comments:
+ # Fetch comments for articles
+ from kb_workflow.crawler.models import ArticleData
+ if isinstance(result.data, ArticleData):
+ comments_result = crawler.crawl_comments(
+ content_type="articles",
+ content_id=result.data.id,
+ include_comments=True
+ )
+ result.comments = comments_result.comments
+ return result
+ else:
+ return CrawlResult(
+ success=False,
+ error=f"Unsupported content type: {content_type}"
+ )
+
+ def _is_cookie_expired(self, result: CrawlResult) -> bool:
+ """Check if the result indicates cookie expiration.
+
+ Args:
+ result: CrawlResult from crawler
+
+ Returns:
+ True if cookie appears to be expired
+ """
+ if not result.success:
+ error_msg = result.error or ""
+ expired_indicators = [
+ "cookie expired",
+ "unauthorized",
+ "forbidden",
+ "需要登录",
+ "请先登录",
+ "登录失效",
+ ]
+ return any(indicator in error_msg.lower() for indicator in expired_indicators)
+ return False
+
+ def _detect_login_page(self, result: CrawlResult) -> bool:
"""Detect if the result indicates a login page redirect.
This happens when the cookie has expired or is invalid.
Args:
- result: Extraction result dictionary
+ result: CrawlResult from crawler
Returns:
True if login page is detected, False otherwise
"""
- if not result:
- return False
-
- title = result.get("title", "")
- content = result.get("content", "")
- content_markdown = result.get("content_markdown", "")
-
- # Check for login-related indicators
- login_indicators = ["登录", "登陆", "sign in", "login"]
- title_lower = title.lower()
-
- # If title contains login keywords
- if any(indicator in title_lower for indicator in login_indicators):
- return True
-
- # If content is suspiciously short (login page has minimal content)
- total_content = content + content_markdown
- if len(total_content.strip()) < 100:
- return True
-
+ if not result.success:
+ error_msg = result.error or ""
+ login_indicators = ["登录", "登陆", "sign in", "login"]
+ return any(indicator in error_msg.lower() for indicator in login_indicators)
return False
- async def fetch_metadata(self, content_id: str, content_type: str, source_url: str = None) -> Dict[str, Any]:
+ def _trigger_qrcode_login(self) -> ZhihuLoginResult:
+ """Trigger QR code login flow.
+
+ Returns:
+ ZhihuLoginResult with success status and cookie
+ """
+ # Use intermediate path from settings or default
+ settings = get_settings()
+ intermediate_path = getattr(settings, 'intermediate_path', '/tmp')
+
+ login = ZhihuLogin(
+ intermediate_path=str(intermediate_path),
+ headless=False, # Show browser for QR code display
+ timeout=120 # 2 minutes to scan
+ )
+ return login.login_by_qrcode()
+
+ def _convert_crawl_result_to_metadata(
+ self,
+ result: CrawlResult,
+ content_type: str,
+ source_url: str
+ ) -> Dict[str, Any]:
+ """Convert CrawlResult to metadata dictionary.
+
+ Args:
+ result: CrawlResult from crawler
+ content_type: Type of content
+ source_url: Original source URL
+
+ Returns:
+ Metadata dictionary
+ """
+ data = result.data
+
+ if isinstance(data, AnswerData):
+ return {
+ "type": "answer",
+ "title": data.question_title,
+ "author": data.metadata.author.name if data.metadata.author else "",
+ "author_url_token": data.metadata.author.url_token if data.metadata.author else None,
+ "content": data.content,
+ "content_markdown": self._html_to_text.handle(data.content),
+ "excerpt": data.excerpt,
+ "question_id": data.question_id,
+ "answer_id": data.id,
+ "voteup_count": data.metadata.voteup_count,
+ "comment_count": data.metadata.comment_count,
+ "created_time": data.metadata.created_time,
+ "updated_time": data.metadata.updated_time,
+ "url": source_url,
+ "has_comments": bool(result.comments),
+ "comment_count_actual": len(result.comments) if result.comments else 0,
+ }
+ elif isinstance(data, ArticleData):
+ return {
+ "type": "zhuanlan",
+ "title": data.title,
+ "author": data.metadata.author.name if data.metadata.author else "",
+ "author_url_token": data.metadata.author.url_token if data.metadata.author else None,
+ "content": data.content,
+ "content_markdown": self._html_to_text.handle(data.content),
+ "excerpt": data.excerpt,
+ "article_id": data.id,
+ "voteup_count": data.metadata.voteup_count,
+ "comment_count": data.metadata.comment_count,
+ "created_time": data.metadata.created_time,
+ "updated_time": data.metadata.updated_time,
+ "url": source_url,
+ "has_comments": bool(result.comments),
+ "comment_count_actual": len(result.comments) if result.comments else 0,
+ }
+ else:
+ return {
+ "type": content_type,
+ "title": "",
+ "author": "",
+ "content": "",
+ "content_markdown": "",
+ "url": source_url,
+ "error": "Unknown data type",
+ }
+
+ def _format_comments_section(self, comments: List[CommentData]) -> str:
+ """Format comments as Markdown.
+
+ Args:
+ comments: List of CommentData objects
+
+ Returns:
+ Markdown-formatted comment section
+ """
+ if not comments:
+ return ""
+
+ lines = []
+ for i, comment in enumerate(comments, 1):
+ author_name = comment.author.name if comment.author else "Anonymous"
+ like_str = f"({comment.like_count} likes)" if comment.like_count > 0 else ""
+ lines.append(f"{i}. **@{author_name}** {like_str}: {comment.content}")
+
+ return "\n".join(lines)
+
+ def _format_with_frontmatter(
+ self,
+ metadata: Dict[str, Any],
+ content: str,
+ content_type: str,
+ comments: Optional[List[CommentData]] = None
+ ) -> str:
+ """Format content with YAML frontmatter.
+
+ Args:
+ metadata: Content metadata dictionary
+ content: Markdown content
+ content_type: 'answer' or 'zhuanlan'
+ comments: Optional list of comments to append
+
+ Returns:
+ Formatted string with YAML frontmatter and content
+ """
+ frontmatter_lines = [
+ "---",
+ f"title: {metadata.get('title', '')}",
+ f"author: {metadata.get('author', '')}",
+ f"source_url: {metadata.get('url', '')}",
+ f"platform: {self.PLATFORM}",
+ f"content_type: {content_type}",
+ f"fetched_at: {datetime.now().isoformat()}",
+ "---",
+ "",
+ ]
+
+ result = "\n".join(frontmatter_lines) + content
+
+ # Append comments section if present
+ if comments:
+ comments_md = self._format_comments_section(comments)
+ if comments_md:
+ result += "\n\n---\n\n## 热门评论\n\n" + comments_md
+
+ return result
+
+ def _load_cached_metadata(self, metadata_file: Path) -> Optional[Dict[str, Any]]:
+ """Load cached metadata from file.
+
+ Args:
+ metadata_file: Path to metadata JSON file
+
+ Returns:
+ Metadata dictionary or None if file doesn't exist or is invalid
+ """
+ import json
+ try:
+ if metadata_file.exists():
+ return json.loads(metadata_file.read_text())
+ except (json.JSONDecodeError, IOError):
+ pass
+ return None
+
+ async def fetch_metadata(
+ self,
+ content_id: str,
+ content_type: str,
+ source_url: str = None
+ ) -> Dict[str, Any]:
"""Fetch metadata for the content.
Args:
@@ -161,42 +390,75 @@ class ZhihuAdapter(PlatformAdapter):
auth_type="zhihu_cookie",
)
+ # Idempotency check: if output_dir is set and content already processed, return cached result
+ if self._output_dir:
+ exists, existing_files = self._idempotency_checker.check_exists(
+ self._output_dir, content_id, self.PLATFORM
+ )
+ if exists:
+ # Load and return cached metadata
+ metadata_file = existing_files[0] if existing_files else None
+ if metadata_file and metadata_file.suffix == ".json":
+ cached = self._load_cached_metadata(metadata_file)
+ if cached:
+ return cached
+
# Build URL based on content type
if content_type == "answer":
- # For answers, try to use source_url first, then fallback to constructed URL
- if source_url:
- url = source_url
- else:
- # Fallback: use answer ID only (may not work for all cases)
- url = f"https://www.zhihu.com/question/0/answer/{content_id}"
+ url = source_url or f"https://www.zhihu.com/question/0/answer/{content_id}"
elif content_type == "zhuanlan":
url = f"https://zhuanlan.zhihu.com/p/{content_id}"
else:
raise ValueError(f"Unsupported content type: {content_type}")
- # Extract content in a separate thread to avoid blocking asyncio
- result = await asyncio.to_thread(self._extract_with_playwright, url)
+ # Crawl content in a separate thread to avoid blocking asyncio
+ result = await asyncio.to_thread(
+ self._crawl_with_crawler,
+ url,
+ content_type,
+ True # include_comments=True by default
+ )
- # Check for login page
- if self._detect_login_page(result):
+ # Check for login page / expired cookie
+ if self._detect_login_page(result) or self._is_cookie_expired(result):
+ # Try QR code login
+ login_result = self._trigger_qrcode_login()
+
+ if login_result.success:
+ # Update cookie and retry
+ self._cookie = login_result.cookie_str
+ # Retry crawl with new cookie
+ result = await asyncio.to_thread(
+ self._crawl_with_crawler,
+ url,
+ content_type,
+ True
+ )
+ else:
+ raise AuthError(
+ step_name="zhihu_metadata",
+ input_summary=f"content_id={content_id}, type={content_type}",
+ auth_type="zhihu_cookie",
+ cause=Exception(f"Login failed: {login_result.error_message}")
+ )
+
+ if not result.success:
raise AuthError(
step_name="zhihu_metadata",
input_summary=f"content_id={content_id}, type={content_type}",
auth_type="zhihu_cookie",
- cause=Exception("Cookie may have expired, please update KB_ZHIHU_COOKIE")
+ cause=Exception(result.error or "Failed to extract content")
)
- if not result:
- raise AuthError(
- step_name="zhihu_metadata",
- input_summary=f"content_id={content_id}, type={content_type}",
- auth_type="zhihu_cookie",
- cause=Exception("Failed to extract content, check if content exists")
- )
+ # Convert to metadata dictionary
+ return self._convert_crawl_result_to_metadata(result, content_type, url)
- return result
-
- async def fetch_content(self, content_id: str, content_type: str, source_url: str = None) -> str:
+ async def fetch_content(
+ self,
+ content_id: str,
+ content_type: str,
+ source_url: str = None
+ ) -> str:
"""Fetch content in Markdown format.
Args:
@@ -205,49 +467,67 @@ class ZhihuAdapter(PlatformAdapter):
source_url: Original source URL (optional, used for answers)
Returns:
- Markdown-formatted content with YAML frontmatter
+ Markdown-formatted content with YAML frontmatter and comments
Raises:
AuthError: If cookie is missing or expired
"""
- # Get metadata (which also validates auth)
- metadata = await self.fetch_metadata(content_id, content_type, source_url=source_url)
+ # Build URL for crawling
+ if content_type == "answer":
+ url = source_url or f"https://www.zhihu.com/question/0/answer/{content_id}"
+ elif content_type == "zhuanlan":
+ url = f"https://zhuanlan.zhihu.com/p/{content_id}"
+ else:
+ raise ValueError(f"Unsupported content type: {content_type}")
+
+ # Crawl content with comments
+ result = await asyncio.to_thread(
+ self._crawl_with_crawler,
+ url,
+ content_type,
+ True
+ )
+
+ # Handle cookie expiration
+ if self._detect_login_page(result) or self._is_cookie_expired(result):
+ login_result = self._trigger_qrcode_login()
+ if login_result.success:
+ self._cookie = login_result.cookie_str
+ result = await asyncio.to_thread(
+ self._crawl_with_crawler,
+ url,
+ content_type,
+ True
+ )
+ else:
+ raise AuthError(
+ step_name="zhihu_content",
+ input_summary=f"content_id={content_id}, type={content_type}",
+ auth_type="zhihu_cookie",
+ cause=Exception(f"Login failed: {login_result.error_message}")
+ )
+
+ if not result.success:
+ raise AuthError(
+ step_name="zhihu_content",
+ input_summary=f"content_id={content_id}, type={content_type}",
+ auth_type="zhihu_cookie",
+ cause=Exception(result.error or "Failed to extract content")
+ )
+
+ # Convert to metadata
+ metadata = self._convert_crawl_result_to_metadata(result, content_type, url)
# Get content markdown
content_markdown = metadata.get("content_markdown", "") or metadata.get("content", "")
- # Format with YAML frontmatter
- return self._format_with_frontmatter(metadata, content_markdown, content_type)
-
- def _format_with_frontmatter(
- self,
- metadata: Dict[str, Any],
- content: str,
- content_type: str
- ) -> str:
- """Format content with YAML frontmatter.
-
- Args:
- metadata: Content metadata dictionary
- content: Markdown content
- content_type: 'answer' or 'zhuanlan'
-
- Returns:
- Formatted string with YAML frontmatter and content
- """
- frontmatter_lines = [
- "---",
- f"title: {metadata.get('title', '')}",
- f"author: {metadata.get('author', '')}",
- f"source_url: {metadata.get('url', '')}",
- f"platform: {self.PLATFORM}",
- f"content_type: {content_type}",
- f"fetched_at: {datetime.now().isoformat()}",
- "---",
- "",
- ]
-
- return "\n".join(frontmatter_lines) + content
+ # Format with YAML frontmatter and comments
+ return self._format_with_frontmatter(
+ metadata,
+ content_markdown,
+ content_type,
+ result.comments
+ )
def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str:
"""Generate output filename for the content.
diff --git a/tests/test_auth/__init__.py b/tests/test_auth/__init__.py
new file mode 100644
index 0000000..d0c3099
--- /dev/null
+++ b/tests/test_auth/__init__.py
@@ -0,0 +1 @@
+"""Tests for auth module"""
diff --git a/tests/test_auth/conftest.py b/tests/test_auth/conftest.py
new file mode 100644
index 0000000..13b8f7a
--- /dev/null
+++ b/tests/test_auth/conftest.py
@@ -0,0 +1,111 @@
+"""Pytest fixtures for auth module tests
+
+Provides mock objects and test fixtures for ZhihuLogin testing.
+"""
+
+import os
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+import pytest
+
+
+@pytest.fixture
+def mock_page():
+ """Mock Playwright Page object."""
+ page = MagicMock()
+ page.goto = MagicMock()
+ page.wait_for_selector = MagicMock()
+ page.query_selector = MagicMock()
+ page.set_default_timeout = MagicMock()
+ return page
+
+
+@pytest.fixture
+def mock_context():
+ """Mock Playwright BrowserContext object."""
+ context = MagicMock()
+ context.cookies = MagicMock(return_value=[])
+ context.add_cookies = MagicMock()
+ context.new_page = MagicMock()
+ context.close = MagicMock()
+ return context
+
+
+@pytest.fixture
+def mock_canvas():
+ """Mock Canvas element for QR code."""
+ canvas = MagicMock()
+ canvas.screenshot = MagicMock(return_value=b"fake_png_bytes")
+ return canvas
+
+
+@pytest.fixture
+def mock_browser(mock_page, mock_context, mock_canvas):
+ """Mock PlaywrightBrowser instance.
+
+ Provides a fully mocked browser that can be used for unit testing
+ without requiring actual browser interaction.
+ """
+ browser = MagicMock()
+
+ # Setup browser properties
+ browser.headless = True
+ browser.timeout = 30000
+ browser.page = mock_page
+ browser.context = mock_context
+
+ # Setup methods
+ browser.start = MagicMock()
+ browser.close = MagicMock()
+ browser.goto = MagicMock()
+ browser.wait_for_selector = MagicMock(return_value=mock_canvas)
+ browser.query_selector = MagicMock(return_value=mock_canvas)
+
+ # Setup context.cookies to return cookie list
+ browser.context.cookies = MagicMock(return_value=[])
+
+ return browser
+
+
+@pytest.fixture
+def sample_cookies():
+ """Sample cookie data for testing.
+
+ Returns a list of cookies including z_c0 for login detection.
+ """
+ return [
+ {"name": "z_c0", "value": "test_z_c0_value", "domain": ".zhihu.com"},
+ {"name": "_xsrf", "value": "test_xsrf_value", "domain": ".zhihu.com"},
+ {"name": "d_c0", "value": "test_d_c0_value", "domain": ".zhihu.com"},
+ ]
+
+
+@pytest.fixture
+def sample_cookies_no_login():
+ """Sample cookies without login cookie (z_c0)."""
+ return [
+ {"name": "_xsrf", "value": "test_xsrf_value", "domain": ".zhihu.com"},
+ {"name": "d_c0", "value": "test_d_c0_value", "domain": ".zhihu.com"},
+ ]
+
+
+@pytest.fixture
+def mock_env_file(tmp_path):
+ """Create a temporary .kb_env file for testing.
+
+ Returns the path to the temporary env file.
+ """
+ env_file = tmp_path / ".kb_env"
+ env_file.write_text("")
+ return env_file
+
+
+@pytest.fixture
+def intermediate_path(tmp_path):
+ """Create a temporary intermediate directory for testing.
+
+ Returns the path to the intermediate directory.
+ """
+ intermediate = tmp_path / "intermediate"
+ intermediate.mkdir(parents=True, exist_ok=True)
+ return str(intermediate)
diff --git a/tests/test_auth/test_cli_login.py b/tests/test_auth/test_cli_login.py
new file mode 100644
index 0000000..f6828bf
--- /dev/null
+++ b/tests/test_auth/test_cli_login.py
@@ -0,0 +1,168 @@
+"""Tests for login CLI subcommand.
+
+Tests the login_command function and CLI integration.
+"""
+
+import json
+from unittest.mock import MagicMock, patch, AsyncMock
+import pytest
+
+from kb_workflow.kb_workflow import login_command
+
+
+class TestLoginCommand:
+ """Test suite for login CLI command."""
+
+ @pytest.fixture
+ def mock_settings(self):
+ """Mock settings with empty cookie."""
+ settings = MagicMock()
+ settings.intermediate_path = "/tmp/intermediate"
+ settings.zhihu_cookie = MagicMock()
+ settings.zhihu_cookie.get_secret_value = MagicMock(return_value="")
+ return settings
+
+ @pytest.fixture
+ def mock_settings_with_cookie(self):
+ """Mock settings with valid cookie."""
+ settings = MagicMock()
+ settings.intermediate_path = "/tmp/intermediate"
+ settings.zhihu_cookie = MagicMock()
+ settings.zhihu_cookie.get_secret_value = MagicMock(
+ return_value="z_c0=test_cookie_value"
+ )
+ return settings
+
+ @pytest.fixture
+ def mock_login_result_success(self):
+ """Mock successful login result."""
+ from kb_workflow.auth.zhihu_login import ZhihuLoginResult
+ return ZhihuLoginResult(
+ success=True,
+ cookie_str="z_c0=valid_cookie; _xsrf=xsrf_value",
+ qrcode_path="/tmp/qrcode.png"
+ )
+
+ @pytest.fixture
+ def mock_validation_result_valid(self):
+ """Mock valid cookie validation result."""
+ from kb_workflow.auth.zhihu_login import CookieValidationResult
+ return CookieValidationResult(
+ valid=True,
+ username="TestUser",
+ user_id="12345678"
+ )
+
+ def test_login_default_mode(
+ self, mock_settings, mock_login_result_success
+ ):
+ """Test 1: login_default_mode - Default starts QR code login."""
+ with patch('kb_workflow.pipeline.config.get_settings', return_value=mock_settings), \
+ patch('kb_workflow.auth.cookie_manager.CookieManager') as MockManager, \
+ patch('kb_workflow.auth.ZhihuLogin') as MockLogin:
+
+ # Setup mock
+ mock_login_instance = MagicMock()
+ mock_login_instance.login_by_qrcode.return_value = mock_login_result_success
+ MockLogin.return_value = mock_login_instance
+ mock_manager = MagicMock()
+ MockManager.return_value = mock_manager
+
+ # Act
+ result = login_command()
+
+ # Assert
+ assert result["success"] is True
+ assert "cookie_str" in result
+ mock_manager.save_cookie.assert_called_once()
+
+ def test_login_import_cookie(self, mock_settings, tmp_path):
+ """Test 2: login_import_cookie - --import-cookie imports JSON."""
+ # Create temp cookie JSON file
+ cookie_file = tmp_path / "cookies.json"
+ cookie_file.write_text(json.dumps([
+ {"name": "z_c0", "value": "imported_value", "domain": ".zhihu.com"}
+ ]))
+
+ with patch('kb_workflow.pipeline.config.get_settings', return_value=mock_settings), \
+ patch('kb_workflow.auth.cookie_manager.CookieManager') as MockManager, \
+ patch('kb_workflow.auth.cookie_manager.parse_json_cookies') as mock_parse:
+
+ mock_parse.return_value = "z_c0=imported_value"
+ mock_manager = MagicMock()
+ MockManager.return_value = mock_manager
+
+ # Act
+ result = login_command(import_cookie=str(cookie_file))
+
+ # Assert
+ assert result["success"] is True
+ assert "imported" in result["message"].lower()
+ mock_manager.save_cookie.assert_called_once_with("z_c0=imported_value")
+
+ def test_login_status_no_cookie(self, mock_settings):
+ """Test 3a: login_status - --status with no cookie returns invalid."""
+ with patch('kb_workflow.pipeline.config.get_settings', return_value=mock_settings):
+
+ # Act
+ result = login_command(status=True)
+
+ # Assert
+ assert result["valid"] is False
+ assert "No cookie" in result["reason"]
+
+ def test_login_status_valid_cookie(
+ self, mock_settings_with_cookie, mock_validation_result_valid
+ ):
+ """Test 3b: login_status - --status with valid cookie returns user info."""
+ with patch('kb_workflow.pipeline.config.get_settings', return_value=mock_settings_with_cookie), \
+ patch('kb_workflow.kb_workflow.asyncio.run') as mock_run, \
+ patch('kb_workflow.auth.ZhihuLogin') as MockLogin:
+
+ # Setup mock - asyncio.run returns the result object directly
+ mock_login_instance = MagicMock()
+ MockLogin.return_value = mock_login_instance
+ mock_run.return_value = mock_validation_result_valid # Return object, not dict
+
+ # Act
+ result = login_command(status=True)
+
+ # Assert
+ assert result["valid"] is True
+ assert result["username"] == "TestUser"
+ assert result["user_id"] == "12345678"
+
+ def test_login_integration(
+ self, mock_settings, mock_login_result_success
+ ):
+ """Test 4: login_integration - Successful login saves cookie."""
+ with patch('kb_workflow.pipeline.config.get_settings', return_value=mock_settings), \
+ patch('kb_workflow.auth.cookie_manager.CookieManager') as MockManager, \
+ patch('kb_workflow.auth.ZhihuLogin') as MockLogin:
+
+ # Setup mock
+ mock_login_instance = MagicMock()
+ mock_login_instance.login_by_qrcode.return_value = mock_login_result_success
+ MockLogin.return_value = mock_login_instance
+ mock_manager = MagicMock()
+ MockManager.return_value = mock_manager
+
+ # Act
+ result = login_command()
+
+ # Assert - Cookie saved after successful login
+ assert result["success"] is True
+ mock_manager.save_cookie.assert_called_once_with(
+ mock_login_result_success.cookie_str
+ )
+
+ def test_login_error_handling(self, mock_settings):
+ """Test 5: login handles exceptions gracefully."""
+ with patch('kb_workflow.pipeline.config.get_settings', side_effect=Exception("Config error")):
+
+ # Act
+ result = login_command()
+
+ # Assert
+ assert result["success"] is False
+ assert "Config error" in result["error"]
diff --git a/tests/test_auth/test_cookie_import.py b/tests/test_auth/test_cookie_import.py
new file mode 100644
index 0000000..521e3f5
--- /dev/null
+++ b/tests/test_auth/test_cookie_import.py
@@ -0,0 +1,87 @@
+"""Tests for Cookie JSON parsing functionality
+
+Tests parse_json_cookies function for Chrome Cookie Editor JSON format.
+"""
+
+import json
+import pytest
+from pathlib import Path
+
+from kb_workflow.auth.cookie_manager import parse_json_cookies
+
+
+class TestParseJsonCookies:
+ """Test suite for parse_json_cookies function"""
+
+ def test_parse_json_cookies(self, tmp_path: Path):
+ """Test parsing Chrome Cookie Editor JSON format"""
+ # Arrange: Create a valid JSON file with Chrome Cookie Editor format
+ cookie_data = [
+ {"name": "z_c0", "value": "token123", "domain": ".zhihu.com"},
+ {"name": "_zap", "value": "zap_value", "domain": ".zhihu.com"},
+ ]
+ json_file = tmp_path / "cookies.json"
+ json_file.write_text(json.dumps(cookie_data), encoding="utf-8")
+
+ # Act
+ result = parse_json_cookies(str(json_file))
+
+ # Assert
+ assert result == "z_c0=token123; _zap=zap_value"
+
+ def test_parse_json_cookies_filters_domain(self, tmp_path: Path):
+ """Test that only zhihu.com domain cookies are kept"""
+ # Arrange: Mix of domains
+ cookie_data = [
+ {"name": "z_c0", "value": "zhihu_token", "domain": ".zhihu.com"},
+ {"name": "other_cookie", "value": "other_value", "domain": ".other.com"},
+ {"name": "_zap", "value": "zap_value", "domain": "zhihu.com"},
+ ]
+ json_file = tmp_path / "cookies.json"
+ json_file.write_text(json.dumps(cookie_data), encoding="utf-8")
+
+ # Act
+ result = parse_json_cookies(str(json_file))
+
+ # Assert: Only zhihu.com cookies included
+ assert "z_c0=zhihu_token" in result
+ assert "_zap=zap_value" in result
+ assert "other_cookie" not in result
+
+ def test_parse_json_cookies_empty(self, tmp_path: Path):
+ """Test that empty array returns empty string"""
+ # Arrange
+ json_file = tmp_path / "cookies.json"
+ json_file.write_text("[]", encoding="utf-8")
+
+ # Act
+ result = parse_json_cookies(str(json_file))
+
+ # Assert
+ assert result == ""
+
+ def test_parse_json_cookies_malformed(self, tmp_path: Path):
+ """Test that malformed JSON raises ValueError"""
+ # Arrange: Invalid JSON content
+ json_file = tmp_path / "cookies.json"
+ json_file.write_text("{not valid json}", encoding="utf-8")
+
+ # Act & Assert
+ with pytest.raises(ValueError, match="Invalid JSON format"):
+ parse_json_cookies(str(json_file))
+
+ def test_parse_json_cookies_file_not_found(self):
+ """Test that non-existent file raises ValueError"""
+ # Act & Assert
+ with pytest.raises(ValueError, match="Cookie file not found"):
+ parse_json_cookies("/non/existent/path/cookies.json")
+
+ def test_parse_json_cookies_not_array(self, tmp_path: Path):
+ """Test that non-array JSON raises ValueError"""
+ # Arrange: JSON object instead of array
+ json_file = tmp_path / "cookies.json"
+ json_file.write_text('{"name": "test"}', encoding="utf-8")
+
+ # Act & Assert
+ with pytest.raises(ValueError, match="Expected JSON array"):
+ parse_json_cookies(str(json_file))
diff --git a/tests/test_auth/test_cookie_persistence.py b/tests/test_auth/test_cookie_persistence.py
new file mode 100644
index 0000000..5b2f417
--- /dev/null
+++ b/tests/test_auth/test_cookie_persistence.py
@@ -0,0 +1,136 @@
+"""Tests for Cookie persistence functionality
+
+Tests CookieManager class for saving and loading cookies from ~/.kb_env file.
+"""
+
+import stat
+import pytest
+from pathlib import Path
+
+from kb_workflow.auth.cookie_manager import CookieManager
+
+
+class TestCookieManager:
+ """Test suite for CookieManager class"""
+
+ def test_save_cookie_creates_file(self, tmp_path: Path):
+ """Test that saving cookie creates the env file"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ manager = CookieManager(env_path=env_file)
+ cookie_str = "z_c0=token123; _zap=zap_value"
+
+ # Act
+ manager.save_cookie(cookie_str)
+
+ # Assert
+ assert env_file.exists()
+ content = env_file.read_text(encoding="utf-8")
+ assert 'ZHIHU_COOKIE="z_c0=token123; _zap=zap_value"' in content
+
+ def test_save_cookie_updates_existing(self, tmp_path: Path):
+ """Test that saving cookie updates existing ZHIHU_COOKIE line"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ env_file.write_text('ZHIHU_COOKIE="old_cookie"\n', encoding="utf-8")
+ manager = CookieManager(env_path=env_file)
+ new_cookie = "z_c0=new_token"
+
+ # Act
+ manager.save_cookie(new_cookie)
+
+ # Assert
+ content = env_file.read_text(encoding="utf-8")
+ assert 'ZHIHU_COOKIE="z_c0=new_token"' in content
+ assert "old_cookie" not in content
+
+ def test_save_cookie_appends_new(self, tmp_path: Path):
+ """Test that saving cookie appends new line when ZHIHU_COOKIE doesn't exist"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ env_file.write_text("OTHER_VAR=value\n", encoding="utf-8")
+ manager = CookieManager(env_path=env_file)
+ cookie_str = "z_c0=token"
+
+ # Act
+ manager.save_cookie(cookie_str)
+
+ # Assert
+ content = env_file.read_text(encoding="utf-8")
+ assert "OTHER_VAR=value" in content
+ assert 'ZHIHU_COOKIE="z_c0=token"' in content
+
+ def test_save_cookie_preserves_other_lines(self, tmp_path: Path):
+ """Test that saving cookie preserves other configuration lines"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ env_file.write_text(
+ "KB_BASE_PATH=/path/to/kb\nOTHER_VAR=value\n",
+ encoding="utf-8"
+ )
+ manager = CookieManager(env_path=env_file)
+ cookie_str = "z_c0=token"
+
+ # Act
+ manager.save_cookie(cookie_str)
+
+ # Assert
+ content = env_file.read_text(encoding="utf-8")
+ assert "KB_BASE_PATH=/path/to/kb" in content
+ assert "OTHER_VAR=value" in content
+ assert 'ZHIHU_COOKIE="z_c0=token"' in content
+
+ def test_save_cookie_sets_file_permissions(self, tmp_path: Path):
+ """Test that saved file has 600 permissions"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ manager = CookieManager(env_path=env_file)
+ cookie_str = "z_c0=token"
+
+ # Act
+ manager.save_cookie(cookie_str)
+
+ # Assert
+ file_stat = env_file.stat()
+ file_mode = stat.filemode(file_stat.st_mode)
+ # Check owner read/write only (600)
+ assert (file_stat.st_mode & 0o777) == 0o600, f"Expected 600, got {oct(file_stat.st_mode & 0o777)}"
+
+ def test_load_cookie_returns_saved(self, tmp_path: Path):
+ """Test that loading cookie returns previously saved value"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ manager = CookieManager(env_path=env_file)
+ cookie_str = "z_c0=token123; _zap=zap_value"
+
+ # Act
+ manager.save_cookie(cookie_str)
+ loaded = manager.load_cookie()
+
+ # Assert
+ assert loaded == "z_c0=token123; _zap=zap_value"
+
+ def test_load_cookie_empty_if_not_exists(self, tmp_path: Path):
+ """Test that loading cookie returns empty string if file doesn't exist"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ manager = CookieManager(env_path=env_file)
+
+ # Act
+ loaded = manager.load_cookie()
+
+ # Assert
+ assert loaded == ""
+
+ def test_load_cookie_empty_if_key_not_found(self, tmp_path: Path):
+ """Test that loading cookie returns empty string if key not in file"""
+ # Arrange
+ env_file = tmp_path / ".kb_env"
+ env_file.write_text("OTHER_VAR=value\n", encoding="utf-8")
+ manager = CookieManager(env_path=env_file)
+
+ # Act
+ loaded = manager.load_cookie()
+
+ # Assert
+ assert loaded == ""
diff --git a/tests/test_auth/test_cookie_validation.py b/tests/test_auth/test_cookie_validation.py
new file mode 100644
index 0000000..e2e69a1
--- /dev/null
+++ b/tests/test_auth/test_cookie_validation.py
@@ -0,0 +1,133 @@
+"""Tests for Cookie validation functionality.
+
+Tests the validate_cookie method in ZhihuLogin class.
+"""
+
+from unittest.mock import MagicMock, patch, AsyncMock
+import pytest
+
+from kb_workflow.auth.zhihu_login import ZhihuLogin, CookieValidationResult
+
+
+class TestCookieValidation:
+ """Test suite for cookie validation."""
+
+ @pytest.fixture
+ def mock_browser_with_request(self):
+ """Mock browser with request API support."""
+ browser = MagicMock()
+ browser.browser = MagicMock()
+ browser.page = MagicMock()
+
+ # Create mock response
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json = AsyncMock(return_value={
+ "id": 12345678,
+ "name": "TestUser",
+ "url_token": "testuser"
+ })
+
+ # Setup page.request.get to return mock response
+ browser.page.request = MagicMock()
+ browser.page.request.get = AsyncMock(return_value=mock_response)
+
+ browser.close = MagicMock()
+ browser.start = MagicMock()
+
+ return browser, mock_response
+
+ @pytest.mark.asyncio
+ async def test_validate_cookie_valid(self, mock_browser_with_request, intermediate_path):
+ """Test 1: validate_cookie_valid - Valid cookie returns valid=True with user info."""
+ # Arrange
+ mock_browser, _ = mock_browser_with_request
+ login = ZhihuLogin(intermediate_path=intermediate_path)
+ login.browser = mock_browser
+
+ # Act
+ result = await login.validate_cookie("z_c0=valid_cookie_value")
+
+ # Assert
+ assert isinstance(result, CookieValidationResult)
+ assert result.valid is True
+ assert result.username == "TestUser"
+ assert result.user_id == "12345678"
+ assert result.reason is None
+
+ @pytest.mark.asyncio
+ async def test_validate_cookie_expired(self, mock_browser_with_request, intermediate_path):
+ """Test 2: validate_cookie_expired - Expired cookie returns valid=False."""
+ # Arrange
+ mock_browser, mock_response = mock_browser_with_request
+ mock_response.status = 401
+
+ login = ZhihuLogin(intermediate_path=intermediate_path)
+ login.browser = mock_browser
+
+ # Act
+ result = await login.validate_cookie("z_c0=expired_cookie_value")
+
+ # Assert
+ assert isinstance(result, CookieValidationResult)
+ assert result.valid is False
+ assert result.reason == "Cookie expired"
+ assert result.username is None
+ assert result.user_id is None
+
+ @pytest.mark.asyncio
+ async def test_validate_cookie_empty(self, intermediate_path):
+ """Test 3: validate_cookie_empty - Empty cookie returns valid=False."""
+ # Arrange
+ login = ZhihuLogin(intermediate_path=intermediate_path)
+ login.browser = MagicMock()
+ login.browser.page = None # No page means no cookie loaded
+
+ # Act
+ result = await login.validate_cookie()
+
+ # Assert
+ assert isinstance(result, CookieValidationResult)
+ assert result.valid is False
+ assert "No cookie" in result.reason
+
+ @pytest.mark.asyncio
+ async def test_validate_cookie_malformed(self, intermediate_path):
+ """Test 4: validate_cookie_malformed - Malformed cookie returns valid=False."""
+ # Arrange
+ mock_browser = MagicMock()
+ mock_browser.browser = MagicMock()
+ mock_browser.page = MagicMock()
+ mock_browser.page.request = MagicMock()
+ mock_browser.page.request.get = AsyncMock(side_effect=Exception("Network error"))
+ mock_browser.close = MagicMock()
+ mock_browser.start = MagicMock()
+
+ login = ZhihuLogin(intermediate_path=intermediate_path)
+ login.browser = mock_browser
+
+ # Act
+ result = await login.validate_cookie("invalid_format_cookie")
+
+ # Assert
+ assert isinstance(result, CookieValidationResult)
+ assert result.valid is False
+ assert "Network error" in result.reason
+
+ @pytest.mark.asyncio
+ async def test_validate_cookie_other_status_code(self, mock_browser_with_request, intermediate_path):
+ """Test 5: validate_cookie with non-200/401 status code."""
+ # Arrange
+ mock_browser, mock_response = mock_browser_with_request
+ mock_response.status = 500
+
+ login = ZhihuLogin(intermediate_path=intermediate_path)
+ login.browser = mock_browser
+
+ # Act
+ result = await login.validate_cookie("z_c0=some_cookie")
+
+ # Assert
+ assert isinstance(result, CookieValidationResult)
+ assert result.valid is False
+ assert "500" in result.reason
diff --git a/tests/test_auth/test_qrcode_login.py b/tests/test_auth/test_qrcode_login.py
new file mode 100644
index 0000000..e15c532
--- /dev/null
+++ b/tests/test_auth/test_qrcode_login.py
@@ -0,0 +1,146 @@
+"""Tests for ZhihuLogin QR code login functionality
+
+Tests the core login flow:
+1. Access login page
+2. Capture QR code screenshot
+3. Poll for login cookie (z_c0)
+4. Extract cookie string on success
+5. Handle timeout
+"""
+
+import pytest
+from unittest.mock import MagicMock, patch, AsyncMock
+from pathlib import Path
+
+
+class TestZhihuLogin:
+ """Test suite for ZhihuLogin class."""
+
+ def test_login_page_access(self, mock_browser):
+ """Test that login page is accessed with correct URL."""
+ from kb_workflow.auth import ZhihuLogin
+
+ login = ZhihuLogin(intermediate_path="/tmp/test")
+ login.browser = mock_browser
+
+ # Simulate calling goto
+ login.browser.goto("https://www.zhihu.com/signin")
+
+ # Verify goto was called with correct URL
+ mock_browser.goto.assert_called_once_with("https://www.zhihu.com/signin")
+
+ def test_qrcode_screenshot(self, mock_browser, mock_canvas, intermediate_path):
+ """Test QR code screenshot is captured and saved correctly."""
+ from kb_workflow.auth import ZhihuLogin
+
+ login = ZhihuLogin(intermediate_path=intermediate_path)
+ login.browser = mock_browser
+
+ # Mock the canvas screenshot
+ mock_browser.query_selector.return_value = mock_canvas
+ mock_canvas.screenshot.return_value = b"fake_png_bytes"
+
+ # Call _save_qrcode_image
+ screenshot_bytes = b"fake_png_bytes"
+ qrcode_path = login._save_qrcode_image(screenshot_bytes)
+
+ # Verify file was created
+ assert qrcode_path.exists()
+ assert qrcode_path.suffix == ".png"
+ assert "zhihu_qrcode" in qrcode_path.name
+
+ def test_cookie_detection_with_z_c0(self, mock_browser, sample_cookies):
+ """Test that login is detected when z_c0 cookie exists."""
+ from kb_workflow.auth import ZhihuLogin
+
+ login = ZhihuLogin(intermediate_path="/tmp/test")
+ login.browser = mock_browser
+
+ # Mock cookies to include z_c0
+ mock_browser.context.cookies.return_value = sample_cookies
+
+ # Check login cookie
+ is_logged_in = login.check_login_cookie()
+
+ assert is_logged_in is True
+
+ def test_cookie_detection_without_z_c0(self, mock_browser, sample_cookies_no_login):
+ """Test that login is NOT detected when z_c0 cookie is missing."""
+ from kb_workflow.auth import ZhihuLogin
+
+ login = ZhihuLogin(intermediate_path="/tmp/test")
+ login.browser = mock_browser
+
+ # Mock cookies without z_c0
+ mock_browser.context.cookies.return_value = sample_cookies_no_login
+
+ # Check login cookie
+ is_logged_in = login.check_login_cookie()
+
+ assert is_logged_in is False
+
+ def test_cookie_extraction(self, mock_browser, sample_cookies):
+ """Test cookie string extraction after successful login."""
+ from kb_workflow.auth import ZhihuLogin
+
+ login = ZhihuLogin(intermediate_path="/tmp/test")
+ login.browser = mock_browser
+
+ # Mock cookies
+ mock_browser.context.cookies.return_value = sample_cookies
+
+ # Get cookie string
+ cookie_str = login.get_cookies_str()
+
+ # Verify format: "name1=value1; name2=value2"
+ assert "z_c0=test_z_c0_value" in cookie_str
+ assert "_xsrf=test_xsrf_value" in cookie_str
+ assert ";" in cookie_str
+
+ def test_login_timeout(self, mock_browser, sample_cookies_no_login):
+ """Test that exception is raised after timeout of polling."""
+ from kb_workflow.auth import ZhihuLogin
+
+ login = ZhihuLogin(intermediate_path="/tmp/test", timeout=2) # Short timeout for test
+ login.browser = mock_browser
+
+ # Mock cookies to always return no login
+ mock_browser.context.cookies.return_value = sample_cookies_no_login
+
+ # Should raise Exception after timeout (reraised by tenacity)
+ with pytest.raises(Exception, match="Not logged in yet"):
+ login._wait_for_login()
+
+
+class TestZhihuLoginResult:
+ """Test suite for ZhihuLoginResult model."""
+
+ def test_result_model_success(self):
+ """Test ZhihuLoginResult model with successful login."""
+ from kb_workflow.auth import ZhihuLoginResult
+
+ result = ZhihuLoginResult(
+ success=True,
+ cookie_str="z_c0=abc123; _xsrf=xyz789",
+ qrcode_path=Path("/tmp/qrcode.png"),
+ error_message=None
+ )
+
+ assert result.success is True
+ assert "z_c0" in result.cookie_str
+ assert result.error_message is None
+
+ def test_result_model_failure(self):
+ """Test ZhihuLoginResult model with failed login."""
+ from kb_workflow.auth import ZhihuLoginResult
+
+ result = ZhihuLoginResult(
+ success=False,
+ cookie_str="",
+ qrcode_path=Path("/tmp/qrcode.png"),
+ error_message="Login timeout after 60 seconds"
+ )
+
+ assert result.success is False
+ assert result.cookie_str == ""
+ assert "timeout" in result.error_message.lower()
diff --git a/tests/test_crawler/__init__.py b/tests/test_crawler/__init__.py
new file mode 100644
index 0000000..1e63172
--- /dev/null
+++ b/tests/test_crawler/__init__.py
@@ -0,0 +1 @@
+"""Tests for Zhihu Crawler Module"""
diff --git a/tests/test_crawler/conftest.py b/tests/test_crawler/conftest.py
new file mode 100644
index 0000000..3337b0b
--- /dev/null
+++ b/tests/test_crawler/conftest.py
@@ -0,0 +1,104 @@
+"""Shared fixtures for Zhihu Crawler tests"""
+
+import pytest
+from unittest.mock import MagicMock, AsyncMock
+
+
+@pytest.fixture
+def mock_page():
+ """Mock Playwright page object.
+
+ Provides a mock page with route method for API interception testing.
+ """
+ page = MagicMock()
+ page.route = MagicMock()
+ page.goto = MagicMock()
+ page.wait_for_selector = MagicMock()
+ page.query_selector = MagicMock()
+ page.close = MagicMock()
+ return page
+
+
+@pytest.fixture
+def mock_browser():
+ """Mock PlaywrightBrowser instance.
+
+ Provides a mock browser with page attribute and lifecycle methods.
+ """
+ browser = MagicMock()
+ browser.page = MagicMock()
+ browser.start = MagicMock()
+ browser.close = MagicMock()
+ browser.goto = MagicMock()
+ return browser
+
+
+@pytest.fixture
+def sample_answer_api_response():
+ """Sample answer API response data.
+
+ Mimics the structure of /api/v4/answers/{id} response.
+ """
+ return {
+ "id": 123456789,
+ "type": "answer",
+ "question": {
+ "id": 987654321,
+ "title": "What is the best way to learn programming?",
+ "type": "question"
+ },
+ "content": "This is a sample answer content in HTML format.
",
+ "excerpt": "This is a sample answer content...",
+ "author": {
+ "id": "user123",
+ "name": "John Doe",
+ "url_token": "john-doe-123",
+ "headline": "Software Engineer"
+ },
+ "voteup_count": 1000,
+ "comment_count": 50,
+ "created_time": 1700000000,
+ "updated_time": 1700100000
+ }
+
+
+@pytest.fixture
+def sample_article_api_response():
+ """Sample article API response data.
+
+ Mimics the structure of zhuanlan article API response.
+ """
+ return {
+ "id": 11223344,
+ "title": "How to Build a Web Scraper",
+ "content": "This article explains how to build a web scraper...
",
+ "excerpt": "This article explains how to build a web scraper...",
+ "author": {
+ "id": "author456",
+ "name": "Jane Smith",
+ "url_token": "jane-smith-456",
+ "headline": "Tech Writer"
+ },
+ "voteup_count": 500,
+ "comment_count": 25,
+ "created_time": 1699900000,
+ "updated_time": 1700000000
+ }
+
+
+@pytest.fixture
+def sample_answer_url():
+ """Sample Zhihu answer URL."""
+ return "https://www.zhihu.com/question/987654321/answer/123456789"
+
+
+@pytest.fixture
+def sample_article_url():
+ """Sample Zhihu zhuanlan article URL."""
+ return "https://zhuanlan.zhihu.com/p/11223344"
+
+
+@pytest.fixture
+def sample_invalid_url():
+ """Sample invalid URL for testing error handling."""
+ return "https://www.example.com/some-page"
diff --git a/tests/test_crawler/test_answer_crawler.py b/tests/test_crawler/test_answer_crawler.py
new file mode 100644
index 0000000..2793e35
--- /dev/null
+++ b/tests/test_crawler/test_answer_crawler.py
@@ -0,0 +1,93 @@
+"""Tests for Zhihu Answer Crawler"""
+
+import pytest
+from unittest.mock import MagicMock, patch, PropertyMock
+
+from kb_workflow.crawler.models import AnswerData, CrawlResult
+
+
+class TestCrawlAnswer:
+ """Tests for crawl_answer() method."""
+
+ def test_crawl_answer_returns_crawl_result(self, mock_browser, sample_answer_url):
+ """Test that crawl_answer returns a CrawlResult object."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ with patch.object(ZhihuCrawler, '_get_cookie', return_value='test_cookie'):
+ with patch('kb_workflow.crawler.zhihu_crawler.PlaywrightBrowser') as mock_pb:
+ mock_pb_instance = MagicMock()
+ mock_pb.return_value = mock_pb_instance
+ mock_pb_instance.page = MagicMock()
+ mock_pb_instance.page.route = MagicMock()
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler.crawl_answer(sample_answer_url)
+
+ assert isinstance(result, CrawlResult)
+
+ def test_crawl_answer_invalid_url_returns_failure(self):
+ """Test that invalid URL returns CrawlResult with success=False."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler.crawl_answer("https://www.example.com/invalid")
+
+ assert result.success is False
+ assert result.error is not None
+ assert "Invalid" in result.error
+
+ def test_crawl_answer_extracts_question_title(
+ self, sample_answer_url, sample_answer_api_response
+ ):
+ """Test that crawl_answer extracts question title from API response."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler, ZhihuApiInterceptor
+
+ # Create interceptor with captured data
+ interceptor = ZhihuApiInterceptor()
+ interceptor.captured_data["answer"] = sample_answer_api_response
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+
+ # Test the parse method directly
+ result_data = crawler._parse_answer_data(
+ sample_answer_api_response,
+ answer_id=123456789,
+ question_id=987654321
+ )
+
+ assert result_data.question_title == "What is the best way to learn programming?"
+
+
+class TestAnswerDataExtraction:
+ """Tests for answer data extraction."""
+
+ def test_parse_answer_data_contains_content(self, sample_answer_api_response):
+ """Test that parsed answer contains content."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler._parse_answer_data(
+ sample_answer_api_response,
+ answer_id=123456789,
+ question_id=987654321
+ )
+
+ assert isinstance(result, AnswerData)
+ assert result.content == "This is a sample answer content in HTML format.
"
+ assert result.excerpt == "This is a sample answer content..."
+
+ def test_parse_answer_data_contains_metadata(self, sample_answer_api_response):
+ """Test that parsed answer contains metadata."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler._parse_answer_data(
+ sample_answer_api_response,
+ answer_id=123456789,
+ question_id=987654321
+ )
+
+ assert result.metadata.voteup_count == 1000
+ assert result.metadata.comment_count == 50
+ assert result.metadata.author is not None
+ assert result.metadata.author.name == "John Doe"
diff --git a/tests/test_crawler/test_api_intercept.py b/tests/test_crawler/test_api_intercept.py
new file mode 100644
index 0000000..90064ec
--- /dev/null
+++ b/tests/test_crawler/test_api_intercept.py
@@ -0,0 +1,78 @@
+"""Tests for API Interception and x-zse-96 Signature Handling"""
+
+import pytest
+from unittest.mock import Mock
+
+from kb_workflow.crawler.zhihu_crawler import ZhihuApiInterceptor
+
+
+class TestApiIntercept:
+ """Tests for API interception and x-zse-96 signature handling."""
+
+ def test_intercept_answer_api(self):
+ """Intercept answer API and capture response data."""
+ interceptor = ZhihuApiInterceptor()
+
+ # Mock route object
+ 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 answer content"}
+ 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):
+ """Intercept article API and capture response data."""
+ interceptor = ZhihuApiInterceptor()
+
+ route = Mock()
+ route.request.url = "https://zhuanlan.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
+ assert interceptor.captured_data["article"]["id"] == 67890
+
+ def test_non_api_requests_continue(self):
+ """Non-API requests continue normally."""
+ interceptor = ZhihuApiInterceptor()
+
+ route = Mock()
+ route.request.url = "https://www.zhihu.com/static/some.js"
+
+ interceptor.handle_route(route)
+
+ # Should call continue_ for non-API requests
+ route.continue_.assert_called_once()
+ route.fulfill.assert_not_called()
+
+ def test_x_zse_96_auto_handled(self):
+ """x-zse-96 signature is automatically handled by browser via route.fetch().
+
+ The interceptor automatically uses the browser's fetch() which includes
+ the x-zse-96 signature headers generated by the browser context.
+ """
+ interceptor = ZhihuApiInterceptor()
+
+ route = Mock()
+ route.request.url = "https://www.zhihu.com/api/v4/answers/123"
+
+ mock_response = Mock()
+ mock_response.json.return_value = {"id": 123, "signed": True}
+ route.fetch.return_value = mock_response
+
+ interceptor.handle_route(route)
+
+ # fetch() is called (which would include browser-generated x-zse-96)
+ route.fetch.assert_called_once()
+ # fulfill() returns response to page
+ route.fulfill.assert_called_once()
diff --git a/tests/test_crawler/test_article_crawler.py b/tests/test_crawler/test_article_crawler.py
new file mode 100644
index 0000000..eda45db
--- /dev/null
+++ b/tests/test_crawler/test_article_crawler.py
@@ -0,0 +1,66 @@
+"""Tests for Zhihu Article Crawler"""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+from kb_workflow.crawler.models import ArticleData, CrawlResult
+
+
+class TestCrawlArticle:
+ """Tests for crawl_article() method."""
+
+ def test_crawl_article_invalid_url_returns_failure(self):
+ """Test that invalid URL returns CrawlResult with success=False."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler.crawl_article("https://www.example.com/invalid")
+
+ assert result.success is False
+ assert result.error is not None
+ assert "Invalid" in result.error
+
+ def test_crawl_article_extracts_title(
+ self, sample_article_api_response
+ ):
+ """Test that crawl_article extracts title from API response."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler._parse_article_data(
+ sample_article_api_response,
+ article_id=11223344
+ )
+
+ assert isinstance(result, ArticleData)
+ assert result.title == "How to Build a Web Scraper"
+
+ def test_crawl_article_extracts_content(
+ self, sample_article_api_response
+ ):
+ """Test that crawl_article extracts content from API response."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler._parse_article_data(
+ sample_article_api_response,
+ article_id=11223344
+ )
+
+ assert result.content == "This article explains how to build a web scraper...
"
+ assert result.excerpt == "This article explains how to build a web scraper..."
+
+ def test_crawl_article_contains_metadata(self, sample_article_api_response):
+ """Test that parsed article contains metadata."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ result = crawler._parse_article_data(
+ sample_article_api_response,
+ article_id=11223344
+ )
+
+ assert result.metadata.voteup_count == 500
+ assert result.metadata.comment_count == 25
+ assert result.metadata.author is not None
+ assert result.metadata.author.name == "Jane Smith"
diff --git a/tests/test_crawler/test_comment_crawler.py b/tests/test_crawler/test_comment_crawler.py
new file mode 100644
index 0000000..629c259
--- /dev/null
+++ b/tests/test_crawler/test_comment_crawler.py
@@ -0,0 +1,182 @@
+"""Tests for Zhihu comment crawling functionality"""
+
+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
+from kb_workflow.crawler.config import CrawlerConfig
+
+
+class TestCommentCrawler:
+ """Tests for comment crawling functionality."""
+
+ @pytest.fixture
+ def crawler(self):
+ """Create ZhihuCrawler instance for testing."""
+ config = CrawlerConfig(proxy_required=False)
+ return ZhihuCrawler(headless=True, config=config)
+
+ @pytest.fixture
+ def mock_comment_response(self):
+ """Sample comment API response data.
+
+ Mimics the structure of /api/v4/comment_v5/{type}/{id}/comments response.
+ """
+ return {
+ "data": [
+ {
+ "id": 111,
+ "content": "This is a hot comment",
+ "author": {"name": "User A", "url_token": "user-a"},
+ "created_time": 1234567890,
+ "like_count": 100,
+ "child_comment_count": 5,
+ "is_author": False
+ },
+ {
+ "id": 222,
+ "content": "Author reply",
+ "author": {"name": "Author", "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 returns CommentsResult."""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch.object(crawler, '_start_browser_with_proxy'):
+ # Mock _page.request.get
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+
+ crawler._page = MagicMock()
+ crawler._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):
+ """Comments use hot (popular) sorting."""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch.object(crawler, '_start_browser_with_proxy'):
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+
+ crawler._page = MagicMock()
+ crawler._page.request.get.return_value = mock_response
+
+ crawler.crawl_comments("answers", 12345)
+
+ # Verify API URL contains order=hot
+ call_args = crawler._page.request.get.call_args
+ assert "order=hot" in call_args[0][0]
+
+ def test_comment_has_required_fields(self, crawler, mock_comment_response):
+ """Each comment contains required fields."""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch.object(crawler, '_start_browser_with_proxy'):
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+
+ crawler._page = MagicMock()
+ crawler._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 == "This is a hot comment"
+ assert comment.author.name == "User A"
+ assert comment.like_count == 100
+
+ def test_no_nested_replies(self, crawler, mock_comment_response):
+ """Do not expand nested replies - only first-level comments."""
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch.object(crawler, '_start_browser_with_proxy'):
+ mock_response = MagicMock()
+ mock_response.status = 200
+ mock_response.json.return_value = mock_comment_response
+
+ crawler._page = MagicMock()
+ crawler._page.request.get.return_value = mock_response
+
+ result = crawler.crawl_comments("answers", 12345)
+
+ # Only return first-level comments, not nested replies
+ assert len(result.comments) == 2
+ # child_count records number of child comments but does not expand
+ assert result.comments[0].child_count == 5
+
+ def test_include_comments_parameter(self, crawler):
+ """include_comments=False does not fetch comments."""
+ result = crawler.crawl_comments("answers", 12345, include_comments=False)
+
+ assert result.comments == []
+ assert result.total_count == 0
+
+ def test_include_comments_in_crawl_answer(self, crawler):
+ """crawl_answer with include_comments=True fetches comments."""
+ mock_answer_data = {
+ "id": 456,
+ "question": {"id": 123, "title": "Test Question"},
+ "content": "Answer content
",
+ "excerpt": "Answer excerpt",
+ "author": {"name": "Author"},
+ "voteup_count": 10,
+ "comment_count": 5
+ }
+
+ with patch.object(crawler, '_get_cookie', return_value="test_cookie"):
+ with patch.object(crawler, '_start_browser_with_proxy'):
+ with patch.object(crawler, '_apply_random_delay'):
+ # Mock ZhihuApiInterceptor
+ from kb_workflow.crawler.zhihu_crawler import ZhihuApiInterceptor
+ mock_interceptor = MagicMock(spec=ZhihuApiInterceptor)
+ mock_interceptor.captured_data = {"answer": mock_answer_data}
+
+ with patch('kb_workflow.crawler.zhihu_crawler.ZhihuApiInterceptor', return_value=mock_interceptor):
+ # Mock page operations
+ crawler._page = MagicMock()
+ crawler._page.goto = MagicMock()
+ crawler._page.route = MagicMock()
+ crawler._page.wait_for_timeout = MagicMock()
+
+ # Mock crawl_comments
+ with patch.object(crawler, 'crawl_comments') as mock_comments:
+ mock_comments.return_value = CommentsResult(
+ content_type="answers",
+ content_id=456,
+ comments=[CommentData(id=1, content="Test comment")],
+ total_count=1
+ )
+
+ with patch.object(crawler, '_close_browser'):
+ result = crawler.crawl_answer(
+ "https://www.zhihu.com/question/123/answer/456",
+ include_comments=True
+ )
+
+ # Verify crawl_comments was called with correct params
+ mock_comments.assert_called_once_with(
+ content_type="answers",
+ content_id=456,
+ include_comments=True
+ )
+
+ # Verify result includes comments
+ assert result.success
+ assert result.comments is not None
+ assert len(result.comments) == 1
+ assert result.comments[0].content == "Test comment"
diff --git a/tests/test_crawler/test_metadata_extraction.py b/tests/test_crawler/test_metadata_extraction.py
new file mode 100644
index 0000000..007577d
--- /dev/null
+++ b/tests/test_crawler/test_metadata_extraction.py
@@ -0,0 +1,109 @@
+"""Tests for metadata extraction from Zhihu API responses"""
+
+import pytest
+
+from kb_workflow.crawler.models import AuthorInfo, ContentMetadata
+
+
+class TestAuthorExtraction:
+ """Tests for author information extraction."""
+
+ def test_parse_author_with_full_data(self):
+ """Test parsing author with all fields present."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ author_data = {
+ "name": "Test User",
+ "url_token": "test-user-123",
+ "headline": "Test headline"
+ }
+
+ result = crawler._parse_author(author_data)
+
+ assert isinstance(result, AuthorInfo)
+ assert result.name == "Test User"
+ assert result.url_token == "test-user-123"
+ assert result.headline == "Test headline"
+
+ def test_parse_author_with_minimal_data(self):
+ """Test parsing author with only required fields."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ author_data = {"name": "Minimal User"}
+
+ result = crawler._parse_author(author_data)
+
+ assert result.name == "Minimal User"
+ assert result.url_token is None
+ assert result.headline is None
+
+ def test_parse_author_with_empty_data(self):
+ """Test parsing author with empty/None data."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+
+ result = crawler._parse_author(None)
+ assert result is None
+
+ result = crawler._parse_author({})
+ assert result.name == ""
+
+
+class TestMetadataExtraction:
+ """Tests for content metadata extraction."""
+
+ def test_parse_metadata_with_full_data(self):
+ """Test parsing metadata with all fields present."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ data = {
+ "author": {"name": "Test Author"},
+ "voteup_count": 100,
+ "comment_count": 20,
+ "created_time": 1700000000,
+ "updated_time": 1700100000
+ }
+
+ result = crawler._parse_metadata(data)
+
+ assert isinstance(result, ContentMetadata)
+ assert result.voteup_count == 100
+ assert result.comment_count == 20
+ assert result.created_time == 1700000000
+ assert result.updated_time == 1700100000
+ assert result.author is not None
+ assert result.author.name == "Test Author"
+
+ def test_parse_metadata_with_defaults(self):
+ """Test parsing metadata with missing optional fields."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ data = {}
+
+ result = crawler._parse_metadata(data)
+
+ assert result.voteup_count == 0
+ assert result.comment_count == 0
+ assert result.created_time is None
+ assert result.updated_time is None
+ assert result.author is None
+
+ def test_parse_metadata_with_zero_values(self):
+ """Test parsing metadata when counts are zero."""
+ from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
+
+ crawler = ZhihuCrawler(headless=True, timeout=10)
+ data = {
+ "voteup_count": 0,
+ "comment_count": 0
+ }
+
+ result = crawler._parse_metadata(data)
+
+ assert result.voteup_count == 0
+ assert result.comment_count == 0
diff --git a/tests/test_crawler/test_proxy_config.py b/tests/test_crawler/test_proxy_config.py
new file mode 100644
index 0000000..206538e
--- /dev/null
+++ b/tests/test_crawler/test_proxy_config.py
@@ -0,0 +1,125 @@
+"""Tests for Crawler Proxy Configuration"""
+
+import os
+import pytest
+from unittest.mock import patch
+
+from kb_workflow.crawler.config import (
+ CrawlerConfig,
+ ProxyConfig,
+ get_crawler_config,
+ reset_crawler_config,
+)
+
+
+class TestProxyConfig:
+ """Tests for ProxyConfig model."""
+
+ def test_parse_proxy_from_https_env(self):
+ """Parse proxy from HTTPS_PROXY environment variable."""
+ reset_crawler_config()
+ with patch.dict(os.environ, {"HTTPS_PROXY": "http://127.0.0.1:7890"}, clear=False):
+ os.environ.pop("HTTP_PROXY", None)
+ 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):
+ """Parse proxy from HTTP_PROXY (HTTPS_PROXY takes priority)."""
+ reset_crawler_config()
+ 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):
+ """Parse proxy with authentication credentials."""
+ reset_crawler_config()
+ with patch.dict(os.environ, {
+ "HTTPS_PROXY": "http://user:pass@proxy.example.com:8080"
+ }, clear=False):
+ os.environ.pop("HTTP_PROXY", None)
+ config = CrawlerConfig.from_env()
+ assert config.proxy.username == "user"
+ assert config.proxy.password == "pass"
+ assert config.proxy.server == "http://proxy.example.com:8080"
+
+ def test_no_proxy_configured(self):
+ """Handle case when no proxy is configured."""
+ reset_crawler_config()
+ # Save current values
+ old_http = os.environ.get("HTTP_PROXY")
+ old_https = os.environ.get("HTTPS_PROXY")
+
+ try:
+ os.environ.pop("HTTP_PROXY", None)
+ os.environ.pop("HTTPS_PROXY", None)
+ config = CrawlerConfig.from_env()
+ assert config.proxy is None
+ finally:
+ # Restore
+ if old_http:
+ os.environ["HTTP_PROXY"] = old_http
+ if old_https:
+ os.environ["HTTPS_PROXY"] = old_https
+
+ def test_proxy_required_validation_fails(self):
+ """Raise ValueError when proxy required but not configured."""
+ config = CrawlerConfig(proxy=None, proxy_required=True)
+ with pytest.raises(ValueError, match="Proxy configuration missing"):
+ config.validate_proxy()
+
+ def test_proxy_required_validation_passes(self):
+ """Pass validation when proxy is configured and required."""
+ config = CrawlerConfig(
+ proxy=ProxyConfig(server="http://127.0.0.1:7890"),
+ proxy_required=True
+ )
+ # Should not raise exception
+ config.validate_proxy()
+
+ def test_proxy_not_required(self):
+ """Pass validation when proxy is not required."""
+ config = CrawlerConfig(proxy=None, proxy_required=False)
+ # Should not raise exception
+ config.validate_proxy()
+
+ def test_to_playwright_format(self):
+ """Convert to Playwright proxy format."""
+ 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"
+ }
+
+ def test_to_playwright_format_no_auth(self):
+ """Convert to Playwright format without authentication."""
+ proxy = ProxyConfig(server="http://127.0.0.1:7890")
+ result = proxy.to_playwright_format()
+ assert result == {"server": "http://127.0.0.1:7890"}
+
+ def test_get_crawler_config_singleton(self):
+ """Test that get_crawler_config returns cached instance."""
+ reset_crawler_config()
+ with patch.dict(os.environ, {"HTTPS_PROXY": "http://127.0.0.1:7890"}, clear=False):
+ os.environ.pop("HTTP_PROXY", None)
+ config1 = get_crawler_config()
+ config2 = get_crawler_config()
+ assert config1 is config2
+
+ def test_default_values(self):
+ """Test default configuration values."""
+ config = CrawlerConfig(proxy=None, proxy_required=False)
+ assert config.min_delay == 1.0
+ assert config.max_delay == 3.0
+ assert config.max_retries == 3
+ assert config.retry_min_wait == 2.0
+ assert config.retry_max_wait == 10.0
diff --git a/tests/test_crawler/test_rate_limiter.py b/tests/test_crawler/test_rate_limiter.py
new file mode 100644
index 0000000..9f9c8b2
--- /dev/null
+++ b/tests/test_crawler/test_rate_limiter.py
@@ -0,0 +1,169 @@
+"""Tests for Rate Limiter and Retry Logic"""
+
+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:
+ """Tests for random delay and retry logic."""
+
+ def test_random_delay_applied(self):
+ """Request has random delay between min_delay and max_delay."""
+ 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
+
+ # Delay should be within range (with small margin for timing)
+ assert 0.1 <= elapsed <= 0.3
+
+ def test_delay_is_random(self):
+ """Delay is randomized, not fixed."""
+ 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)
+
+ # Delays should have variation
+ assert max(delays) - min(delays) > 0.01
+
+ def test_exponential_backoff_retry(self):
+ """Retry uses exponential backoff with increasing intervals."""
+ 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())
+ # Simulate a network error - raise exception to trigger retry
+ raise ConnectionError("Simulated network error")
+
+ # Mock the delay to speed up test
+ with patch.object(crawler, '_apply_random_delay'):
+ result = crawler._crawl_with_retry(failing_func)
+
+ # Should have failed after retries
+ assert result.success is False
+ # Should have attempted max_retries times
+ assert len(call_times) == 3 # Initial + 2 retries
+
+ def test_retry_success_after_failure(self):
+ """Retry succeeds if function eventually succeeds."""
+ config = CrawlerConfig(
+ max_retries=3,
+ retry_min_wait=0.05,
+ retry_max_wait=0.2,
+ proxy_required=False
+ )
+ crawler = ZhihuCrawler(config=config)
+
+ call_count = [0]
+
+ def eventually_succeeds():
+ call_count[0] += 1
+ if call_count[0] < 2:
+ raise Exception("Simulated failure")
+ return Mock(success=True, content_type="test")
+
+ # Mock _apply_random_delay to avoid actual delay
+ with patch.object(crawler, '_apply_random_delay'):
+ result = crawler._crawl_with_retry(eventually_succeeds)
+
+ assert result.success is True
+ assert call_count[0] == 2 # Failed once, succeeded on second try
+
+ def test_crawl_with_retry_returns_result(self):
+ """_crawl_with_retry returns CrawlResult from wrapped function."""
+ config = CrawlerConfig(proxy_required=False)
+ crawler = ZhihuCrawler(config=config)
+
+ from kb_workflow.crawler.models import CrawlResult
+
+ expected_result = CrawlResult(success=True, content_type="test")
+
+ def success_func():
+ return expected_result
+
+ with patch.object(crawler, '_apply_random_delay'):
+ result = crawler._crawl_with_retry(success_func)
+
+ assert result.success is True
+ assert result.content_type == "test"
+
+
+class TestProxyIntegration:
+ """Tests for proxy configuration integration."""
+
+ def test_no_proxy_with_proxy_required_raises(self):
+ """Raises ValueError when proxy_required=True but no proxy configured."""
+ config = CrawlerConfig(proxy=None, proxy_required=True)
+ crawler = ZhihuCrawler(config=config)
+
+ with pytest.raises(ValueError, match="Proxy configuration missing"):
+ crawler._start_browser_with_proxy("test_cookie")
+
+ def test_no_proxy_with_proxy_not_required_passes(self):
+ """Succeeds when proxy_required=False and no proxy configured."""
+ config = CrawlerConfig(proxy=None, proxy_required=False)
+ crawler = ZhihuCrawler(config=config)
+
+ # Mock the browser startup to avoid actual browser launch
+ with patch.object(crawler, '_playwright') as mock_pw:
+ mock_pb_instance = MagicMock()
+ mock_pw.return_value = mock_pb_instance
+ mock_pb_instance.chromium.launch.return_value = MagicMock()
+
+ # Should not raise
+ crawler._start_browser_with_proxy("test_cookie")
+
+ def test_proxy_applied_to_browser_context(self):
+ """Proxy is applied to browser context when configured."""
+ from kb_workflow.crawler.config import ProxyConfig
+
+ config = CrawlerConfig(
+ proxy=ProxyConfig(server="http://127.0.0.1:7890"),
+ proxy_required=True
+ )
+ crawler = ZhihuCrawler(config=config)
+
+ # Mock browser components
+ with patch('kb_workflow.crawler.zhihu_crawler.sync_playwright') as mock_sp:
+ mock_pw = MagicMock()
+ mock_sp.return_value.start.return_value = mock_pw
+
+ mock_browser = MagicMock()
+ mock_pw.chromium.launch.return_value = mock_browser
+
+ mock_context = MagicMock()
+ mock_browser.new_context.return_value = mock_context
+
+ crawler._start_browser_with_proxy("test_cookie")
+
+ # Verify proxy was passed to new_context
+ call_kwargs = mock_browser.new_context.call_args[1]
+ assert "proxy" in call_kwargs
+ assert call_kwargs["proxy"]["server"] == "http://127.0.0.1:7890"
diff --git a/tests/test_platforms/test_zhihu.py b/tests/test_platforms/test_zhihu.py
index 02bc4f7..61aca91 100644
--- a/tests/test_platforms/test_zhihu.py
+++ b/tests/test_platforms/test_zhihu.py
@@ -6,18 +6,245 @@ Tests cover:
- Cookie-based authentication
- Login page detection
- AuthError handling
+
+Phase 09 - Pipeline Integration:
+- ZhihuAdapter now uses ZhihuCrawler instead of ZhihuPlaywrightExtractor
+- Comments are included by default and appended to content
+- Cookie expiration triggers QR code login
+- Idempotency via IdempotencyChecker
"""
import pytest
from unittest.mock import MagicMock, patch, PropertyMock
import asyncio
+from pathlib import Path
+from datetime import datetime
from kb_workflow.pipeline.platforms.zhihu import ZhihuAdapter
from kb_workflow.pipeline.errors import AuthError
+# =============================================================================
+# Mock Fixtures for ZhihuCrawler Integration (Phase 09)
+# =============================================================================
-class TestZhihuAdapterAnswerContent:
- """Tests for ZhihuAdapter answer extraction"""
+
+@pytest.fixture
+def mock_crawl_answer_result():
+ """Mock CrawlResult from ZhihuCrawler.crawl_answer().
+
+ Simulates successful answer crawl with metadata and content.
+ Used for testing ZhihuAdapter integration with new crawler.
+ """
+ from kb_workflow.crawler.models import (
+ CrawlResult, AnswerData, AuthorInfo, ContentMetadata, CommentData, CommentAuthor
+ )
+
+ return CrawlResult(
+ success=True,
+ content_type="answer",
+ data=AnswerData(
+ id=789012,
+ question_id=123456,
+ question_title="Test Question Title for Pipeline Integration",
+ content="This is the answer content from ZhihuCrawler.
",
+ excerpt="This is the answer content from ZhihuCrawler.",
+ metadata=ContentMetadata(
+ author=AuthorInfo(
+ name="Test Author",
+ url_token="test-author-123",
+ headline="Test Author Headline"
+ ),
+ voteup_count=1234,
+ comment_count=56,
+ created_time=1705312800, # 2024-01-15 10:00:00 UTC
+ updated_time=1705399200 # 2024-01-16 10:00:00 UTC
+ )
+ ),
+ comments=[
+ CommentData(
+ id=111,
+ content="Great answer! Very helpful.",
+ author=CommentAuthor(name="Commenter1", url_token="commenter-1"),
+ created_time=1705316400,
+ like_count=42,
+ child_count=3,
+ is_author=False
+ ),
+ CommentData(
+ id=222,
+ content="Thanks for sharing!",
+ author=CommentAuthor(name="Commenter2", url_token="commenter-2"),
+ created_time=1705320000,
+ like_count=15,
+ child_count=0,
+ is_author=False
+ )
+ ],
+ error=None
+ )
+
+
+@pytest.fixture
+def mock_crawl_article_result():
+ """Mock CrawlResult from ZhihuCrawler.crawl_article().
+
+ Simulates successful zhuanlan article crawl.
+ Used for testing ZhihuAdapter integration with new crawler.
+ """
+ from kb_workflow.crawler.models import (
+ CrawlResult, ArticleData, AuthorInfo, ContentMetadata, CommentData, CommentAuthor
+ )
+
+ return CrawlResult(
+ success=True,
+ content_type="article",
+ data=ArticleData(
+ id=123456,
+ title="Test Article Title for Pipeline Integration",
+ content="Article Content
This is the article content.
",
+ excerpt="Article Content This is the article content.",
+ metadata=ContentMetadata(
+ author=AuthorInfo(
+ name="Test Article Author",
+ url_token="test-article-author",
+ headline="Writer and Developer"
+ ),
+ voteup_count=5678,
+ comment_count=89,
+ created_time=1705226400, # 2024-01-14 10:00:00 UTC
+ updated_time=None
+ )
+ ),
+ comments=[
+ CommentData(
+ id=333,
+ content="Excellent article!",
+ author=CommentAuthor(name="Reader1", url_token="reader-1"),
+ created_time=1705230000,
+ like_count=28,
+ child_count=1,
+ is_author=False
+ )
+ ],
+ error=None
+ )
+
+
+@pytest.fixture
+def mock_crawl_no_comments_result():
+ """Mock CrawlResult without comments."""
+ from kb_workflow.crawler.models import (
+ CrawlResult, AnswerData, AuthorInfo, ContentMetadata
+ )
+
+ return CrawlResult(
+ success=True,
+ content_type="answer",
+ data=AnswerData(
+ id=789012,
+ question_id=123456,
+ question_title="Test Question No Comments",
+ content="Answer without comments.
",
+ excerpt="Answer without comments.",
+ metadata=ContentMetadata(
+ author=AuthorInfo(name="Test Author", url_token="test-author"),
+ voteup_count=100,
+ comment_count=0,
+ created_time=1705312800,
+ updated_time=None
+ )
+ ),
+ comments=[],
+ error=None
+ )
+
+
+@pytest.fixture
+def mock_comments_list():
+ """Mock list of CommentData for testing comment formatting.
+
+ Provides a realistic set of comments with various attributes.
+ """
+ from kb_workflow.crawler.models import CommentData, CommentAuthor
+
+ return [
+ CommentData(
+ id=1001,
+ content="This is the first comment with some content.",
+ author=CommentAuthor(name="User1", url_token="user-1"),
+ created_time=1705316400,
+ like_count=100,
+ child_count=5,
+ is_author=False
+ ),
+ CommentData(
+ id=1002,
+ content="This is the second comment, from the author.",
+ author=CommentAuthor(name="AuthorName", url_token="author-token"),
+ created_time=1705320000,
+ like_count=50,
+ child_count=2,
+ is_author=True
+ ),
+ CommentData(
+ id=1003,
+ content="Third comment here.",
+ author=CommentAuthor(name="User3", url_token="user-3"),
+ created_time=1705323600,
+ like_count=25,
+ child_count=0,
+ is_author=False
+ )
+ ]
+
+
+@pytest.fixture
+def mock_zhihu_login_result():
+ """Mock ZhihuLoginResult from ZhihuLogin.login_by_qrcode().
+
+ Simulates successful QR code login result.
+ Used for testing cookie refresh flow.
+ """
+ from kb_workflow.auth.zhihu_login import ZhihuLoginResult
+ from pathlib import Path
+
+ return ZhihuLoginResult(
+ success=True,
+ cookie_str="z_c0=test_cookie_value; _xsrf=test_xsrf; d_c0=test_d_c0",
+ qrcode_path=Path("/tmp/test_qrcode.png"),
+ error_message=None
+ )
+
+
+@pytest.fixture
+def mock_crawl_failed_result():
+ """Mock failed CrawlResult for error handling tests."""
+ from kb_workflow.crawler.models import CrawlResult
+
+ return CrawlResult(
+ success=False,
+ content_type="answer",
+ data=None,
+ comments=None,
+ error="Cookie expired or invalid"
+ )
+
+
+@pytest.fixture
+def mock_idempotency_checker():
+ """Mock IdempotencyChecker for testing idempotent behavior."""
+ mock = MagicMock()
+ mock.is_processed.return_value = False
+ return mock
+
+
+# =============================================================================
+# Basic Adapter Tests
+# =============================================================================
+
+
+class TestZhihuAdapterBasic:
+ """Tests for basic ZhihuAdapter functionality"""
def test_get_content_type_answer(self):
"""Test content type detection for answer URLs"""
@@ -52,32 +279,90 @@ class TestZhihuAdapterAnswerContent:
assert adapter.get_output_filename("789012", "answer", "metadata.json") == "zhihu_789012_metadata.json"
assert adapter.get_output_filename("123456", "zhuanlan", "content.md") == "zhihu_123456_content.md"
- @pytest.mark.asyncio
- async def test_fetch_metadata_answer(self, mock_zhihu_answer_result):
- """Test fetching answer metadata"""
+ def test_platform_constant(self):
+ """Test PLATFORM constant"""
adapter = ZhihuAdapter()
- adapter._cookie = "test_cookie" # Set cookie directly
+ adapter._cookie = "test" # Avoid config loading
+ assert adapter.PLATFORM == "zhihu"
- with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_answer_result):
- metadata = await adapter.fetch_metadata("789012", "answer")
- assert metadata["title"] == "Test Question Title"
+# =============================================================================
+# ZhihuCrawler Integration Tests (Phase 09)
+# =============================================================================
+
+
+class TestZhihuAdapterCrawlerIntegration:
+ """Tests for ZhihuAdapter integration with ZhihuCrawler.
+
+ These tests verify the adapter correctly uses the new ZhihuCrawler
+ instead of ZhihuPlaywrightExtractor.
+ """
+
+ @pytest.mark.asyncio
+ async def test_fetch_metadata_answer(self, mock_crawl_answer_result):
+ """Test fetching answer metadata via ZhihuCrawler.
+
+ Verifies:
+ - ZhihuCrawler.crawl_answer() is called with include_comments=True
+ - Metadata is correctly extracted from CrawlResult
+ - Output format matches expected adapter response
+ """
+ adapter = ZhihuAdapter()
+ adapter._cookie = "test_cookie"
+
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_answer_result):
+ metadata = await adapter.fetch_metadata("789012", "answer", source_url="https://www.zhihu.com/question/123456/answer/789012")
+
+ assert metadata["title"] == "Test Question Title for Pipeline Integration"
assert metadata["author"] == "Test Author"
assert metadata["type"] == "answer"
+ assert metadata["voteup_count"] == 1234
@pytest.mark.asyncio
- async def test_fetch_content_answer(self, mock_zhihu_answer_result):
- """Test fetching answer content as Markdown"""
+ async def test_fetch_metadata_article(self, mock_crawl_article_result):
+ """Test fetching zhuanlan article metadata via ZhihuCrawler.
+
+ Verifies:
+ - ZhihuCrawler.crawl_article() is called with correct URL
+ - Metadata is correctly extracted from CrawlResult
+ - Output format matches expected adapter response
+ """
adapter = ZhihuAdapter()
- adapter._cookie = "test_cookie" # Set cookie directly
+ adapter._cookie = "test_cookie"
- with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_answer_result):
- content = await adapter.fetch_content("789012", "answer")
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_article_result):
+ metadata = await adapter.fetch_metadata("123456", "zhuanlan")
- assert "---" in content # YAML frontmatter
+ assert metadata["title"] == "Test Article Title for Pipeline Integration"
+ assert metadata["author"] == "Test Article Author"
+ assert metadata["type"] == "zhuanlan"
+
+ @pytest.mark.asyncio
+ async def test_fetch_content_with_comments(self, mock_crawl_answer_result):
+ """Test fetching content with comments appended.
+
+ Verifies:
+ - Content is fetched with include_comments=True
+ - Comments are appended to content in correct Markdown format
+ - Comment format: numbered list with author mention
+ """
+ adapter = ZhihuAdapter()
+ adapter._cookie = "test_cookie"
+
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_answer_result):
+ content = await adapter.fetch_content("789012", "answer", source_url="https://www.zhihu.com/question/123456/answer/789012")
+
+ # Verify YAML frontmatter
+ assert "---" in content
assert "title:" in content
- assert "Test Question Title" in content
- assert "This is the answer content" in content
+
+ # Verify main content
+ assert "Test Question Title for Pipeline Integration" in content
+
+ # Verify comments section is appended
+ assert "## 热门评论" in content
+ assert "@Commenter1" in content
+ assert "Great answer!" in content
@pytest.mark.asyncio
async def test_fetch_metadata_no_cookie_raises_auth_error(self):
@@ -91,80 +376,230 @@ class TestZhihuAdapterAnswerContent:
assert exc_info.value.auth_type == "zhihu_cookie"
-class TestZhihuAdapterZhuanlanContent:
- """Tests for ZhihuAdapter zhuanlan (article) extraction"""
-
- @pytest.mark.asyncio
- async def test_fetch_metadata_zhuanlan(self, mock_zhihu_zhuanlan_result):
- """Test fetching zhuanlan metadata"""
- adapter = ZhihuAdapter()
- adapter._cookie = "test_cookie" # Set cookie directly
-
- with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_zhuanlan_result):
- metadata = await adapter.fetch_metadata("123456", "zhuanlan")
-
- assert metadata["title"] == "Test Article Title"
- assert metadata["author"] == "Test Author"
- assert metadata["type"] == "zhuanlan"
-
- @pytest.mark.asyncio
- async def test_fetch_content_zhuanlan(self, mock_zhihu_zhuanlan_result):
- """Test fetching zhuanlan content as Markdown"""
- adapter = ZhihuAdapter()
- adapter._cookie = "test_cookie" # Set cookie directly
-
- with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_zhuanlan_result):
- content = await adapter.fetch_content("123456", "zhuanlan")
-
- assert "---" in content # YAML frontmatter
- assert "title:" in content
- assert "Test Article Title" in content
- assert "This is the article content" in content
-
- @pytest.mark.asyncio
- async def test_detect_login_page_raises_auth_error(self):
- """Test that login page detection raises AuthError"""
- adapter = ZhihuAdapter()
- adapter._cookie = "expired_cookie" # Set cookie directly
-
- # Simulate login page response (empty content or login title)
- login_page_result = {
- "type": "answer",
- "title": "登录 - 知乎",
- "content": "",
- "content_markdown": "",
- }
-
- with patch.object(adapter, '_extract_with_playwright', return_value=login_page_result):
- with pytest.raises(AuthError) as exc_info:
- await adapter.fetch_metadata("789012", "answer")
-
- assert exc_info.value.auth_type == "zhihu_cookie"
- assert "过期" in str(exc_info.value.cause) or "expired" in str(exc_info.value.cause).lower()
-
- @pytest.mark.asyncio
- async def test_detect_short_content_raises_auth_error(self):
- """Test that very short content (< 100 chars) raises AuthError"""
- adapter = ZhihuAdapter()
- adapter._cookie = "test_cookie"
-
- # Simulate response with very short content
- short_content_result = {
- "type": "answer",
- "title": "Some Title",
- "content": "x", # Very short content
- "content_markdown": "",
- }
-
- with patch.object(adapter, '_extract_with_playwright', return_value=short_content_result):
- with pytest.raises(AuthError) as exc_info:
- await adapter.fetch_metadata("789012", "answer")
-
- assert exc_info.value.auth_type == "zhihu_cookie"
+# =============================================================================
+# Cookie Expiration and Auto-Login Tests (Task 2)
+# =============================================================================
class TestZhihuAdapterCookieHandling:
- """Tests for cookie handling"""
+ """Tests for cookie handling and auto-login"""
+
+ @pytest.mark.asyncio
+ async def test_cookie_valid_returns_data(self, mock_crawl_answer_result):
+ """Test that valid cookie allows normal data retrieval"""
+ adapter = ZhihuAdapter()
+ adapter._cookie = "valid_cookie"
+
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_answer_result):
+ metadata = await adapter.fetch_metadata("789012", "answer")
+
+ assert metadata["title"] == "Test Question Title for Pipeline Integration"
+
+ @pytest.mark.asyncio
+ async def test_cookie_expired_triggers_login(self, mock_crawl_failed_result, mock_zhihu_login_result, mock_crawl_answer_result):
+ """Test that expired cookie triggers QR code login flow.
+
+ Verifies:
+ - CrawlResult with success=False triggers login
+ - ZhihuLogin.login_by_qrcode() is called
+ - New cookie is used to retry the fetch
+ """
+ adapter = ZhihuAdapter()
+ adapter._cookie = "expired_cookie"
+
+ # First call fails (cookie expired), second succeeds after login
+ crawl_results = [mock_crawl_failed_result, mock_crawl_answer_result]
+
+ with patch.object(adapter, '_crawl_with_crawler', side_effect=crawl_results):
+ with patch.object(adapter, '_trigger_qrcode_login', return_value=mock_zhihu_login_result):
+ metadata = await adapter.fetch_metadata("789012", "answer")
+
+ # Verify login was triggered and cookie was updated
+ assert adapter._cookie == mock_zhihu_login_result.cookie_str
+
+ @pytest.mark.asyncio
+ async def test_login_retry_success(self, mock_crawl_failed_result, mock_zhihu_login_result, mock_crawl_answer_result):
+ """Test that after successful login, retry succeeds"""
+ adapter = ZhihuAdapter()
+ adapter._cookie = "expired_cookie"
+
+ crawl_results = [mock_crawl_failed_result, mock_crawl_answer_result]
+
+ with patch.object(adapter, '_crawl_with_crawler', side_effect=crawl_results):
+ with patch.object(adapter, '_trigger_qrcode_login', return_value=mock_zhihu_login_result):
+ content = await adapter.fetch_content("789012", "answer")
+
+ # Verify login was triggered and cookie was updated
+ assert adapter._cookie == mock_zhihu_login_result.cookie_str
+ assert "Test Question Title for Pipeline Integration" in content
+
+ @pytest.mark.asyncio
+ async def test_login_failure_raises_error(self, mock_crawl_failed_result):
+ """Test that failed login raises appropriate error"""
+ from kb_workflow.auth.zhihu_login import ZhihuLoginResult
+
+ adapter = ZhihuAdapter()
+ adapter._cookie = "expired_cookie"
+
+ failed_login = ZhihuLoginResult(
+ success=False,
+ cookie_str="",
+ error_message="Login timeout"
+ )
+
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_failed_result):
+ with patch.object(adapter, '_trigger_qrcode_login', return_value=failed_login):
+ with pytest.raises(AuthError) as exc_info:
+ await adapter.fetch_metadata("789012", "answer")
+
+ assert exc_info.value.auth_type == "zhihu_cookie"
+
+
+# =============================================================================
+# Comment Formatting Tests (Task 3)
+# =============================================================================
+
+
+class TestZhihuAdapterCommentFormatting:
+ """Tests for comment formatting functionality"""
+
+ @pytest.mark.asyncio
+ async def test_comment_formatting_with_comments(self, mock_crawl_answer_result):
+ """Test that comments are formatted with proper Markdown structure.
+
+ Verifies:
+ - Comments section starts with "---" separator
+ - "## 热门评论" heading is present
+ - Each comment is numbered with bold author name
+ """
+ adapter = ZhihuAdapter()
+ adapter._cookie = "test_cookie"
+
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_answer_result):
+ content = await adapter.fetch_content("789012", "answer")
+
+ # Check comment section structure
+ assert "---" in content
+ assert "## 热门评论" in content
+
+ # Check comment formatting: numbered list with author
+ assert "1. **@Commenter1**" in content or "1. **@Commenter1**" in content
+ assert "Great answer!" in content
+
+ @pytest.mark.asyncio
+ async def test_comment_formatting_no_comments(self, mock_crawl_no_comments_result):
+ """Test that no comments section is added when there are no comments.
+
+ Verifies:
+ - No "---" separator for empty comments
+ - No "## 热门评论" heading when comments list is empty
+ """
+ adapter = ZhihuAdapter()
+ adapter._cookie = "test_cookie"
+
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_no_comments_result):
+ content = await adapter.fetch_content("789012", "answer")
+
+ # No comments section should be present
+ assert "## 热门评论" not in content
+
+ def test_format_comments_section(self, mock_comments_list):
+ """Test internal _format_comments_section method.
+
+ Verifies:
+ - Comments are formatted as numbered list
+ - Author name is bold/mentioned
+ - Like count is included
+ - Format matches expected style:
+ 1. **@AuthorName** (100 likes): Comment content...
+ """
+ adapter = ZhihuAdapter()
+ adapter._cookie = "test"
+
+ result = adapter._format_comments_section(mock_comments_list)
+
+ # Check format structure
+ assert "1. **@User1**" in result
+ assert "(100 likes)" in result or "100 likes" in result
+ assert "This is the first comment" in result
+
+ # Check author comment indicator
+ assert "@AuthorName" in result
+
+
+# =============================================================================
+# Idempotency Tests (Task 3)
+# =============================================================================
+
+
+class TestZhihuAdapterIdempotency:
+ """Tests for idempotent execution"""
+
+ @pytest.mark.asyncio
+ async def test_idempotent_execution_skips_processed(self, mock_crawl_answer_result, tmp_path):
+ """Test that idempotent check prevents reprocessing when output_dir is set.
+
+ Verifies:
+ - If output_dir is set and content is already processed, cached result is returned
+ - No duplicate fetches are made
+ """
+ import json
+
+ adapter = ZhihuAdapter(output_dir=tmp_path)
+ adapter._cookie = "test_cookie"
+
+ # Create existing files to simulate already processed content
+ metadata_file = tmp_path / "zhihu_789012_metadata.json"
+ content_file = tmp_path / "zhihu_789012_content.md"
+ cached_metadata = {
+ "title": "Cached Title",
+ "author": "Cached Author",
+ "type": "answer"
+ }
+ metadata_file.write_text(json.dumps(cached_metadata))
+ content_file.write_text("Cached content")
+
+ # When content is already processed, should return cached result
+ with patch.object(adapter, '_crawl_with_crawler') as mock_crawl:
+ metadata = await adapter.fetch_metadata("789012", "answer")
+
+ # Should NOT have called the crawler since content is cached
+ mock_crawl.assert_not_called()
+
+ # Should return cached metadata
+ assert metadata["title"] == "Cached Title"
+
+ @pytest.mark.asyncio
+ async def test_different_content_processes_independently(self, mock_crawl_answer_result, mock_crawl_article_result):
+ """Test that different content IDs are processed independently.
+
+ Verifies:
+ - Processing one content doesn't affect processing of another
+ - Each content_id is tracked separately
+ """
+ adapter = ZhihuAdapter()
+ adapter._cookie = "test_cookie"
+
+ # Process answer
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_answer_result):
+ metadata_answer = await adapter.fetch_metadata("789012", "answer")
+
+ # Process article
+ with patch.object(adapter, '_crawl_with_crawler', return_value=mock_crawl_article_result):
+ metadata_article = await adapter.fetch_metadata("123456", "zhuanlan")
+
+ # Both should return their respective data
+ assert metadata_answer["title"] == "Test Question Title for Pipeline Integration"
+ assert metadata_article["title"] == "Test Article Title for Pipeline Integration"
+
+
+# =============================================================================
+# Cookie Settings Tests
+# =============================================================================
+
+
+class TestZhihuAdapterCookieSettings:
+ """Tests for cookie loading from settings"""
def test_cookie_from_settings(self):
"""Test cookie is loaded from settings"""
@@ -175,9 +610,3 @@ class TestZhihuAdapterCookieHandling:
adapter = ZhihuAdapter()
assert adapter.cookie == "test_cookie_value"
-
- def test_platform_constant(self):
- """Test PLATFORM constant"""
- adapter = ZhihuAdapter()
- adapter._cookie = "test" # Avoid config loading
- assert adapter.PLATFORM == "zhihu"