MediaCrawler 接入豌豆HTTP代理：app_key 配置、IP 池缓存与自动换 IP 全解析

MediaCrawler 接入豌豆HTTP代理：app_key 配置、IP 池缓存与自动换 IP 全解析

【免费下载链接】MediaCrawler小红书笔记 | 评论爬虫、抖音视频 | 评论爬虫、快手视频 | 评论爬虫、B 站视频 ｜ 评论爬虫、微博帖子 ｜ 评论爬虫、百度贴吧帖子 ｜ 百度贴吧评论回复爬虫 | 知乎问答文章｜评论爬虫项目地址: https://gitcode.com/GitHub_Trending/me/MediaCrawler

本文基于 MediaCrawler 仓库中的 豌豆HTTP使用文档 展开，完整讲解如何在 MediaCrawler 中接入豌豆HTTP 代理：从注册实名、获取 app_key，到通过环境变量初始化代理实例，再到源码级的 IP 拉取、Redis 缓存、过期检测与自动刷新机制。读完后你将能够独立配置豌豆HTTP 作为 MediaCrawler 的代理 IP 供应商，并理解代理池在爬虫请求链路中的完整工作原理。

一、准备豌豆HTTP 代理 IP 信息

豌豆HTTP 代理目前只支持企业用户（文档原文声明）。使用流程分为三步：注册实名、领取免费试用并选择套餐、获取开放接口密钥 app_key。

1. 注册并实名认证

访问豌豆HTTP 官网注册账号并完成实名认证。在国内使用代理 IP 服务必须实名，这是后续领取试用额度与调用开放接口的前提。

2. 领取免费试用与选择套餐

注册后可以从官网入口领取免费试用，如下图所示：

试用额度用尽后，需要根据自身爬取量选择对应套餐：

3. 获取 app_key

登录官网后，进入「个人中心 → 开放接口」页面，即可找到 app_key，如下图所示。这个 app_key 就是 MediaCrawler 初始化豌豆HTTP 代理实例时唯一需要的凭据参数：

二、在 MediaCrawler 中初始化豌豆HTTP 代理

原文档给出的最小初始化示例如下，只需要一个参数app_key，推荐通过环境变量注入，避免把密钥硬编码进代码：

# 文件地址： proxy/providers/wandou_http_proxy.py # -*- coding: utf-8 -*- def new_wandou_http_proxy() -> WanDouHttpProxy: """ 构造豌豆HTTP实例 Returns: """ return WanDouHttpProxy( app_key=os.getenv( "wandou_app_key", "你的豌豆HTTP app_key" ), # 通过环境变量的方式获取豌豆HTTP app_key )

查看仓库中的 实际工厂函数实现，它对上面示例做了增强：同时兼容两种环境变量命名格式，且优先使用大写格式：

def new_wandou_http_proxy() -> WanDouHttpProxy: """ Supports two environment variable naming formats: 1. Uppercase format: WANDOU_APP_KEY 2. Lowercase format: wandou_app_key Prioritize uppercase format, use lowercase format if not exists """ # Support both uppercase and lowercase environment variable formats, prioritize uppercase app_key = os.getenv("WANDOU_APP_KEY") or os.getenv("wandou_app_key", "your_wandou_http_app_key") return WanDouHttpProxy(app_key=app_key)

因此实际使用时只需设置环境变量（两种方式任选其一）：

export WANDOU_APP_KEY="你的豌豆HTTP app_key" # 推荐，大写格式优先 # 或 export wandou_app_key="你的豌豆HTTP app_key"

WanDouHttpProxy构造参数说明（见 构造函数）：

构造时会固定代理品牌名为WANDOUHTTP，请求接口地址为https://api.wandouapp.com/（GET 请求，携带app_key与num两个查询参数），并创建一个IpCache实例用于 IP 缓存。

三、源码解析：get_proxy的拉取与缓存策略

WanDouHttpProxy.get_proxy 是豌豆HTTP 代理的核心方法，实现逻辑可以概括为「缓存优先、差额补取、逐个落缓存」：

1. 缓存优先：先调用self.ip_cache.load_all_ip(proxy_brand_name="WANDOUHTTP")从缓存加载所有未过期 IP。如果缓存中可用数量>= num，直接返回前num个，不再请求远程接口；
2. 差额补取：若缓存不足，计算need_get_count = num - len(ip_cache_list)，并以min(need_get_count, 100)作为本次请求的num参数（豌豆HTTP 单次最多返回 100 个 IP），发起 GET 请求；
3. 响应解析与落缓存：当响应code == 200时，遍历data列表，把每项转换为统一的IpInfoModel（ip、port、user/password为空字符串、expired_time_ts由expire_time字符串解析为 Unix 时间戳），再以WANDOUHTTP_{ip}_{port}为 key 写入缓存，TTL 为「过期时间戳 − 当前时间戳」，即 IP 到期后缓存条目自动失效；
4. 错误处理：code != 200时抛出IpGetError，并对两个典型错误码给出明确提示：
   • 10001：通用错误，需查看msg字段的具体错误信息（常见原因包括 app_key 错误、欠费等）；
   • 10048：无可用套餐，即账号下没有生效中的代理套餐或额度已耗尽。

