From 9e1452d2a9bea9e4d3c88ad886a800424155b7c8 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Tue, 17 Mar 2026 22:14:47 +0800 Subject: [PATCH] feat(08-03): integrate proxy, delay, and retry into ZhihuCrawler - Add CrawlerConfig with ProxyConfig, delay, and retry settings - Integrate proxy configuration via _start_browser_with_proxy() - Add random delay via _apply_random_delay() - Add retry logic via _crawl_with_retry using tenacity - Add ZhihuApiInterceptor tests for x-zse-96 handling - All tests pass --- kb_workflow/crawler/zhihu_crawler.py | 224 +++++++++++++++++++---- tests/test_crawler/test_api_intercept.py | 78 ++++++++ tests/test_crawler/test_rate_limiter.py | 169 +++++++++++++++++ 3 files changed, 434 insertions(+), 37 deletions(-) create mode 100644 tests/test_crawler/test_api_intercept.py create mode 100644 tests/test_crawler/test_rate_limiter.py diff --git a/kb_workflow/crawler/zhihu_crawler.py b/kb_workflow/crawler/zhihu_crawler.py index c928140..b424da4 100644 --- a/kb_workflow/crawler/zhihu_crawler.py +++ b/kb_workflow/crawler/zhihu_crawler.py @@ -2,13 +2,21 @@ 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 +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 @@ -22,13 +30,15 @@ from .models import ( 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). + 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 @@ -40,6 +50,9 @@ class ZhihuApiInterceptor: 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 """ @@ -66,7 +79,11 @@ class ZhihuCrawler: """Zhihu Content Crawler. Crawls Zhihu answers and articles using Playwright API interception. - Reuses Cookie management from Phase 7 auth module. + 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 @@ -82,17 +99,29 @@ class ZhihuCrawler: "?order=hot&limit=20" ) - def __init__(self, headless: bool = True, timeout: int = 30): + 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. @@ -103,6 +132,121 @@ class ZhihuCrawler: 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. @@ -211,11 +355,7 @@ class ZhihuCrawler: ) try: - self.browser = PlaywrightBrowser( - headless=self.headless, - timeout=self.timeout - ) - self.browser.start(cookie=self._get_cookie()) + self._start_browser_with_proxy(self._get_cookie()) # Build comment API URL api_url = self.COMMENT_API_TEMPLATE.format( @@ -224,7 +364,7 @@ class ZhihuCrawler: ) # Use Playwright's request API to directly call the API - response = self.browser.page.request.get(api_url) + response = self._page.request.get(api_url) if response.status != 200: return CommentsResult( @@ -272,11 +412,10 @@ class ZhihuCrawler: ) finally: - if self.browser: - self.browser.close() + self._close_browser() def crawl_answer(self, url: str, include_comments: bool = False) -> CrawlResult: - """Crawl Zhihu answer. + """Crawl Zhihu answer with retry support. Args: url: Zhihu answer URL (e.g., https://www.zhihu.com/question/123/answer/456) @@ -286,6 +425,20 @@ class ZhihuCrawler: 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") @@ -293,18 +446,14 @@ class ZhihuCrawler: question_id, answer_id = int(match.group(1)), int(match.group(2)) try: - self.browser = PlaywrightBrowser( - headless=self.headless, - timeout=self.timeout - ) - self.browser.start(cookie=self._get_cookie()) + self._start_browser_with_proxy(self._get_cookie()) self.interceptor = ZhihuApiInterceptor() - # Set up route interception - self.browser.page.route("**/api/v4/**", self.interceptor.handle_route) + # 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.browser.goto(url, wait_until="networkidle") + self._page.goto(url, wait_until="networkidle") # Wait for data capture time.sleep(2) @@ -330,10 +479,6 @@ class ZhihuCrawler: # Fetch comments if requested if include_comments: - # Need to close current browser first as crawl_comments creates its own - self.browser.close() - self.browser = None - comments_result = self.crawl_comments( content_type="answers", content_id=answer_id, @@ -347,11 +492,10 @@ class ZhihuCrawler: return CrawlResult(success=False, error=str(e)) finally: - if self.browser: - self.browser.close() + self._close_browser() def crawl_article(self, url: str) -> CrawlResult: - """Crawl Zhihu zhuanlan article. + """Crawl Zhihu zhuanlan article with retry support. Args: url: Zhihu zhuanlan URL (e.g., https://zhuanlan.zhihu.com/p/123) @@ -359,6 +503,17 @@ class ZhihuCrawler: 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") @@ -366,16 +521,12 @@ class ZhihuCrawler: article_id = int(match.group(1)) try: - self.browser = PlaywrightBrowser( - headless=self.headless, - timeout=self.timeout - ) - self.browser.start(cookie=self._get_cookie()) + self._start_browser_with_proxy(self._get_cookie()) self.interceptor = ZhihuApiInterceptor() - self.browser.page.route("**/api/v4/**", self.interceptor.handle_route) + self._page.route("**/api/v4/**", self.interceptor.handle_route) - self.browser.goto(url, wait_until="networkidle") + self._page.goto(url, wait_until="networkidle") time.sleep(2) @@ -396,8 +547,7 @@ class ZhihuCrawler: return CrawlResult(success=False, error=str(e)) finally: - if self.browser: - self.browser.close() + self._close_browser() def _crawl_article_from_dom(self, url: str, article_id: int) -> CrawlResult: """Crawl article using DOM parsing as fallback. 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_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"