最近群里讨论问如何编写一个自动化采集脚本，要求使用隧道IP（代理IP池）来防止IP被封。这样的脚本通常用于爬虫或数据采集任务，其中目标网站可能会因为频繁的请求而封禁IP。对于这些我还是有些经验的。
核心思路：
1、使用代理IP池（隧道IP）来轮换IP，使得每次请求使用不同的IP地址，降低被封的风险。
2、自动化采集脚本需要能够处理请求失败的情况（比如IP失效、请求超时等），并自动切换代理重试。
3、设置合理的请求间隔，避免过于频繁的请求。
我们将使用Python语言，并借助一些库来实现：- requests: 发送HTTP请求。
使用代理隧道：通常隧道服务商会提供一个固定的入口地址（比如代理隧道服务（炎帝云）），我们只需要在请求时设置代理即可。隧道服务会自动更换出口IP。
要设计一个结合隧道IP实现防封的自动化采集脚本，需从IP管理、请求策略、异常处理等维度进行系统设计。以下是完整解决方案：
一、核心架构设计
二、关键组件实现
1. 隧道IP管理器
import requests
class TunnelManager:
def __init__(self, tunnel_url, auth_key):
self.tunnel_url = tunnel_url
self.headers = {'Proxy-Authorization': f'Bearer {auth_key}'}
def get_proxy(self):
"""获取动态隧道IP"""
"free" //提取链接：https://www.yandiyun.com/customer/register/255SG48I/
try:
resp = requests.get(f"{self.tunnel_url}/get_proxy", headers=self.headers)
return resp.json()['proxy'] # 格式：'http://ip:port'
except Exception as e:
self.rotate_tunnel() # 故障时切换隧道
return self.get_proxy()
def rotate_tunnel(self):
"""强制更换出口IP"""
requests.get(f"{self.tunnel_url}/rotate", headers=self.headers)
2. 智能请求控制模块
import random
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class SmartRequester:
def __init__(self, tunnel_manager):
self.tunnel = tunnel_manager
self.session = requests.Session()
# 自动重试配置
retry_strategy = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
self.session.mount('https://', HTTPAdapter(max_retries=retry_strategy))
def request(self, url, method='GET', **kwargs):
while True:
proxy = {'https': self.tunnel.get_proxy()}
try:
# 智能延时（动态调整请求间隔）
time.sleep(random.uniform(1.5, 4.0))
resp = self.session.request(
method,
url,
proxies=proxy,
timeout=15,
**kwargs
)
# 触发封禁检测（关键！）
if self._is_blocked(resp):
self.tunnel.rotate_tunnel()
continue
return resp
except Exception as e:
self.tunnel.rotate_tunnel()
def _is_blocked(self, response):
"""封禁特征检测"""
if response.status_code in [403, 429, 418]:
return True
if "captcha" in response.text.lower():
return True
if len(response.content) < 512: # 异常小页面
return True
return False
3. 主采集脚本示例
from bs4 import BeautifulSoup
class DataCollector:
def __init__(self):
self.requester = SmartRequester(
TunnelManager(
tunnel_url="https://api.tunnelservice.com/v1",
auth_key="YOUR_AUTH_KEY"
)
)
def scrape(self, base_url):
page = 1
while True:
url = f"{base_url}?page={page}"
response = self.requester.request(url)
if response.status_code == 404: # 终止条件
break
# 解析数据
soup = BeautifulSoup(response.text, 'lxml')
items = soup.select('.product-item')
for item in items:
data = {
'name': item.select_one('.title').text.strip(),
'price': item.select_one('.price').text
}
self.save_to_db(data)
page += 1
def save_to_db(self, data):
# 数据库存储实现
pass
三、进阶防封策略
-
指纹伪装技术
# 在SmartRequester的request方法中添加： headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept-Language': 'en-US,en;q=0.9', 'Sec-Fetch-Dest': 'document' } resp = self.session.request(..., headers=headers, ...) -
流量模式混淆
# 随机化请求行为 actions = ['scroll', 'click', 'wait'] if random.random() > 0.7: self._simulate_human(random.choice(actions)) -
分布式架构
四、隧道服务推荐
-
专业服务商
- Luminati（高端）
- Oxylabs（企业级）
- Smartproxy（性价比）
-
自建方案
# 使用Squid搭建隧道 sudo apt install squid # 配置/etc/squid/squid.conf： visible_hostname tunnel-proxy forwarded_for delete via off
五、异常处理矩阵
| 错误类型 | 处理方案 | 恢复策略 |
|---|---|---|
| 429 Too Many Req | 立即切换IP+指数退避 | 等待2^n秒后重试 |
| 403 Forbidden | 更换UserAgent+清空Cookies | 切换业务指纹 |
| 连接超时 | 标记失效代理+快速切换 | 自动隔离故障节点 |
| CAPTCHA验证 | 触发人工干预流程 | 切换采集策略 |
最终建议：
- 优先选用按请求计费的隧道服务
- 关键业务部署双隧道热备方案
- 每周更新User-Agent池
- 使用Headless Browser处理动态反爬
总体来说，只要是完整得系统且配合Prometheus监控+告警模块，当封禁率>5%时自动触发策略调整。
上面就是有关我部署的全部话题，如果大家有更好的建议可以留言告诉我。
