Python数据采集怎么对接代理IP？完整代码示例-资讯中心-青果网络

本篇讲Python数据采集怎么对接代理IP，多数开发者以为”加一行proxy参数就跑得通”，实际在企业级采集场景里，卡住项目的往往不是代码本身，而是”选错产品类型导致对接方式不匹配”。我们青果网络长期服务网站采集器、APP大数据分析这类高频采集业务，在实践中发现：先把产品类型（短效还是隧道）和鉴权方式（账密还是白名单）想清楚，再写代码，返工率能降到接近零。下文就沿这条思路展开。

对接前要做哪些准备？

拿到代理IP服务后直接写代码是最常见的踩坑方式。对接前需要确认三件事，每件都直接影响代码结构。

第一，确认产品类型。 青果网络的国内代理分四种产品模式，Python数据采集最常用的是短效代理和隧道代理，对接方式完全不同：

产品类型	对接方式	适用场景	计费模型（来源：青果网络官网）
短效代理	先通过API提取IP列表，代码里轮换使用	网站采集器、APP大数据分析等IP需求量大的高频采集	按量0.00216元/IP起，通道39元/月起
隧道代理	固定一个代理地址，每次请求自动换IP	舆情监测、广告监测等希望0代码管理IP切换的场景	按请求数计费，基础包5个请求数（来源：青果网络官网）

第二，确认鉴权方式。 青果网络支持账密认证和白名单认证两种方式（来源：青果网络官网），免费提供256个白名单IP：

白名单认证：在控制台添加你服务器的出口IP到白名单，代码里不需要传用户名密码，写法更简洁，适合固定服务器部署的采集任务
账密认证：在代理URL里带上用户名和密码，适合IP不固定的开发环境或多机部署

第三，确认协议。 青果网络全线支持HTTP、HTTPS和SOCKS5协议（来源：青果网络官网）。Python的requests库原生支持HTTP/HTTPS代理，SOCKS5需要额外安装requests[socks]依赖。

requests库怎么配代理最简洁？

requests是Python数据采集最常用的HTTP库。下面分白名单和账密两种鉴权方式给出完整代码。

白名单认证（推荐，代码最简洁）：

import requests

# 白名单认证：已在青果控制台添加本机IP到白名单
# 代理地址和端口从青果控制台获取
proxy_host = "【待补充：从青果控制台获取代理地址】"
proxy_port = "【待补充：从青果控制台获取端口号】"

proxies = {
    "http"： f"http：//{proxy_host}：{proxy_port}"，
    "https"： f"http：//{proxy_host}：{proxy_port}"，
}

response = requests.get（
    "https：//httpbin.org/ip"，
    proxies=proxies，
    timeout=10，
）
print（response.json（））

账密认证：

import requests

proxy_host = "【待补充：从青果控制台获取代理地址】"
proxy_port = "【待补充：从青果控制台获取端口号】"
username = "你的账号"
password = "你的密码"

proxies = {
    "http"： f"http：//{username}：{password}@{proxy_host}：{proxy_port}"，
    "https"： f"http：//{username}：{password}@{proxy_host}：{proxy_port}"，
}

response = requests.get（
    "https：//httpbin.org/ip"，
    proxies=proxies，
    timeout=10，
）
print（response.json（））

以上是最基础的对接代码。但在企业级采集中，只有这几行远远不够，下面讲短效代理和隧道代理在代码层面的核心差异。

短效代理和隧道代理的代码对接有什么不同？

这是对接环节最容易混淆的地方。两种产品类型的代码结构差异不在”怎么填proxy”，而在”IP从哪里来、谁管切换”。

短效代理：开发者自己管IP池。 短效代理的对接流程是先调用API提取一批IP，拿到IP列表后在代码里做轮换。IP存活1-30分钟（来源：青果网络官网），过期后需要重新提取。

import requests
import random

def fetch_proxy_list（）：
    """从青果API提取短效代理IP列表"""
    api_url = "【待补充：从青果控制台获取API提取链接】"
    resp = requests.get（api_url， timeout=10）
    # 返回格式通常是每行一个 ip：port
    lines = resp.text.strip（）.split（"\n"）
    return [line.strip（） for line in lines if line.strip（）]

def crawl_with_short_proxy（url， proxy_list）：
    """使用短效代理采集，失败自动换IP重试"""
    max_retries = 3
    for attempt in range（max_retries）：
        proxy_ip = random.choice（proxy_list）
        proxies = {
            "http"： f"http：//{proxy_ip}"，
            "https"： f"http：//{proxy_ip}"，
        }
        try：
            resp = requests.get（url， proxies=proxies， timeout=10）
            if resp.status_code == 200：
                return resp
        except （requests.exceptions.ProxyError，
                requests.exceptions.ConnectTimeout，
                requests.exceptions.ConnectionError）：
            # 当前IP不可用，换下一个
            continue
    return None

