1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

# fetcher/free_proxy.py
import re
import asyncio
import aiohttp
from typing import List, Tuple
from abc import ABC, abstractmethod


class BaseFetcher(ABC):
    """代理采集器基类"""

    @abstractmethod
    async def fetch(self) -> List[Tuple[str, int, str]]:
        """
        返回代理列表，每个元素为 (ip, port, protocol)
        protocol: 'http', 'https', 'socks5'
        """
        pass


class KuaiDailiFetcher(BaseFetcher):
    """快代理采集器"""

    BASE_URL = "https://www.kuaidaili.com/free/inha/{page}/"

    async def fetch(self) -> List[Tuple[str, int, str]]:
        proxies = []
        async with aiohttp.ClientSession() as session:
            for page in range(1, 6):  # 采集前 5 页
                try:
                    url = self.BASE_URL.format(page=page)
                    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                        html = await resp.text()
                        # 解析 IP 和端口
                        ips = re.findall(
                            r'<td data-title="IP">([\d.]+)</td>', html
                        )
                        ports = re.findall(
                            r'<td data-title="PORT">(\d+)</td>', html
                        )
                        for ip, port in zip(ips, ports):
                            proxies.append((ip, int(port), "http"))
                    await asyncio.sleep(2)  # 避免被反爬
                except Exception as e:
                    print(f"[KuaiDaili] 采集第 {page} 页失败: {e}")
        return proxies


class Ip3366Fetcher(BaseFetcher):
    """云代理采集器"""

    BASE_URL = "http://www.ip3366.net/free/?stype=1&page={page}"

    async def fetch(self) -> List[Tuple[str, int, str]]:
        proxies = []
        async with aiohttp.ClientSession() as session:
            for page in range(1, 4):
                try:
                    url = self.BASE_URL.format(page=page)
                    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                        html = await resp.text()
                        items = re.findall(
                            r'<tr>\s*<td>([\d.]+)</td>\s*<td>(\d+)</td>',
                            html
                        )
                        for ip, port in items:
                            proxies.append((ip, int(port), "http"))
                    await asyncio.sleep(2)
                except Exception as e:
                    print(f"[Ip3366] 采集第 {page} 页失败: {e}")
        return proxies

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# fetcher/paid_proxy.py
import aiohttp
from typing import List, Tuple


class PaidProxyFetcher(BaseFetcher):
    """付费代理 API 采集器"""

    def __init__(self, api_url: str, api_key: str):
        self.api_url = api_url
        self.api_key = api_key

    async def fetch(self) -> List[Tuple[str, int, str]]:
        proxies = []
        params = {
            "key": self.api_key,
            "num": 100,          # 每次提取数量
            "protocol": 1,       # 1=HTTP, 2=HTTPS, 5=SOCKS5
            "anonymity": 2,      # 2=高匿
            "format": "json",
        }
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(self.api_url, params=params) as resp:
                    data = await resp.json()
                    for item in data.get("data", []):
                        proxies.append((
                            item["ip"],
                            item["port"],
                            item.get("protocol", "http")
                        ))
            except Exception as e:
                print(f"[PaidProxy] 采集失败: {e}")
        return proxies


class FetcherScheduler:
    """采集调度器：管理所有采集器并汇总结果"""

    def __init__(self, fetchers: list):
        self.fetchers = fetchers

    async def fetch_all(self) -> List[Tuple[str, int, str]]:
        tasks = [f.fetch() for f in self.fetchers]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        all_proxies = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"采集器 {i} 出错: {result}")
            else:
                all_proxies.extend(result)
                print(f"采集器 {i} 获取 {len(result)} 个代理")

        # 去重
        unique = list(set(all_proxies))
        print(f"汇总: 共 {len(all_proxies)} 个，去重后 {len(unique)} 个")
        return unique

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

# checker/validator.py
import time
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import Optional


@dataclass
class ProxyInfo:
    """代理详细信息"""
    ip: str
    port: int
    protocol: str
    speed: float = 0.0          # 响应时间（秒）
    anonymity: str = "unknown"  # transparent / anonymous / high_anonymous
    country: str = "unknown"
    is_valid: bool = False


