Upload proxy_pool.py

proxy_pool.py

@@ -2,7 +2,7 @@ import requests
 import time
 import logging
 import random
-from threading import Lock
+from threading import Lock, RLock
 from collections import defaultdict
 from datetime import datetime, timedelta
 
@@ -11,61 +11,82 @@ logger = logging.getLogger(__name__)
 class ProxyPool:
     def __init__(self):
         self.proxies = []
-        self.proxy_usage = defaultdict(list)  # 记录每个IP的使用时间
-        self.lock = Lock()
+        self.proxy_usage = defaultdict(list)
+        self.lock = RLock()
         self.last_fetch_time = datetime.min
-        self.fetch_interval = 300  # 5分钟更新一次代理池
+        self.fetch_interval = 300
+        self.last_cleanup_time = time.time()
+        self.cleanup_interval = 60
         
+    def _cleanup_usage_records(self):
+        """清理过期的使用记录"""
+        current_time = time.time()
+        if current_time - self.last_cleanup_time < self.cleanup_interval:
+            return
+            
+        with self.lock:
+            for ip in list(self.proxy_usage.keys()):
+                self.proxy_usage[ip] = [t for t in self.proxy_usage[ip] 
+                                      if current_time - t < 1.0]
+                if not self.proxy_usage[ip]:
+                    del self.proxy_usage[ip]
+            self.last_cleanup_time = current_time
+
     def _parse_proxy_list(self, text):
         """解析代理列表文本"""
         proxies = []
-        for line in text.strip().split(','):
-            line = line.strip()
-            if not line:
-                continue
+        try:
+            lines = [line.strip() for line in text.strip().split(',') if line.strip()]
             
-            try:
-                if line.startswith('http://'):
-                    proxy_type = 'http'
-                    proxy = line[7:]
-                elif line.startswith('socks4://'):
-                    proxy_type = 'socks4'
-                    proxy = line[9:]
-                elif line.startswith('socks5://'):
-                    proxy_type = 'socks5'
-                    proxy = line[9:]
-                else:
+            for line in lines:
+                try:
+                    if line.startswith('http://'):
+                        proxy_type = 'http'
+                        proxy = line[7:]
+                    elif line.startswith('socks4://'):
+                        proxy_type = 'socks4'
+                        proxy = line[9:]
+                    elif line.startswith('socks5://'):
+                        proxy_type = 'socks5'
+                        proxy = line[9:]
+                    else:
+                        continue
+                    
+                    if ':' in proxy:
+                        ip, port = proxy.split(':')
+                        proxies.append({
+                            'type': proxy_type,
+                            'ip': ip.strip(),
+                            'port': int(port.strip())
+                        })
+                except Exception as e:
+                    logger.error(f"解析代理失败: {line}, 错误: {e}")
                     continue
-                
-                if ':' in proxy:
-                    ip, port = proxy.split(':')
-                    proxies.append({
-                        'type': proxy_type,
-                        'ip': ip,
-                        'port': int(port)
-                    })
-            except Exception as e:
-                logger.error(f"解析代理失败: {line}, 错误: {e}")
-                continue
-                
+                    
+        except Exception as e:
+            logger.error(f"解析代理列表失败: {e}")
         return proxies
 
     def _fetch_proxies(self):
         """从API获取代理列表"""
         try:
-            # 获取混合代理池（HTTP/HTTPS各5个）
             urls = [
                 "https://269900.xyz/fetch_http?count=5",
-                "https://269900.xyz/fetch_https?count=5"
+                "https://269900.xyz/fetch_https?count=5",
+                "https://269900.xyz/fetch_socks5?count=5"
             ]
             
             new_proxies = []
             for url in urls:
-                response = requests.get(url, timeout=10)
-                if response.status_code == 200:
-                    proxies = self._parse_proxy_list(response.text)
-                    new_proxies.extend(proxies)
-                    logger.info(f"成功获取 {len(proxies)} 个代理从 {url}")
+                try:
+                    response = requests.get(url, timeout=5)
+                    if response.status_code == 200:
+                        proxies = self._parse_proxy_list(response.text)
+                        new_proxies.extend(proxies)
+                        logger.info(f"成功获取 {len(proxies)} 个代理从 {url}")
+                except requests.RequestException as e:
+                    logger.error(f"获取代理失败 ({url}): {e}")
+                    continue
             
             return new_proxies
         except Exception as e:
@@ -74,57 +95,81 @@ class ProxyPool:
 
     def update_proxy_pool(self):
         """更新代理池"""
+        current_time = datetime.now()
+        should_update = False
+        
         with self.lock:
-            current_time = datetime.now()
             if (current_time - self.last_fetch_time).total_seconds() >= self.fetch_interval:
-                new_proxies = self._fetch_proxies()
-                if new_proxies:
+                should_update = True
+                self.last_fetch_time = current_time
+        
+        if should_update:
+            new_proxies = self._fetch_proxies()
+            if new_proxies:
+                with self.lock:
                     self.proxies = new_proxies
-                    self.last_fetch_time = current_time
                     logger.info(f"代理池更新成功，当前代理数量: {len(self.proxies)}")
 
     def can_use_proxy(self, proxy_ip):
-        """检查代理IP是否可以使用（限制每秒最多5次请求）"""
+        """检查代理IP是否可以使用"""
         current_time = time.time()
+        
         with self.lock:
-            # 清理旧的使用记录
-            self.proxy_usage[proxy_ip] = [t for t in self.proxy_usage[proxy_ip] 
-                                        if current_time - t < 1.0]
+            usage_times = self.proxy_usage[proxy_ip]
+            usage_times = [t for t in usage_times if current_time - t < 1.0]
+            self.proxy_usage[proxy_ip] = usage_times
             
-            # 检查是否超过限制
-            if len(self.proxy_usage[proxy_ip]) >= 5:
+            if len(usage_times) >= 5:
                 return False
             
-            # 添加新的使用记录
             self.proxy_usage[proxy_ip].append(current_time)
             return True
 
     def get_proxy(self):
         """获取一个可用的代理"""
+        self._cleanup_usage_records()
         self.update_proxy_pool()
         
-        with self.lock:
-            if not self.proxies:
-                return None
-            
-            # 随机尝试最多5个代理
-            for _ in range(min(5, len(self.proxies))):
-                proxy = random.choice(self.proxies)
-                proxy_ip = proxy.get('ip')
+        for _ in range(3):
+            with self.lock:
+                if not self.proxies:
+                    return None
                 
-                if proxy_ip and self.can_use_proxy(proxy_ip):
+                available_proxies = []
+                for proxy in self.proxies:
+                    proxy_ip = proxy.get('ip')
+                    if proxy_ip and self.can_use_proxy(proxy_ip):
+                        available_proxies.append(proxy)
+                
+                if available_proxies:
+                    proxy = random.choice(available_proxies)
                     proxy_type = proxy.get('type', 'http').lower()
-                    proxy_str = f"{proxy_type}://{proxy_ip}:{proxy.get('port')}"
+                    proxy_str = f"{proxy_type}://{proxy['ip']}:{proxy['port']}"
                     return {
                         'http': proxy_str,
                         'https': proxy_str if proxy_type == 'http' else None
                     }
             
-            return None
+            time.sleep(0.1)
+        
+        return None
 
     def remove_proxy(self, proxy):
         """移除无效的代理"""
+        if not proxy:
+            return
+            
         with self.lock:
-            if proxy in self.proxies:
-                self.proxies.remove(proxy)
-                logger.info(f"移除无效代理: {proxy}")
+            try:
+                proxy_str = proxy.get('http') or proxy.get('https')
+                if not proxy_str:
+                    return
+                    
+                for p in self.proxies[:]:
+                    current_proxy_str = f"{p['type']}://{p['ip']}:{p['port']}"
+                    if current_proxy_str == proxy_str:
+                        self.proxies.remove(p)
+                        logger.info(f"移除无效代理: {proxy_str}")
+                        break
+            except Exception as e:
+                logger.error(f"移除代理时出错: {e}")