#现代网络爬虫代理技术完全指南

import requests

def quick_check_anonymity(proxy):
    """
    使用 httpbin 检测代理匿名程度
    :param proxy: 代理地址，格式 'http://ip:port' 或 'socks5://ip:port'
    """
    test_url = "http://httpbin.org/headers"
    try:
        resp = requests.get(
            test_url,
            proxies={"http": proxy, "https": proxy},
            timeout=10
        )
        headers = resp.json()["headers"]
        
        if "X-Forwarded-For" in headers and headers["X-Forwarded-For"]:
            return "⚠️ 普通匿名：暴露了部分IP信息"
        elif "Via" in headers:
            return "❌ 透明代理：完全暴露了真实IP"
        else:
            return "✅ 高匿名：目标服务器完全看不到任何代理痕迹"
    except Exception as e:
        return f"💀 无效代理：{str(e)}"

import random
from concurrent.futures import ThreadPoolExecutor
import redis
import requests
import time

class SimpleProxyPool:
    def __init__(self, redis_host="localhost", redis_port=6379):
        self.redis = redis.Redis(host=redis_host, port=redis_port,
                                 db=0, decode_responses=True)
        self.test_url = "http://www.baidu.com"   # 验证用目标，换海外站点亦可
        self.valid_key = "proxies:valid"         # zset，score代表成功次数
        self.pending_key = "proxies:pending"     # set，待验证代理

    def add_proxies(self, proxy_list):
        """批量添加待验证代理"""
        self.redis.sadd(self.pending_key, *proxy_list)

    def _validate_single(self, proxy):
        """单个代理验证，成功则增加score，失败则移出有效集"""
        try:
            requests.get(self.test_url, proxies={"http": proxy}, timeout=3)
            # 每验证成功一次，score +1
            self.redis.zincrby(self.valid_key, 1, proxy)
        except Exception:
            self.redis.zrem(self.valid_key, proxy)
            self.redis.srem(self.pending_key, proxy)

    def run_validation(self, threads=10):
        """多线程验证所有待验证 + 已有代理"""
        candidates = (list(self.redis.smembers(self.pending_key)) +
                      list(self.redis.zrange(self.valid_key, 0, -1)))
        # 去重
        candidates = list(set(candidates))
        with ThreadPoolExecutor(max_workers=threads) as executor:
            executor.map(self._validate_single, candidates)

    def get_best_proxy(self):
        """获取当前成功率最高的代理（用于对稳定性要求极高的任务）"""
        top = self.redis.zrevrange(self.valid_key, 0, 0)
        return top[0] if top else None

    def get_random_proxy(self, min_score=1):
        """随机获取一个得分 >= min_score 的代理，实现简单轮换"""
        candidates = self.redis.zrangebyscore(self.valid_key, min_score, float("inf"))
        return random.choice(candidates) if candidates else None

# ———— 使用示例 ————
if __name__ == "__main__":
    pool = SimpleProxyPool()
    # 添加你在供应商处购买的代理，格式为 "http://user:pass@ip:port"
    pool.add_proxies(["http://127.0.0.1:7890"])  # 仅示例，请替换为真实代理

    # 每10分钟自动验证一轮
    while True:
        pool.run_validation()
        valid_count = pool.redis.zcard(pool.valid_key)
        print(f"当前有效代理数：{valid_count}")
        time.sleep(60 * 10)