IP池纯净度测试：用httpbin + 自建检测服务质量

 import requests
 ​
 def check_exit_ip(proxy_addr):
     try:
         resp = requests.get(
             "https://httpbin.org/ip",
             proxies={"https": f"http://{proxy_addr}"},
             timeout=10
         )
         data = resp.json()
         return data.get("origin", "")
     except Exception as e:
         return f"FAIL: {e}"

 def check_header_leak(proxy_addr):
     resp = requests.get(
         "https://httpbin.org/headers",
         proxies={"https": f"http://{proxy_addr}"},
         timeout=10
     )
     headers = resp.json().get("headers", {})
     leak_fields = ["X-Forwarded-For", "Via", "X-Real-Ip", "Proxy-Connection"]
     leaks = {k: v for k, v in headers.items() if k in leak_fields}
     return leaks  # 空字典 = 没泄露

 def check_latency(proxy_addr, target_url="https://httpbin.org/delay/0"):
     start = time.time()
     try:
         resp = requests.get(
             target_url,
             proxies={"https": f"http://{proxy_addr}"},
             timeout=15
         )
         latency_ms = (time.time() - start) * 1000
         return round(latency_ms, 1)
     except Exception:
         return -1  # 超时或失败

 # 检测服务端（Flask 示例）
 from flask import Flask, request, jsonify
 import geoip2.database
 ​
 app = Flask(__name__)
 reader = geoip2.database.Reader('/path/to/GeoLite2-City.mmdb')
 asn_reader = geoip2.database.Reader('/path/to/GeoLite2-ASN.mmdb')
 ​
 @app.route('/check')
 def check():
     ip = request.remote_addr
     try:
         geo = reader.city(ip)
         asn = asn_reader.asn(ip)
         return jsonify({
             "ip": ip,
             "country": geo.country.iso_code,
             "city": geo.city.name,
             "asn_number": asn.autonomous_system_number,
             "asn_org": asn.autonomous_system_organization,
             "is_datacenter": is_datacenter_asn(asn.autonomous_system_number),
         })
     except Exception as e:
         return jsonify({"ip": ip, "error": str(e)})
 ​
 def is_datacenter_asn(asn_number):
     """粗略判断：常见数据中心ASN列表"""
     dc_asns = {
         14061,   # DigitalOcean
         16509,   # Amazon/AWS
         15169,   # Google
         8075,    # Microsoft/Azure
         13335,   # Cloudflare
         20473,   # Choopa/Vultr
         63949,   # Linode
         # ... 补充更多
     }
     return asn_number in dc_asns

 def analyze_subnet_diversity(ip_list):
     """统计 /24 子网段分布"""
     from collections import Counter
     subnets = Counter()
     for ip in ip_list:
         parts = ip.split(".")
         subnet = ".".join(parts[:3])
         subnets[subnet] += 1
 ​
     total = len(ip_list)
     top10 = subnets.most_common(10)
     concentration = sum(c for _, c in top10) / total  # 前10个子网占总量的比例
 ​
     return {
         "total_ips": total,
         "unique_subnets": len(subnets),
         "top10_concentration": round(concentration, 3),
         "top10_subnets": top10,
     }

 def calc_purity_score(check_result):
     score = 100
 ​
     # 1. 连通性（基础分，不通直接0分）
     if not check_result["connectable"]:
         return 0
 ​
     # 2. 请求头泄露（扣20分）
     if check_result["header_leaks"]:
         score -= 20
 ​
     # 3. 地理位置不符（扣15分）
     if not check_result["geo_match"]:
         score -= 15
 ​
     # 4. 数据中心IP（扣10分，非绝对扣分，取决于场景）
     if check_result["is_datacenter"]:
         score -= 10
 ​
     # 5. 子网集中度过高（扣10分）
     if check_result["subnet_concentration"] > 0.3:
         score -= 10
 ​
     # 6. 延迟过高（>2000ms 扣10分，>5000ms 扣20分）
     latency = check_result["latency_ms"]
     if latency > 5000:
         score -= 20
     elif latency > 2000:
         score -= 10
 ​
     # 7. HTTPS 不支持（扣15分）
     if not check_result["https_supported"]:
         score -= 15
 ​
     # 8. 并发不稳定（10并发成功率<80% 扣10分）
     if check_result["concurrent_success_rate"] < 0.8:
         score -= 10
 ​
     # 9. DNS泄露（扣15分）
     if check_result["dns_leak"]:
         score -= 15
 ​
     return max(score, 0)

 import asyncio
 import aiohttp
 ​
 async def batch_check(ip_list, check_url, max_concurrent=50):
     sem = asyncio.Semaphore(max_concurrent)
     results = []
 ​
     async def check_one(ip):
         async with sem:
             try:
                 async with aiohttp.ClientSession() as session:
                     proxy_url = f"http://{ip}"
                     start = time.time()
                     async with session.get(
                         check_url,
                         proxy=proxy_url,
                         timeout=aiohttp.ClientTimeout(total=15)
                     ) as resp:
                         latency = (time.time() - start) * 1000
                         data = await resp.json()
                         results.append({
                             "proxy": ip,
                             "status": resp.status,
                             "latency_ms": round(latency, 1),
                             "exit_ip": data.get("origin", data.get("ip", "")),
                             "success": True,
                         })
             except Exception as e:
                 results.append({
                     "proxy": ip,
                     "success": False,
                     "error": str(e),
                 })
 ​
     tasks = [check_one(ip) for ip in ip_list]
     await asyncio.gather(*tasks)
     return results