Python设置代理IP的常用方法与实践指南

import requests
 
# 定义代理配置
proxies = {
    'http': 'http://127.0.0.1:8080',
    'https': 'https://127.0.0.1:8080'
}
 
# 使用代理发送请求
try:
    response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=10)
    print(f"通过代理访问，返回IP: {response.json()}")
except requests.exceptions.RequestException as e:
    print(f"请求失败: {e}")

import requests
 
# 需要用户名密码的代理
proxies_with_auth = {
    'http': 'http://username:password@proxy.example.com:8080',
    'https': 'https://username:password@proxy.example.com:8080'
}
 
response = requests.get('https://httpbin.org/ip', proxies=proxies_with_auth)
print(response.text)

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
 
# 创建会话并配置重试策略
session = requests.Session()
 
# 配置重试策略
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
 
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
 
# 代理配置
proxies = {
    'http': 'http://127.0.0.1:8080',
    'https': 'https://127.0.0.1:8080'
}
 
# 发送请求
response = session.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

import urllib.request
import urllib.error
 
# 创建代理处理器
proxy_handler = urllib.request.ProxyHandler({
    'http': 'http://127.0.0.1:8080',
    'https': 'https://127.0.0.1:8080'
})
 
# 创建opener
opener = urllib.request.build_opener(proxy_handler)
 
# 安装opener
urllib.request.install_opener(opener)
 
try:
    # 使用代理发送请求
    response = urllib.request.urlopen('https://httpbin.org/ip')
    print(response.read().decode('utf-8'))
except urllib.error.URLError as e:
    print(f"请求失败: {e}")

import urllib.request
import base64
 
# 创建密码管理器
password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
 
# 添加用户名和密码
proxy_url = 'proxy.example.com:8080'
username = 'your_username'
password = 'your_password'
password_mgr.add_password(None, proxy_url, username, password)
 
# 创建代理认证处理器
proxy_auth_handler = urllib.request.ProxyBasicAuthHandler(password_mgr)
 
# 创建并安装opener
opener = urllib.request.build_opener(proxy_auth_handler)
urllib.request.install_opener(opener)
 
# 发送请求
response = urllib.request.urlopen('https://httpbin.org/ip')
print(response.read().decode('utf-8'))

import aiohttp
import asyncio
 
async def fetch_with_proxy():
    # 代理配置
    proxy = 'http://127.0.0.1:8080'
    
    # 创建TCP连接器
    connector = aiohttp.TCPConnector(ssl=False)
    
    # 创建会话
    async with aiohttp.ClientSession(connector=connector) as session:
        try:
            # 使用代理发送异步请求
            async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
                data = await response.json()
                print(f"通过代理访问，返回IP: {data}")
        except aiohttp.ClientError as e:
            print(f"请求失败: {e}")
 
# 运行异步函数
asyncio.run(fetch_with_proxy())

import aiohttp
import asyncio
from aiohttp import BasicAuth
 
async def fetch_with_auth_proxy():
    # 代理认证信息
    proxy_auth = BasicAuth('username', 'password')
    proxy = 'http://proxy.example.com:8080'
    
    # 创建会话
    async with aiohttp.ClientSession() as session:
        try:
            # 使用带认证的代理发送请求
            async with session.get(
                'https://httpbin.org/ip',
                proxy=proxy,
                proxy_auth=proxy_auth
            ) as response:
                data = await response.json()
                print(f"认证代理访问成功: {data}")
        except aiohttp.ClientError as e:
            print(f"请求失败: {e}")
 
asyncio.run(fetch_with_auth_proxy())

import aiohttp
import asyncio
import random
from typing import List
 
