kb-skills/.planning/phases/08-crawler/08-03-PLAN.md

phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, user_setup, must_haves

phase	plan	type	wave	depends_on	files_modified	autonomous	requirements	user_setup	must_haves
08-crawler	03	execute	2	08-01	kb_workflow/crawler/config.py kb_workflow/crawler/zhihu_crawler.py tests/test_crawler/test_proxy_config.py tests/test_crawler/test_rate_limiter.py tests/test_crawler/test_api_intercept.py	true	NET-01 NET-02 NET-03	service why env_vars dashboard_config proxy 防止 IP 被知乎封禁 name source HTTP_PROXY 代理服务地址，如 http://127.0.0.1:7890 name source HTTPS_PROXY 代理服务地址，如 http://127.0.0.1:7890	truths artifacts key_links 用户可以通过环境变量配置代理 IP 不配置代理时，爬虫拒绝运行并提示错误 每次请求间有随机延迟（1-3秒） 请求失败时有限重试（3次），间隔递增 x-zse-96 签名由浏览器自动处理，无需手动逆向 path provides contains kb_workflow/crawler/config.py CrawlerConfig 配置类，代理和延迟设置 class CrawlerConfig path provides contains kb_workflow/crawler/zhihu_crawler.py 代理应用和重试逻辑 def _get_proxy_config path provides exports tests/test_crawler/test_proxy_config.py 代理配置测试 test_proxy_required path provides exports tests/test_crawler/test_rate_limiter.py 频率控制测试 test_random_delay from to via pattern CrawlerConfig HTTP_PROXY/HTTPS_PROXY 环境变量读取 os.environ.get.*PROXY from to via pattern ZhihuCrawler browser.new_context(proxy=...) Playwright 代理配置 new_context.*proxy

实现网络反爬措施：强制代理配置、请求频率控制、有限重试机制，以及利用浏览器自动处理 x-zse-96 签名。

Purpose: 降低被封风险，提高爬取稳定性 Output: CrawlerConfig + 代理/延迟/重试逻辑 + 测试

<execution_context> @/home/starmen1265/.claude/get-shit-done/workflows/execute-plan.md @/home/starmen1265/.claude/get-shit-done/templates/summary.md </execution_context>

@.planning/PROJECT.md @.planning/ROADMAP.md @.planning/STATE.md @.planning/phases/08-crawler/08-CONTEXT.md @.planning/phases/08-crawler/08-RESEARCH.md @.planning/phases/08-crawler/08-01-PLAN.md

From kb_workflow/crawler/zhihu_crawler.py (Plan 01):

class ZhihuCrawler:
    def __init__(self, headless: bool = True, timeout: int = 30)
    def _get_cookie(self) -> str
    def crawl_answer(self, url: str) -> CrawlResult
    def crawl_article(self, url: str) -> CrawlResult

代理配置模式 (from RESEARCH.md):

import os
from urllib.parse import urlparse

def get_proxy_config():
    proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
    if not proxy_url:
        return None

    parsed = urlparse(proxy_url)
    config = {"server": f"{parsed.scheme}://{parsed.hostname}:{parsed.port}"}

    if parsed.username and parsed.password:
        config["username"] = parsed.username
        config["password"] = parsed.password

    return config

重试模式 (from RESEARCH.md):

from tenacity import Retrying, stop_after_attempt, wait_exponential

for attempt in Retrying(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    reraise=True
):
    with attempt:
        # 爬取逻辑

Task 1: 创建 CrawlerConfig 配置类 kb_workflow/crawler/config.py, tests/test_crawler/test_proxy_config.py 创建爬虫配置类，处理代理和延迟设置。

1. 创建 kb_workflow/crawler/config.py:

"""Crawler configuration

Handles proxy settings, rate limiting, and retry configuration.
"""

import os
from urllib.parse import urlparse
from typing import Optional, Dict, Any

from pydantic import BaseModel, Field


class ProxyConfig(BaseModel):
    """代理配置"""
    server: str = Field(..., description="代理服务器地址")
    username: Optional[str] = Field(None, description="代理用户名")
    password: Optional[str] = Field(None, description="代理密码")

    def to_playwright_format(self) -> Dict[str, Any]:
        """转换为 Playwright 代理格式"""
        config = {"server": self.server}
        if self.username:
            config["username"] = self.username
        if self.password:
            config["password"] = self.password
        return config


