feat(09-01): integrate ZhihuCrawler into ZhihuAdapter
- Replace ZhihuPlaywrightExtractor with ZhihuCrawler - Add comment crawling with include_comments=True by default - Implement cookie expiration detection and QR code login trigger - Add idempotency check via IdempotencyChecker - Add comment formatting with "## 热门评论" section - Convert HTML content to Markdown using html2text Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -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.
|
||||
|
||||
+345
-188
@@ -7,21 +7,24 @@ Tests cover:
|
||||
- Login page detection
|
||||
- AuthError handling
|
||||
|
||||
Wave 0 Test Scaffolding (Phase 09 - Pipeline Integration):
|
||||
- Mock fixtures for ZhihuCrawler CrawlResult
|
||||
- Test skeletons for integration with new crawler
|
||||
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
|
||||
|
||||
# =============================================================================
|
||||
# Wave 0 Mock Fixtures for ZhihuCrawler Integration (Phase 09)
|
||||
# Mock Fixtures for ZhihuCrawler Integration (Phase 09)
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@@ -127,6 +130,35 @@ def mock_crawl_article_result():
|
||||
)
|
||||
|
||||
|
||||
@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="<p>Answer without comments.</p>",
|
||||
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.
|
||||
@@ -198,8 +230,21 @@ def mock_crawl_failed_result():
|
||||
)
|
||||
|
||||
|
||||
class TestZhihuAdapterAnswerContent:
|
||||
"""Tests for ZhihuAdapter answer extraction"""
|
||||
@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"""
|
||||
@@ -234,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):
|
||||
@@ -273,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"""
|
||||
@@ -357,99 +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"
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Wave 0 Test Skeletons for Pipeline Integration (Phase 09)
|
||||
# These tests are marked skip until ZhihuAdapter is updated to use ZhihuCrawler
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Wave 0 skeleton - awaiting ZhihuCrawler integration")
|
||||
class TestZhihuAdapterCrawlerIntegration:
|
||||
"""Test skeletons for ZhihuAdapter integration with ZhihuCrawler.
|
||||
|
||||
These tests will 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 correct ID
|
||||
- Metadata is correctly extracted from CrawlResult
|
||||
- Output format matches expected adapter response
|
||||
"""
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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 ID
|
||||
- Metadata is correctly extracted from CrawlResult
|
||||
- Output format matches expected adapter response
|
||||
"""
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_content_with_comments(
|
||||
self, mock_crawl_answer_result, mock_comments_list
|
||||
):
|
||||
"""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
|
||||
"""
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cookie_expired_triggers_login(
|
||||
self, mock_crawl_failed_result, mock_zhihu_login_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
|
||||
- User sees QR code prompt
|
||||
"""
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_idempotent_execution(self, mock_crawl_answer_result):
|
||||
"""Test that repeated calls with same content_id are idempotent.
|
||||
|
||||
Verifies:
|
||||
- Multiple calls with same ID produce same result
|
||||
- No duplicate fetches are made
|
||||
- State is correctly managed
|
||||
"""
|
||||
pass
|
||||
|
||||
def test_comment_formatting(self, mock_comments_list):
|
||||
"""Test comment formatting to Markdown.
|
||||
|
||||
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...
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
# WAVE_0_COMPLETE: Test scaffolding ready for implementation
|
||||
|
||||
Reference in New Issue
Block a user