快连VPN服务器负载实时查看与低延迟节点自动切换脚本编写教程

# server_fetcher.py
import requests
import json
import re

def fetch_server_list():
    """
    获取快连VPN服务器列表及基础信息。
    此处为示例URL，实际需根据情况替换。
    """
    try:
        # 示例：假设本地有一个服务提供服务器信息
        # 实际情况可能是解析本地文件或抓取网页
        response = requests.get('http://127.0.0.1:8080/servers', timeout=10)
        if response.status_code == 200:
            servers = json.loads(response.text)
            return servers # 期望返回格式：[{'name':'US-NY-01', 'host':'ny1.kuailian.com', ...}, ...]
        else:
            print("无法从本地服务获取服务器列表。")
            return []
    except Exception as e:
        print(f"获取服务器列表时发生错误：{e}")
        # 作为后备，可以返回一个硬编码的常用服务器列表
        backup_list = [
            {'name': '美国-纽约-01', 'host': 'us-ny.kuailianvpn.com', 'region': 'US'},
            {'name': '日本-东京-01', 'host': 'jp-tk.kuailianvpn.com', 'region': 'JP'},
            {'name': '德国-法兰克福-01', 'host': 'de-ff.kuailianvpn.com', 'region': 'EU'},
            # ... 添加更多
        ]
        return backup_list

# ping_tester.py
import ping3
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def ping_server(host, count=3, timeout=2):
    """
    Ping指定主机，返回平均延迟和丢包率。
    """
    delays = []
    lost = 0
    for i in range(count):
        try:
            delay = ping3.ping(host, timeout=timeout, unit='ms')
            if delay is not None and delay > 0:
                delays.append(delay)
            else:
                lost += 1
        except Exception:
            lost += 1
        time.sleep(0.1) # 短暂间隔，避免过于密集

    avg_delay = sum(delays) / len(delays) if delays else 9999
    loss_rate = lost / count
    return avg_delay, loss_rate

def batch_ping_servers(server_list, max_workers=10):
    """
    并发Ping多个服务器，提高效率。
    """
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_host = {executor.submit(ping_server, s['host']): s for s in server_list}
        for future in as_completed(future_to_host):
            server = future_to_host[future]
            try:
                avg_delay, loss_rate = future.result()
                results[server['name']] = {
                    'host': server['host'],
                    'delay': avg_delay,
                    'loss': loss_rate,
                    'score': 0 # 待计算
                }
            except Exception as e:
                print(f"测试服务器 {server['name']} 时出错：{e}")
    return results

# evaluator.py
def calculate_score(server_info, delay_weight=0.7, loss_weight=0.3, preferred_regions=['JP', 'US']):
    """
    计算服务器综合得分。
    """
    delay = server_info['delay']
    loss = server_info['loss']
    
    # 简单归一化：假设延迟超过300ms就很差，取倒数并缩放
    # 防止除零，延迟加1
    normalized_delay_score = 100 / (delay + 1) if delay < 300 else 10 / (delay + 1)
    
    # 丢包率直接转化为得分
    loss_score = (1 - loss) * 100
    
    # 基础得分
    base_score = (delay_weight * normalized_delay_score) + (loss_weight * loss_score)
    
    # 区域偏好（可选）
    region = server_info.get('region', '')
    if region in preferred_regions:
        base_score += 10 # 给予偏好区域额外加分
    
    server_info['score'] = round(base_score, 2)
    return server_info

def select_best_server(server_results):
    """
    从测试结果中选择综合得分最高的服务器。
    """
    if not server_results:
        return None
    for name, info in server_results.items():
        calculate_score(info)
    best_server = max(server_results.items(), key=lambda x: x[1]['score'])
    return best_server # 返回 (服务器名, 详细信息)

# vpn_controller.py (示例：假设存在CLI)
import subprocess
import sys
import os