class AsyncProxyPool:
    def __init__(self, proxies: List[str]):
        self.proxies = proxies
        self.failed_proxies = set()
    
    def get_random_proxy(self) -> str:
        available_proxies = [p for p in self.proxies if p not in self.failed_proxies]
        if not available_proxies:
            raise Exception("无可用的代理")
        return random.choice(available_proxies)
    
    def mark_proxy_failed(self, proxy: str):
        self.failed_proxies.add(proxy)
        print(f"代理 {proxy} 标记为失败")
    
    async def fetch_with_retry(self, url: str, max_retries: int = 3):
        for attempt in range(max_retries):
            proxy = self.get_random_proxy()
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(url, proxy=proxy, timeout=10) as response:
                        if response.status == 200:
                            return await response.text()
                        else:
                            print(f"代理 {proxy} 返回状态码: {response.status}")
            except Exception as e:
                print(f"代理 {proxy} 请求失败: {e}")
                self.mark_proxy_failed(proxy)
        
        raise Exception("所有代理都失败")
 
# 使用示例
async def main():
    proxies = [
        'http://127.0.0.1:8080',
        'http://127.0.0.1:8081',
        'http://127.0.0.1:8082'
    ]
    
    proxy_pool = AsyncProxyPool(proxies)
    
    try:
        result = await proxy_pool.fetch_with_retry('https://httpbin.org/ip')
        print("请求成功:", result[:100])
    except Exception as e:
        print("最终失败:", e)
 
asyncio.run(main())

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import time
 
def setup_chrome_proxy():
    # Chrome选项配置
    chrome_options = Options()
    
    # 代理设置
    proxy = '127.0.0.1:8080'
    chrome_options.add_argument(f'--proxy-server={proxy}')
    
    # 其他常用选项
    chrome_options.add_argument('--disable-gpu')
    chrome_options.add_argument('--no-sandbox')
    chrome_options.add_argument('--disable-dev-shm-usage')
    
    # 创建驱动
    driver = webdriver.Chrome(options=chrome_options)
    
    try:
        # 访问测试页面
        driver.get('https://httpbin.org/ip')
        time.sleep(3)
        
        # 获取页面内容
        page_source = driver.page_source
        print(f"页面内容: {page_source[:200]}")
        
    finally:
        driver.quit()
 
setup_chrome_proxy()

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import zipfile
import os
 
def create_proxy_auth_extension(proxy_host, proxy_port, username, password):
    """创建带认证的代理扩展"""
    
    manifest_json = """
    {
        "version": "1.0.0",
        "manifest_version": 2,
        "name": "Chrome Proxy",
        "permissions": [
            "proxy",
            "tabs",
            "unlimitedStorage",
            "storage",
            "<all_urls>",
            "webRequest",
            "webRequestBlocking"
        ],
        "background": {
            "scripts": ["background.js"],
            "persistent": true
        },
        "minimum_chrome_version":"22.0.0"
    }
    """
    
    background_js = f"""
    var config = {{
            mode: "fixed_servers",
            rules: {{
              singleProxy: {{
                scheme: "http",
                host: "{proxy_host}",
                port: parseInt({proxy_port})
              }},
              bypassList: ["localhost"]
            }}
          }};
 
    chrome.proxy.settings.set({{value: config, scope: "regular"}}, function() {{}});
 
    function callbackFn(details) {{
        return {{
            authCredentials: {{
                username: "{username}",
                password: "{password}"
            }}
        }};
    }}
 
    chrome.webRequest.onAuthRequired.addListener(
                callbackFn,
                {{urls: ["<all_urls>"]}},
                ['blocking']
    );
    """
    
    # 创建扩展文件
    pluginfile = 'proxy_auth_plugin.zip'
    
    with zipfile.ZipFile(pluginfile, 'w') as zp:
        zp.writestr("manifest.json", manifest_json)
        zp.writestr("background.js", background_js)
    
    return pluginfile
 
def setup_auth_proxy():
    # 代理信息
    proxy_host = 'proxy.example.com'
    proxy_port = 8080
    username = 'your_username'
    password = 'your_password'
    
    # 创建认证扩展
    pluginfile = create_proxy_auth_extension(proxy_host, proxy_port, username, password)
    
    # Chrome选项
    chrome_options = Options()
    chrome_options.add_extension(pluginfile)
    
    # 创建驱动
    driver = webdriver.Chrome(options=chrome_options)
    
    try:
        driver.get('https://httpbin.org/ip')
        time.sleep(3)
        print(f"页面标题: {driver.title}")
        
    finally:
        driver.quit()
        # 清理扩展文件
        if os.path.exists(pluginfile):
            os.remove(pluginfile)
 
