Python 爬虫代理实战 —— 从一行代码到生产级代理池 🕵️#

1
同一个 IP，短时间请求太频繁 → 标记 → 限制 → 封禁

1
import requests
2

3
proxies = {
4
    "http": "http://1.2.3.4:8080",
5
    "https": "http://1.2.3.4:8080",
6
}
7

8
resp = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=10)
9
print(resp.json())   # {"origin": "1.2.3.4"} ← 网站看到的你的 IP

1
proxies = {
2
    "http": "http://user123:pass456@1.2.3.4:8080",
3
    "https": "http://user123:pass456@1.2.3.4:8080",
4
}

1
# httpx 同步版
2
import httpx
3
with httpx.Client(proxy="http://1.2.3.4:8080", timeout=10) as client:
4
    r = client.get("https://httpbin.org/ip")
5
    print(r.json())
6

7
# aiohttp 异步版
8
import aiohttp, asyncio
9

10
async def main():
11
    async with aiohttp.ClientSession() as session:
12
        async with session.get("https://httpbin.org/ip",
13
                               proxy="http://1.2.3.4:8080") as resp:
14
            print(await resp.json())
15

16
asyncio.run(main())

1
import requests
2

3
proxies = {"http": "http://1.2.3.4:8080", "https": "http://1.2.3.4:8080"}
4
r = requests.get("https://httpbin.org/headers", proxies=proxies, timeout=10)
5
headers = r.json()["headers"]
6

7
print("X-Forwarded-For" in headers)  # True = 透明代理，真IP泄露！
8
print("Via" in headers)              # True = 匿名代理，被标记为代理流量
9
# 两个都是 False = 高匿，放心用

1
# 需要 pip install requests[socks]
2
proxies = {
3
    "http": "socks5h://user:pass@1.2.3.4:1080",
4
    "https": "socks5h://user:pass@1.2.3.4:1080",
5
}

1
{"code": 0, "data": [{"ip": "1.2.3.4", "port": 8080, "expire": "2026-08-17 18:00:00"}]}

1
import requests
2

3
API_URL = "https://代理厂商的提取接口?num=50&type=json"
4

5
def fetch_proxies(num=50):
6
    """从厂商API拉一批代理，返回 ['ip:port', ...] 列表"""
7
    try:
8
        r = requests.get(API_URL, params={"num": num}, timeout=15)
9
        data = r.json()
10
        return [f"{item['ip']}:{item['port']}" for item in data.get("data", [])]
11
    except Exception:
12
        return []   # 拉取失败返回空，让上层重试

1
# 你的代码里永远写同一个入口，IP 却是千变万化的
2
proxies = {"http": "http://user:pass@tunnel.example.com:8888",
3
           "https": "http://user:pass@tunnel.example.com:8888"}

1
import time
2
import requests
3

4
def validate_proxy(proxy, timeout=8):
5
    """体检一个代理，返回 (可用?, 延迟, 匿名级别)"""
6
    proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"}
7
    try:
8
        start = time.time()
9
        r = requests.get("https://httpbin.org/ip",
10
                         proxies=proxies, timeout=timeout)
11
        latency = time.time() - start
12

13
        if r.status_code != 200:
14
            return False, None, None
15

16
        # 关键校验：网站看到的IP必须等于代理IP
17
        seen = r.json().get("origin", "").split(",")[0].strip()
18
        if seen != proxy.split(":")[0]:
19
            return False, None, None   # 代理没生效，走了直连！
20

21
        # 匿名性检测
22
        r2 = requests.get("https://httpbin.org/headers",
23
                          proxies=proxies, timeout=timeout)
24
        h = r2.json().get("headers", {})
25
        if "X-Forwarded-For" in h or "Via" in h:
26
            level = "anonymous"   # 匿名或透明，凑合能用但风险高
27
        else:
28
            level = "elite"       # 高匿，合格
29
        return True, round(latency, 2), level
30
    except Exception:
31
        return False, None, None

1
from concurrent.futures import ThreadPoolExecutor, as_completed
2

3
def validate_batch(proxy_list, workers=20):
4
    """并发体检，返回 [(proxy, latency, level), ...] 通过者列表"""
5
    good = []
6
    with ThreadPoolExecutor(max_workers=workers) as ex:
7
        futures = {ex.submit(validate_proxy, p): p for p in proxy_list}
8
        for f in as_completed(futures):
9
            proxy = futures[f]
10
            ok, latency, level = f.result()
11
            if ok:
12
                good.append((proxy, latency, level))
13
    return good

1
import random
2

3
pool = ["1.1.1.1:8080", "2.2.2.2:8080", "3.3.3.3:8080"]
4
proxy = random.choice(pool)

1
import itertools
2

3
pool = ["1.1.1.1:8080", "2.2.2.2:8080", "3.3.3.3:8080"]
4
rr = itertools.cycle(pool)      # 无限循环迭代器
5