def connect_to_server(server_name):
    """
    通过命令行切换连接到指定服务器。
    此函数为示例，命令参数需要根据快连VPN实际命令行工具调整。
    """
    # 示例命令，绝对路径更可靠
    cli_path = "/Applications/KuaiLian VPN.app/Contents/Resources/cli" # macOS示例
    if sys.platform == "win32":
        cli_path = r"C:\Program Files\KuaiLian VPN\kuailian.exe"
    
    if not os.path.exists(cli_path):
        print(f"未找到CLI工具：{cli_path}，尝试使用系统默认路径或别名")
        cli_path = "kuailian" # 假设已加入系统PATH
    
    command = [cli_path, "connect", "--server", server_name]
    try:
        result = subprocess.run(command, capture_output=True, text=True, timeout=30)
        if result.returncode == 0:
            print(f"成功切换到服务器：{server_name}")
            return True
        else:
            print(f"切换失败。错误输出：{result.stderr}")
            return False
    except FileNotFoundError:
        print("未找到快连VPN命令行工具，请检查安装。")
        return False
    except subprocess.TimeoutExpired:
        print("切换命令执行超时。")
        return False

# main_script.py
import time
import logging
from datetime import datetime
# 导入自定义模块
from server_fetcher import fetch_server_list
from ping_tester import batch_ping_servers
from evaluator import select_best_server
from vpn_controller import connect_to_server

# 配置日志
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s',
                    handlers=[logging.FileHandler('vpn_auto_switch.log'), logging.StreamHandler()])

CURRENT_CONNECTED_SERVER = None # 记录当前连接的服务器
CHECK_INTERVAL = 300 # 检查间隔，单位秒（例如每5分钟检查一次）
SCORE_THRESHOLD = 5 # 只有当最佳节点比当前节点得分高出此阈值时才切换

def main_loop():
    global CURRENT_CONNECTED_SERVER
    logging.info("快连VPN自动切换脚本启动。")
    
    while True:
        try:
            # 1. 获取服务器列表
            servers = fetch_server_list()
            if not servers:
                logging.warning("未获取到服务器列表，等待下次循环。")
                time.sleep(CHECK_INTERVAL)
                continue
                
            # 2. 批量测试延迟和丢包
            logging.info(f"开始测试 {len(servers)} 个服务器...")
            test_results = batch_ping_servers(servers)
            
            # 3. 选择最佳服务器
            best_server_info = select_best_server(test_results)
            if not best_server_info:
                logging.error("未能选出最佳服务器。")
                time.sleep(CHECK_INTERVAL)
                continue
                
            best_name, best_data = best_server_info
            
            # 4. 决策与切换
            if CURRENT_CONNECTED_SERVER is None:
                # 首次运行，直接连接最佳节点
                logging.info(f"初始连接至最佳节点：{best_name} (得分：{best_data['score']})")
                if connect_to_server(best_name):
                    CURRENT_CONNECTED_SERVER = best_name
            else:
                # 非首次，比较得分
                current_score = test_results.get(CURRENT_CONNECTED_SERVER, {}).get('score', 0)
                if best_data['score'] > (current_score + SCORE_THRESHOLD):
                    logging.info(f"发现更优节点：{best_name} (得分：{best_data['score']}) > 当前节点：{CURRENT_CONNECTED_SERVER} (得分：{current_score})，准备切换。")
                    if connect_to_server(best_name):
                        CURRENT_CONNECTED_SERVER = best_name
                else:
                    logging.debug(f"当前节点 {CURRENT_CONNECTED_SERVER} 仍是最优或差异不大，保持连接。")
            
            # 5. 记录本次检查结果
            logging.info(f"本轮检查完成。当前连接：{CURRENT_CONNECTED_SERVER}。最佳可用：{best_name}。等待 {CHECK_INTERVAL} 秒后继续。\n")
            
        except KeyboardInterrupt:
            logging.info("用户中断脚本执行。")
            break
        except Exception as e:
            logging.error(f"主循环发生未知错误：{e}", exc_info=True)
            
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    main_loop()