setup_auth_proxy()

import requests
import asyncio
import aiohttp
from typing import List, Dict
import time
 
class ProxyValidator:
    def __init__(self, test_url: str = 'https://httpbin.org/ip'):
        self.test_url = test_url
        self.timeout = 10
    
    def validate_proxy(self, proxy: Dict[str, str]) -> Dict:
        """验证单个代理的有效性"""
        result = {
            'proxy': proxy,
            'is_valid': False,
            'response_time': 0,
            'error': None
        }
        
        try:
            start_time = time.time()
            response = requests.get(
                self.test_url,
                proxies=proxy,
                timeout=self.timeout
            )
            result['response_time'] = time.time() - start_time
            
            if response.status_code == 200:
                result['is_valid'] = True
                result['response_data'] = response.json()
            else:
                result['error'] = f"HTTP状态码: {response.status_code}"
                
        except requests.exceptions.ConnectTimeout:
            result['error'] = "连接超时"
        except requests.exceptions.ProxyError:
            result['error'] = "代理错误"
        except requests.exceptions.SSLError:
            result['error'] = "SSL证书错误"
        except Exception as e:
            result['error'] = f"未知错误: {str(e)}"
        
        return result
    
    async def validate_proxy_async(self, proxy: Dict[str, str]) -> Dict:
        """异步验证代理"""
        result = {
            'proxy': proxy,
            'is_valid': False,
            'response_time': 0,
            'error': None
        }
        
        try:
            start_time = time.time()
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    self.test_url,
                    proxy=list(proxy.values())[0],
                    timeout=aiohttp.ClientTimeout(total=self.timeout)
                ) as response:
                    result['response_time'] = time.time() - start_time
                    
                    if response.status == 200:
                        result['is_valid'] = True
                        result['response_data'] = await response.json()
                    else:
                        result['error'] = f"HTTP状态码: {response.status}"
                        
        except asyncio.TimeoutError:
            result['error'] = "连接超时"
        except Exception as e:
            result['error'] = f"验证失败: {str(e)}"
        
        return result
    
    def validate_proxy_list(self, proxies: List[Dict[str, str]]) -> List[Dict]:
        """批量验证代理列表"""
        results = []
        for proxy in proxies:
            result = self.validate_proxy(proxy)
            results.append(result)
            print(f"代理 {proxy} 验证结果: {'有效' if result['is_valid'] else '无效'}")
        
        return results
    
    async def validate_proxy_list_async(self, proxies: List[Dict[str, str]]) -> List[Dict]:
        """异步批量验证代理"""
        tasks = [self.validate_proxy_async(proxy) for proxy in proxies]
        results = await asyncio.gather(*tasks)
        
        for result in results:
            proxy = result['proxy']
            print(f"代理 {proxy} 验证结果: {'有效' if result['is_valid'] else '无效'}")
        
        return results
 
# 使用示例
validator = ProxyValidator()
 
# 测试代理列表
test_proxies = [
    {'http': 'http://127.0.0.1:8080', 'https': 'https://127.0.0.1:8080'},
    {'http': 'http://invalid.proxy:8080', 'https': 'https://invalid.proxy:8080'}
]
 
# 同步验证
print("=== 同步验证 ===")
sync_results = validator.validate_proxy_list(test_proxies)
 
# 异步验证
print("\n=== 异步验证 ===")
asyncio.run(validator.validate_proxy_list_async(test_proxies))

import requests
import time
import random
from typing import Optional, Callable
from functools import wraps
 