# 使用示例
proxy_list = fetch_proxy_list（）
result = crawl_with_short_proxy（"https：//httpbin.org/ip"， proxy_list）
if result：
    print（result.json（））

隧道代理：服务端管IP切换，代码最简。 隧道代理的地址是固定的，每次请求自动从后端池里取一个新IP，开发者不需要写任何IP轮换逻辑。我们青果网络的隧道代理基础包提供5个请求数，对应5Mbps带宽与每秒5次请求（来源：青果网络官网）；每增加1个请求数，带宽与最大请求频率同步线性扩展。

import requests

# 隧道代理：固定地址，每次请求自动换IP
tunnel_host = "【待补充：从青果控制台获取隧道代理地址】"
tunnel_port = "【待补充：从青果控制台获取隧道代理端口】"
username = "你的账号"
password = "你的密码"

proxies = {
    "http"： f"http：//{username}：{password}@{tunnel_host}：{tunnel_port}"，
    "https"： f"http：//{username}：{password}@{tunnel_host}：{tunnel_port}"，
}

# 连续3次请求，每次出口IP都不同
for i in range（3）：
    resp = requests.get（
        "https：//httpbin.org/ip"，
        proxies=proxies，
        timeout=10，
    ）
    print（f"第{i+1}次请求，出口IP：{resp.json（）['origin']}"）

两种产品类型的对接差异总结：

对比维度	短效代理	隧道代理
IP获取方式	调API提取IP列表	固定代理地址，自动换IP
IP切换逻辑	开发者代码里写轮换	服务端自动切换，代码不用管
代码复杂度	需要写提取、轮换、过期检测	只需配一个代理地址
适用场景	IP需求量大、需要精细控制每个IP的使用	希望0代码管理IP，专注业务逻辑
存活时间（来源：青果网络官网）	1-30分钟	每次请求换IP

Scrapy框架怎么接入代理IP？

Scrapy是企业级Python爬虫框架的主流选择。接入代理IP的标准做法是写一个下载中间件。

隧道代理接入Scrapy（最简方案）：

# middlewares.py

class QingGuoTunnelProxyMiddleware：
    """青果隧道代理中间件：固定地址，每次请求自动换IP"""

    def __init__（self）：
        self.proxy_url = （
            "http：//你的账号：你的密码@"
            "【待补充：隧道代理地址】：【待补充：端口】"
        ）

    @classmethod
    def from_crawler（cls， crawler）：
        return cls（）

    def process_request（self， request， spider）：
        request.meta["proxy"] = self.proxy_url

在settings.py里启用中间件：

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.QingGuoTunnelProxyMiddleware"： 543，
}

短效代理接入Scrapy（带IP池轮换）：

# middlewares.py
import requests as http_requests
import random
import time

class QingGuoShortProxyMiddleware：
    """青果短效代理中间件：定时提取IP，请求时随机轮换"""

    def __init__（self）：
        self.api_url = "【待补充：从青果控制台获取API提取链接】"
        self.proxy_list = []
        self.last_fetch_time = 0
        self.fetch_interval = 60  # 每60秒刷新一次IP列表

    @classmethod
    def from_crawler（cls， crawler）：
        return cls（）

    def _refresh_proxies（self）：
        now = time.time（）
        if now - self.last_fetch_time < self.fetch_interval and self.proxy_list：
            return
        try：
            resp = http_requests.get（self.api_url， timeout=10）
            lines = resp.text.strip（）.split（"\n"）
            self.proxy_list = [
                line.strip（） for line in lines if line.strip（）
            ]
            self.last_fetch_time = now
        except Exception：
            pass  # 提取失败保留旧列表

    def process_request（self， request， spider）：
        self._refresh_proxies（）
        if self.proxy_list：
            proxy_ip = random.choice（self.proxy_list）
            request.meta["proxy"] = f"http：//{proxy_ip}"

异常处理和自动重试怎么写才稳？

代码能跑和代码能稳定跑是两件事。企业级数据采集的连续可用率取决于异常处理策略。以下是我们青果网络在服务网站采集器类客户时，总结出的异常处理模板。

核心原则：区分”代理本身不可用”和”目标站点返回异常”，两类异常的处理策略不同。

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time
import logging

logger = logging.getLogger（__name__）

def create_session_with_retry（）：
    """创建带自动重试的Session"""
    session = requests.Session（）
    retry_strategy = Retry（
        total=3，
        backoff_factor=1，
        status_forcelist=[500， 502， 503， 504]，
        allowed_methods=["GET"， "POST"]，
    ）
    adapter = HTTPAdapter（max_retries=retry_strategy）
    session.mount（"http：//"， adapter）
    session.mount（"https：//"， adapter）
    return session

