utils/proxy_manager.py

import urllib.parse
import random
import time
import requests as std_requests
from datetime import datetime
import yaml
import os
import re
import threading
from concurrent.futures import ThreadPoolExecutor

CLASH_API_URL = ""
LOCAL_PROXY_URL = ""
ENABLE_NODE_SWITCH = False
POOL_MODE = False
FASTEST_MODE = False
PROXY_GROUP_NAME = "节点选择"
CLASH_SECRET = ""
NODE_BLACKLIST = []
_IS_IN_DOCKER = os.path.exists('/.dockerenv')
_global_switch_lock = threading.Lock()
_last_switch_time = 0
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
BASE_DIR = os.path.dirname(CURRENT_DIR)

def format_docker_url(url: str) -> str:
    """智能检测：如果在 Docker 中运行，自动把 127.0.0.1 转为宿主机魔法地址"""
    if not url or not isinstance(url, str):
        return url
    if _IS_IN_DOCKER:
        if "127.0.0.1" in url:
            return url.replace("127.0.0.1", "host.docker.internal")
        if "localhost" in url:
            return url.replace("localhost", "host.docker.internal")
    return url

def reload_proxy_config():
    global CLASH_API_URL, LOCAL_PROXY_URL, ENABLE_NODE_SWITCH, POOL_MODE, \
           FASTEST_MODE, PROXY_GROUP_NAME, CLASH_SECRET, NODE_BLACKLIST
    config_dir = os.path.join(BASE_DIR, "data")
    config_path = os.path.join(config_dir, "config.yaml")
    if not os.path.exists(config_path):
        print(f"[{ts()}] [WARNING] 配置文件 {config_path} 不存在，使用默认代理设置。")
        conf_data = {}
    else:
        with open(config_path, "r", encoding="utf-8") as f:
            conf_data = yaml.safe_load(f) or {}

    clash_conf = conf_data.get("clash_proxy_pool", {})
    ENABLE_NODE_SWITCH = clash_conf.get("enable", False)
    POOL_MODE = clash_conf.get("pool_mode", False)
    FASTEST_MODE = clash_conf.get("fastest_mode", False)
    CLASH_API_URL = format_docker_url(clash_conf.get("api_url", "http://127.0.0.1:9090"))
    LOCAL_PROXY_URL = format_docker_url(clash_conf.get("test_proxy_url", "http://127.0.0.1:7890"))
    
    PROXY_GROUP_NAME = clash_conf.get("group_name", "节点选择")
    CLASH_SECRET = clash_conf.get("secret", "")
    NODE_BLACKLIST = clash_conf.get("blacklist", ["港", "HK", "台", "TW", "中国", "CN"])
   
    print(f"[{ts()}] [系统] 代理管理模块配置已同步更新。")

def ts() -> str:
    """获取当前时间戳字符串，用于日志"""
    return datetime.now().strftime("%H:%M:%S")

def clean_for_log(text: str) -> str:
    """用于日志输出：过滤掉字符串中的国旗、飞机、火箭等 Emoji 符号"""
    emoji_pattern = re.compile(
        r'[\U0001F1E6-\U0001F1FF]'
        r'|[\U0001F300-\U0001F6FF]'
        r'|[\U0001F900-\U0001F9FF]'
        r'|[\U00002600-\U000027BF]'
        r'|[\uFE0F]'
    )
    return emoji_pattern.sub('', text).strip()

def get_display_name(proxy_url: str) -> str:
    """统一日志脱敏：将 URL 转换为 [X号机] 或隐藏域名"""
    if not proxy_url:
        return "全局单机"
    try:
        parsed = urllib.parse.urlparse(proxy_url)
        if parsed.port and 41000 < parsed.port <= 41050:
            return f"{parsed.port - 41000}号机"
        return f"端口:{parsed.port}"
    except:
        return "未知通道"

def get_api_url_for_proxy(proxy_url: str) -> str:
    """根据开关决定是独立容器 API，还是使用固定 API"""
    if not POOL_MODE or not proxy_url:
        return CLASH_API_URL
    try:
        parsed = urllib.parse.urlparse(proxy_url)
        port = parsed.port
        if port and 41000 < port <= 41050:
            api_port = port + 1000
            return format_docker_url(f"http://{parsed.hostname}:{api_port}")
    except Exception:
        pass
    return CLASH_API_URL

def test_proxy_liveness(proxy_url=None):
    """测试当前代理是否可用 (脱敏)"""
    raw_url = proxy_url if proxy_url else LOCAL_PROXY_URL
    target_proxy = format_docker_url(raw_url)
    proxies = {"http": target_proxy, "https": target_proxy}
    display_name = get_display_name(proxy_url if proxy_url else LOCAL_PROXY_URL)
    
    try:
        res = std_requests.get("https://cloudflare.com/cdn-cgi/trace", proxies=proxies, timeout=5)
        if res.status_code == 200:
            loc = "UNKNOWN"
            for line in res.text.split('\n'):
                if line.startswith("loc="):
                    loc = line.split("=")[1].strip()

            blocked_regions = ["CN", "HK"]
            if loc in blocked_regions:
                print(f"[{ts()}] [代理测活] {display_name} 地区受限 ({loc})，弃用！")
                return False
                
            print(f"[{ts()}] [代理测活] {display_name} 成功！地区 ({loc})，延迟: {res.elapsed.total_seconds():.2f}s")
            return True
        return False
    except Exception:
        print(f"[{ts()}] [代理测活] {display_name} 链路中断或超时。")
        return False