class ProxyValidator:
    """异步代理验证器"""

    # 用于检测 IP 和匿名度的 API
    CHECK_URL = "https://httpbin.org/ip"
    TIMEOUT = 10  # 秒
    CONCURRENT_LIMIT = 100  # 最大并发数

    def __init__(self, local_ip: str):
        self.local_ip = local_ip
        self.semaphore = asyncio.Semaphore(self.CONCURRENT_LIMIT)

    async def validate_single(self, proxy: ProxyInfo) -> ProxyInfo:
        """验证单个代理"""
        async with self.semaphore:
            proxy_url = f"{proxy.protocol}://{proxy.ip}:{proxy.port}"
            start_time = time.time()
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(
                        self.CHECK_URL,
                        proxy=proxy_url,
                        timeout=aiohttp.ClientTimeout(total=self.TIMEOUT),
                    ) as resp:
                        if resp.status == 200:
                            data = await resp.json()
                            proxy.speed = round(time.time() - start_time, 2)
                            proxy.is_valid = True
                            # 匿名度检测
                            origin_ip = data.get("origin", "")
                            if self.local_ip not in origin_ip:
                                proxy.anonymity = "high_anonymous"
                            else:
                                proxy.anonymity = "transparent"
            except Exception:
                proxy.is_valid = False
            return proxy

    async def validate_batch(self, proxies: list) -> list:
        """批量验证代理"""
        tasks = [self.validate_single(p) for p in proxies]
        results = await asyncio.gather(*tasks)
        valid = [p for p in results if p.is_valid]
        print(f"验证完成: {len(proxies)} 个中 {len(valid)} 个可用")
        return valid

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# checker/geo.py
import aiohttp


class GeoChecker:
    """代理地理位置检测"""

    GEO_API = "http://ip-api.com/json/{ip}?lang=zh-CN"

    async def check_geo(self, ip: str) -> dict:
        """查询 IP 地理信息"""
        try:
            async with aiohttp.ClientSession() as session:
                url = self.GEO_API.format(ip=ip)
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                    data = await resp.json()
                    if data.get("status") == "success":
                        return {
                            "country": data.get("country", ""),
                            "region": data.get("regionName", ""),
                            "city": data.get("city", ""),
                            "isp": data.get("isp", ""),
                        }
        except Exception:
            pass
        return {"country": "unknown", "region": "", "city": "", "isp": ""}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

# storage/redis_store.py
import json
import random
import redis
from typing import Optional, List


class ProxyStore:
    """Redis 代理存储"""

    PROXY_KEY = "proxy:pool"            # 有序集合主键
    PROXY_DETAIL_PREFIX = "proxy:detail:"  # 代理详情 Hash
    INITIAL_SCORE = 50                  # 新代理初始分数
    MAX_SCORE = 100
    MIN_SCORE = 0

    def __init__(self, host="localhost", port=6379, db=0, password=None):
        self.redis = redis.Redis(
            host=host, port=port, db=db, password=password,
            decode_responses=True
        )

    def add(self, proxy_str: str, detail: dict = None) -> bool:
        """添加新代理（如果已存在则跳过）"""
        if not self.redis.zscore(self.PROXY_KEY, proxy_str):
            self.redis.zadd(self.PROXY_KEY, {proxy_str: self.INITIAL_SCORE})
            if detail:
                self.redis.hset(
                    f"{self.PROXY_DETAIL_PREFIX}{proxy_str}",
                    mapping=detail
                )
                # 设置详情 TTL，24 小时后自动清理
                self.redis.expire(
                    f"{self.PROXY_DETAIL_PREFIX}{proxy_str}", 86400
                )
            return True
        return False

    def increase_score(self, proxy_str: str, delta: int = 1):
        """提升代理分数（验证成功时调用）"""
        score = self.redis.zscore(self.PROXY_KEY, proxy_str)
        if score is not None:
            new_score = min(self.MAX_SCORE, score + delta)
            self.redis.zadd(self.PROXY_KEY, {proxy_str: new_score})

    def decrease_score(self, proxy_str: str, delta: int = 1):
        """降低代理分数（验证失败时调用），低于阈值则删除"""
        score = self.redis.zscore(self.PROXY_KEY, proxy_str)
        if score is not None:
            new_score = score - delta
            if new_score <= self.MIN_SCORE:
                self.remove(proxy_str)
            else:
                self.redis.zadd(self.PROXY_KEY, {proxy_str: new_score})

    def remove(self, proxy_str: str):
        """删除代理"""
        self.redis.zrem(self.PROXY_KEY, proxy_str)
        self.redis.delete(f"{self.PROXY_DETAIL_PREFIX}{proxy_str}")

    def get_random(self, min_score: int = 80) -> Optional[str]:
        """获取一个高分随机代理"""
        # 先尝试获取满分代理
        top_proxies = self.redis.zrangebyscore(
            self.PROXY_KEY, min_score, self.MAX_SCORE
        )
        if top_proxies:
            return random.choice(top_proxies)
        # 降级：获取任意可用代理
        all_proxies = self.redis.zrangebyscore(
            self.PROXY_KEY, self.MIN_SCORE + 1, self.MAX_SCORE
        )
        if all_proxies:
            return random.choice(all_proxies)
        return None

    def get_top(self, count: int = 10) -> List[tuple]:
        """获取评分最高的 N 个代理"""
        return self.redis.zrevrangebyscore(
            self.PROXY_KEY, self.MAX_SCORE, self.MIN_SCORE,
            start=0, num=count, withscores=True
        )

    def count(self) -> int:
        """获取代理总数"""
        return self.redis.zcard(self.PROXY_KEY)

    def count_valid(self, min_score: int = 50) -> int:
        """获取有效代理数（分数高于阈值）"""
        return self.redis.zcount(self.PROXY_KEY, min_score, self.MAX_SCORE)

    def cleanup_expired(self, min_score: int = 10):
        """清理低分代理"""
        removed = self.redis.zremrangebyscore(
            self.PROXY_KEY, 0, min_score
        )
        print(f"清理了 {removed} 个低分代理")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# api/server.py