class CrawlerConfig(BaseModel):
    """爬虫配置"""

    # 代理设置
    proxy: Optional[ProxyConfig] = Field(None, description="代理配置")
    proxy_required: bool = Field(default=True, description="是否强制要求代理")

    # 请求频率
    min_delay: float = Field(default=1.0, description="最小延迟（秒）")
    max_delay: float = Field(default=3.0, description="最大延迟（秒）")

    # 重试设置
    max_retries: int = Field(default=3, description="最大重试次数")
    retry_min_wait: float = Field(default=2.0, description="重试最小等待（秒）")
    retry_max_wait: float = Field(default=10.0, description="重试最大等待（秒）")

    @classmethod
    def from_env(cls) -> "CrawlerConfig":
        """从环境变量加载配置"""
        proxy = cls._parse_proxy_env()
        return cls(proxy=proxy)

    @staticmethod
    def _parse_proxy_env() -> Optional[ProxyConfig]:
        """解析代理环境变量"""
        proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")

        if not proxy_url:
            return None

        parsed = urlparse(proxy_url)
        server = f"{parsed.scheme}://{parsed.hostname}"

        if parsed.port:
            server += f":{parsed.port}"

        return ProxyConfig(
            server=server,
            username=parsed.username,
            password=parsed.password
        )

    def validate_proxy(self) -> None:
        """验证代理配置

        Raises:
            ValueError: 如果 proxy_required=True 但未配置代理
        """
        if self.proxy_required and not self.proxy:
            raise ValueError(
                "代理配置缺失！请设置环境变量 HTTP_PROXY 或 HTTPS_PROXY。\n"
                "示例: export HTTPS_PROXY=http://127.0.0.1:7890"
            )


def get_crawler_config() -> CrawlerConfig:
    """获取爬虫配置单例"""
    return CrawlerConfig.from_env()

2. 创建 tests/test_crawler/test_proxy_config.py:

import os
import pytest
from unittest.mock import patch

from kb_workflow.crawler.config import (
    CrawlerConfig,
    ProxyConfig,
    get_crawler_config
)


class TestProxyConfig:

    def test_parse_proxy_from_https_env(self):
        """从 HTTPS_PROXY 解析代理"""
        with patch.dict(os.environ, {"HTTPS_PROXY": "http://127.0.0.1:7890"}):
            config = CrawlerConfig.from_env()
            assert config.proxy is not None
            assert config.proxy.server == "http://127.0.0.1:7890"

    def test_parse_proxy_from_http_env(self):
        """从 HTTP_PROXY 解析代理（HTTPS_PROXY 优先）"""
        with patch.dict(os.environ, {
            "HTTP_PROXY": "http://127.0.0.1:8080",
            "HTTPS_PROXY": "http://127.0.0.1:7890"
        }, clear=False):
            config = CrawlerConfig.from_env()
            assert config.proxy.server == "http://127.0.0.1:7890"

    def test_parse_proxy_with_auth(self):
        """解析带认证的代理"""
        with patch.dict(os.environ, {
            "HTTPS_PROXY": "http://user:pass@proxy.example.com:8080"
        }):
            config = CrawlerConfig.from_env()
            assert config.proxy.username == "user"
            assert config.proxy.password == "pass"

    def test_no_proxy_configured(self):
        """未配置代理"""
        with patch.dict(os.environ, {}, clear=True):
            # 清除所有代理相关环境变量
            os.environ.pop("HTTP_PROXY", None)
            os.environ.pop("HTTPS_PROXY", None)
            config = CrawlerConfig.from_env()
            assert config.proxy is None

    def test_proxy_required_validation_fails(self):
        """强制代理但未配置时抛出异常"""
        config = CrawlerConfig(proxy=None, proxy_required=True)
        with pytest.raises(ValueError, match="代理配置缺失"):
            config.validate_proxy()

    def test_proxy_required_validation_passes(self):
        """配置代理后验证通过"""
        config = CrawlerConfig(
            proxy=ProxyConfig(server="http://127.0.0.1:7890"),
            proxy_required=True
        )
        config.validate_proxy()  # 不应抛出异常

    def test_proxy_not_required(self):
        """不强制代理时验证通过"""
        config = CrawlerConfig(proxy=None, proxy_required=False)
        config.validate_proxy()  # 不应抛出异常

    def test_to_playwright_format(self):
        """转换为 Playwright 格式"""
        proxy = ProxyConfig(
            server="http://127.0.0.1:7890",
            username="user",
            password="pass"
        )
        result = proxy.to_playwright_format()
        assert result == {
            "server": "http://127.0.0.1:7890",
            "username": "user",
            "password": "pass"
        }

