starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(10): research proxy pool domain

Browse Source
This commit is contained in:
starmen1265
2026-03-18 16:18:38 +08:00
parent 1b58bf9364
commit fe22f94358
1 changed files with 599 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.planning/phases/10-proxy-pool/10-RESEARCH.md
+599
View File
@@ -0,0 +1,599 @@
# Phase 10: Proxy Pool (代理池管理) - Research
**Researched:** 2026-03-18
**Domain:** Python Proxy Pool Management / Playwright Proxy Integration
**Confidence:** HIGH
## Summary
本研究针对代理池基础设施的实现方案，重点解决多代理配置、自动轮换、健康检查和失效回退四个核心需求。基于现有 `kb_workflow/crawler/config.py` 的单代理实现，推荐采用 **ProxyPool 类 + 被动健康检测 + 冷却恢复** 的架构模式。
**Primary recommendation:** 新建 `kb_workflow/crawler/proxy_pool.py` 模块，实现 ProxyPool 类，使用 itertools.cycle 进行顺序轮换，通过惰性检查实现代理冷却恢复，与现有 CrawlerConfig 无缝集成。
<user_constraints>
## User Constraints (from CONTEXT.md)
### Implementation Decisions (Locked)
#### 代理池配置方式
- **配置来源**: YAML 文件（`~/.kb_proxies.yaml`）
- **配置格式**: 简单 URL 列表，认证信息嵌入 URL
```yaml
proxies:
- "http://user:pass@proxy1.example.com:8080"
- "http://proxy2.example.com:3128"
- "socks5://user:pass@proxy3.example.com:1080"
```
- **向后兼容**: 如果未配置代理池文件，自动回退到现有的 `HTTP_PROXY` 环境变量（单代理模式）
- **支持的协议**: HTTP、HTTPS、SOCKS5
#### 轮换策略
- **策略**: 顺序轮换（Round-Robin）
- **时机**: 每次请求自动轮换到下一个代理
- **实现**: 维护一个索引指针，每次请求后递增并循环
#### 健康检查机制
- **检测方式**: 被动检测（请求失败时标记）
- **失效判断**: 连续 3 次请求失败后标记为失效
- **失败类型**: 网络超时、连接失败、代理返回错误
#### 失效代理处理
- **策略**: 暂时移除 + 冷却恢复
- **冷却时间**: 5 分钟后自动恢复到轮换池
- **全失效处理**: 所有代理都失效时，回退到第一个代理继续尝试请求
### Claude's Discretion
- 冷却时间的具体实现方式（定时器 vs 惰性检查）
- 代理统计信息的记录格式（可选功能）
- 日志输出的详细程度
- 失效代理的具体判断条件（HTTP 状态码范围）
### Deferred Ideas (OUT OF SCOPE)
- **API 获取代理** — 从代理服务商 API 自动获取代理
- **加权轮换** — 根据代理质量/速度加权
- **代理统计面板** — 展示每个代理的成功率、响应时间
- **配置热加载** — 运行时检测配置文件变化并重新加载
</user_constraints>
<phase_requirements>
## Phase Requirements
| ID | Description | Research Support |
|----|-------------|-----------------|
| PROXY-01 | 用户可以配置代理池 — 支持从文件或环境变量加载多个代理地址 | YAML 配置 + Pydantic BaseModel 解析 + 向后兼容环境变量 |
| PROXY-02 | 系统自动轮换代理 — 每次请求自动选择下一个可用代理 | itertools.cycle + 索引指针实现 Round-Robin |
| PROXY-03 | 代理健康检查 — 检测代理可用性，自动移除失效代理 | 被动检测（失败计数）+ 连续3次失败标记失效 |
| PROXY-04 | 代理失败回退 — 当前代理失败时自动切换到下一个代理 | 失效代理跳过 + 冷却恢复 + 全失效回退策略 |
</phase_requirements>
## Standard Stack
### Core
| Library | Version | Purpose | Why Standard |
|---------|---------|---------|--------------|
| Pydantic | 2.11+ | ProxyConfig 数据模型 | 项目已使用；类型安全；支持 SecretStr 敏感字段 |
| pydantic-settings | 2.8+ | 配置管理 | 项目已使用；支持 YAML/环境变量加载 |
| PyYAML | 6.0+ | YAML 配置解析 | Python 标准库；轻量级 |
| itertools (stdlib) | - | cycle() 实现轮换 | 零依赖；高效无限迭代器 |
### Supporting
| Library | Version | Purpose | When to Use |
|---------|---------|---------|-------------|
| tenacity | 8.5+ | 重试机制（现有） | 代理失败时的重试策略 |
| loguru | 0.7+ | 日志记录 | 代理状态变化日志 |
### Alternatives Considered
| Instead of | Could Use | Tradeoff |
|------------|-----------|----------|
| 自定义 ProxyPool | rotating-proxy 库 | 代码质量差、维护不活跃、不支持自定义配置源 |
| 自定义 ProxyPool | scrapy-rotating-proxies | 与 Scrapy 绑定、不适用于 Playwright |
| 被动检测 | 主动健康检查 | 主动检测增加延迟和复杂度；被动检测足够满足需求 |
| 惰性冷却检查 | 定时器后台任务 | 定时器需要线程/异步；惰性检查更简单、无并发问题 |
**Installation:**
```bash
# 项目已有依赖，无需新增
# Pydantic, pydantic-settings, PyYAML 均已安装
```
## Architecture Patterns
### Recommended Project Structure
```
kb_workflow/crawler/
├── config.py # 现有：CrawlerConfig, ProxyConfig（扩展）
├── proxy_pool.py # 新增：ProxyPool, ProxyEntry, ProxyPoolConfig
├── zhihu_crawler.py # 修改：使用 ProxyPool
└── models.py # 现有：CrawlResult 等
```
### Pattern 1: ProxyPool 类设计
**What:** 代理池核心类，管理多个代理的生命周期
**When to use:** 每个爬虫实例一个 ProxyPool，处理该爬虫的所有代理相关逻辑
**Example:**
```python
# Source: 基于现有代码模式和 Infatica 最佳实践
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional, Iterator
from itertools import cycle
from pathlib import Path
import yaml
from pydantic import BaseModel
import logging
logger = logging.getLogger(__name__)
class ProxyEntry(BaseModel):
"""单个代理的配置和状态"""
url: str # 完整 URL，包含认证信息
fail_count: int = 0 # 连续失败次数
last_fail_time: Optional[datetime] = None # 最后失败时间
is_available: bool = True # 当前是否可用
def record_success(self):
"""记录请求成功"""
self.fail_count = 0
self.is_available = True
def record_failure(self):
"""记录请求失败"""
self.fail_count += 1
self.last_fail_time = datetime.now()
if self.fail_count >= 3:
self.is_available = False
logger.warning(f"Proxy marked as unavailable: {self.url[:30]}...")
def check_recovery(self, cooldown_seconds: int = 300) -> bool:
"""检查是否可以从冷却中恢复（惰性检查）"""
if self.is_available:
return True
if self.last_fail_time is None:
return True
elapsed = (datetime.now() - self.last_fail_time).total_seconds()
if elapsed >= cooldown_seconds:
self.is_available = True
self.fail_count = 0
logger.info(f"Proxy recovered after {elapsed:.0f}s cooldown")
return True
return False
class ProxyPoolConfig(BaseModel):
"""代理池配置"""
proxies: list[str] = [] # 代理 URL 列表
cooldown_seconds: int = 300 # 冷却时间（秒）
max_fail_count: int = 3 # 最大连续失败次数
class ProxyPool:
"""代理池管理器
功能：
- 从 YAML 文件或环境变量加载代理列表
- Round-Robin 轮换
- 被动健康检测
- 冷却恢复
"""
def __init__(self, config: ProxyPoolConfig):
self.config = config
self._entries: list[ProxyEntry] = [
ProxyEntry(url=url) for url in config.proxies
]
self._current_index = 0
self._cycle: Optional[Iterator] = None
if self._entries:
self._cycle = cycle(self._entries)
@classmethod
def from_yaml(cls, path: Path) -> "ProxyPool":
"""从 YAML 文件加载代理池"""
if not path.exists():
logger.debug(f"Proxy config file not found: {path}")
return cls(ProxyPoolConfig())
with open(path) as f:
data = yaml.safe_load(f) or {}
proxies = data.get("proxies", [])
return cls(ProxyPoolConfig(proxies=proxies))
@classmethod
def from_env_fallback(cls) -> "ProxyPool":
"""从环境变量加载单代理（向后兼容）"""
import os
proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
if proxy_url:
return cls(ProxyPoolConfig(proxies=[proxy_url]))
return cls(ProxyPoolConfig())
def get_next_proxy(self) -> Optional[ProxyEntry]:
"""获取下一个可用代理
实现逻辑：
1. 惰性检查当前代理是否可恢复
2. 跳过不可用的代理
3. 所有代理都失效时，回退到第一个代理
"""
if not self._entries:
return None
if len(self._entries) == 1:
# 单代理模式：总是返回该代理
return self._entries[0]
# 尝试找到可用代理
checked = set()
while len(checked) < len(self._entries):
entry = self._entries[self._current_index]
self._current_index = (self._current_index + 1) % len(self._entries)
checked.add(self._current_index)
# 惰性检查恢复
if entry.check_recovery(self.config.cooldown_seconds):
return entry
# 所有代理都失效，回退到第一个代理强制使用
logger.warning("All proxies unavailable, falling back to first proxy")
return self._entries[0]
def has_proxies(self) -> bool:
"""检查是否有配置的代理"""
return len(self._entries) > 0
def get_available_count(self) -> int:
"""获取当前可用代理数量"""
return sum(
1 for e in self._entries
if e.check_recovery(self.config.cooldown_seconds)
)
```
### Pattern 2: CrawlerConfig 集成
**What:** 扩展现有 CrawlerConfig 支持 ProxyPool
**When to use:** 爬虫初始化时加载代理池配置
**Example:**
```python
# Source: 基于现有 config.py 扩展
from pathlib import Path
def get_crawler_config_with_pool() -> tuple["CrawlerConfig", ProxyPool]:
"""获取爬虫配置和代理池
优先级：
1. ~/.kb_proxies.yaml（代理池）
2. HTTP_PROXY/HTTPS_PROXY（单代理，向后兼容）
"""
proxy_file = Path.home() / ".kb_proxies.yaml"
proxy_pool = ProxyPool.from_yaml(proxy_file)
if not proxy_pool.has_proxies():
# 回退到环境变量
proxy_pool = ProxyPool.from_env_fallback()
# 原有配置逻辑
config = CrawlerConfig.from_env()
return config, proxy_pool
```
### Pattern 3: ZhihuCrawler 集成
**What:** 修改 ZhihuCrawler 使用 ProxyPool
**When to use:** 每次启动浏览器时从代理池获取代理
**Example:**
```python
# Source: 基于现有 zhihu_crawler.py 修改
class ZhihuCrawler:
def __init__(
self,
headless: bool = True,
timeout: int = 30,
config: Optional[CrawlerConfig] = None,
proxy_pool: Optional[ProxyPool] = None
):
self.headless = headless
self.timeout = timeout
self.config = config or get_crawler_config()
self.proxy_pool = proxy_pool # 新增
# ...
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 = {
'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...'
}
# 从代理池获取代理
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()
context_options["service_workers"] = "block"
self._context = self._browser.new_context(**context_options)
# ...
def _on_request_success(self):
"""请求成功时通知代理池"""
if hasattr(self, '_current_proxy') and self._current_proxy:
self._current_proxy.record_success()
def _on_request_failure(self):
"""请求失败时通知代理池"""
if hasattr(self, '_current_proxy') and self._current_proxy:
self._current_proxy.record_failure()
```
### Anti-Patterns to Avoid
- **每次请求创建新浏览器** — Playwright 浏览器启动成本高，应在同一代理上处理多个请求后再切换
- **主动健康检查** — 增加延迟；被动检测足够
- **硬编码冷却时间** — 应通过配置文件可配置
- **忽略 SOCKS5 代理** — 需要额外依赖 `requests[socks]`，Playwright 原生支持
## Don't Hand-Roll
| Problem | Don't Build | Use Instead | Why |
|---------|-------------|-------------|-----|
| 代理轮换逻辑 | 手写 while 循环 | `itertools.cycle()` | 边界处理复杂、容易死循环 |
| 配置解析 | 手写正则提取 URL | `urllib.parse.urlparse()` | URL 格式多变、边界情况多 |
| 代理认证提取 | 字符串 split | Pydantic BaseModel | 类型安全、自动验证 |
| 冷却时间计算 | 手写时间差 | `datetime.timedelta` | 时区、精度问题 |
**Key insight:** 代理池看似简单，但边界情况极多（URL 解析、认证、协议差异、失败重试）。使用标准库和 Pydantic 可以避免大部分陷阱。
## Common Pitfalls
### Pitfall 1: 代理配置不完整导致真实 IP 泄漏
**What goes wrong:** 只配置了 HTTP 代理但未配置 HTTPS 代理，或 Playwright proxy 配置不完整，导致部分请求绕过代理暴露真实 IP。
**Why it happens:** 开发者通常只关注主要请求的代理配置，忽略了 Playwright 的 `proxy` 参数需要正确格式。
**How to avoid:**
```python
# 正确的 Playwright 代理配置
context_options = {
"proxy": {
"server": "http://proxy.example.com:8080",
"username": "user", # 可选
"password": "pass", # 可选
}
}
```
**Warning signs:**
- 使用代理后仍然收到 403 错误
- 目标平台显示真实地理位置
- IP 检测 API 返回真实 IP 而非代理 IP
### Pitfall 2: 固定间隔轮换被识别为机器人
**What goes wrong:** 每次请求后立即切换代理，顺序固定（A→B→C→A），被反爬系统识别为自动化行为。
**Why it happens:** Round-Robin 是最简单的实现，但对于个人知识库场景（低频爬取），顺序轮换配合随机请求延迟已经足够。
**How to avoid:** 本项目采用 **会话级轮换**（每个浏览器会话使用一个代理），而非请求级轮换。结合 Phase 11 的随机延迟（1-5秒），风险可控。
**Warning signs:**
- 新代理 IP 很快被封
- 频繁收到 CAPTCHA 挑战
### Pitfall 3: 代理池全失效时无限循环
**What goes wrong:** 所有代理都被标记为失效后，`get_next_proxy()` 返回 None 或进入死循环。
**Why it happens:** 只考虑了正常情况，没有处理边界条件。
**How to avoid:**
```python
def get_next_proxy(self) -> Optional[ProxyEntry]:
# ... 轮换逻辑 ...
# 所有代理都失效时，回退到第一个代理
logger.warning("All proxies unavailable, falling back to first proxy")
return self._entries[0] # 强制使用第一个
```
### Pitfall 4: 代理 URL 解析错误
**What goes wrong:** SOCKS5 代理格式为 `socks5://user:pass@host:port`，但解析时只处理了 HTTP 格式。
**Why it happens:** 不同协议的 URL 格式略有差异，手动解析容易遗漏。
**How to avoid:**
```python
from urllib.parse import urlparse
def parse_proxy_url(url: str) -> dict:
"""解析代理 URL 为 Playwright 格式"""
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
```
## Code Examples
### 配置文件示例 (~/.kb_proxies.yaml)
```yaml
# KB Skills 代理池配置
# 支持格式：http://[user:pass@]host:port, socks5://[user:pass@]host:port
proxies:
# 本地代理（Clash/V2Ray）
- "http://127.0.0.1:7890"
# 带认证的 HTTP 代理
- "http://user:password@proxy.example.com:8080"
# SOCKS5 代理
- "socks5://127.0.0.1:1080"
# 可选：冷却时间（秒），默认 300
cooldown_seconds: 300
# 可选：最大连续失败次数，默认 3
max_fail_count: 3
```
### 代理池初始化
```python
# Source: 推荐的初始化模式
from pathlib import Path
from kb_workflow.crawler.proxy_pool import ProxyPool
def init_proxy_pool() -> ProxyPool:
"""初始化代理池，优先从配置文件加载"""
proxy_file = Path.home() / ".kb_proxies.yaml"
# 优先级 1: YAML 配置文件
if proxy_file.exists():
pool = ProxyPool.from_yaml(proxy_file)
if pool.has_proxies():
return pool
# 优先级 2: 环境变量（向后兼容）
pool = ProxyPool.from_env_fallback()
return pool
```
### 爬虫使用代理池
```python
# Source: ZhihuCrawler 集成示例
from kb_workflow.crawler.config import get_crawler_config
from kb_workflow.crawler.proxy_pool import ProxyPool, init_proxy_pool
# 初始化
config = get_crawler_config()
proxy_pool = init_proxy_pool()
# 创建爬虫
crawler = ZhihuCrawler(
headless=True,
timeout=30,
config=config,
proxy_pool=proxy_pool
)
# 爬取时代理自动轮换
result = crawler.crawl_answer("https://www.zhihu.com/question/123/answer/456")
```
## State of the Art
| Old Approach | Current Approach | When Changed | Impact |
|--------------|------------------|--------------|--------|
| 单代理硬编码 | 环境变量配置 | v1.2 | 更灵活，支持不同环境 |
| 环境变量单代理 | YAML 配置文件代理池 | v1.3 (Phase 10) | 支持多代理、轮换、健康检查 |
| 主动健康检查 | 被动检测 | v1.3 (Phase 10) | 减少延迟，简化实现 |
| 立即移除失效代理 | 冷却恢复 | v1.3 (Phase 10) | 代理可恢复使用，提高利用率 |
**Deprecated/outdated:**
- `rotating-proxy` 库：2020年后无更新，代码质量差
- `scrapy-rotating-proxies`：与 Scrapy 绑定，不适用于 Playwright
## Open Questions
1. **冷却检查时机：惰性检查 vs 定时器？**
- What we know: 惰性检查在 `get_next_proxy()` 时执行，无需额外线程/异步
- What's unclear: 高并发场景下惰性检查的性能影响
- Recommendation: 本项目为单线程顺序执行，惰性检查足够；如未来需要并发，可迁移到定时器
2. **代理统计信息是否需要持久化？**
- What we know: 当前设计只在内存中维护状态
- What's unclear: 是否需要将代理成功率持久化到文件
- Recommendation: v1.3 暂不持久化；如后续需要统计面板，再添加
## Validation Architecture
### Test Framework
| Property | Value |
|----------|-------|
| Framework | pytest 8.3+ |
| Config file | pytest.ini (pythonpath = ., testpaths = tests) |
| Quick run command | `pytest tests/test_crawler/test_proxy_pool.py -x` |
| Full suite command | `pytest tests/test_crawler/ -v` |
### Phase Requirements → Test Map
| Req ID | Behavior | Test Type | Automated Command | File Exists? |
|--------|----------|-----------|-------------------|-------------|
| PROXY-01 | 从 YAML 文件加载代理池 | unit | `pytest tests/test_crawler/test_proxy_pool.py::test_load_from_yaml -x` | ❌ Wave 0 |
| PROXY-01 | 从环境变量加载代理（向后兼容） | unit | `pytest tests/test_crawler/test_proxy_pool.py::test_load_from_env -x` | ❌ Wave 0 |
| PROXY-02 | Round-Robin 轮换 | unit | `pytest tests/test_crawler/test_proxy_pool.py::test_round_robin -x` | ❌ Wave 0 |
| PROXY-03 | 连续3次失败标记失效 | unit | `pytest tests/test_crawler/test_proxy_pool.py::test_fail_threshold -x` | ❌ Wave 0 |
| PROXY-03 | 冷却恢复 | unit | `pytest tests/test_crawler/test_proxy_pool.py::test_cooldown_recovery -x` | ❌ Wave 0 |
| PROXY-04 | 全失效回退 | unit | `pytest tests/test_crawler/test_proxy_pool.py::test_all_unavailable_fallback -x` | ❌ Wave 0 |
| PROXY-04 | 代理切换后继续请求 | integration | `pytest tests/test_crawler/test_proxy_pool.py::test_proxy_switch -x` | ❌ Wave 0 |
### Sampling Rate
- **Per task commit:** `pytest tests/test_crawler/test_proxy_pool.py -x`
- **Per wave merge:** `pytest tests/test_crawler/ -v`
- **Phase gate:** Full suite green before `/gsd:verify-work`
### Wave 0 Gaps
- [ ] `tests/test_crawler/test_proxy_pool.py` — 代理池核心逻辑测试
- [ ] `tests/test_crawler/conftest.py` — 添加 proxy_pool fixture（已存在，可能需要扩展）
- [ ] 无需框架安装 — pytest 已配置
## Sources
### Primary (HIGH confidence)
- [Infatica: Master Proxy Rotation with Python's Requests Library](https://infatica.io/blog/python-proxy-requests/) — Round-Robin 实现、代理轮换最佳实践
- [Playwright Python Documentation - Network](https://playwright.dev/python/docs/network) — Playwright 代理配置官方文档
- [tenacity Documentation](https://tenacity.readthedocs.io/) — Python 重试库官方文档（现有依赖）
### Secondary (MEDIUM confidence)
- [Oxylabs: How to Rotate Proxies in Python](https://oxylabs.io/blog/rotate-proxies-python) — 代理轮换策略概述
- [Stack Overflow: Round Robin proxy rotation](https://stackoverflow.com/questions/74110824/how-to-make-different-session-proxy-with-each-request) — 实现代码示例
### Tertiary (LOW confidence)
- [DEV.to: Proxy Rotation for Web Scraping in 2026](https://dev.to/agenthustler/proxy-rotation-for-web-scraping-in-2026-the-complete-guide-with-code-23d) — 通用指南（需验证）
### Code Analysis (HIGH confidence)
- `kb_workflow/crawler/config.py` — 现有代理配置实现
- `kb_workflow/crawler/zhihu_crawler.py` — 现有爬虫实现
- `.planning/research/PITFALLS.md` — 反爬虫陷阱研究
- `.planning/research/STACK.md` — 技术栈研究
## Metadata
**Confidence breakdown:**
- Standard stack: HIGH — 基于现有项目依赖和官方文档
- Architecture: HIGH — 基于现有代码模式和最佳实践
- Pitfalls: HIGH — 基于项目 PITFALLS.md 和社区经验
**Research date:** 2026-03-18
**Valid until:** 30 days (稳定技术栈)