from fastapi import FastAPI, Query, HTTPException
from storage.redis_store import ProxyStore

app = FastAPI(title="Proxy Pool API", version="1.0")
store = ProxyStore()


@app.get("/get")
async def get_proxy(
    min_score: int = Query(80, description="最低分数阈值"),
    protocol: str = Query(None, description="指定协议: http/https/socks5"),
):
    """随机获取一个高质量代理"""
    proxy = store.get_random(min_score=min_score)
    if not proxy:
        raise HTTPException(status_code=503, detail="代理池为空")
    return {"proxy": proxy, "protocol": protocol or "http"}


@app.get("/get_batch")
async def get_batch(
    count: int = Query(10, description="获取数量"),
    min_score: int = Query(80, description="最低分数阈值"),
):
    """批量获取代理"""
    proxies = store.get_top(count=count)
    return {
        "count": len(proxies),
        "proxies": [{"proxy": p, "score": s} for p, s in proxies],
    }


@app.get("/count")
async def get_count():
    """查看代理池状态"""
    return {
        "total": store.count(),
        "valid": store.count_valid(min_score=50),
        "high_quality": store.count_valid(min_score=80),
    }


@app.post("/report")
async def report_proxy(proxy: str, is_valid: bool):
    """爬虫反馈代理状态"""
    if is_valid:
        store.increase_score(proxy)
    else:
        store.decrease_score(proxy, delta=10)
    return {"status": "ok"}


# 启动: uvicorn api.server:app --host 0.0.0.0 --port 5000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# api/weighted_selector.py
import random
from typing import Optional


def weighted_random_proxy(store) -> Optional[str]:
    """
    基于分数的加权随机选择。
    分数越高的代理被选中的概率越大。
    """
    # 获取所有代理及其分数
    proxies_with_scores = store.redis.zrangebyscore(
        store.PROXY_KEY, 10, store.MAX_SCORE, withscores=True
    )
    if not proxies_with_scores:
        return None

    proxies = [p for p, _ in proxies_with_scores]
    weights = [s for _, s in proxies_with_scores]

    # random.choices 支持权重选择
    selected = random.choices(proxies, weights=weights, k=1)
    return selected[0]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

# scorer/quality.py
from dataclasses import dataclass


@dataclass
class ProxyMetrics:
    """代理性能指标"""
    total_requests: int = 0       # 总请求次数
    success_count: int = 0        # 成功次数
    fail_count: int = 0           # 失败次数
    avg_speed: float = 0.0        # 平均响应时间（秒）
    last_check_time: float = 0.0  # 上次检测时间戳
    consecutive_fails: int = 0    # 连续失败次数


