代理IP全解析：从原理到自建代理池的实战指南

<table id="list">
  <tbody>
    <tr>
      <td>123.123.123.123</td> <!-- IP -->
      <td>8080</td>              <!-- 端口 -->
      <td>高匿</td>              <!-- 匿名度 -->
      <td>HTTP</td>              <!-- 类型 -->
    </tr>
  </tbody>
</table>

from lxml import etree
import httpx
 
headers = {'User-Agent': 'Mozilla/5.0'}
url = "https://www.kuaidaili.com/free/inha/1/"
res = httpx.get(url, headers=headers)
con = etree.HTML(res.text)
ip_list = con.xpath('//*[@id="list"]/table/tbody/tr/td/text()')
 
proxies = []
for i in range(len(ip_list)//4):
    proxies.append({
        "ip": ip_list[i*4],
        "port": ip_list[i*4+1],
        "anonymity": ip_list[i*4+2],
        "type": ip_list[i*4+3]
    })

import requests
 
API_URL = "http://api.example.com/proxies?count=10&type=https"
response = requests.get(API_URL)
proxies = response.text.strip().split('\n')

def test_proxy(proxy):
    proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"}
    try:
        res = requests.get("http://httpbin.org/ip", proxies=proxies, timeout=5)
        return res.json()["origin"] == proxy.split(":")[0]
    except:
        return False

import sqlite3
 
conn = sqlite3.connect("proxy_pool.db")
cursor = conn.cursor()
cursor.execute("""
    CREATE TABLE IF NOT EXISTS proxies (
        id INTEGER PRIMARY KEY,
        ip TEXT UNIQUE,
        port INTEGER,
        anonymity TEXT,
        type TEXT,
        last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        is_valid INTEGER DEFAULT 1
    )
""")

import redis
 
r = redis.Redis(host='localhost', port=6379, db=0)
 
# 存储代理
r.hset("proxy:123.123.123.123:8080", mapping={
    "anonymity": "high",
    "type": "https",
    "last_checked": "2025-08-11T12:00:00"
})
 
# 获取可用代理
valid_proxies = [k.decode() for k in r.keys("proxy:*") if r.hget(k, "is_valid") == b"1"]

import random
 
def get_random_proxy():
    valid_proxies = [p for p in all_proxies if p["is_valid"]]
    return random.choice(valid_proxies) if valid_proxies else None

def calculate_weight(proxy):
    # 响应时间越短，权重越高
    speed_weight = 1 / (proxy["avg_response_time"] + 0.1)
    # 成功率越高，权重越高
    success_weight = proxy["success_rate"] ** 2
    return speed_weight * success_weight
 
def get_weighted_proxy():
    weighted_proxies = [(p, calculate_weight(p)) for p in all_proxies]
    total_weight = sum(w for _, w in weighted_proxies)
    rand = random.uniform(0, total_weight)
    current_sum = 0
    for proxy, weight in weighted_proxies:
        current_sum += weight
        if current_sum > rand:
            return proxy
    return None

from apscheduler.schedulers.blocking import BlockingScheduler
 
def update_proxies():
    new_proxies = fetch_proxies_from_api()
    for proxy in new_proxies:
        if not r.exists(f"proxy:{proxy}"):
            store_proxy(proxy)
    clean_invalid_proxies()
 
scheduler = BlockingScheduler()
scheduler.add_job(update_proxies, 'interval', hours=1)
scheduler.start()

# HELP proxy_pool_size Current size of proxy pool
# TYPE proxy_pool_size gauge
proxy_pool_size{type="https"} 152
proxy_pool_size{type="socks5"} 47
 
# HELP proxy_request_duration Proxy request duration in seconds
# TYPE proxy_request_duration histogram
proxy_request_duration_bucket{le="0.1"} 1245
proxy_request_duration_bucket{le="0.5"} 1892

def post_to_social_media(account, content):
    proxy = get_proxy_by_account(account)
    session = requests.Session()
    session.proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"}
    
    # 模拟人类操作延迟
    time.sleep(random.uniform(5, 15))
    
    response = session.post(
        "https://api.socialmedia.com/posts",
        json={"content": content},
        headers={
            "User-Agent": random.choice(USER_AGENTS),
            "X-Forwarded-For": generate_random_ip()
        }
    )
    return response.json()