做爬虫或者数据采集的朋友,大概率都碰到过这种糟心事:刚从服务商那儿提出来的IP,测的时候好好的,结果一上业务就报错;有的IP更离谱,用了没几分钟就挂了,你还得一个个去排查到底是哪个出了问题。
说到底,问题就出在没有一套系统帮你盯着这些代理IP的状态。手头上要是只有三五个IP,人工测一测、记一记还能凑合。可要是几十上百个IP轮着用,靠人去盯根本不现实,漏一个就可能导致整批任务失败。
今天这篇文章,就手把手教大家用Python搭一个简单但完整的代理IP可用性日志系统,从检测、记录到数据分析一条龙搞定。代码不复杂,看完就能上手用。

一、先想清楚:这个系统要记录哪些东西
搭系统之前得先想明白,我们到底需要知道什么?其实就三个问题:
这个IP现在能不能用?
能用的话,响应速度快不快?
放到历史里看,这个IP表现稳不稳定?
想清楚这三个问题,日志字段就好设计了。每条记录至少包含下面这些信息:
| 字段 | 说明 |
|---|---|
timestamp | 检测时间 |
proxy_ip | 代理IP地址(含端口) |
status | 可用/不可用 |
latency_ms | 响应延迟(毫秒) |
target_url | 检测目标网站 |
error_msg | 失败时的错误信息(如有) |
存储方面不用一上来就搞很复杂。IP数量不多的话,CSV文件完全够用,读写都方便;后面数据量大了,再迁移到SQLite,查询分析会更顺手。
二、核心代码:从检测到记录一步步来
1. 先写一个IP可用性检测函数
这是整个系统的地基。思路很简单:用 requests 库通过代理去访问一个稳定的网站,看能不能正常返回结果。
import requests
import time
import csv
from datetime import datetime
def check_proxy(proxy, target_url="http://www.baidu.com", timeout=5):
"""
检测单个代理IP是否可用
返回 (是否可用, 延迟毫秒数, 错误信息)
"""
proxies = {
"http": f"http://{proxy}",
"https": f"http://{proxy}"
}
try:
start_time = time.time()
response = requests.get(target_url, proxies=proxies, timeout=timeout)
end_time = time.time()
latency_ms = round((end_time - start_time) * 1000, 2)
if response.status_code == 200:
return True, latency_ms, None
else:
return False, None, f"HTTP {response.status_code}"
except requests.exceptions.Timeout:
return False, None, "Timeout"
except requests.exceptions.ConnectionError:
return False, None, "Connection Error"
except Exception as e:
return False, None, str(e)这里有两个细节要注意。一是检测目标要选稳定的公共服务,别拿那些本身就经常抽风的网站来测,不然IP没问题,你反而误判了。二是超时时间根据自己业务来定,普通检测5到8秒就够了,设太短容易误杀,设太长又拖慢整体检测速度。
2. 把检测结果写进日志
检测完了不能光看一眼就过,得把结果存下来,后面才好分析。
def write_log(proxy, is_alive, latency_ms, error_msg, log_file="proxy_log.csv"):
"""
将检测结果写入CSV日志文件
"""
timestamp = datetime.now().isoformat()
row = {
"timestamp": timestamp,
"proxy": proxy,
"status": "alive" if is_alive else "dead",
"latency_ms": latency_ms if is_alive else "",
"error": error_msg if not is_alive else ""
}
# 文件不存在就先写表头
try:
with open(log_file, 'x', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=row.keys())
writer.writeheader()
except FileExistsError:
pass
# 追加写入
with open(log_file, 'a', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=row.keys())
writer.writerow(row)3. 批量检测加定时调度,实现自动化
单个IP检测搞定了,接下来就是把整个代理池跑一遍,再配上定时任务,让它自己转起来。
import logging
from apscheduler.schedulers.blocking import BlockingScheduler
logging.basicConfig(level=logging.INFO)
def scan_proxy_pool(proxy_list):
"""
批量检测代理池中的所有IP,并记录日志
"""
for proxy in proxy_list:
is_alive, latency, error = check_proxy(proxy)
write_log(proxy, is_alive, latency, error)
if is_alive:
logging.info(f"✅ 可用 | {proxy} | 延迟: {latency}ms")
else:
logging.info(f"❌ 失效 | {proxy} | {error}")
# 从代理服务商API获取IP列表(换成你自己的)
def fetch_proxy_list():
# 示例:requests.get("https://api.proxyprovider.com/getips")
return ["192.168.1.1:8080", "192.168.1.2:8080"]
# 每5分钟检测一次
scheduler = BlockingScheduler()
scheduler.add_job(
lambda: scan_proxy_pool(fetch_proxy_list()),
'interval',
minutes=5,
id='proxy_scan'
)
scheduler.start()检测频率自己把握。短效代理变化快,可以设得密一点,3到5分钟一次;长效代理稳定性好,15分钟甚至半小时一次都没问题。
4. IP多了就上并发,别傻等
代理池要是有几十个上百个IP,串行检测一个一个等,黄花菜都凉了。用 concurrent.futures 开并发,效率能提一大截。
from concurrent.futures import ThreadPoolExecutor, as_completed
def scan_proxy_pool_concurrent(proxy_list, max_workers=10):
"""
并发检测代理池,大幅提升效率
"""
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_proxy = {
executor.submit(check_proxy, proxy): proxy
for proxy in proxy_list
}
for future in as_completed(future_to_proxy):
proxy = future_to_proxy[future]
is_alive, latency, error = future.result()
write_log(proxy, is_alive, latency, error)并发数别开太猛,10到20个线程通常就够了。开太多反而可能把自己网络打满,或者触发代理服务商的限流。
三、日志攒起来了,怎么从中挖出有用信息
光记日志还不够,得会从数据里看门道。下面这几个函数,日常分析基本够用了。
1. 查某个IP最近的检测记录
想知道某个IP最近表现怎么样,拉最近N次记录看看就清楚了。
import pandas as pd
def get_recent_status(proxy, log_file="proxy_log.csv", n=10):
"""查询某个IP最近N次的检测结果"""
df = pd.read_csv(log_file)
return df[df['proxy'] == proxy].tail(n)2. 算一算IP的可用率
单次检测说明不了什么,可用率才是硬指标。比如最近7天这个IP有多少比例是活着的,一眼就能看出靠不靠谱。
from datetime import timedelta
def calculate_availability(proxy, log_file="proxy_log.csv", days=7):
"""计算某个IP最近N天的可用率"""
df = pd.read_csv(log_file)
df['timestamp'] = pd.to_datetime(df['timestamp'])
cutoff = datetime.now() - timedelta(days=days)
recent = df[(df['proxy'] == proxy) & (df['timestamp'] >= cutoff)]
if len(recent) == 0:
return None
alive_count = len(recent[recent['status'] == 'alive'])
return round(alive_count / len(recent) * 100, 2)3. 把表现最差的IP揪出来
代理池里总有那么几个拖后腿的,用这个函数把可用率最低的几个IP排出来,该淘汰淘汰,该换服务商换服务商。
def find_worst_proxies(log_file="proxy_log.csv", top_n=5):
"""找出可用率最低的top N个IP"""
df = pd.read_csv(log_file)
results = {}
for proxy in df['proxy'].unique():
proxy_df = df[df['proxy'] == proxy]
alive = len(proxy_df[proxy_df['status'] == 'alive'])
results[proxy] = round(alive / len(proxy_df) * 100, 2)
return sorted(results.items(), key=lambda x: x[1])[:top_n]四、想再专业点?可视化和告警安排上
上面这套已经能解决大部分问题了。如果想让系统更完善,还可以往两个方向延伸。
一是接可视化面板。 用 Prometheus + Grafana 搭一个实时监控仪表盘,成功率趋势、平均响应时间、错误类型分布这些指标都能直观看到,比自己翻CSV舒服多了。
二是加告警机制。 定个阈值,比如某个IP最近10次检测的可用率低于95%,就自动发邮件或者微信通知你。实现起来也不复杂,在检测脚本里加一段判断逻辑就行,省得自己天天去盯数据。
五、总结
回头看,这套代理IP可用性日志系统其实就干了三件事:
定时检测——隔一段时间就把代理池里的IP挨个验一遍,看连通性和响应速度
记录日志——每次检测的时间、IP、状态、延迟、错误信息都存下来
查询分析——从历史数据里算可用率、找异常IP、反过来优化你的代理调度策略
代码加起来也就一百多行,算不上什么高深技术。但真正跑起来之后,你会发现它的价值很大——以前是"凭感觉"判断哪个IP好用,现在是拿数据说话,哪些IP稳定、哪个时间段质量高、哪些节点该换掉,心里都清清楚楚。
做爬虫这行有句话说得好:代理IP用得好,不如监控做得好。这套日志系统就是你代理管理的"仪表盘",有了它,再也不用为IP突然失效抓瞎了。