pytest tests/test_crawler/test_proxy_config.py -x --tb=short - CrawlerConfig 类定义完成 - 从 HTTP_PROXY/HTTPS_PROXY 读取代理 - proxy_required=True 时未配置代理抛出异常 - 所有测试通过 Task 2: 整合代理和延迟到 ZhihuCrawler kb_workflow/crawler/zhihu_crawler.py, tests/test_crawler/test_rate_limiter.py, tests/test_crawler/test_api_intercept.py - Test 1: 未配置代理且 proxy_required=True 时抛出异常 - Test 2: 请求间有随机延迟 (1-3 秒) - Test 3: 失败请求重试 3 次，间隔递增 - Test 4: API 拦截正确处理 x-zse-96 签名（由浏览器自动处理） 更新 ZhihuCrawler 类，整合代理配置、随机延迟和重试机制。

1. 更新 kb_workflow/crawler/zhihu_crawler.py:

import random
import time
from typing import Optional, Dict, Any
from playwright.sync_api import Route, Response
from tenacity import Retrying, stop_after_attempt, wait_exponential, RetryError

from kb_workflow.metadata.zhihu_extractor import PlaywrightBrowser
from kb_workflow.pipeline.config import get_settings
from .models import AnswerData, ArticleData, ContentMetadata, AuthorInfo, CrawlResult
from .config import CrawlerConfig, get_crawler_config


