proxy_manager.py

#!/usr/bin/env python3
"""
高性能多代理管理器
支持多代理轮换、健康检查、自动故障转移
"""

import threading
import time
import random
import requests
from urllib.parse import urlparse
from logger import setup_logger


class ProxyManager:
    def __init__(self, proxy_sources=None, max_failures=3, health_check_interval=60, api_refresh_interval=300, wait_for_postlist_check=True, postlist_check_timeout=120, min_postlist_proxies_before_api=0):
        """
        初始化代理管理器
        :param proxy_sources: 代理源列表 (文件路径或API接口)
        :param max_failures: 最大失败次数
        :param health_check_interval: 健康检查间隔(秒)
        :param api_refresh_interval: API代理刷新间隔(秒)
        :param wait_for_postlist_check: 是否等待 postlist 文件检查完成
        :param postlist_check_timeout: postlist 检查超时时间(秒)
        :param min_postlist_proxies_before_api: 在使用 API 前 postlist 需要的最少可用代理数
        """
        self.logger = setup_logger('ProxyManager')
        self.proxy_sources = proxy_sources or []
        self.max_failures = max_failures
        self.health_check_interval = health_check_interval
        self.api_refresh_interval = api_refresh_interval
        
        # postlist 检查配置
        self.wait_for_postlist_check = wait_for_postlist_check
        self.postlist_check_timeout = postlist_check_timeout
        self.min_postlist_proxies_before_api = min_postlist_proxies_before_api
        
        # 代理池和状态管理
        self.proxy_pool = []
        self.proxy_stats = {}  # {proxy_url: {'failures': 0, 'last_used': time, 'active': True}}
        self.current_proxy_index = 0
        self.lock = threading.RLock()
        
        # 健康检查线程
        self.health_check_thread = None
        self.api_refresh_thread = None
        self.running = False
        self.last_api_refresh = 0
        self.min_api_interval = 60  # 最小API调用间隔(秒)
        
        # postlist 检查状态
        self.postlist_check_completed = False
        self.postlist_check_event = threading.Event()
        self.file_sources = []
        self.api_sources = []
        
        self._separate_proxy_sources()
        
        self.logger.info(f"代理管理器初始化完成，最大失败次数: {max_failures}")
        if self.wait_for_postlist_check and self.file_sources:
            self.logger.info(f"已启用 postlist 优先检查模式，文件源: {len(self.file_sources)}个，API源: {len(self.api_sources)}个")

    def _separate_proxy_sources(self):
        """分离文件源和API源"""
        self.file_sources = []
        self.api_sources = []
        
        for source in self.proxy_sources:
            if source.startswith('http://') or source.startswith('https://'):
                self.api_sources.append(source)
            else:
                self.file_sources.append(source)
    
    def _wait_for_postlist_check_completion(self):
        """等待 postlist 文件检查完成"""
        if not self.wait_for_postlist_check or not self.file_sources:
            self.postlist_check_completed = True
            self.postlist_check_event.set()
            return True
        
        self.logger.info(f"等待 postlist 文件检查完成，超时时间: {self.postlist_check_timeout}秒")
        
        # 等待检查完成或超时
        completed = self.postlist_check_event.wait(timeout=self.postlist_check_timeout)
        
        if completed:
            self.logger.info("✅ postlist 文件检查已完成")
        else:
            self.logger.warning(f"⚠️  postlist 文件检查超时 ({self.postlist_check_timeout}秒)，继续执行")
            self.postlist_check_completed = True
            self.postlist_check_event.set()
        
        return True

    def _check_postlist_and_trigger_api_if_needed(self):
        """检查 postlist 文件代理状态，在需要时触发 API 获取"""
        with self.lock:
            # 统计当前可用代理数量
            active_proxies = sum(1 for stats in self.proxy_stats.values() 
                               if stats.get('active', True) and stats.get('failures', 0) < self.max_failures)
            
            total_proxies = len(self.proxy_pool)
            
            self.logger.info(f"postlist 检查状态: {active_proxies}/{total_proxies} 个代理可用")
            
            # 检查是否需要从 API 获取
            need_api_fetch = (
                active_proxies < self.min_postlist_proxies_before_api or  # 可用代理数不足
                (total_proxies == 0 and self.api_sources)  # 没有任何代理且有 API 源
            )
            
            if need_api_fetch and self.api_sources:
                self.logger.info(f"postlist 可用代理不足 ({active_proxies} < {self.min_postlist_proxies_before_api})，开始从 API 获取...")
                
                # 从 API 获取代理
                api_proxies = []
                for api_source in self.api_sources:
                    self.logger.info(f"从 API 获取代理: {api_source}")
                    proxies = self.load_proxies_from_api(api_source)
                    api_proxies.extend(proxies)
                
                if api_proxies:
                    # 添加到代理池
                    existing_proxies = set(self.proxy_pool)
                    new_proxies = [proxy for proxy in api_proxies if proxy not in existing_proxies]
                    
                    if new_proxies:
                        self.proxy_pool.extend(new_proxies)
                        
                        # 初始化新代理的统计信息
                        for proxy in new_proxies:
                            self.proxy_stats[proxy] = {
                                'failures': 0,
                                'last_used': 0,
                                'active': True,
                                'response_time': 0
                            }
                        
                        self.logger.info(f"✅ 从 API 获取了 {len(new_proxies)} 个新代理，代理池总数: {len(self.proxy_pool)}")
                    else:
                        self.logger.info("API 返回的代理都已存在于代理池中")
                else:
                    self.logger.warning("⚠️  API 源没有返回任何代理")
            else:
                self.logger.info(f"postlist 代理足够，无需从 API 获取 ({active_proxies} >= {self.min_postlist_proxies_before_api})")
            
            # 标记 postlist 检查已完成
            self.postlist_check_completed = True
            self.postlist_check_event.set()
            self.logger.info("✅ postlist 检查已完成，后续将正常进行代理管理")

    def load_proxies_from_file(self, file_path):
        """从文件加载代理列表"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                proxies = []
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#'):
                        # 格式: user:pass@host:port
                        if '@' in line:
                            proxy_url = f"http://{line}"
                            proxies.append(proxy_url)
                
                self.logger.info(f"从文件 {file_path} 加载了 {len(proxies)} 个代理")
                return proxies
        except Exception as e:
            self.logger.error(f"加载代理文件失败 {file_path}: {e}")
            return []

    def load_proxies_from_api(self, api_url, headers=None):
        """从API接口获取代理列表"""
        try:
            response = requests.get(api_url, headers=headers or {}, timeout=10)
            response.raise_for_status()
            
            # 获取响应文本数据
            proxy_data = response.text.strip()
            
            if not proxy_data:
                self.logger.warning(f"API返回空数据: {api_url}")
                return []
            
            # 按行分割代理数据
            proxy_list = proxy_data.split('\n') if '\n' in proxy_data else [proxy_data]
            proxies = []
            
            # 正则表达式匹配 user:password@host:port 格式
            import re
            pattern = re.compile(r'^(?P<user>[^:]+):(?P<password>[^@]+)@(?P<host>[^:]+):(?P<port>\d+)$')
            
            for proxy in proxy_list:
                proxy = proxy.strip()
                if not proxy:
                    continue
                    
                match = pattern.match(proxy)
                if match:
                    # 匹配到完整格式: user:password@host:port
                    user = match.group('user')
                    password = match.group('password')
                    host = match.group('host')
                    port = match.group('port')
                    
                    proxy_url = f"http://{user}:{password}@{host}:{port}"
                    proxies.append(proxy_url)
                    self.logger.debug(f"解析代理: {host}:{port} (用户: {user})")
                    
                else:
                    # 兼容 ip:port 格式
                    parts = proxy.split(":")
                    if len(parts) == 2:
                        host, port = parts
                        if host and port.isdigit():
                            proxy_url = f"http://{host}:{port}"
                            proxies.append(proxy_url)
                            self.logger.debug(f"解析代理: {host}:{port} (无认证)")
                        else:
                            self.logger.warning(f"无效的代理格式: {proxy}")
                    else:
                        self.logger.warning(f"无法解析代理格式: {proxy}")
            
            self.logger.info(f"从API {api_url} 获取了 {len(proxies)} 个代理")
            return proxies
            
        except Exception as e:
            self.logger.error(f"从API获取代理失败 {api_url}: {e}")
            return []

    def refresh_proxy_pool(self, is_initial_load=False):
        """刷新代理池
        :param is_initial_load: 是否为初始加载，只有初始加载才会使用 postlist
        """
        with self.lock:
            new_proxies = []
            
            # 只有初始加载时才使用 postlist 优先检查
            if is_initial_load and self.wait_for_postlist_check and not self.postlist_check_completed:
                self.logger.info("初始加载: 优先加载 postlist 文件代理...")
                # 只加载文件源
                for file_source in self.file_sources:
                    self.logger.info(f"加载 postlist 文件: {file_source}")
                    proxies = self.load_proxies_from_file(file_source)
                    new_proxies.extend(proxies)
            else:
                # 刷新模式，只使用 API 源
                if not is_initial_load:
                    self.logger.info("刷新代理池: 只使用 API 源...")
                    for api_source in self.api_sources:
                        self.logger.info(f"从 API 刷新代理: {api_source}")
                        proxies = self.load_proxies_from_api(api_source)
                        new_proxies.extend(proxies)
                else:
                    # 初始加载但未启用 postlist 优先模式，加载所有源
                    self.logger.info("初始加载: 加载所有代理源...")
                    for source in self.proxy_sources:
                        if source.startswith('http://') or source.startswith('https://'):
                            # API接口
                            proxies = self.load_proxies_from_api(source)
                        else:
                            # 文件路径
                            proxies = self.load_proxies_from_file(source)
                        
                        new_proxies.extend(proxies)
            
            # 更新代理池
            old_count = len(self.proxy_pool)
            self.proxy_pool = list(set(new_proxies))  # 去重
            
            # 初始化新代理的统计信息
            for proxy in self.proxy_pool:
                if proxy not in self.proxy_stats:
                    self.proxy_stats[proxy] = {
                        'failures': 0,
                        'last_used': 0,
                        'active': True,
                        'response_time': 0
                    }
            
            # 清理不存在的代理统计
            active_proxies = set(self.proxy_pool)
            self.proxy_stats = {k: v for k, v in self.proxy_stats.items() if k in active_proxies}
            
            self.logger.info(f"代理池已刷新: {old_count} -> {len(self.proxy_pool)} 个代理")
            
            # 如果是 postlist 优先模式且尚未完成检查，检查是否需要从 API 获取
            if self.wait_for_postlist_check and not self.postlist_check_completed:
                self._check_postlist_and_trigger_api_if_needed()

    def _refresh_from_api_sources(self):
        """仅从 API 源刷新代理池 - 不使用 postlist"""
        with self.lock:
            # 检查是否距离上次 API 调用太近
            current_time = time.time()
            if current_time - self.last_api_refresh < self.min_api_interval:
                self.logger.debug(f"距离上次 API 调用太近，跳过 (间隔: {current_time - self.last_api_refresh:.0f}秒)")
                return False
            
            api_sources = [source for source in self.proxy_sources 
                          if source.startswith('http://') or source.startswith('https://')]
            
            if not api_sources:
                self.logger.debug("没有配置API代理源")
                return False
            
            new_proxies = []
            for api_source in api_sources:
                self.logger.info(f"从API源获取新代理: {api_source}")
                proxies = self.load_proxies_from_api(api_source)
                new_proxies.extend(proxies)
            
            if not new_proxies:
                self.logger.warning("API源没有返回新代理")
                return False
            
            # 过滤掉已存在的代理，只添加新的
            existing_proxies = set(self.proxy_pool)
            fresh_proxies = [proxy for proxy in new_proxies if proxy not in existing_proxies]
            
            if not fresh_proxies:
                self.logger.info("API返回的代理都已存在于代理池中")
                return False
            
            # 添加新代理到代理池
            self.proxy_pool.extend(fresh_proxies)
            
            # 初始化新代理的统计信息
            for proxy in fresh_proxies:
                self.proxy_stats[proxy] = {
                    'failures': 0,
                    'last_used': 0,
                    'active': True,
                    'response_time': 0
                }
            
            self.logger.info(f"从API获取了 {len(fresh_proxies)} 个新代理，代理池总数: {len(self.proxy_pool)}")
            return True

    def get_current_proxy(self):
        """获取当前可用的代理"""
        with self.lock:
            if not self.proxy_pool:
                return None
            
            # 首先清理已禁用的代理
            self._cleanup_failed_proxies()
            
            if not self.proxy_pool:
                self.logger.error("清理后没有任何可用代理")
                return None
            
            # 确保索引在有效范围内
            if self.current_proxy_index >= len(self.proxy_pool):
                self.current_proxy_index = 0
            
            # 寻找可用的代理
            attempts = 0
            start_index = self.current_proxy_index
            
            while attempts < len(self.proxy_pool):
                proxy = self.proxy_pool[self.current_proxy_index]
                stats = self.proxy_stats.get(proxy, {})
                
                # 严格检查代理是否可用
                if (stats.get('active', True) and 
                    stats.get('failures', 0) < self.max_failures):
                    # 更新使用时间
                    self.proxy_stats[proxy]['last_used'] = time.time()
                    self.logger.debug(f"选择代理: {proxy} (失败{stats.get('failures', 0)}次)")
                    return proxy
                else:
                    # 记录为什么跳过这个代理
                    self.logger.debug(f"跳过代理: {proxy} (活跃:{stats.get('active', True)}, 失败:{stats.get('failures', 0)}/{self.max_failures})")
                
                # 切换到下一个代理
                self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxy_pool)
                attempts += 1
                
                # 避免无限循环
                if self.current_proxy_index == start_index and attempts > 0:
                    break
            
            # 所有代理都不可用，尝试从 API 获取新代理 (不使用 postlist)
            self.logger.warning("所有代理都不可用，尝试从 API 获取新代理 (不使用 postlist)")
            
            # 尝试刷新代理池（只尝试一次，避免递归）
            if self._refresh_from_api_sources():
                self.logger.info("成功从API获取新代理")
                # 重置索引到新代理
                self.current_proxy_index = len(self.proxy_pool) - 1  # 指向最后一个新代理
                
                # 再次尝试选择（非递归）
                for i in range(len(self.proxy_pool)):
                    proxy = self.proxy_pool[i]
                    stats = self.proxy_stats.get(proxy, {})
                    if (stats.get('active', True) and 
                        stats.get('failures', 0) < self.max_failures):
                        self.current_proxy_index = i
                        self.proxy_stats[proxy]['last_used'] = time.time()
                        self.logger.info(f"选择新代理: {proxy}")
                        return proxy
            
            self.logger.error("没有任何可用代理")
            return None

    def report_proxy_failure(self, proxy_url):
        """报告代理失败"""
        with self.lock:
            if proxy_url in self.proxy_stats:
                self.proxy_stats[proxy_url]['failures'] += 1
                failures = self.proxy_stats[proxy_url]['failures']
                
                self.logger.warning(f"代理失败 {proxy_url}: {failures}/{self.max_failures}")
                
                if failures >= self.max_failures:
                    # 立即禁用代理
                    self.proxy_stats[proxy_url]['active'] = False
                    self.logger.error(f"代理已禁用 {proxy_url}: 失败次数过多")
                    
                    # 如果当前正在使用这个失败的代理，立即切换
                    if (self.proxy_pool and 
                        self.current_proxy_index < len(self.proxy_pool) and
                        self.proxy_pool[self.current_proxy_index] == proxy_url):
                        self.logger.info(f"当前代理已失败，立即切换: {proxy_url}")
                        self.switch_to_next_proxy()
                    
                    # 如果失败次数过多，立即从代理池中移除
                    if failures >= self.max_failures * 1.5:  # 降低移除阈值
                        self.logger.error(f"代理失败次数过多，立即从池中移除: {proxy_url}")
                        if proxy_url in self.proxy_pool:
                            # 记录移除前的索引
                            removed_index = self.proxy_pool.index(proxy_url)
                            self.proxy_pool.remove(proxy_url)
                            
                            # 调整当前代理索引
                            if self.current_proxy_index > removed_index:
                                self.current_proxy_index -= 1
                            elif self.current_proxy_index >= len(self.proxy_pool) and self.proxy_pool:
                                self.current_proxy_index = 0
                        
                        if proxy_url in self.proxy_stats:
                            del self.proxy_stats[proxy_url]

    def report_proxy_success(self, proxy_url, response_time=0):
        """报告代理成功"""
        with self.lock:
            if proxy_url in self.proxy_stats:
                # 重置失败计数
                self.proxy_stats[proxy_url]['failures'] = 0
                self.proxy_stats[proxy_url]['active'] = True
                self.proxy_stats[proxy_url]['response_time'] = response_time
                self.proxy_stats[proxy_url]['last_used'] = time.time()

    def switch_to_next_proxy(self):
        """切换到下一个可用代理"""
        with self.lock:
            if len(self.proxy_pool) <= 1:
                return
            
            old_index = self.current_proxy_index
            old_proxy = self.proxy_pool[old_index] if old_index < len(self.proxy_pool) else "Unknown"
            
            # 寻找下一个可用的代理
            attempts = 0
            while attempts < len(self.proxy_pool):
                self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxy_pool)
                
                # 避免回到原来的代理
                if self.current_proxy_index == old_index:
                    break
                
                proxy = self.proxy_pool[self.current_proxy_index]
                stats = self.proxy_stats.get(proxy, {})
                
                # 检查这个代理是否可用
                if (stats.get('active', True) and 
                    stats.get('failures', 0) < self.max_failures):
                    self.logger.info(f"代理切换: {old_proxy} -> {proxy}")
                    return
                
                attempts += 1
            
            # 如果没有找到可用代理，至少切换到下一个
            self.current_proxy_index = (old_index + 1) % len(self.proxy_pool)
            new_proxy = self.proxy_pool[self.current_proxy_index]
            self.logger.warning(f"强制代理切换: {old_proxy} -> {new_proxy} (可能不可用)")

    def check_proxy_health(self, proxy_url, timeout=10):
        """检查单个代理的健康状态 - 宽松版本"""
        try:
            proxy_dict = {
                'http': proxy_url,
                'https': proxy_url
            }
            
            self.logger.debug(f"开始检查代理: {proxy_url}")
            
            # 尝试多个测试URL，只要有一个成功就算可用
            test_urls = [
                'https://httpbin.org/ip',
            ]
            
            success_count = 0
            for test_url in test_urls:
                try:
                    if self._test_proxy_connection(proxy_dict, test_url, timeout//len(test_urls)):
                        success_count += 1
                        self.logger.debug(f"代理测试成功: {proxy_url} -> {test_url}")
                        break  # 有一个成功就够了
                except Exception as e:
                    self.logger.debug(f"代理测试失败: {proxy_url} -> {test_url}: {e}")
                    continue
            
            if success_count > 0:
                # 至少有一个测试成功
                self.report_proxy_success(proxy_url)
                return True
            else:
                # 所有测试都失败
                self.logger.warning(f"代理所有测试都失败: {proxy_url}")
                self.report_proxy_failure(proxy_url)
                return False
                
        except Exception as e:
            self.logger.debug(f"代理健康检查异常 {proxy_url}: {e}")
            self.report_proxy_failure(proxy_url)
            return False

    def _test_proxy_connection(self, proxy_dict, test_url, timeout):
        """测试代理连接到指定URL"""
        try:
            start_time = time.time()
            
            response = requests.get(
                test_url,
                proxies=proxy_dict,
                timeout=timeout,
                headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'},
                verify=False  # 忽略SSL证书验证
            )
            
            response_time = time.time() - start_time
            
            if response.status_code == 200:
                self.logger.debug(f"代理连接成功 {test_url}: {response_time:.2f}秒")
                return True
            else:
                self.logger.debug(f"代理连接失败 {test_url}: 状态码 {response.status_code}")
                return False
                
        except requests.exceptions.SSLError as e:
            # SSL错误，但不立即废弃，让上层决定
            self.logger.debug(f"SSL错误 {test_url}: {e}")
            raise e  # 重新抛出异常，让上层处理
                
        except requests.exceptions.ProxyError as e:
            # 代理错误
            self.logger.debug(f"代理错误 {test_url}: {e}")
            return False
            
        except requests.exceptions.ConnectTimeout as e:
            # 连接超时
            self.logger.debug(f"连接超时 {test_url}: {e}")
            return False
            
        except requests.exceptions.ConnectionError as e:
            # 连接错误，检查是否是严重的 SSL 相关错误
            error_msg = str(e).lower()
            ssl_error_keywords = [
                'ssl_error_syscall', 'ssl_connect', 'wrong version number',
                'error:0a00010b', 'ssl_error_ssl', 'ssl_handshake_failure'
            ]
            if any(keyword in error_msg for keyword in ssl_error_keywords):
                self.logger.debug(f"连接 SSL 错误 {test_url}: {e}")
                raise e  # 重新抛出异常，让上层处理
            else:
                self.logger.debug(f"连接错误 {test_url}: {e}")
                return False
                
        except Exception as e:
            self.logger.debug(f"未知错误 {test_url}: {e}")
            return False

    def _mark_proxy_as_dead(self, proxy_url):
        """立即标记代理为死亡状态"""
        with self.lock:
            if proxy_url in self.proxy_stats:
                self.proxy_stats[proxy_url]['failures'] = self.max_failures  # 直接设为最大失败次数
                self.proxy_stats[proxy_url]['active'] = False
                self.logger.error(f"代理已立即废弃 {proxy_url}: 检测到严重错误")

    def _is_serious_ssl_error(self, exception):
        """判断是否是严重的 SSL 错误"""
        error_msg = str(exception).lower()
        serious_ssl_keywords = [
            'ssl_error_syscall',
            'ssl_connect',
            'certificate verify failed',
            'ssl handshake failed',
            'error:0a00010b:ssl routines::wrong version number',  # 新增：您遇到的错误
            'openssl/3.0.13: error:0a00010b',  # 新增：更具体的错误
            'wrong version number',  # 新增：更广泛的匹配
            'ssl_error_ssl',
            'ssl_handshake_failure',
            'sslv3_alert_handshake_failure',
            'ssl_wrong_version_number',
            'ssl version or cipher mismatch',
            'ssl3_get_record:wrong version number',
            'ssl routines:ssl23_get_server_hello:unknown protocol'
        ]
        
        return any(keyword in error_msg for keyword in serious_ssl_keywords)

    def _cleanup_failed_proxies(self):
        """清理失败次数过多的代理"""
        with self.lock:
            original_count = len(self.proxy_pool)
            
            # 找出需要移除的代理
            proxies_to_remove = []
            for proxy in self.proxy_pool:
                stats = self.proxy_stats.get(proxy, {})
                failures = stats.get('failures', 0)
                active = stats.get('active', True)
                
                # 移除失败次数过多或被标记为不活跃的代理
                if failures >= self.max_failures or not active:
                    proxies_to_remove.append(proxy)
                    self.logger.info(f"移除失败代理: {proxy} (失败{failures}次, 活跃:{active})")
            
            # 从代理池中移除
            for proxy in proxies_to_remove:
                if proxy in self.proxy_pool:
                    self.proxy_pool.remove(proxy)
                if proxy in self.proxy_stats:
                    del self.proxy_stats[proxy]
            
            # 调整当前代理索引
            if self.current_proxy_index >= len(self.proxy_pool):
                self.current_proxy_index = 0
            
            removed_count = len(proxies_to_remove)
            remaining_count = len(self.proxy_pool)
            
            if removed_count > 0:
                self.logger.info(f"清理完成: 移除{removed_count}个失败代理, 剩余{remaining_count}个代理")
            
            # 如果剩余代理太少，尝试从 API 获取更多 (不使用 postlist)
            if remaining_count < 3:
                self.logger.warning(f"剩余代理过少({remaining_count}个)，尝试从 API 获取更多 (不使用 postlist)")
                self._refresh_from_api_sources()



    def health_check_worker(self):
        """健康检查工作线程"""
        while self.running:
            try:
                with self.lock:
                    proxies_to_check = list(self.proxy_pool)
                
                # 随机检查顺序，避免总是检查同一个代理
                random.shuffle(proxies_to_check)
                
                for proxy in proxies_to_check:
                    if not self.running:
                        break
                    
                    # 只检查失败次数较多的代理
                    stats = self.proxy_stats.get(proxy, {})
                    if stats.get('failures', 0) > 0:
                        self.check_proxy_health(proxy)
                    
                    time.sleep(1)  # 避免过于频繁的检查
                
                # 等待下次检查，但要能快速响应停止信号
                for _ in range(self.health_check_interval):
                    if not self.running:
                        break
                    time.sleep(1)
                
            except Exception as e:
                self.logger.error(f"健康检查线程错误: {e}")
                for _ in range(10):
                    if not self.running:
                        break
                    time.sleep(1)

    def api_refresh_worker(self):
        """API代理刷新工作线程"""
        while self.running:
            try:
                # 检查是否需要刷新
                current_time = time.time()
                if current_time - self.last_api_refresh >= self.api_refresh_interval:
                    
                    # 检查当前可用代理数量
                    with self.lock:
                        active_count = sum(1 for stats in self.proxy_stats.values() 
                                         if stats.get('active', True) and stats.get('failures', 0) < self.max_failures)
                    
                    # 只在真正需要时才刷新代理池
                    total_count = len(self.proxy_pool)
                    
                    if total_count == 0:
                        # 代理池为空，立即从 API 刷新代理
                        self.logger.info("代理池为空，从 API 获取代理 (不使用 postlist)")
                        if self._refresh_from_api_sources():
                            self.last_api_refresh = current_time
                    elif active_count < total_count * 0.3:
                        # 可用代理少于30%，主动从 API 刷新
                        self.logger.info(f"可用代理不足 ({active_count}/{total_count})，从 API 刷新代理池 (不使用 postlist)")
                        if self._refresh_from_api_sources():
                            self.last_api_refresh = current_time
                    elif active_count < 5:
                        # 可用代理少于5个，补充一些
                        self.logger.info(f"可用代理较少 ({active_count}个)，从 API 补充代理 (不使用 postlist)")
                        if self._refresh_from_api_sources():
                            self.last_api_refresh = current_time
                    else:
                        # 代理充足，跳过本次刷新
                        self.logger.debug(f"代理充足 ({active_count}/{total_count})，跳过API刷新")
                        self.last_api_refresh = current_time  # 更新时间，避免频繁检查
                
                # 等待下次检查，但要能快速响应停止信号
                for _ in range(60):  # 每分钟检查一次
                    if not self.running:
                        break
                    time.sleep(1)
                
            except Exception as e:
                self.logger.error(f"API刷新线程错误: {e}")
                for _ in range(30):
                    if not self.running:
                        break
                    time.sleep(1)

    def start(self, skip_initial_check=False):
        """启动代理管理器"""
        self.running = True
        
        # 初始加载代理，标记为初始加载
        self.refresh_proxy_pool(is_initial_load=True)
        
        # 如果启用了 postlist 优先检查，先检查 postlist 文件代理
        if self.wait_for_postlist_check and self.file_sources and self.proxy_pool:
            self.logger.info("正在检查 postlist 文件代理可用性...")
            self._initial_proxy_check()
            # 检查完成后，触发 API 获取检查
            if not self.postlist_check_completed:
                self._check_postlist_and_trigger_api_if_needed()
        elif not skip_initial_check and self.proxy_pool:
            # 正常模式的代理检查
            self.logger.info("正在检查代理可用性...")
            self._initial_proxy_check()
        
        # 启动健康检查线程
        if self.health_check_interval > 0:
            self.health_check_thread = threading.Thread(target=self.health_check_worker, daemon=True)
            self.health_check_thread.start()
            self.logger.info("健康检查线程已启动")
        
        # 启动API刷新线程
        if self.api_refresh_interval > 0:
            self.api_refresh_thread = threading.Thread(target=self.api_refresh_worker, daemon=True)
            self.api_refresh_thread.start()
            self.logger.info("API刷新线程已启动")
        
        self.logger.info("代理管理器已启动")

    def _initial_proxy_check(self):
        """启动时检查所有代理的可用性"""
        import concurrent.futures
        import threading
        
        total_proxies = len(self.proxy_pool)
        self.logger.info(f"开始检查 {total_proxies} 个代理的可用性...")
        
        # 使用线程池并发检查代理 - 增加并发数
        max_workers = min(50, total_proxies)  # 最多50个并发线程
        checked_count = 0
        valid_count = 0
        
        def check_single_proxy(proxy_url):
            """检查单个代理"""
            nonlocal checked_count, valid_count
            
            try:
                # 适中的检查时间
                is_valid = self.check_proxy_health(proxy_url, timeout=12)
                
                with threading.Lock():
                    checked_count += 1
                    if is_valid:
                        valid_count += 1
                    
                    # 显示进度
                    if checked_count % 10 == 0 or checked_count == total_proxies:
                        self.logger.info(f"检查进度: {checked_count}/{total_proxies}, 可用: {valid_count}")
                
                return proxy_url, is_valid
                
            except Exception as e:
                with threading.Lock():
                    checked_count += 1
                self.logger.debug(f"检查代理异常 {proxy_url}: {e}")
                return proxy_url, False
        
        # 并发检查所有代理
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_proxy = {executor.submit(check_single_proxy, proxy): proxy 
                             for proxy in self.proxy_pool.copy()}
            
            # 等待所有检查完成
            for future in concurrent.futures.as_completed(future_to_proxy):
                try:
                    proxy_url, is_valid = future.result()
                except Exception as e:
                    self.logger.debug(f"代理检查任务异常: {e}")
        
        # 等待所有状态更新完成，然后重新统计
        time.sleep(0.2)  # 增加等待时间确保所有状态更新完成
        
        with self.lock:
            final_active = sum(1 for stats in self.proxy_stats.values() 
                             if stats.get('active', True) and stats.get('failures', 0) < self.max_failures)
        
        self.logger.info(f"✅ 代理检查完成: {final_active}/{total_proxies} 个代理可用")
        
        # 显示详细的失败统计
        failed_count = sum(1 for stats in self.proxy_stats.values() 
                          if not stats.get('active', True) or stats.get('failures', 0) >= self.max_failures)
        if failed_count > 0:
            self.logger.info(f"   其中 {failed_count} 个代理不可用 (失败或被禁用)")
        
        # 如果检查过程统计和最终统计不一致，记录调试信息
        if final_active != valid_count:
            self.logger.warning(f"⚠️  统计差异: 检查过程{valid_count}个，最终统计{final_active}个")
        
        # 如果可用代理太少，尝试从API获取更多
        if final_active < 3:
            self.logger.warning(f"可用代理过少 ({final_active}个)，尝试从API获取更多代理")
            self._refresh_from_api_sources()

    def stop(self):
        """停止代理管理器"""
        self.running = False
        
        # 快速停止线程，减少等待时间
        if self.health_check_thread and self.health_check_thread.is_alive():
            self.health_check_thread.join(timeout=1)
        if self.api_refresh_thread and self.api_refresh_thread.is_alive():
            self.api_refresh_thread.join(timeout=1)
            
        self.logger.info("代理管理器已停止")

    def get_stats(self):
        """获取代理统计信息"""
        with self.lock:
            stats = {
                'total_proxies': len(self.proxy_pool),
                'active_proxies': sum(1 for stats in self.proxy_stats.values() 
                                     if stats.get('active', True) and stats.get('failures', 0) < self.max_failures),
                'current_proxy': self.proxy_pool[self.current_proxy_index] if self.proxy_pool else None,
                'proxy_details': dict(self.proxy_stats)
            }
            return stats

    def add_proxy_source(self, source):
        """添加代理源"""
        if source not in self.proxy_sources:
            self.proxy_sources.append(source)
            self.logger.info(f"添加代理源: {source}")

    def remove_proxy_source(self, source):
        """移除代理源"""
        if source in self.proxy_sources:
            self.proxy_sources.remove(source)
            self.logger.info(f"移除代理源: {source}")