kb-skills/.planning/phases/10-proxy-pool/10-02-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
10-proxy-pool	02	execute	2	10-01	kb_workflow/crawler/proxy_pool.py kb_workflow/crawler/zhihu_crawler.py tests/test_crawler/test_proxy_pool.py	true	PROXY-03 PROXY-04		truths artifacts key_links 代理连续 3 次失败后自动标记为失效 失效代理 5 分钟后自动恢复到轮换池 当前代理失败时自动切换到下一个可用代理 所有代理都失效时回退到第一个代理继续请求 path provides min_lines kb_workflow/crawler/proxy_pool.py 健康检查、冷却恢复、代理状态管理 150 path provides contains kb_workflow/crawler/zhihu_crawler.py ProxyPool 集成，代理失败时切换 proxy_pool from to via pattern kb_workflow/crawler/zhihu_crawler.py kb_workflow/crawler/proxy_pool.py ProxyPool.get_next_proxy() proxy_pool.get_next_proxy from to via pattern ZhihuCrawler._start_browser_with_proxy ProxyEntry 获取代理并传递给 Playwright proxy_entry.url

实现代理健康检查和失效切换机制，并集成到 ZhihuCrawler 中。

Purpose: 当代理失败时自动切换到备用代理，提升爬取稳定性 Output: 健康检查逻辑 + ZhihuCrawler 代理池集成

<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/10-proxy-pool/10-01-SUMMARY.md

From kb_workflow/crawler/proxy_pool.py:

class ProxyEntry(BaseModel):
    url: str
    fail_count: int = 0
    last_fail_time: Optional[datetime] = None
    is_available: bool = True

class ProxyPoolConfig(BaseModel):
    proxies: list[str] = []
    cooldown_seconds: int = 300
    max_fail_count: int = 3

class ProxyPool:
    def __init__(self, config: ProxyPoolConfig): ...
    def get_next_proxy(self) -> Optional[ProxyEntry]: ...
    def has_proxies(self) -> bool: ...
    @classmethod
    def from_yaml(cls, path: Path) -> "ProxyPool": ...
    @classmethod
    def from_env_fallback(cls) -> "ProxyPool": ...

From kb_workflow/crawler/zhihu_crawler.py (current):

class ZhihuCrawler:
    def __init__(
        self,
        headless: bool = True,
        timeout: int = 30,
        config: Optional[CrawlerConfig] = None
    ): ...

    def _start_browser_with_proxy(self, cookie: str) -> None:
        # Currently uses self.config.proxy directly
        if self.config.proxy:
            context_options["proxy"] = self.config.proxy.to_playwright_format()

Task 1: 实现健康检查与冷却恢复 kb_workflow/crawler/proxy_pool.py, tests/test_crawler/test_proxy_pool.py - Test fail threshold: 连续调用 record_failure() 3 次后，is_available 变为 False - Test success resets count: 调用 record_success() 后，fail_count 重置为 0 - Test cooldown recovery: 失效代理在冷却时间（300秒）后 check_recovery() 返回 True - Test cooldown not elapsed: 失效代理在冷却时间内 check_recovery() 返回 False - Test all unavailable fallback: 所有代理都失效时 get_next_proxy() 返回第一个代理 - Test get available count: get_available_count() 返回正确的可用代理数 扩展 ProxyEntry 类添加健康检查方法：

1. record_success(self): 记录成功，重置 fail_count 为 0，设置 is_available = True
2. record_failure(self): 记录失败，fail_count += 1，记录 last_fail_time，如果 fail_count >= max_fail_count 则设置 is_available = False
3. check_recovery(self, cooldown_seconds: int) -> bool: 惰性检查，如果 is_available = True 返回 True；如果冷却时间已过，重置状态并返回 True；否则返回 False

扩展 ProxyPool 类：

1. get_next_proxy(self) -> Optional[ProxyEntry]: 修改轮换逻辑

   • 跳过 is_available = False 且未过冷却时间的代理
   • 在循环中调用 entry.check_recovery(self.config.cooldown_seconds)
   • 所有代理都不可用时，回退到 entries[0] 并记录 warning 日志

2. get_available_count(self) -> int: 返回当前可用代理数量

更新测试文件添加：

• test_fail_threshold
• test_success_resets_count
• test_cooldown_recovery
• test_cooldown_not_elapsed (需要 mock datetime 或使用很短的冷却时间)
• test_all_unavailable_fallback
• test_get_available_count