class ZhihuCrawler:
    """知乎内容爬虫

    支持:
    - 强制代理配置
    - 随机请求延迟
    - 有限重试
    - 自动处理 x-zse-96 签名
    """

    # ... URL patterns 等现有代码 ...

    def __init__(
        self,
        headless: bool = True,
        timeout: int = 30,
        config: Optional[CrawlerConfig] = None
    ):
        self.headless = headless
        self.timeout = timeout
        self.config = config or get_crawler_config()
        self.browser: Optional[PlaywrightBrowser] = None
        self.interceptor: Optional[ZhihuApiInterceptor] = None

    def _get_cookie(self) -> str:
        """获取 Cookie"""
        settings = get_settings()
        return settings.zhihu_cookie.get_secret_value()

    def _apply_random_delay(self) -> None:
        """应用随机延迟"""
        delay = random.uniform(self.config.min_delay, self.config.max_delay)
        time.sleep(delay)

    def _start_browser_with_proxy(self, cookie: str) -> None:
        """启动浏览器并应用代理配置"""
        # 验证代理
        self.config.validate_proxy()

        from playwright.sync_api import sync_playwright

        self.playwright = sync_playwright().start()
        browser_args = {"headless": self.headless}

        # 应用代理
        context_args = {
            'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }

        if self.config.proxy:
            context_args["proxy"] = self.config.proxy.to_playwright_format()

        # 阻止 Service Worker 干扰 API 拦截
        context_args["service_workers"] = "block"

        self.browser = self.playwright.chromium.launch(**browser_args)
        self.context = self.browser.new_context(**context_args)

        # 添加 Cookie
        if cookie:
            self._add_cookies(cookie)

        self.page = self.context.new_page()
        self.page.set_default_timeout(self.timeout * 1000)

    def _add_cookies(self, cookie_str: str) -> None:
        """添加 Cookie 到浏览器上下文"""
        cookies = []
        for item in cookie_str.split(';'):
            item = item.strip()
            if '=' in item:
                name, value = item.split('=', 1)
                cookies.append({
                    'name': name.strip(),
                    'value': value.strip(),
                    'domain': '.zhihu.com',
                    'path': '/',
                    'secure': True,
                    'httpOnly': False,
                    'sameSite': 'Lax'
                })
        self.context.add_cookies(cookies)

    def _close_browser(self) -> None:
        """关闭浏览器"""
        if hasattr(self, 'page') and self.page:
            self.page.close()
        if hasattr(self, 'context') and self.context:
            self.context.close()
        if hasattr(self, 'browser') and self.browser:
            self.browser.close()
        if hasattr(self, 'playwright'):
            self.playwright.stop()

    def _crawl_with_retry(self, crawl_func, *args, **kwargs):
        """带重试的爬取"""
        try:
            for attempt in Retrying(
                stop=stop_after_attempt(self.config.max_retries),
                wait=wait_exponential(
                    multiplier=1,
                    min=self.config.retry_min_wait,
                    max=self.config.retry_max_wait
                ),
                reraise=True
            ):
                with attempt:
                    # 随机延迟
                    self._apply_random_delay()
                    return crawl_func(*args, **kwargs)
        except RetryError as e:
            return CrawlResult(
                success=False,
                error=f"重试 {self.config.max_retries} 次后仍失败: {str(e)}"
            )

    def crawl_answer(self, url: str, include_comments: bool = False) -> CrawlResult:
        """爬取知乎回答（带重试）"""
        return self._crawl_with_retry(self._crawl_answer_impl, url, include_comments)

    def _crawl_answer_impl(self, url: str, include_comments: bool = False) -> CrawlResult:
        """爬取知乎回答实现"""
        import re

        match = re.search(self.ANSWER_PATTERN, url)
        if not match:
            return CrawlResult(success=False, error="Invalid answer URL")

        question_id, answer_id = match.groups()

        try:
            self._start_browser_with_proxy(self._get_cookie())
            self.interceptor = ZhihuApiInterceptor()

            # 设置路由拦截 - x-zse-96 签名由浏览器自动处理
            self.page.route("**/api/v4/**", self.interceptor.handle_route)

            # 访问页面触发 API 请求
            self.page.goto(url, wait_until="networkidle")

            # 等待数据捕获
            time.sleep(2)

            if "answer" not in self.interceptor.captured_data:
                return CrawlResult(success=False, error="Failed to capture answer data")

            answer_data = self.interceptor.captured_data["answer"]

            result_data = AnswerData(
                id=int(answer_id),
                question_id=int(question_id),
                question_title=answer_data.get("question", {}).get("title", ""),
                content=answer_data.get("content", ""),
                excerpt=answer_data.get("excerpt", ""),
                metadata=self._parse_metadata(answer_data)
            )

            result = CrawlResult(success=True, content_type="answer", data=result_data)

            # 可选：爬取评论
            if include_comments:
                comments_result = self.crawl_comments(
                    content_type="answers",
                    content_id=int(answer_id)
                )
                result.comments = comments_result.comments

            return result

        except Exception as e:
            return CrawlResult(success=False, error=str(e))
        finally:
            self._close_browser()

    def crawl_article(self, url: str, include_comments: bool = False) -> CrawlResult:
        """爬取知乎专栏文章（带重试）"""
        return self._crawl_with_retry(self._crawl_article_impl, url, include_comments)

    def _crawl_article_impl(self, url: str, include_comments: bool = False) -> CrawlResult:
        """爬取知乎专栏文章实现"""
        # ... 类似 crawl_answer 的实现 ...
        pass

    # crawl_comments 方法保持不变
    # ...

2. 创建 tests/test_crawler/test_rate_limiter.py:

import pytest
import time
from unittest.mock import Mock, patch, MagicMock

from kb_workflow.crawler.zhihu_crawler import ZhihuCrawler
from kb_workflow.crawler.config import CrawlerConfig


class TestRateLimiter:

    def test_random_delay_applied(self):
        """请求间有随机延迟"""
        config = CrawlerConfig(
            min_delay=0.1,
            max_delay=0.2,
            proxy_required=False
        )
        crawler = ZhihuCrawler(config=config)

        start = time.time()
        crawler._apply_random_delay()
        elapsed = time.time() - start

        assert 0.1 <= elapsed <= 0.25  # 允许少量误差

    def test_delay_is_random(self):
        """延迟是随机的"""
        config = CrawlerConfig(
            min_delay=0.05,
            max_delay=0.15,
            proxy_required=False
        )
        crawler = ZhihuCrawler(config=config)

        delays = []
        for _ in range(10):
            start = time.time()
            crawler._apply_random_delay()
            delays.append(time.time() - start)

        # 延迟应该有变化
        assert max(delays) - min(delays) > 0.01

    def test_exponential_backoff_retry(self):
        """重试间隔递增"""
        config = CrawlerConfig(
            max_retries=3,
            retry_min_wait=0.1,
            retry_max_wait=1.0,
            proxy_required=False
        )
        crawler = ZhihuCrawler(config=config)

        call_times = []

        def failing_func():
            call_times.append(time.time())
            raise Exception("Simulated failure")

        result = crawler._crawl_with_retry(failing_func)

        assert result.success is False
        assert "重试 3 次后仍失败" in result.error
        assert len(call_times) == 3  # 初始 + 2 次重试

3. 创建 tests/test_crawler/test_api_intercept.py:

import pytest
from unittest.mock import Mock, MagicMock

from kb_workflow.crawler.zhihu_crawler import ZhihuApiInterceptor


class TestApiIntercept:

    def test_intercept_answer_api(self):
        """拦截答案 API"""
        interceptor = ZhihuApiInterceptor()

        # 模拟 route 对象
        route = Mock()
        route.request.url = "https://www.zhihu.com/api/v4/answers/12345"

        mock_response = Mock()
        mock_response.json.return_value = {"id": 12345, "content": "test"}
        route.fetch.return_value = mock_response

        interceptor.handle_route(route)

        assert "answer" in interceptor.captured_data
        assert interceptor.captured_data["answer"]["id"] == 12345

    def test_intercept_article_api(self):
        """拦截文章 API"""
        interceptor = ZhihuApiInterceptor()

        route = Mock()
        route.request.url = "https://www.zhihu.com/api/v4/articles/67890"

        mock_response = Mock()
        mock_response.json.return_value = {"id": 67890, "title": "test article"}
        route.fetch.return_value = mock_response

        interceptor.handle_route(route)

        assert "article" in interceptor.captured_data

    def test_non_api_requests_continue(self):
        """非 API 请求继续正常加载"""
        interceptor = ZhihuApiInterceptor()

        route = Mock()
        route.request.url = "https://www.zhihu.com/static/some.js"

        interceptor.handle_route(route)

        route.continue_.assert_called_once()
        route.fulfill.assert_not_called()

    def test_x_zse_96_auto_handled(self):
        """x-zse-96 签名由浏览器自动处理（通过 route.fetch）"""
        interceptor = ZhihuApiInterceptor()

        route = Mock()
        route.request.url = "https://www.zhihu.com/api/v4/answers/123"

        interceptor.handle_route(route)

        # fetch() 会自动带上浏览器生成的签名
        route.fetch.assert_called_once()
        # fulfill() 返回响应给页面
        route.fulfill.assert_called_once()

pytest tests/test_crawler/test_rate_limiter.py tests/test_crawler/test_api_intercept.py -x --tb=short - 代理配置整合到 ZhihuCrawler - 随机延迟 1-3 秒生效 - 失败重试 3 次，间隔递增 - x-zse-96 签名由浏览器自动处理 - 所有测试通过 **自动化验证:** ```bash pytest tests/test_crawler/ -x -v ```

代理配置验证:

# 不配置代理时应报错
unset HTTP_PROXY HTTPS_PROXY
python -c "from kb_workflow.crawler import ZhihuCrawler; ZhihuCrawler().crawl_answer('https://www.zhihu.com/question/1/answer/1')"
# 应输出: ValueError: 代理配置缺失！

<success_criteria>

1. 代理配置从 HTTP_PROXY/HTTPS_PROXY 读取
2. proxy_required=True 时未配置代理抛出异常
3. 请求间有随机延迟 (1-3 秒)
4. 失败请求重试 3 次，间隔递增
5. x-zse-96 签名由浏览器自动处理
6. 所有测试通过 </success_criteria>

After completion, create `.planning/phases/08-crawler/08-03-SUMMARY.md`