class ProxyScorer:
    """
    代理综合评分器

    最终分数 = 成功率得分 * 0.4 + 速度得分 * 0.3 + 稳定性得分 * 0.3
    """

    # 各维度权重
    WEIGHT_SUCCESS_RATE = 0.4
    WEIGHT_SPEED = 0.3
    WEIGHT_STABILITY = 0.3

    # 速度等级阈值（秒）
    SPEED_EXCELLENT = 1.0
    SPEED_GOOD = 3.0
    SPEED_ACCEPTABLE = 5.0

    def calculate_score(self, metrics: ProxyMetrics) -> int:
        """计算代理综合分数（0-100）"""

        # 1. 成功率得分
        if metrics.total_requests == 0:
            success_score = 50  # 新代理默认
        else:
            rate = metrics.success_count / metrics.total_requests
            success_score = rate * 100

        # 2. 速度得分
        if metrics.avg_speed <= self.SPEED_EXCELLENT:
            speed_score = 100
        elif metrics.avg_speed <= self.SPEED_GOOD:
            speed_score = 70
        elif metrics.avg_speed <= self.SPEED_ACCEPTABLE:
            speed_score = 40
        else:
            speed_score = 10

        # 3. 稳定性得分（连续失败次数越多，扣分越严重）
        if metrics.consecutive_fails == 0:
            stability_score = 100
        elif metrics.consecutive_fails <= 2:
            stability_score = 60
        elif metrics.consecutive_fails <= 5:
            stability_score = 20
        else:
            stability_score = 0

        # 加权计算
        final_score = (
            success_score * self.WEIGHT_SUCCESS_RATE
            + speed_score * self.WEIGHT_SPEED
            + stability_score * self.WEIGHT_STABILITY
        )
        return max(0, min(100, int(final_score)))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

# middlewares.py
import requests
import logging
from scrapy.exceptions import IgnoreRequest

class ProxyMiddleware:
    def __init__(self, proxy_pool_url):
        self.proxy_pool_url = proxy_pool_url
        self.logger = logging.getLogger(__name__)

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            proxy_pool_url=crawler.settings.get('PROXY_POOL_URL')
        )

    def _get_random_proxy(self):
        try:
            response = requests.get(self.proxy_pool_url)
            if response.status_code == 200:
                return response.text.strip()
        except requests.ConnectionError:
            return None
        return None

    def process_request(self, request, spider):
        # 如果请求已经设置代理（例如特定请求），则跳过
        if request.meta.get('proxy'):
            return

        proxy = self._get_random_proxy()
        if proxy:
            self.logger.debug(f"Using proxy: {proxy}")
            # 设置代理，格式: http://user:pass@ip:端口 或 http://ip:端口
            request.meta['proxy'] = f"http://{proxy}"
        else:
            self.logger.warning("No proxy available from pool!")

    def process_response(self, request, response, spider):
        # 如果遇到验证码、封禁等状态码
        if response.status in [403, 429]:
            self.logger.warning(
                f"Proxy {request.meta.get('proxy')} banned "
                f"(Status {response.status}), retrying..."
            )
            # 标记该代理失效（可选：调用接口报告该代理坏）
            # self._report_bad_proxy(request.meta.get('proxy'))

            # 删除当前代理设置，重新调度请求（会再次经过 process_request 换新代理）
            del request.meta['proxy']
            return request.replace(dont_filter=True)

        return response

    def process_exception(self, request, exception, spider):
        # 处理连接超时、DNS 错误等
        self.logger.error(f"Proxy {request.meta.get('proxy')} failed: {exception}")

        # 换代理重试
        if 'proxy' in request.meta:
            del request.meta['proxy']
        return request.replace(dont_filter=True)

1
2
3
4
5
6
7
8

# settings.py
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.ProxyMiddleware': 543,
    # 禁用 Scrapy 默认 UserAgent 和重试中间件（视情况而定）
    # 'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,
}

PROXY_POOL_URL = 'http://localhost:5000/get'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

# re_tools/anti_crawl.py
import requests
from itertools import cycle


class ProxyRotator:
    """
    逆向工程专用代理轮转器。
    适用于：API 接口批量测试、签名算法验证、数据批量拉取等场景。
    """

    def __init__(self, pool_api: str = "http://localhost:5000"):
        self.pool_api = pool_api

    def get_proxy(self, protocol: str = "http") -> dict:
        """从代理池获取一个代理，返回 requests 可用格式"""
        try:
            resp = requests.get(f"{self.pool_api}/get")
            proxy_str = resp.json().get("proxy")
            if proxy_str:
                return {
                    "http": f"http://{proxy_str}",
                    "https": f"http://{proxy_str}",
                }
        except Exception:
            pass
        return {}

    def report_result(self, proxy_str: str, success: bool):
        """向代理池反馈使用结果"""
        try:
            requests.post(
                f"{self.pool_api}/report",
                params={"proxy": proxy_str, "is_valid": success}
            )
        except Exception:
            pass

    def request_with_retry(self, url: str, max_retries: int = 3, **kwargs) -> requests.Response:
        """带代理轮转的请求，自动重试"""
        for attempt in range(max_retries):
            proxies = self.get_proxy()
            try:
                resp = requests.get(
                    url, proxies=proxies, timeout=10, **kwargs
                )
                if resp.status_code == 200:
                    self.report_result(
                        list(proxies.values())[0], True
                    )
                    return resp
            except Exception:
                if proxies:
                    self.report_result(
                        list(proxies.values())[0], False
                    )
        raise Exception(f"请求失败，已重试 {max_retries} 次")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# frida_proxy_hook.py