关键代码如下：

async def get_proxy(self, num: int) -> List[IpInfoModel]: # Prioritize getting IP from cache ip_cache_list = self.ip_cache.load_all_ip( proxy_brand_name=self.proxy_brand_name ) if len(ip_cache_list) >= num: return ip_cache_list[:num] # If the quantity in cache is insufficient, get from IP provider to supplement, then store in cache need_get_count = num - len(ip_cache_list) self.params.update({"num": min(need_get_count, 100)}) # Maximum 100 ip_infos = [] async with make_async_client() as client: url = self.api_path + "?" + urlencode(self.params) utils.logger.info(f"[WanDouHttpProxy.get_proxy] get ip proxy url:{url}") response = await client.get(url, headers={...}) res_dict: Dict = response.json() if res_dict.get("code") == 200: data: List[Dict] = res_dict.get("data", []) current_ts = utils.get_unix_timestamp() for ip_item in data: ip_info_model = IpInfoModel( ip=ip_item.get("ip"), port=ip_item.get("port"), user="", # 豌豆HTTP 无需账号密码认证 password="", expired_time_ts=utils.get_unix_time_from_time_str( ip_item.get("expire_time") ), ) ip_key = f"WANDOUHTTP_{ip_info_model.ip}_{ip_info_model.port}" self.ip_cache.set_ip(ip_key, ip_info_model.model_dump_json(), ex=ip_info_model.expired_time_ts - current_ts) else: error_code = res_dict.get("code") if error_code == 10001: error_msg = "General error, check msg content for specific error information" elif error_code == 10048: error_msg = "No available package" raise IpGetError(f"{error_msg} (code: {error_code})") return ip_cache_list + ip_infos

这里有两点值得注意：

• 无需账密：与部分需要user:password认证的代理不同，豌豆HTTP 返回的 IP 免认证，因此在 ProxyIpPool 校验逻辑 中会走http://{ip}:{port}的无认证分支，代理自动刷新 Mixin 生成 httpx 代理 URL 时同理；
• 缓存依赖 Redis：IpCache 内部通过CacheFactory按 Redis 缓存类型创建客户端，load_all_ip以WANDOUHTTP_*为 key 模式批量读取未过期 IP，过期清理由缓存后端（Redis）的 TTL 机制自动完成。若你的 Redis 不可用，IP 缓存会退化为每次向远程接口请求。

四、代理池如何消费 wandouhttp 供应商

拿到WanDouHttpProxy实例后，并不是爬虫直接使用它，而是先构建一个统一的 IP 代理池。整条链路如下：

1. 供应商注册表

在 proxy_ip_pool.py 中，所有供应商在模块加载时即被实例化并注册到IpProxyProvider字典：

IpProxyProvider: Dict[str, ProxyProvider] = { ProviderNameEnum.KUAI_DAILI_PROVIDER.value: new_kuai_daili_proxy(), ProviderNameEnum.WANDOU_HTTP_PROVIDER.value: new_wandou_http_proxy(), ProviderNameEnum.STATIC_PROVIDER.value: StaticProxyProvider(), }

其中豌豆HTTP 对应的 key 是wandouhttp（定义见 ProviderNameEnum）。注意new_wandou_http_proxy()在模块导入时就会执行，因此环境变量WANDOU_APP_KEY需要在启动进程前设置好，否则构造时会取到占位符your_wandou_http_app_key。

2. 相关配置项

在 config/base_config.py 中控制代理行为的配置项：

# Whether to enable IP proxy ENABLE_IP_PROXY = False # Number of proxy IP pools IP_PROXY_POOL_COUNT = 2 # Proxy IP provider name IP_PROXY_PROVIDER_NAME = "kuaidaili" # kuaidaili | wandouhttp | static # Static proxy configuration (used when IP_PROXY_PROVIDER_NAME is set to "static") STATIC_PROXY_URL = ""

接入豌豆HTTP 时，需要修改为：

ENABLE_IP_PROXY = True IP_PROXY_POOL_COUNT = 20 # 按需求调整池内 IP 数量 IP_PROXY_PROVIDER_NAME = "wandouhttp"

另外，DISABLE_SSL_VERIFY 仅在使用会注入自签名证书的企业代理、Burp Suite、mitmproxy 等中间人代理时才设为 True，普通 HTTP 转发型代理保持 False 即可。

