快连VPN的API密钥管理与自动化脚本（Python/Shell）调用实战教程

# vpn_manager.py
import os
import requests
import json
from typing import Optional, Dict, Any

class KuaiLianVPNClient:
    """快连VPN API客户端类"""
    
    def __init__(self, api_base: str = "https://api.kuailian.com/v1", api_key: Optional[str] = None):
        """
        初始化客户端。
        :param api_base: API基础地址，请替换为快连官方地址。
        :param api_key: API密钥。优先使用参数，其次从环境变量读取。
        """
        self.api_base = api_base.rstrip('/')
        self.api_key = api_key or os.environ.get('KUAILIAN_API_KEY')
        if not self.api_key:
            raise ValueError("API密钥未提供。请通过参数传入或设置环境变量 'KUAILIAN_API_KEY'")
        self.headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """内部请求方法，处理通用逻辑"""
        url = f"{self.api_base}/{endpoint.lstrip('/')}"
        try:
            resp = self.session.request(method, url, **kwargs)
            resp.raise_for_status()  # 如果状态码不是200，抛出HTTPError异常
            return resp.json()
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            if hasattr(e, 'response') and e.response is not None:
                print(f"响应状态码: {e.response.status_code}")
                print(f"响应内容: {e.response.text}")
            return {}

# 示例：初始化客户端（密钥从环境变量读取）
# client = KuaiLianVPNClient()

# 在 KuaiLianVPNClient 类中添加方法
    def get_server_list(self) -> list:
        """获取所有可用服务器列表"""
        data = self._make_request('GET', 'servers')
        # 假设返回数据格式： {"servers": [{“id”: “us-sfo-01”, “name”: “美国-旧金山”, “load”: 45, “country”: “US”}, ...]}
        return data.get('servers', [])

    def get_fastest_server(self) -> Optional[Dict]:
        """根据服务器负载选择最优服务器（示例逻辑）"""
        servers = self.get_server_list()
        if not servers:
            return None
        # 简单逻辑：选择负载最低的服务器
        fastest = min(servers, key=lambda s: s.get('load', 100))
        return fastest

    def connect_to_server(self, server_id: str) -> bool:
        """连接到指定服务器"""
        payload = {'server_id': server_id}
        data = self._make_request('POST', 'connection/connect', json=payload)
        # 假设成功返回： {"status": "connecting", “message”: “正在连接...”}
        return data.get('status') in ['connecting', 'connected']

    def disconnect(self) -> bool:
        """断开当前连接"""
        data = self._make_request('POST', 'connection/disconnect')
        return data.get('status') == 'disconnected'

    def get_connection_status(self) -> Dict:
        """获取当前连接状态"""
        return self._make_request('GET', 'connection')

# 使用示例
if __name__ == "__main__":
    client = KuaiLianVPNClient()
    
    # 1. 获取并显示服务器列表
    print("=== 服务器列表 ===")
    servers = client.get_server_list()
    for srv in servers[:5]:  # 显示前5个
        print(f"ID: {srv.get('id')}, 名称: {srv.get('name')}, 负载: {srv.get('load')}%")
    
    # 2. 自动选择并连接到最快服务器
    print("\n=== 尝试连接到最优服务器 ===")
    best = client.get_fastest_server()
    if best:
        print(f"已选择服务器: {best.get('name')}")
        if client.connect_to_server(best.get('id')):
            print("连接命令已发送成功。")
        else:
            print("连接请求失败。")
    else:
        print("无法获取服务器列表。")
    
    # 3. 检查状态
    status = client.get_connection_status()
    print(f"\n当前连接状态: {status}")

# monitor_vpn.py
import time
from vpn_manager import KuaiLianVPNClient  # 导入上面定义的类

def monitor_and_ensure_connection(client, check_interval=60, max_retries=3):
    """监控VPN连接，如果断开则自动重连到最优服务器"""
    retry_count = 0
    while True:
        status = client.get_connection_status()
        is_connected = status.get('connected', False)
        current_ip = status.get('ip', '未知')
        
        print(f"[{time.ctime()}] 状态: {'已连接' if is_connected else '未连接'}, IP: {current_ip}")
        
        if not is_connected and retry_count < max_retries:
            print("连接已断开，尝试重新连接...")
            best = client.get_fastest_server()
            if best and client.connect_to_server(best.get('id')):
                print(f"已发起连接到 {best.get('name')} 的请求。")
                retry_count += 1
            else:
                print("获取服务器或发起连接失败。")
        elif is_connected:
            retry_count = 0  # 连接成功，重置重试计数
        
        time.sleep(check_interval)  # 等待一段时间后再次检查

if __name__ == "__main__":
    vpn_client = KuaiLianVPNClient()
    try:
        monitor_and_ensure_connection(vpn_client, check_interval=300)  # 每5分钟检查一次
    except KeyboardInterrupt:
        print("\n监控已由用户中断。")

#!/bin/bash

# 快连VPN Shell API控制脚本
# 请将您的API密钥设置到环境变量 KUAILIAN_API_KEY
API_BASE="https://api.kuailian.com/v1"
API_KEY="${KUAILIAN_API_KEY}"

if [ -z "$API_KEY" ]; then
    echo "错误: 请设置环境变量 KUAILIAN_API_KEY"
    exit 1
fi

AUTH_HEADER="Authorization: Bearer $API_KEY"

# 函数：发送API请求并处理JSON响应
_api_call() {
    local method=$1
    local endpoint=$2
    local data=$3
    local url="$API_BASE/$endpoint"
    
    local curl_cmd="curl -s -X $method -H '$AUTH_HEADER' -H 'Content-Type: application/json'"
    if [ -n "$data" ]; then
        curl_cmd+=" -d '$data'"
    fi
    curl_cmd+=" '$url'"
    
    eval $curl_cmd | jq .
}

# 1. 获取服务器列表
list_servers() {
    echo "获取服务器列表..."
    _api_call GET "servers"
}

# 2. 获取连接状态
get_status() {
    echo "获取连接状态..."
    _api_call GET "connection"
}

# 3. 连接到特定服务器（示例服务器ID）
connect_to() {
    local server_id=$1
    if [ -z "$server_id" ]; then
        echo "用法: $0 connect <server_id>"
        echo "提示: 先运行 '$0 list' 查看服务器ID"
        exit 1
    fi
    echo "正在连接到服务器 $server_id ..."
    _api_call POST "connection/connect" "{\"server_id\": \"$server_id\"}"
}

# 4. 断开连接
disconnect() {
    echo "断开VPN连接..."
    _api_call POST "connection/disconnect"
}

# 主命令解析
case "$1" in
    "list")
        list_servers
        ;;
    "status")
        get_status
        ;;
    "connect")
        connect_to "$2"
        ;;
    "disconnect")
        disconnect
        ;;
    *)
        echo "用法: $0 {list|status|connect <server_id>|disconnect}"
        exit 1
        ;;
esac

# 赋予执行权限
chmod +x vpn_control.sh

# 设置API密钥（当前会话有效）
export KUAILIAN_API_KEY="your_api_key_here"

# 查看服务器列表
./vpn_control.sh list

# 连接到特定服务器（假设服务器ID是us-nyc-01）
./vpn_control.sh connect us-nyc-01

# 查看当前状态
./vpn_control.sh status

# 断开连接
./vpn_control.sh disconnect