"""
Frida 脚本：Hook OkHttp 设置动态代理
配合代理池使用，每次请求使用不同代理 IP
"""

FRIDA_SCRIPT = """
Java.perform(function() {
    var Proxy = Java.use('java.net.Proxy');
    var ProxyType = Java.use('java.net.Proxy$Type');
    var InetSocketAddress = Java.use('java.net.InetSocketAddress');
    var OkHttpClient = Java.use('okhttp3.OkHttpClient$Builder');

    // Hook OkHttpClient.Builder.build()
    OkHttpClient.build.implementation = function() {
        // 从代理池 API 获取代理（通过 Java 发 HTTP 请求）
        var proxyHost = "PROXY_HOST";
        var proxyPort = PROXY_PORT;

        var addr = InetSocketAddress.$new(proxyHost, proxyPort);
        var proxy = Proxy.$new(ProxyType.HTTP.value, addr);
        this.proxy(proxy);

        console.log("[*] OkHttp proxy set to: " + proxyHost + ":" + proxyPort);
        return this.build();
    };
});
"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

# monitor/metrics.py
from prometheus_client import Gauge, Counter, Histogram, start_http_server

# 代理池容量指标
POOL_TOTAL = Gauge(
    "proxy_pool_total", "代理池中代理总数"
)
POOL_VALID = Gauge(
    "proxy_pool_valid", "可用代理数量"
)
POOL_HIGH_QUALITY = Gauge(
    "proxy_pool_high_quality", "高质量代理数量（分数>80）"
)

# 采集器指标
FETCH_COUNT = Counter(
    "proxy_fetch_total", "采集到的代理总数", ["source"]
)
FETCH_ERRORS = Counter(
    "proxy_fetch_errors_total", "采集失败次数", ["source"]
)

# 验证器指标
CHECK_DURATION = Histogram(
    "proxy_check_duration_seconds", "单个代理验证耗时"
)
CHECK_RESULTS = Counter(
    "proxy_check_results_total", "验证结果统计", ["result"]
)

# API 指标
API_REQUESTS = Counter(
    "proxy_api_requests_total", "API 请求次数", ["endpoint"]
)
API_EMPTY_RESPONSES = Counter(
    "proxy_api_empty_total", "代理池为空的响应次数"
)


def update_pool_metrics(store):
    """定期更新池容量指标"""
    POOL_TOTAL.set(store.count())
    POOL_VALID.set(store.count_valid(min_score=50))
    POOL_HIGH_QUALITY.set(store.count_valid(min_score=80))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# alertmanager/rules.yml
groups:
  - name: proxy_pool_alerts
    rules:
      - alert: ProxyPoolLow
        expr: proxy_pool_valid < 50
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "可用代理数量不足"
          description: "代理池可用代理仅剩 {{ $value }} 个，请检查采集器"

      - alert: ProxyPoolEmpty
        expr: proxy_pool_valid == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "代理池为空"
          description: "代理池中没有可用代理，爬虫任务可能中断"

      - alert: FetcherDown
        expr: rate(proxy_fetch_total[10m]) == 0
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "采集器停止工作"
          description: "过去 10 分钟没有新代理被采集"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

# docker-compose.yml
version: "3.8"
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  fetcher:
    build: ./fetcher
    depends_on:
      - redis
    environment:
      - REDIS_HOST=redis
      - FETCH_INTERVAL=300
    restart: always

  checker:
    build: ./checker
    depends_on:
      - redis
    environment:
      - REDIS_HOST=redis
      - CHECK_INTERVAL=60
      - CONCURRENT_LIMIT=200
    restart: always
    deploy:
      replicas: 2  # 可水平扩展

  api:
    build: ./api
    depends_on:
      - redis
    ports:
      - "5000:5000"
    environment:
      - REDIS_HOST=redis
    restart: always

  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitor/prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    depends_on:
      - prometheus

volumes:
  redis_data:

1
2
3
4
5
6
7
8

# Tunnel Proxy Example
import base64

def process_request(self, request, spider):
    request.meta['proxy'] = "http://proxy.vendor.com:8000"
    # 某些厂商要求在头部通过 Proxy-Authorization 认证
    auth = base64.b64encode(b"user:pass").decode()
    request.headers['Proxy-Authorization'] = f"Basic {auth}"