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_platforms/test_zhihu.py b/tests/test_platforms/test_zhihu.py index 4b69354..61aca91 100644 --- a/tests/test_platforms/test_zhihu.py +++ b/tests/test_platforms/test_zhihu.py @@ -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="

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. @@ -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