注意：冷却时间测试可以使用 cooldown_seconds=1 来加速测试。 pytest tests/test_crawler/test_proxy_pool.py -x - ProxyEntry.record_success() 和 record_failure() 方法正常工作 - 连续 3 次失败后代理被标记为失效 - 冷却时间后代理自动恢复 - 所有代理失效时回退到第一个代理 - 所有测试通过

Task 2: ZhihuCrawler 集成 ProxyPool kb_workflow/crawler/zhihu_crawler.py, kb_workflow/crawler/proxy_pool.py, tests/test_crawler/test_proxy_pool.py - Test crawler with proxy pool: ZhihuCrawler 接受 proxy_pool 参数并使用代理池中的代理 - Test proxy switch on failure: 代理失败时自动切换到下一个代理（集成测试） 修改 ZhihuCrawler 类以集成 ProxyPool：

1. 构造函数添加 proxy_pool 参数:

def __init__(
    self,
    headless: bool = True,
    timeout: int = 30,
    config: Optional[CrawlerConfig] = None,
    proxy_pool: Optional[ProxyPool] = None  # 新增
):
    ...
    self.proxy_pool = proxy_pool
    self._current_proxy: Optional[ProxyEntry] = None  # 跟踪当前代理

2. 修改 _start_browser_with_proxy 方法:

def _start_browser_with_proxy(self, cookie: str) -> None:
    self.config.validate_proxy()
    self._playwright = sync_playwright().start()
    self._browser = self._playwright.chromium.launch(headless=self.headless)

    context_options = {...}

    # 优先使用代理池
    if self.proxy_pool and self.proxy_pool.has_proxies():
        proxy_entry = self.proxy_pool.get_next_proxy()
        if proxy_entry:
            proxy_config = self._parse_proxy_url(proxy_entry.url)
            context_options["proxy"] = proxy_config
            self._current_proxy = proxy_entry
    elif self.config.proxy:
        # 向后兼容：使用单代理配置
        context_options["proxy"] = self.config.proxy.to_playwright_format()
    ...

3. 添加 _parse_proxy_url 辅助方法:

@staticmethod
def _parse_proxy_url(url: str) -> Dict[str, Any]:
    """解析代理 URL 为 Playwright 格式"""
    from urllib.parse import urlparse
    parsed = urlparse(url)
    server = f"{parsed.scheme}://{parsed.hostname}"
    if parsed.port:
        server += f":{parsed.port}"
    config = {"server": server}
    if parsed.username:
        config["username"] = parsed.username
    if parsed.password:
        config["password"] = parsed.password
    return config

4. 添加代理状态反馈方法:

def _on_request_success(self) -> None:
    """请求成功时通知代理池"""
    if self._current_proxy:
        self._current_proxy.record_success()

def _on_request_failure(self) -> None:
    """请求失败时通知代理池"""
    if self._current_proxy:
        self._current_proxy.record_failure()

5. 更新导入:

from .proxy_pool import ProxyPool, ProxyEntry

注意：实际的代理切换在浏览器重启时自动发生（每次 _start_browser_with_proxy 调用都会从代理池获取新代理）。

添加集成测试（使用 mock）验证 ZhihuCrawler 与 ProxyPool 的集成。 pytest tests/test_crawler/test_proxy_pool.py -x - ZhihuCrawler 构造函数接受 proxy_pool 参数 - _start_browser_with_proxy 优先使用代理池 - 代理失败时自动标记，下次请求使用新代理 - 向后兼容现有单代理配置 - 所有测试通过

**Wave 2 验证检查：** 1. `pytest tests/test_crawler/test_proxy_pool.py -x` 全部通过 2. 代理连续 3 次失败后被标记为失效 3. 失效代理 5 分钟后自动恢复 4. ZhihuCrawler 正确集成 ProxyPool 5. 所有代理失效时回退到第一个代理

<success_criteria>

• 用户可以从 YAML 文件加载多个代理地址（PROXY-01）
• 用户可以从环境变量加载单代理（PROXY-01）
• 每次请求自动选择下一个可用代理（PROXY-02）
• 代理连续 3 次失败后自动标记为失效（PROXY-03）
• 失效代理 5 分钟后自动恢复（PROXY-03）
• 当前代理失败时自动切换到备用代理（PROXY-04）
• 所有代理失效时回退到第一个代理（PROXY-04） </success_criteria>

After completion, create `.planning/phases/10-proxy-pool/10-02-SUMMARY.md`