def smart_switch_node(proxy_url=None):
    global _last_switch_time
    if not ENABLE_NODE_SWITCH:
        return True

    # 如果是独立代理池模式，互相不影响，不需要锁
    if POOL_MODE and proxy_url:
        return _do_smart_switch(proxy_url)

    with _global_switch_lock:
        if time.time() - _last_switch_time < 10:
            print(f"[{ts()}] [代理池] 其他线程刚完成切换，跳过本次请求...")
            return True

        success = _do_smart_switch(proxy_url)
        if success:
            _last_switch_time = time.time()
        return success

def _do_smart_switch(proxy_url=None):
    """智能切换节点并测活的核心逻辑 (脱敏)"""
    if not ENABLE_NODE_SWITCH:
        return True
        
    current_api_url = get_api_url_for_proxy(proxy_url)
    headers = {"Authorization": f"Bearer {CLASH_SECRET}"} if CLASH_SECRET else {}
    
    display_name = get_display_name(proxy_url)

    api_display = get_display_name(current_api_url).replace("号机", "号API")
    
    try:
        resp = std_requests.get(f"{current_api_url}/proxies", headers=headers, timeout=5)
        if resp.status_code != 200:
            print(f"[{ts()}] [ERROR] 无法连接 Clash API ({api_display})，请检查容器状态。")
            return False
            
        proxies_data = resp.json().get('proxies', {})

        actual_group_name = None
        for key in proxies_data.keys():
            if PROXY_GROUP_NAME in key and isinstance(proxies_data[key], dict) and 'all' in proxies_data[key]:
                actual_group_name = key
                break
                
        if not actual_group_name:
            print(f"[{ts()}] [ERROR] {display_name} 找不到策略组关键词 '{PROXY_GROUP_NAME}'")
            return False
            
        safe_group_name = urllib.parse.quote(actual_group_name)
        all_nodes = proxies_data[actual_group_name].get('all', [])
        
        valid_nodes = [
            n for n in all_nodes 
            if not any(kw.upper() in n.upper() for kw in NODE_BLACKLIST)
        ]
        
        if not valid_nodes:
            print(f"[{ts()}] [ERROR] {display_name} 过滤后无可用节点，请检查黑名单。")
            return False

        if FASTEST_MODE:
            print(f"\n[{ts()}] [代理池] {display_name} 开启优选模式，并发测速 {len(valid_nodes)} 个节点...")
            
            session = std_requests.Session()
            
            def trigger_delay(n):
                enc_n = urllib.parse.quote(n, safe="")
                try:
                    session.get(
                        f"{current_api_url}/proxies/{enc_n}/delay?timeout=2000&url=http://www.gstatic.com/generate_204", 
                        headers=headers, timeout=2.5
                    )
                except:
                    pass

            thread_count = min(10, len(valid_nodes))
            with ThreadPoolExecutor(max_workers=thread_count) as executor:
                executor.map(trigger_delay, valid_nodes)
                
            session.close()
                
            time.sleep(1.5)
            
            try:
                resp2 = std_requests.get(f"{current_api_url}/proxies", headers=headers, timeout=5)
                if resp2.status_code == 200:
                    p_data = resp2.json().get('proxies', {})
                    best_node = None
                    min_delay = float('inf')
                    
                    for n in valid_nodes:
                        history = p_data.get(n, {}).get("history", [])
                        if history:
                            delay = history[-1].get("delay", 0)
                            if 0 < delay < min_delay:
                                min_delay = delay
                                best_node = n
                    
                    if best_node:
                        print(f"[{ts()}] [代理池] {display_name} 测速完成，最快节点: [{clean_for_log(best_node)}] ({min_delay}ms)")
                        switch_resp = std_requests.put(
                            f"{current_api_url}/proxies/{safe_group_name}", 
                            headers=headers, json={"name": best_node}, timeout=5
                        )
                        if switch_resp.status_code == 204:
                            time.sleep(1)
                            if test_proxy_liveness(proxy_url):
                                return True
                            print(f"[{ts()}] [代理池] {display_name} 最快节点测活失败，回退到随机抽卡模式...")
                    else:
                        print(f"[{ts()}] [代理池] {display_name} 所有节点均超时，回退到随机抽卡模式...")
            except Exception as e:
                print(f"[{ts()}] [代理池] {display_name} 优选模式异常: {e}，回退到随机抽卡模式...")

        max_retries = 10
        for i in range(1, max_retries + 1):
            selected_node = random.choice(valid_nodes)
            
            print(f"\n[{ts()}] [代理池] {display_name} 尝试切换节点: [{clean_for_log(selected_node)}] ({i}/{max_retries})")
            
            switch_resp = std_requests.put(
                f"{current_api_url}/proxies/{safe_group_name}", 
                headers=headers, json={"name": selected_node}, timeout=5
            )
            
            if switch_resp.status_code == 204:
                time.sleep(1.5)
                if test_proxy_liveness(proxy_url):
                    return True
                print(f"[{ts()}] [代理池] {display_name} 测活失败，重新抽卡...")
            else:
                print(f"[{ts()}] [代理池] {display_name} 指令下发失败 (HTTP {switch_resp.status_code})。")
                
        print(f"\n[{ts()}] [代理池] {display_name} 连续 10 次抽卡均不可用！")
        return False
        
    except Exception as e:
        print(f"[{ts()}] [ERROR] {display_name} 切换节点异常: {e}")
        return False

reload_proxy_config()