6
for _ in range(10):
7
    proxy = next(rr)            # 1,2,3,1,2,3,1,2,3,1 绝对均匀

1
import random
2

3
# (代理, 权重)，权重可以是 10/延迟 之类的分数
4
weighted_pool = [("1.1.1.1:8080", 10), ("2.2.2.2:8080", 6), ("3.3.3.3:8080", 3)]
5

6
proxies = [p for p, w in weighted_pool]
7
weights = [w for p, w in weighted_pool]
8
proxy = random.choices(proxies, weights=weights, k=1)[0]

1
import requests
2

3
def get_with_retry(url, pool, max_tries=3):
4
    """拿代理发请求，失败就换代理重试，最多试3个代理"""
5
    for _ in range(max_tries):
6
        proxy = random.choice(pool)
7
        try:
8
            r = requests.get(url,
9
                             proxies={"http": f"http://{proxy}",
10
                                      "https": f"http://{proxy}"},
11
                             timeout=10)
12
            if r.status_code == 200:
13
                return r
14
        except Exception:
15
            continue      # 这个代理死了，换下一个
16
    return None           # 池子里连试3个都不行

1
import random
2

3
def crawl_task(urls):
4
    """一个任务用一个IP，从一而终"""
5
    proxy = random.choice(pool)
6
    proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"}
7
    session = requests.Session()      # 顺便复用TCP连接，快
8
    for url in urls:
9
        r = session.get(url, proxies=proxies, timeout=10)
10
        # 处理 r ...

1
# mini_pool.py —— 单机版代理池，一个文件搞定
2
import sqlite3
3
import threading
4
import time
5
import random
6
from concurrent.futures import ThreadPoolExecutor, as_completed
7

8
# ---- 存储层：SQLite ----
9
class MiniPool:
10
    def __init__(self, db="mini_pool.db"):
11
        self.conn = sqlite3.connect(db, check_same_thread=False)
12
        self.lock = threading.Lock()
13
        self.conn.execute("""
14
            CREATE TABLE IF NOT EXISTS proxies (
15
                proxy TEXT PRIMARY KEY,
16
                score INTEGER DEFAULT 100,
17
                latency REAL,
18
                level TEXT,
19
                updated_at REAL
20
            )""")
21
        self.conn.commit()
22

23
    def add(self, proxy, score=100, latency=0, level="elite"):
24
        with self.lock:
25
            self.conn.execute(
26
                "INSERT OR REPLACE INTO proxies VALUES (?,?,?,?,?)",
27
                (proxy, score, latency, level, time.time()))
28
            self.conn.commit()
29

30
    def get(self):
31
        """取一个代理：优先高分，同分随机"""
32
        with self.lock:
33
            cur = self.conn.execute(
34
                "SELECT proxy FROM proxies WHERE score >= 60 "
35
                "ORDER BY score DESC, RANDOM() LIMIT 1")
36
            row = cur.fetchone()
37
        return row[0] if row else None
38

39
    def punish(self, proxy, penalty=20):
40
        """代理用坏了：扣分"""
41
        with self.lock:
42
            self.conn.execute(
43
                "UPDATE proxies SET score = score - ? WHERE proxy = ?",
44
                (penalty, proxy))
45
            self.conn.commit()
46

47
    def remove(self, proxy):
48
        with self.lock:
49
            self.conn.execute("DELETE FROM proxies WHERE proxy = ?", (proxy,))
50
            self.conn.commit()
51

52
    def sweep(self, min_score=40):
53
        """大扫除：清掉低分和过期代理"""
54
        with self.lock:
55
            self.conn.execute("DELETE FROM proxies WHERE score < ?", (min_score,))
56
            self.conn.commit()
57

58
    def count(self):
59
        with self.lock:
60
            return self.conn.execute("SELECT COUNT(*) FROM proxies").fetchone()[0]

1
# 接在 mini_pool.py 后面
2
pool = MiniPool()
3

4
def replenish():
5
    """库存低于50就进货，然后批量体检入库"""
6
    if pool.count() >= 50:
7
        return
8
    fresh = fetch_proxies(num=50)          # 第4.2节的函数
9
    good = validate_batch(fresh, workers=20)  # 第5.3节的函数
10
    for proxy, latency, level in good:
11
        score = 100 if level == "elite" else 70
12
        pool.add(proxy, score=score, latency=latency, level=level)
13
    print(f"进货 {len(fresh)} 个，存活 {len(good)} 个，库存 {pool.count()}")
14

15
def crawl_with_pool(url):
16
    """爬虫侧用法：取代理 → 请求 → 失败就惩罚"""
17
    for _ in range(3):
18
        proxy = pool.get()
19
        if not proxy:
20
            replenish()        # 池子空了，先补货
21
            continue
22
        try:
23
            r = requests.get(url,
24
                proxies={"http": f"http://{proxy}", "https": f"http://{proxy}"},
25
                timeout=10)
26
            if r.status_code == 200:
27
                return r
28
        except Exception:
29
            pool.punish(proxy) # 它死了，扣分
30
    return None
31

32
# 启动时先补一轮，之后每5分钟补一次（生产环境用定时任务）
33
replenish()
34
while True:
35
    time.sleep(300)
36
    replenish()
37
    pool.sweep(min_score=40)

1
# storage.py —— Redis 存储层
2
import redis
3

4
class RedisStorage:
5
    def __init__(self, host="127.0.0.1", port=6379, db=0, key="proxy_pool:zset"):
6
        self.r = redis.Redis(host=host, port=port, db=db, decode_responses=True)
7
        self.key = key
8

9
    def add(self, proxy, score=100):
10
        """新代理入库（重复的只更新分数）"""
11
        self.r.zadd(self.key, {proxy: score})
12

13
    def get(self):
14
        """取分数最高的一个，并顺手 -1 分（自然降权）"""
15
        top = self.r.zrevrange(self.key, 0, 0)
16
        if not top:
17
            return None
18
        proxy = top[0]
19
        self.r.zincrby(self.key, -1, proxy)
20
        return proxy
21

22
    def punish(self, proxy, penalty=30):
23
        """爬虫上报代理挂了：狠扣"""
24
        self.r.zincrby(self.key, -penalty, proxy)
25

26
    def revive(self, proxy):
27
        """体检通过：分数拉回80"""
28
        self.r.zadd(self.key, {proxy: 80})
29

30
    def remove(self, proxy):
31
        self.r.zrem(self.key, proxy)
32

33
    def sweep(self, min_score=50):
34
        """定时大扫除"""
35
        self.r.zremrangebyscore(self.key, "-inf", min_score)
36

37
    def count(self):
38
        return self.r.zcard(self.key)
39

40
    def all_with_scores(self):
41
        return self.r.zrevrange(self.key, 0, -1, withscores=True)

1
proxy_pool/
2
├── config.py        # 配置：Redis地址、厂商API、阈值
3
├── getter.py        # 获取器：从厂商API进货
4
├── validator.py     # 验证器：并发体检
5
├── storage.py       # 存储层：Redis ZSet（上面已给）
6
├── scheduler.py     # 调度器：定时补给+清洗
7
├── api.py           # HTTP服务：对外提供代理
8
└── main.py          # 入口：一键启动所有模块

1
REDIS_HOST = "127.0.0.1"
2
REDIS_PORT = 6379
3

4
VENDOR_API = "https://代理厂商的提取接口?num=100&type=json"
5
MIN_STOCK = 50        # 库存低于这个数就进货
6
MAX_STOCK = 200       # 库存上限
7
VALIDATE_WORKERS = 30 # 并发体检线程数
8
SWEEP_INTERVAL = 300  # 清洗间隔(秒)
9
GET_INTERVAL = 600    # 补给检查间隔(秒)
10
MIN_SCORE = 50        # 低于该分淘汰

1
import requests
2
from config import VENDOR_API
3

4
def fetch(num=100):
5
    """从厂商API拉代理。失败返回空列表，调度器稍后重试"""
6
    try:
7
        r = requests.get(VENDOR_API, params={"num": num}, timeout=20)
8
        data = r.json()
9
        return [f"{x['ip']}:{x['port']}" for x in data.get("data", [])]
10
    except Exception as e:
11
        print(f"[getter] 拉取失败: {e}")
12
        return []

1
import requests
2
from concurrent.futures import ThreadPoolExecutor, as_completed
3
from config import VALIDATE_WORKERS
4

5
def validate_one(proxy, timeout=8):
6
    """返回 (proxy, ok, latency, level)"""
7
    proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"}
8
    try:
9
        import time
10
        t0 = time.time()
11
        r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=timeout)
12
        latency = time.time() - t0
13
        if r.status_code != 200:
14
            return proxy, False, None, None
15
        if r.json().get("origin", "").split(",")[0].strip() != proxy.split(":")[0]:
16
            return proxy, False, None, None
17
        r2 = requests.get("https://httpbin.org/headers", proxies=proxies, timeout=timeout)
18
        h = r2.json().get("headers", {})
19
        level = "elite" if ("X-Forwarded-For" not in h and "Via" not in h) else "anonymous"
20
        return proxy, True, round(latency, 2), level
21
    except Exception:
22
        return proxy, False, None, None
23

24
def validate_batch(proxies):
25
    """并发体检一批，返回通过的 [(proxy, latency, level)]"""
26
    good = []
27
    with ThreadPoolExecutor(max_workers=VALIDATE_WORKERS) as ex:
28
        futs = [ex.submit(validate_one, p) for p in proxies]
29
        for f in as_completed(futs):
30
            proxy, ok, latency, level = f.result()
31
            if ok:
32
                good.append((proxy, latency, level))
33
    return good

1
import time
2
from getter import fetch
3
from validator import validate_batch
4
from storage import RedisStorage
5
from config import (MIN_STOCK, MAX_STOCK, MIN_SCORE,
6
                    SWEEP_INTERVAL, GET_INTERVAL)
7

8
storage = RedisStorage()
9

10
def replenish():
11
    """库存不够就进货 → 体检 → 入库"""
12
    if storage.count() >= MIN_STOCK:
13
        return
14
    need = MAX_STOCK - storage.count()
15
    fresh = fetch(num=need)
16
    if not fresh:
17
        print("[scheduler] 没拉到新货")
18
        return
19
    good = validate_batch(fresh)
20
    for proxy, latency, level in good:
21
        score = 100 if level == "elite" else 70
22
        storage.add(proxy, score=score)
23
    print(f"[scheduler] 进货{len(fresh)} 存活{len(good)} 库存{storage.count()}")
24

25
def run():
26
    """主循环：补给 + 定期体检在库代理 + 大扫除"""
27
    last_get = last_sweep = 0
28
    while True:
29
        now = time.time()
30
        if now - last_get > GET_INTERVAL:
31
            replenish()
32
            last_get = now
33
        if now - last_sweep > SWEEP_INTERVAL:
34
            # 对在库代理做一轮体检，坏的扣分，好的复活
35
            for proxy, score in storage.all_with_scores():
36
                _, ok, _, _ = validate_one(proxy)
37
                if ok:
38
                    storage.revive(proxy)
39
                else:
40
                    storage.punish(proxy, penalty=40)
41
            storage.sweep(min_score=MIN_SCORE)
42
            last_sweep = now
43
            print(f"[scheduler] 清洗完成，库存{storage.count()}")
44
        time.sleep(10)

1
# pip install fastapi uvicorn
2
from fastapi import FastAPI, Query
3
from storage import RedisStorage
4

5
app = FastAPI(title="Proxy Pool API")
6
storage = RedisStorage()
7

8
@app.get("/get")
9
def get_proxy():
10
    """取一个代理：爬虫每次请求前调用"""
11
    proxy = storage.get()
12
    if not proxy:
13
        return {"code": 0, "msg": "池子空了，稍后再试"}
14
    return {"code": 1, "proxy": proxy}
15

16
@app.get("/count")
17
def count():
18
    return {"code": 1, "count": storage.count()}
19

20
@app.delete("/bad")
21
def report_bad(proxy: str = Query(..., description="挂掉的代理 ip:port")):
22
    """爬虫上报坏代理"""
23
    storage.punish(proxy)
24
    return {"code": 1, "msg": "已惩罚"}

1
import threading
2
import uvicorn
3
from scheduler import run as scheduler_run
4
from api import app
5

6
if __name__ == "__main__":
7
    # 调度器在后台线程跑
8
    threading.Thread(target=scheduler_run, daemon=True).start()
9
    # API 服务在前台
10
    uvicorn.run(app, host="0.0.0.0", port=8000)

1
import requests
2

3
def get_proxy():
4
    r = requests.get("http://127.0.0.1:8000/get", timeout=3)
5
    data = r.json()
6
    return data["proxy"] if data["code"] == 1 else None
7

8
def report_bad(proxy):
9
    requests.delete("http://127.0.0.1:8000/bad", params={"proxy": proxy}, timeout=3)
10

11
def crawl(url):
12
    for _ in range(3):
13
        proxy = get_proxy()
14
        if not proxy:
15
            continue
16
        try:
17
            r = requests.get(url,
18
                proxies={"http": f"http://{proxy}", "https": f"http://{proxy}"},
19
                timeout=10)
20
            if r.status_code == 200:
21
                return r
22
        except Exception:
23
            report_bad(proxy)   # 坏了上报，池子会惩罚它
24
    return None

1
# monitor.py —— 每5分钟查一次水位，低了就喊人
2
import time, requests
3

4
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=你的token"
5

6
def check_and_alert():
7
    r = requests.get("http://127.0.0.1:8000/count", timeout=5)
8
    count = r.json()["count"]
9
    if count < 20:
10
        requests.post(DINGTALK_WEBHOOK, json={
11
            "msgtype": "text",
12
            "text": {"content": f"⚠️ 代理池水位告警: 当前仅剩 {count} 个代理！"}
13
        })
14
    return count
15

16
while True:
17
    try:
18
        print(f"[monitor] 水位 {check_and_alert()}")
19
    except Exception as e:
20
        print(f"[monitor] 检查失败: {e}")
21
    time.sleep(300)