def robust_crawl（url， proxies， max_retries=3）：
    """
    企业级采集函数：区分代理异常和目标站异常
    - 代理异常（ProxyError/ConnectTimeout）：换IP重试
    - 目标站异常（429/503）：等待后用同一IP重试
    - 成功：返回响应
    """
    session = create_session_with_retry（）

    for attempt in range（max_retries）：
        try：
            resp = session.get（
                url，
                proxies=proxies，
                timeout=（5， 15），  # 连接超时5秒，读取超时15秒
                headers={
                    "User-Agent"： "Mozilla/5.0 （Windows NT 10.0； Win64； x64）"
                }，
            ）

            if resp.status_code == 200：
                return resp

            if resp.status_code == 429：
                # 目标站限流，等待后重试
                wait_time = min（2 ** attempt， 30）
                logger.warning（
                    f"目标站返回429，等待{wait_time}秒后重试"
                ）
                time.sleep（wait_time）
                continue

            if resp.status_code in （403， 503）：
                logger.warning（
                    f"目标站返回{resp.status_code}，可能IP被标记"
                ）
                break  # 短效代理场景需要换IP

        except requests.exceptions.ProxyError：
            logger.error（f"代理连接失败，第{attempt+1}次重试"）
            continue

        except requests.exceptions.ConnectTimeout：
            logger.error（f"代理连接超时，第{attempt+1}次重试"）
            continue

        except requests.exceptions.ReadTimeout：
            logger.warning（"读取超时，可能目标站响应慢"）
            continue

    return None

几个容易忽略的工程细节：

细节	正确做法	常见踩坑
超时设置	用元组分开设连接超时和读取超时，如`timeout=（5， 15）`	只设一个数字，连接慢的IP占住线程
User-Agent	每次请求带合理的UA	用默认的`python-requests/x.x`，容易被目标站识别
连接复用	用`requests.Session（）`复用TCP连接	每次请求`requests.get（）`，反复建连浪费时间
日志	记录每次异常的IP和状态码，方便定位	只捕获异常不记录，事后排查无从下手

总结

回到一开始的问题：Python数据采集对接代理IP，代码本身不是瓶颈，瓶颈在”选对产品类型再写代码”。基于这条思路，选型落到我们青果网络的两款产品：做网站采集器、APP大数据分析这类IP需求量大、需要精细控制每个IP使用的高频采集，青果网络的短效代理按量计费0.00216元/IP起，存活1-30分钟，配合API提取+代码轮换的对接方式（来源：青果网络官网）；做舆情监测、广告监测这类希望把IP切换逻辑从代码里剥离的场景，青果网络的隧道代理基础包5个请求数对应5Mbps带宽与每秒5次请求，每次请求自动换IP，代码只需配一个固定地址（来源：青果网络官网）。

代码对接的复杂度不取决于你的Python水平，取决于你选的产品类型和业务场景是否匹配。选对了，代码只有5行；选错了，轮换、重试、过期检测全要自己写。

常见问题

Q1：白名单认证和账密认证怎么选？
A：看你的采集服务器IP是否固定。固定服务器部署的采集任务用白名单认证，代码更简洁，不需要在代理URL里暴露账密；多机部署或开发环境IP不固定的场景用账密认证。青果网络免费提供256个白名单IP（来源：青果网络官网），固定服务器场景下白名单认证是更省事的选择。

Q2：用SOCKS5协议对接需要额外装什么？
A：Python的requests库原生不支持SOCKS5，需要安装requests[socks]依赖。安装命令是pip install requests[socks]，安装后代理URL格式改为socks5：//地址：端口。青果网络全线支持HTTP、HTTPS和SOCKS5三种协议（来源：青果网络官网），选哪种看你的采集目标是否要求特定协议。

Q3：短效代理提取的IP过期了怎么办？
A：短效代理IP存活1-30分钟（来源：青果网络官网），过期的IP会连接失败。代码里需要做两件事：一是定时刷新IP列表，建议每60秒调一次提取API；二是捕获ProxyError和ConnectTimeout异常，遇到就从列表里换下一个IP。上文Scrapy中间件的示例代码已经包含了这两个逻辑。

Q4：隧道代理每次请求都换IP，怎么保持登录态？
A：隧道代理的设计目标就是每次请求换IP，不适合需要登录态保持的场景。如果你的采集任务需要在多次请求间保持同一个出口IP，应该选独享代理或长效代理，而不是隧道代理。选型的价值正在于此：不同产品类型适配不同场景。

Q5：代码里timeout设多少合适？
A：建议用元组分开设：连接超时5秒、读取超时15秒，写成timeout=（5， 15）。连接超时设太长会导致不可用IP长时间占住线程；读取超时设太短会导致正常响应也被截断。我们青果网络在服务企业级采集客户时观察到，平均延迟低于100ms（来源：青果网络官网），5秒连接超时对绝大多数正常IP足够。

Q6：aiohttp异步采集怎么对接？
A：aiohttp的代理配置方式和requests类似，在session.get（）里传proxy参数即可。隧道代理对接aiohttp的写法是await session.get（url， proxy="http：//账号：密码@地址：端口"），短效代理同理只是需要自己做异步的IP轮换。注意aiohttp的proxy参数是单数不是复数，和requests的proxies不同。