class SmartProxyManager:
    def __init__(self, proxies: list, max_retries: int = 3, retry_delay: float = 1.0):
        self.proxies = proxies
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.failed_proxies = set()
        self.proxy_stats = {}
    
    def get_working_proxy(self) -> Optional[Dict[str, str]]:
        """获取可用的代理"""
        available_proxies = [p for p in self.proxies if str(p) not in self.failed_proxies]
        
        if not available_proxies:
            return None
        
        # 优先选择成功率高的代理
        sorted_proxies = sorted(
            available_proxies,
            key=lambda x: self.proxy_stats.get(str(x), {}).get('success_rate', 0),
            reverse=True
        )
        
        return sorted_proxies[0]
    
    def mark_proxy_failed(self, proxy: Dict[str, str]):
        """标记代理为失败"""
        proxy_str = str(proxy)
        self.failed_proxies.add(proxy_str)
        
        # 更新统计信息
        if proxy_str not in self.proxy_stats:
            self.proxy_stats[proxy_str] = {'success_count': 0, 'fail_count': 0}
        
        self.proxy_stats[proxy_str]['fail_count'] += 1
    
    def mark_proxy_success(self, proxy: Dict[str, str]):
        """标记代理为成功"""
        proxy_str = str(proxy)
        
        if proxy_str not in self.proxy_stats:
            self.proxy_stats[proxy_str] = {'success_count': 0, 'fail_count': 0}
        
        self.proxy_stats[proxy_str]['success_count'] += 1
    
    def get_proxy_success_rate(self, proxy: Dict[str, str]) -> float:
        """获取代理成功率"""
        proxy_str = str(proxy)
        stats = self.proxy_stats.get(proxy_str, {'success_count': 0, 'fail_count': 0})
        
        total = stats['success_count'] + stats['fail_count']
        if total == 0:
            return 0.0
        
        return stats['success_count'] / total
    
    def smart_request(self, url: str, **kwargs) -> Optional[requests.Response]:
        """智能请求，自动处理代理失败和重试"""
        
        for attempt in range(self.max_retries):
            proxy = self.get_working_proxy()
            
            if not proxy:
                print("没有可用的代理")
                return None
            
            try:
                print(f"尝试使用代理 {proxy} (第{attempt + 1}次)")
                
                # 添加代理到kwargs
                kwargs['proxies'] = proxy
                kwargs['timeout'] = kwargs.get('timeout', 10)
                
                response = requests.get(url, **kwargs)
                
                if response.status_code == 200:
                    self.mark_proxy_success(proxy)
                    print(f"请求成功！代理成功率: {self.get_proxy_success_rate(proxy):.2%}")
                    return response
                else:
                    print(f"HTTP状态码异常: {response.status_code}")
                    self.mark_proxy_failed(proxy)
                    
            except requests.exceptions.RequestException as e:
                print(f"请求异常: {e}")
                self.mark_proxy_failed(proxy)
            
            # 重试延迟
            if attempt < self.max_retries - 1:
                delay = self.retry_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"等待 {delay:.1f} 秒后重试...")
                time.sleep(delay)
        
        return None
 
# 使用示例
proxies = [
    {'http': 'http://127.0.0.1:8080', 'https': 'https://127.0.0.1:8080'},
    {'http': 'http://proxy1.example.com:8080', 'https': 'https://proxy1.example.com:8080'},
    {'http': 'http://proxy2.example.com:8080', 'https': 'https://proxy2.example.com:8080'}
]
 
manager = SmartProxyManager(proxies)
 
# 智能请求
response = manager.smart_request('https://httpbin.org/ip')
if response:
    print(f"最终成功！响应: {response.json()}")
else:
    print("所有代理都失败")