3. 创建池子与取 IP 逻辑

create_ip_pool 根据config.IP_PROXY_PROVIDER_NAME从注册表取出供应商并创建ProxyIpPool，随后立即load_proxies()拉取一批 IP 入池；若配置了一个注册表中不存在的 provider 名，会抛出带可选值列表的ValueError（这一点有专门的回归测试 test_create_ip_pool_unknown_provider_raises_value_error 保障）。

池子的取 IP 行为由 ProxyIpPool.get_proxy 决定：

• 池空时自动reload_proxies重新拉取；
• 使用random.choice随机抽取一个 IP 并立即从池中移除（用后即弃，避免重复暴露同一出口 IP）；
• 若启用enable_validate_ip，会先通过探测地址验证 IP 是否可用，不可用则借助tenacity重试（最多 3 次、间隔 1 秒）重新抽取；
• 抽取结果保存为current_proxy，供后续请求复用。

4. 过期检测与自动刷新

豌豆HTTP 的每个 IP 都带expire_time，IpInfoModel.is_expired 以「缓冲秒数」（默认 30 秒）提前判定过期，避免在 IP 临死时刻发起请求失败：当前时间戳 ≥ 过期时间戳 − buffer 即视为过期。ProxyIpPool.get_or_refresh_proxy 则是各平台客户端在每次请求前的标准入口——当前代理未过期就复用，过期则重新从池中抽取。

平台客户端通过继承 ProxyRefreshMixin 接入该机制：在__init__中调用init_proxy_pool(proxy_ip_pool)注入池引用，在请求方法前调用_refresh_proxy_if_expired()，该方法检测到代理过期后会拉取新 IP 并自动更新 httpx 的self.proxyURL。

整体工作流程可参考 代理 IP 使用流程图 中的示意：配置供应商 → 拉取 IP 入池 → 随机抽取 → 过期校验/刷新 → 请求失败换 IP 重试。

五、验证与实操建议

1. 独立验证代理池：仓库提供了 test/test_proxy_ip_pool.py，其中test_ip_pool可真实拉取 IP，test_ip_expiration_standalone与test_proxy_pool_auto_refresh不依赖真实供应商即可验证过期判定与自动刷新逻辑，适合在配置完WANDOU_APP_KEY后先跑一遍确认链路通畅；
2. 环境准备：先export WANDOU_APP_KEY=...，再修改config/base_config.py的ENABLE_IP_PROXY = True与IP_PROXY_PROVIDER_NAME = "wandouhttp"，最后按 代理使用 的整体说明启动爬虫；
3. 额度与批量上限：豌豆HTTP 单次最多返回 100 个 IP（源码中min(need_get_count, 100)硬限制），IP_PROXY_POOL_COUNT设置过大时会自动分批请求或依赖缓存补齐；
4. 排错要点：日志中出现code: 10001优先检查 app_key 是否配置正确、账号是否欠费；code: 10048表示账号下无可用套餐，需先到官网续费或领取试用；日志中[WanDouHttpProxy.get_proxy] get ip proxy url:...会打印真实请求 URL，可用于确认app_key是否被正确读取；
5. 缓存观察：缓存 key 形如WANDOUHTTP_{ip}_{port}，可在 Redis 中直接查看当前池内还有哪些未过期 IP 及其剩余 TTL。

六、合规与注意事项

• 本文所有接入方式仅适用于学习与研究目的。项目各源文件头部声明明确要求：不得用于商业用途、不得进行大规模爬取或对目标平台造成运营干扰、应遵守目标平台使用条款与 robots.txt 规则，并合理控制请求频率（详见 LICENSE）；
• 代理 IP 的作用是降低因本地出口 IP 被风控导致的请求失败概率，并不改变「控制请求频率」这一合规前提，CRAWLER_MAX_SLEEP_SEC、MAX_CONCURRENCY_NUM等限速配置应配合使用；
• 若你已有自建或静态代理，也可以将IP_PROXY_PROVIDER_NAME设为static并通过STATIC_PROXY_URL直接指定代理地址（支持http://user:password@host:port格式，见 StaticProxyProvider），与豌豆HTTP 走的是同一套代理池抽象，可随时切换对比。

【免费下载链接】MediaCrawler小红书笔记 | 评论爬虫、抖音视频 | 评论爬虫、快手视频 | 评论爬虫、B 站视频 ｜ 评论爬虫、微博帖子 ｜ 评论爬虫、百度贴吧帖子 ｜ 百度贴吧评论回复爬虫 | 知乎问答文章｜评论爬虫项目地址: https://gitcode.com/GitHub_Trending/me/MediaCrawler