feat: Implement a proxy pool with concurrent testing and integrate proxy management commands into the Telegram bot.

.gitignore

@@ -38,3 +38,4 @@ autogptplus_drission_oai.py
 accounts.json
 team-reg-go
 CodexAuth
+.agent/rules/use.md

auto_gpt_team.py

@@ -53,6 +53,14 @@ except ImportError:
     StripePaymentAPI = None
     api_payment_with_retry = None
 
+# 导入代理池模块
+try:
+    import proxy_pool
+    PROXY_POOL_AVAILABLE = True
+except ImportError:
+    proxy_pool = None
+    PROXY_POOL_AVAILABLE = False
+
 # ================= 配置加载 =================
 try:
     import tomllib
@@ -2306,8 +2314,13 @@ def run_single_registration_api(progress_callback=None, step_callback=None, prox
     if not ibans:
         return {"success": False, "error": "没有可用的 IBAN，请先通过 /iban_add 导入"}
     
-    # 使用配置的代理或传入的代理
+    # 使用代理: 优先传入参数 > 代理池轮询 > 配置静态代理
-    use_proxy = proxy or API_PROXY or None
+    if proxy:
+        use_proxy = proxy
+    elif PROXY_POOL_AVAILABLE and proxy_pool.get_proxy_count() > 0:
+        use_proxy = proxy_pool.get_next_proxy()
+    else:
+        use_proxy = API_PROXY or None
     if use_proxy:
         log_status("代理", f"使用代理: {use_proxy}")
     
@@ -2428,13 +2441,14 @@ def run_single_registration_api(progress_callback=None, step_callback=None, prox
     return {"success": False, "error": f"重试 {max_user_retries} 次后仍失败: {last_error}"}
 
 
-def run_single_registration_auto(progress_callback=None, step_callback=None, mode: str = None) -> dict:
+def run_single_registration_auto(progress_callback=None, step_callback=None, mode: str = None, proxy: str = None) -> dict:
     """自动选择模式执行注册
     
     Args:
         progress_callback: 进度回调
         step_callback: 步骤回调
         mode: 强制指定模式 ("browser" / "api")，None 则使用配置
+        proxy: 指定代理地址，None 则由代理池或配置决定
         
     Returns:
         dict: 注册结果
@@ -2445,7 +2459,7 @@ def run_single_registration_auto(progress_callback=None, step_callback=None, mod
         if not API_MODE_AVAILABLE:
             log_status("警告", "协议模式不可用，回退到浏览器模式")
             return run_single_registration(progress_callback, step_callback)
-        return run_single_registration_api(progress_callback, step_callback)
+        return run_single_registration_api(progress_callback, step_callback, proxy=proxy)
     else:
         return run_single_registration(progress_callback, step_callback)
 

proxy_pool.py

@@ -0,0 +1,312 @@
+"""
+代理池管理模块
+- 从 proxy.txt 加载代理
+- 并发测试代理可用性
+- 线程安全的轮询分配
+"""
+
+import os
+import time
+import threading
+import concurrent.futures
+from pathlib import Path
+from urllib.parse import urlparse
+
+# 尝试导入 curl_cffi (更好的指纹伪装)
+try:
+    from curl_cffi import requests as curl_requests
+    CURL_AVAILABLE = True
+except ImportError:
+    curl_requests = None
+    CURL_AVAILABLE = False
+
+import requests
+
+BASE_DIR = Path(__file__).parent
+PROXY_FILE = BASE_DIR / "proxy.txt"
+
+# 测试目标 URL
+TEST_URL = "https://api.openai.com/v1/models"
+TEST_TIMEOUT = 10  # 秒
+
+
+def parse_proxy_url(proxy_url: str) -> dict | None:
+    """解析代理 URL，返回结构化信息
+    
+    支持格式:
+        http://host:port
+        http://username:password@host:port
+        socks5://host:port
+        socks5://username:password@host:port
+    
+    Returns:
+        dict: {"url": str, "scheme": str, "host": str, "port": int, "username": str, "password": str}
+        None: 格式无效
+    """
+    proxy_url = proxy_url.strip()
+    if not proxy_url:
+        return None
+    
+    # 确保有 scheme
+    if not proxy_url.startswith(("http://", "https://", "socks5://", "socks4://")):
+        proxy_url = "http://" + proxy_url
+    
+    try:
+        parsed = urlparse(proxy_url)
+        if not parsed.hostname or not parsed.port:
+            return None
+        
+        return {
+            "url": proxy_url,
+            "scheme": parsed.scheme,
+            "host": parsed.hostname,
+            "port": parsed.port,
+            "username": parsed.username or "",
+            "password": parsed.password or "",
+        }
+    except Exception:
+        return None
+
+
+def load_proxies() -> list[str]:
+    """从 proxy.txt 加载代理列表
+    
+    Returns:
+        list[str]: 代理 URL 列表
+    """
+    if not PROXY_FILE.exists():
+        return []
+    
+    proxies = []
+    try:
+        with open(PROXY_FILE, "r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                # 跳过空行和注释
+                if not line or line.startswith("#"):
+                    continue
+                # 验证格式
+                parsed = parse_proxy_url(line)
+                if parsed:
+                    proxies.append(parsed["url"])
+    except Exception:
+        pass
+    
+    return proxies
+
+
+def save_proxies(proxies: list[str]):
+    """保存代理列表到 proxy.txt (保留文件头部注释)
+    
+    Args:
+        proxies: 代理 URL 列表
+    """
+    header_lines = []
+    
+    # 读取原文件头部注释
+    if PROXY_FILE.exists():
+        try:
+            with open(PROXY_FILE, "r", encoding="utf-8") as f:
+                for line in f:
+                    if line.strip().startswith("#") or not line.strip():
+                        header_lines.append(line.rstrip())
+                    else:
+                        break
+        except Exception:
+            pass
+    
+    try:
+        with open(PROXY_FILE, "w", encoding="utf-8") as f:
+            # 写入头部注释
+            if header_lines:
+                f.write("\n".join(header_lines) + "\n")
+            # 写入代理
+            for proxy in proxies:
+                f.write(proxy + "\n")
+    except Exception as e:
+        print(f"[ProxyPool] 保存代理文件失败: {e}")
+
+
+def test_single_proxy(proxy_url: str, timeout: int = TEST_TIMEOUT) -> bool:
+    """测试单个代理是否可用
+    
+    Args:
+        proxy_url: 代理 URL
+        timeout: 超时秒数
+        
+    Returns:
+        bool: 代理是否可用
+    """
+    proxies_dict = {"http": proxy_url, "https": proxy_url}
+    
+    try:
+        if CURL_AVAILABLE:
+            # 使用 curl_cffi (更好的指纹)
+            resp = curl_requests.head(
+                TEST_URL,
+                proxies=proxies_dict,
+                timeout=timeout,
+                verify=False,
+                impersonate="edge",
+            )
+        else:
+            # 回退到 requests
+            resp = requests.head(
+                TEST_URL,
+                proxies=proxies_dict,
+                timeout=timeout,
+                verify=False,
+            )
+        # 任何响应都算成功 (包括 401/403，说明代理本身是通的)
+        return True
+    except Exception:
+        return False
+
+
+class ProxyPool:
+    """线程安全的代理池"""
+    
+    def __init__(self):
+        self._working_proxies: list[str] = []
+        self._index = 0
+        self._lock = threading.Lock()
+        self._last_test_time: float = 0
+        self._last_test_results: dict = {}  # {total, alive, removed}
+    
+    def reload(self) -> int:
+        """从文件重新加载代理
+        
+        Returns:
+            int: 加载的代理数量
+        """
+        with self._lock:
+            self._working_proxies = load_proxies()
+            self._index = 0
+            return len(self._working_proxies)
+    
+    def test_and_clean(self, concurrency: int = 20, timeout: int = TEST_TIMEOUT) -> dict:
+        """并发测试所有代理，移除不可用的
+        
+        Args:
+            concurrency: 并发数
+            timeout: 单个代理超时秒数
+            
+        Returns:
+            dict: {"total": int, "alive": int, "removed": int, "duration": float}
+        """
+        # 先从文件加载最新
+        all_proxies = load_proxies()
+        if not all_proxies:
+            self._last_test_results = {"total": 0, "alive": 0, "removed": 0, "duration": 0}
+            return self._last_test_results
+        
+        total = len(all_proxies)
+        start_time = time.time()
+        
+        # 并发测试
+        alive_proxies = []
+        dead_proxies = []
+        
+        with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
+            future_to_proxy = {
+                executor.submit(test_single_proxy, proxy, timeout): proxy
+                for proxy in all_proxies
+            }
+            
+            for future in concurrent.futures.as_completed(future_to_proxy):
+                proxy = future_to_proxy[future]
+                try:
+                    is_alive = future.result()
+                    if is_alive:
+                        alive_proxies.append(proxy)
+                    else:
+                        dead_proxies.append(proxy)
+                except Exception:
+                    dead_proxies.append(proxy)
+        
+        duration = time.time() - start_time
+        
+        # 更新工作代理池
+        with self._lock:
+            self._working_proxies = alive_proxies
+            self._index = 0
+        
+        # 保存存活的代理到文件 (移除死亡代理)
+        if dead_proxies:
+            save_proxies(alive_proxies)
+        
+        self._last_test_time = time.time()
+        self._last_test_results = {
+            "total": total,
+            "alive": len(alive_proxies),
+            "removed": len(dead_proxies),
+            "duration": round(duration, 1),
+        }
+        
+        return self._last_test_results
+    
+    def get_next_proxy(self) -> str | None:
+        """获取下一个代理 (轮询)
+        
+        Returns:
+            str: 代理 URL，池为空时返回 None
+        """
+        with self._lock:
+            if not self._working_proxies:
+                return None
+            proxy = self._working_proxies[self._index % len(self._working_proxies)]
+            self._index += 1
+            return proxy
+    
+    def get_proxy_count(self) -> int:
+        """获取当前可用代理数量"""
+        with self._lock:
+            return len(self._working_proxies)
+    
+    def get_status(self) -> dict:
+        """获取代理池状态
+        
+        Returns:
+            dict: {"count": int, "last_test_time": float, "last_test_results": dict}
+        """
+        with self._lock:
+            return {
+                "count": len(self._working_proxies),
+                "proxies": list(self._working_proxies),
+                "last_test_time": self._last_test_time,
+                "last_test_results": self._last_test_results,
+            }
+
+
+# ============ 全局单例 ============
+_pool = ProxyPool()
+
+
+def get_pool() -> ProxyPool:
+    """获取全局代理池实例"""
+    return _pool
+
+
+def reload_proxies() -> int:
+    """重新加载代理"""
+    return _pool.reload()
+
+
+def test_and_clean_proxies(concurrency: int = 20) -> dict:
+    """并发测试并清理代理"""
+    return _pool.test_and_clean(concurrency=concurrency)
+
+
+def get_next_proxy() -> str | None:
+    """获取下一个代理 (轮询)"""
+    return _pool.get_next_proxy()
+
+
+def get_proxy_count() -> int:
+    """获取可用代理数量"""
+    return _pool.get_proxy_count()
+
+
+def get_proxy_status() -> dict:
+    """获取代理池状态"""
+    return _pool.get_status()

telegram_bot.py

@@ -197,6 +197,10 @@ class ProvisionerBot:
             ("schedule", self.cmd_schedule),
             ("schedule_config", self.cmd_schedule_config),
             ("schedule_status", self.cmd_schedule_status),
+            # 代理池管理
+            ("proxy_status", self.cmd_proxy_status),
+            ("proxy_test", self.cmd_proxy_test),
+            ("proxy_reload", self.cmd_proxy_reload),
         ]
         for cmd, handler in handlers:
             self.app.add_handler(CommandHandler(cmd, handler))
@@ -362,6 +366,10 @@ class ProvisionerBot:
             BotCommand("schedule", "定时调度器 开关"),
             BotCommand("schedule_config", "调度器参数配置"),
             BotCommand("schedule_status", "调度器运行状态"),
+            # 代理池
+            BotCommand("proxy_status", "代理池状态"),
+            BotCommand("proxy_test", "测试并清理代理"),
+            BotCommand("proxy_reload", "重新加载代理"),
         ]
         try:
             await self.app.bot.set_my_commands(commands)
@@ -451,6 +459,11 @@ class ProvisionerBot:
 /schedule_config - 配置调度器参数
 /schedule_status - 查看调度器运行状态
 
+<b>🌐 代理池管理:</b>
+/proxy_status - 查看代理池状态
+/proxy_test - 测试并清理不可用代理
+/proxy_reload - 从 proxy.txt 重新加载代理
+
 <b>💡 示例:</b>
 <code>/list</code> - 查看所有待处理账号
 <code>/run 0</code> - 处理第一个 Team
@@ -3949,6 +3962,45 @@ class ProvisionerBot:
         import json
         import threading
         
+        # ===== 代理池预检测 =====
+        try:
+            import proxy_pool
+            pool_count = proxy_pool.get_proxy_count()
+            if pool_count == 0:
+                # 尝试加载
+                pool_count = proxy_pool.reload_proxies()
+            
+            if pool_count > 0:
+                await self.app.bot.send_message(
+                    chat_id,
+                    f"🌐 <b>代理池预检测</b>\n\n"
+                    f"正在测试 {pool_count} 个代理 (20 并发)...",
+                    parse_mode="HTML"
+                )
+                loop = asyncio.get_event_loop()
+                test_result = await loop.run_in_executor(
+                    self.executor,
+                    lambda: proxy_pool.test_and_clean_proxies(concurrency=20)
+                )
+                await self.app.bot.send_message(
+                    chat_id,
+                    f"✅ <b>代理池就绪</b>\n\n"
+                    f"总计: {test_result['total']} | "
+                    f"存活: {test_result['alive']} | "
+                    f"移除: {test_result['removed']}\n"
+                    f"耗时: {test_result['duration']}s",
+                    parse_mode="HTML"
+                )
+                if test_result['alive'] == 0:
+                    await self.app.bot.send_message(
+                        chat_id,
+                        "⚠️ 所有代理都不可用，将使用直连或静态代理",
+                    )
+        except ImportError:
+            pass
+        except Exception as e:
+            log.warning(f"代理池预检测异常: {e}")
+        
         results = []
         success_count = 0
         fail_count = 0
@@ -6332,6 +6384,126 @@ class ProvisionerBot:
         except Exception as e:
             await update.message.reply_text(f"❌ 添加 IBAN 失败: {e}")
 
+    # ==================== 代理池管理 ====================
+
+    @admin_only
+    async def cmd_proxy_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
+        """查看代理池状态"""
+        try:
+            import proxy_pool
+
+            status = proxy_pool.get_proxy_status()
+            count = status["count"]
+            last_test = status["last_test_results"]
+
+            lines = [f"<b>🌐 代理池状态</b>\n"]
+            lines.append(f"可用代理: {count} 个")
+
+            if last_test:
+                lines.append(f"\n<b>上次测试:</b>")
+                lines.append(f"  总计: {last_test.get('total', 0)}")
+                lines.append(f"  存活: {last_test.get('alive', 0)}")
+                lines.append(f"  移除: {last_test.get('removed', 0)}")
+                lines.append(f"  耗时: {last_test.get('duration', 0)}s")
+
+            if status["last_test_time"]:
+                from datetime import datetime
+                test_time = datetime.fromtimestamp(status["last_test_time"])
+                lines.append(f"  时间: {test_time.strftime('%H:%M:%S')}")
+
+            if count > 0:
+                # 显示前5个代理 (脱敏)
+                lines.append(f"\n<b>代理列表</b> (前 5 个):")
+                for i, proxy in enumerate(status["proxies"][:5]):
+                    # 脱敏: 隐藏密码部分
+                    display = proxy
+                    if "@" in proxy:
+                        parts = proxy.split("@")
+                        scheme_auth = parts[0]
+                        host_part = parts[1]
+                        # 取 scheme 部分
+                        if "://" in scheme_auth:
+                            scheme = scheme_auth.split("://")[0]
+                            display = f"{scheme}://***@{host_part}"
+                        else:
+                            display = f"***@{host_part}"
+                    lines.append(f"  {i+1}. <code>{display}</code>")
+                if count > 5:
+                    lines.append(f"  ... 还有 {count - 5} 个")
+            else:
+                lines.append(f"\n💡 将代理添加到 <code>proxy.txt</code> 然后使用 /proxy_reload 加载")
+
+            await update.message.reply_text("\n".join(lines), parse_mode="HTML")
+
+        except ImportError:
+            await update.message.reply_text("❌ proxy_pool 模块未找到")
+        except Exception as e:
+            await update.message.reply_text(f"❌ 获取代理池状态失败: {e}")
+
+    @admin_only
+    async def cmd_proxy_test(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
+        """测试并清理代理"""
+        try:
+            import proxy_pool
+
+            # 先加载最新
+            pool_count = proxy_pool.reload_proxies()
+            if pool_count == 0:
+                await update.message.reply_text(
+                    "📭 <b>代理池为空</b>\n\n"
+                    "请将代理添加到 <code>proxy.txt</code> 后重试",
+                    parse_mode="HTML"
+                )
+                return
+
+            msg = await update.message.reply_text(
+                f"🔄 <b>正在测试代理</b>\n\n"
+                f"代理数量: {pool_count}\n"
+                f"并发: 20\n"
+                f"请稍候...",
+                parse_mode="HTML"
+            )
+
+            loop = asyncio.get_event_loop()
+            result = await loop.run_in_executor(
+                self.executor,
+                lambda: proxy_pool.test_and_clean_proxies(concurrency=20)
+            )
+
+            await msg.edit_text(
+                f"✅ <b>代理测试完成</b>\n\n"
+                f"总计: {result['total']}\n"
+                f"存活: {result['alive']} ✅\n"
+                f"移除: {result['removed']} ❌\n"
+                f"耗时: {result['duration']}s\n\n"
+                f"{'💡 不可用代理已从 proxy.txt 中移除' if result['removed'] > 0 else '🎉 所有代理均可用'}",
+                parse_mode="HTML"
+            )
+
+        except ImportError:
+            await update.message.reply_text("❌ proxy_pool 模块未找到")
+        except Exception as e:
+            await update.message.reply_text(f"❌ 代理测试失败: {e}")
+
+    @admin_only
+    async def cmd_proxy_reload(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
+        """从 proxy.txt 重新加载代理"""
+        try:
+            import proxy_pool
+
+            count = proxy_pool.reload_proxies()
+            await update.message.reply_text(
+                f"✅ <b>代理池已重新加载</b>\n\n"
+                f"已加载: {count} 个代理\n\n"
+                f"{'💡 使用 /proxy_test 测试代理可用性' if count > 0 else '📭 proxy.txt 为空或不存在'}",
+                parse_mode="HTML"
+            )
+
+        except ImportError:
+            await update.message.reply_text("❌ proxy_pool 模块未找到")
+        except Exception as e:
+            await update.message.reply_text(f"❌ 重新加载失败: {e}")
+
     # ==================== 定时调度器 ====================
 
     @admin_only
@@ -6868,6 +7040,39 @@ class ProvisionerBot:
         import json
         import threading
 
+        # ===== 代理池预检测 =====
+        try:
+            import proxy_pool
+            pool_count = proxy_pool.get_proxy_count()
+            if pool_count == 0:
+                pool_count = proxy_pool.reload_proxies()
+            
+            if pool_count > 0:
+                log.info(f"[Scheduler] 代理池预检测: 测试 {pool_count} 个代理...")
+                loop = asyncio.get_event_loop()
+                test_result = await loop.run_in_executor(
+                    self.executor,
+                    lambda: proxy_pool.test_and_clean_proxies(concurrency=20)
+                )
+                log.info(f"[Scheduler] 代理池: 存活 {test_result['alive']}/{test_result['total']}，移除 {test_result['removed']}")
+                
+                if chat_id:
+                    try:
+                        await self.app.bot.send_message(
+                            chat_id,
+                            f"🌐 <b>代理池预检测</b>\n\n"
+                            f"存活: {test_result['alive']}/{test_result['total']} | "
+                            f"移除: {test_result['removed']} | "
+                            f"耗时: {test_result['duration']}s",
+                            parse_mode="HTML"
+                        )
+                    except Exception:
+                        pass
+        except ImportError:
+            pass
+        except Exception as e:
+            log.warning(f"代理池预检测异常: {e}")
+
         results = []
         success_count = 0
         fail_count = 0