import asyncio
import aiohttp
import time
import random
import json
import sqlite3
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import logging
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
class ProxyPool:
    """高性能代理池管理器"""
    
    def __init__(self, db_path: str = 'proxy_pool.db', max_concurrent_tests: int = 10):
        self.db_path = db_path
        self.max_concurrent_tests = max_concurrent_tests
        self.test_url = 'https://httpbin.org/ip'
        self.timeout = 15
        self._init_database()
    
    def _init_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS proxies (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                proxy TEXT UNIQUE NOT NULL,
                protocol TEXT NOT NULL,
                ip TEXT NOT NULL,
                port INTEGER NOT NULL,
                is_valid BOOLEAN DEFAULT 1,
                response_time REAL,
                success_count INTEGER DEFAULT 0,
                fail_count INTEGER DEFAULT 0,
                last_tested TIMESTAMP,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_proxy ON proxies(proxy);
            CREATE INDEX IF NOT EXISTS idx_valid ON proxies(is_valid);
            CREATE INDEX IF NOT EXISTS idx_last_tested ON proxies(last_tested);
        ''')
        
        conn.commit()
        conn.close()
    
    def add_proxy(self, proxy: str, protocol: str = 'http') -> bool:
        """添加代理到数据库"""
        try:
            # 解析代理信息
            if '@' in proxy:
                # 格式: username:password@ip:port
                auth_part, addr_part = proxy.split('@')
                ip, port = addr_part.split(':')
            else:
                # 格式: ip:port
                ip, port = proxy.split(':')
            
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                INSERT OR REPLACE INTO proxies 
                (proxy, protocol, ip, port, updated_at)
                VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
            ''', (proxy, protocol, ip, int(port)))
            
            conn.commit()
            conn.close()
            
            logger.info(f"代理 {proxy} 已添加到数据库")
            return True
            
        except Exception as e:
            logger.error(f"添加代理失败: {e}")
            return False
    
    def add_proxies_batch(self, proxies: List[Dict[str, str]]):
        """批量添加代理"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        for proxy_info in proxies:
            proxy = proxy_info.get('proxy')
            protocol = proxy_info.get('protocol', 'http')
            
            if proxy:
                try:
                    if '@' in proxy:
                        auth_part, addr_part = proxy.split('@')
                        ip, port = addr_part.split(':')
                    else:
                        ip, port = proxy.split(':')
                    
                    cursor.execute('''
                        INSERT OR REPLACE INTO proxies 
                        (proxy, protocol, ip, port, updated_at)
                        VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
                    ''', (proxy, protocol, ip, int(port)))
                    
                except Exception as e:
                    logger.error(f"添加代理 {proxy} 失败: {e}")
        
        conn.commit()
        conn.close()
        logger.info(f"批量添加 {len(proxies)} 个代理完成")
    
    async def test_proxy_async(self, session: aiohttp.ClientSession, proxy_info: Dict) -> Dict:
        """异步测试单个代理"""
        proxy = proxy_info['proxy']
        protocol = proxy_info['protocol']
        
        result = {
            'proxy': proxy,
            'is_valid': False,
            'response_time': 0,
            'error': None
        }
        
        try:
            start_time = time.time()
            proxy_url = f"{protocol}://{proxy}"
            
            async with session.get(
                self.test_url,
                proxy=proxy_url,
                timeout=aiohttp.ClientTimeout(total=self.timeout)
            ) as response:
                result['response_time'] = time.time() - start_time
                
                if response.status == 200:
                    result['is_valid'] = True
                    data = await response.json()
                    result['response_data'] = data
                else:
                    result['error'] = f"HTTP状态码: {response.status}"
                    
        except asyncio.TimeoutError:
            result['error'] = "连接超时"
        except aiohttp.ClientError as e:
            result['error'] = f"客户端错误: {str(e)}"
        except Exception as e:
            result['error'] = f"未知错误: {str(e)}"
        
        return result
    
    async def test_all_proxies_async(self):
        """异步测试所有代理"""
        # 获取所有需要测试的代理
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT proxy, protocol FROM proxies 
            WHERE is_valid = 1 OR 
            (last_tested IS NULL OR last_tested < datetime('now', '-1 hour'))
        ''')
        
        proxies_to_test = cursor.fetchall()
        conn.close()
        
        if not proxies_to_test:
            logger.info("没有需要测试的代理")
            return
        
        logger.info(f"开始测试 {len(proxies_to_test)} 个代理")
        
        # 创建会话并限制并发数
        connector = aiohttp.TCPConnector(limit=self.max_concurrent_tests)
        
        async with aiohttp.ClientSession(connector=connector) as session:
            # 分批处理，避免一次性创建过多任务
            batch_size = self.max_concurrent_tests
            
            for i in range(0, len(proxies_to_test), batch_size):
                batch = proxies_to_test[i:i + batch_size]
                
                # 创建测试任务
                tasks = []
                for proxy, protocol in batch:
                    proxy_info = {'proxy': proxy, 'protocol': protocol}
                    tasks.append(self.test_proxy_async(session, proxy_info))
                
                # 执行测试
                results = await asyncio.gather(*tasks, return_exceptions=True)
                
                # 更新数据库
                await self._update_proxy_results(results)
                
                logger.info(f"完成第 {i//batch_size + 1} 批测试")
                
                # 短暂延迟，避免对测试服务器造成压力
                await asyncio.sleep(1)
        
        logger.info("代理测试完成")
    
    async def _update_proxy_results(self, results: List):
        """更新代理测试结果到数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        for result in results:
            if isinstance(result, Exception):
                logger.error(f"测试任务异常: {result}")
                continue
            
            proxy = result['proxy']
            is_valid = result['is_valid']
            response_time = result.get('response_time', 0)
            
            if is_valid:
                cursor.execute('''
                    UPDATE proxies 
                    SET is_valid = 1, response_time = ?, success_count = success_count + 1,
                        last_tested = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
                    WHERE proxy = ?
                ''', (response_time, proxy))
            else:
                cursor.execute('''
                    UPDATE proxies 
                    SET is_valid = 0, fail_count = fail_count + 1,
                        last_tested = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
                    WHERE proxy = ?
                ''', (proxy,))
            
            logger.info(f"代理 {proxy} 测试结果: {'有效' if is_valid else '无效'}")
        
        conn.commit()
        conn.close()
    
    def get_valid_proxies(self, limit: int = 10) -> List[Dict]:
        """获取有效的代理列表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT proxy, protocol, response_time, success_count, fail_count
            FROM proxies 
            WHERE is_valid = 1
            ORDER BY response_time ASC, success_count DESC
            LIMIT ?
        ''', (limit,))
        
        proxies = []
        for row in cursor.fetchall():
            proxy, protocol, response_time, success_count, fail_count = row
            
            # 计算成功率
            total = success_count + fail_count
            success_rate = success_count / total if total > 0 else 0
            
            proxies.append({
                'proxy': proxy,
                'protocol': protocol,
                'response_time': response_time or 999,
                'success_rate': success_rate
            })
        
        conn.close()
        return proxies
    
    def get_random_proxy(self, weighted: bool = True) -> Optional[Dict[str, str]]:
        """获取随机代理"""
        valid_proxies = self.get_valid_proxies(limit=50)
        
        if not valid_proxies:
            return None
        
        if weighted:
            # 基于成功率加权的随机选择
            weights = [p['success_rate'] for p in valid_proxies]
            if sum(weights) == 0:
                # 如果所有权重都是0，使用均等权重
                weights = [1] * len(valid_proxies)
            
            selected = random.choices(valid_proxies, weights=weights, k=1)[0]
        else:
            selected = random.choice(valid_proxies)
        
        return {
            'http': f"{selected['protocol']}://{selected['proxy']}",
            'https': f"{selected['protocol']}://{selected['proxy']}"
        }
    
    def get_proxy_stats(self) -> Dict:
        """获取代理池统计信息"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 总体统计
        cursor.execute('SELECT COUNT(*) FROM proxies')
        total = cursor.fetchone()[0]
        
        cursor.execute('SELECT COUNT(*) FROM proxies WHERE is_valid = 1')
        valid = cursor.fetchone()[0]
        
        cursor.execute('SELECT COUNT(*) FROM proxies WHERE is_valid = 0')
        invalid = cursor.fetchone()[0]
        
        # 平均响应时间
        cursor.execute('SELECT AVG(response_time) FROM proxies WHERE is_valid = 1 AND response_time IS NOT NULL')
        avg_response_time = cursor.fetchone()[0] or 0
        
        conn.close()
        
        return {
            'total': total,
            'valid': valid,
            'invalid': invalid,
            'valid_rate': valid / total if total > 0 else 0,
            'avg_response_time': round(avg_response_time, 2)
        }
    
    def cleanup_invalid_proxies(self, days: int = 7):
        """清理长期无效的代理"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            DELETE FROM proxies 
            WHERE is_valid = 0 AND 
            (last_tested < datetime('now', '-' || ? || ' days') OR last_tested IS NULL)
        ''', (days,))
        
        deleted_count = cursor.rowcount
        conn.commit()
        conn.close()
        
        logger.info(f"清理了 {deleted_count} 个长期无效的代理")
        return deleted_count
 
# 使用示例
async def main():
    # 创建代理池
    proxy_pool = ProxyPool()
    
    # 添加一些测试代理
    test_proxies = [
        {'proxy': '127.0.0.1:8080', 'protocol': 'http'},
        {'proxy': '127.0.0.1:8081', 'protocol': 'http'},
        {'proxy': '127.0.0.1:8082', 'protocol': 'http'}
    ]
    
    proxy_pool.add_proxies_batch(test_proxies)
    
    # 测试所有代理
    await proxy_pool.test_all_proxies_async()
    
    # 获取统计信息
    stats = proxy_pool.get_proxy_stats()
    print(f"代理池统计: {stats}")
    
    # 获取有效代理
    valid_proxies = proxy_pool.get_valid_proxies(limit=5)
    print(f"有效代理: {valid_proxies}")
    
    # 获取随机代理
    random_proxy = proxy_pool.get_random_proxy()
    print(f"随机代理: {random_proxy}")
 
# 运行示例
# asyncio.run(main())

# 推荐的代理配置结构
PROXY_CONFIG = {
    'rotation_enabled': True,      # 启用代理轮换
    'retry_on_failure': True,      # 失败时重试
    'max_retries': 3,             # 最大重试次数
    'request_timeout': 15,          # 请求超时时间
    'retry_delay': 1,               # 重试延迟（秒）
    'success_rate_threshold': 0.8,  # 成功率阈值
    'response_time_threshold': 5    # 响应时间阈值（秒）
}

import logging
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class ProxyMetrics:
    """代理性能指标"""
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    avg_response_time: float = 0.0
    blocked_count: int = 0
 
class ProxyMonitor:
    """代理性能监控器"""
    
    def __init__(self):
        self.metrics = ProxyMetrics()
        self.logger = logging.getLogger(__name__)
    
    def record_request(self, success: bool, response_time: float, blocked: bool = False):
        """记录请求结果"""
        self.metrics.total_requests += 1
        
        if success:
            self.metrics.successful_requests += 1
            # 更新平均响应时间
            total_time = self.metrics.avg_response_time * (self.metrics.total_requests - 1) + response_time
            self.metrics.avg_response_time = total_time / self.metrics.total_requests
        else:
            self.metrics.failed_requests += 1
            if blocked:
                self.metrics.blocked_count += 1
    
    def get_success_rate(self) -> float:
        """获取成功率"""
        if self.metrics.total_requests == 0:
            return 0.0
        return self.metrics.successful_requests / self.metrics.total_requests
    
    def get_report(self) -> dict:
        """获取监控报告"""
        return {
            'total_requests': self.metrics.total_requests,
            'successful_requests': self.metrics.successful_requests,
            'failed_requests': self.metrics.failed_requests,
            'success_rate': f"{self.get_success_rate():.2%}",
            'avg_response_time': f"{self.metrics.avg_response_time:.2f}s",
            'blocked_count': self.metrics.blocked_count,
            'block_rate': f"{self.metrics.blocked_count / max(self.metrics.total_requests, 1):.2%}"
        }
    
    def should_alert(self, threshold: float = 0.7) -> bool:
        """判断是否需要告警"""
        return self.get_success_rate() < threshold

# 使用连接池复用连接
session = requests.Session()
adapter = requests.adapters.HTTPAdapter(
    pool_connections=100,    # 连接池大小
    pool_maxsize=100,       # 最大连接数
    max_retries=3           # 重试次数
)
session.mount('http://', adapter)
session.mount('https://', adapter)

# 使用信号量控制并发数
import asyncio
from asyncio import Semaphore
 
class ConcurrentProxyManager:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = Semaphore(max_concurrent)
    
    async def fetch_with_proxy(self, url: str, proxy: str):
        async with self.semaphore:
            # 实际的请求逻辑
            async with aiohttp.ClientSession() as session:
                async with session.get(url, proxy=proxy) as response:
                    return await response.text()