快连VPN API接口实战：自动化节点切换与连接状态监控脚本编写

import requests
import time
import os
from typing import Dict, List

# 从环境变量读取API密钥，确保安全
API_BASE_URL = "https://api.kuailianvpn.com/v1"
API_KEY = os.getenv("KUAILIAN_API_KEY")

def get_auth_header():
    """构造认证请求头。"""
    return {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

def fetch_server_list() -> List[Dict]:
    """获取所有可用服务器节点列表。"""
    try:
        response = requests.get(f"{API_BASE_URL}/servers", headers=get_auth_header(), timeout=10)
        response.raise_for_status()  # 检查HTTP错误
        return response.json().get('data', [])
    except requests.exceptions.RequestException as e:
        print(f"获取节点列表失败: {e}")
        return []

def ping_server(hostname: str) -> float:
    """简易延迟测试（使用TCP握手时间模拟ping）。注意：生产环境可能需要ICMP权限或使用HTTP ping。"""
    import socket
    start = time.time()
    try:
        socket.create_connection((hostname, 443), timeout=2).close()
        return (time.time() - start) * 1000  # 转换为毫秒
    except (socket.timeout, socket.error):
        return float('inf')  # 超时或错误，返回无限大延迟

def connect_to_server(server_id: str) -> bool:
    """请求连接到指定ID的服务器。"""
    data = {"server_id": server_id}
    try:
        response = requests.post(f"{API_BASE_URL}/connection/connect", headers=get_auth_header(), json=data, timeout=15)
        if response.status_code == 200:
            print(f"已成功请求连接到节点 {server_id}")
            return True
        else:
            print(f"连接请求失败: {response.status_code} - {response.text}")
            return False
    except requests.exceptions.RequestException as e:
        print(f"连接请求异常: {e}")
        return False

def auto_switch_to_best_node():
    """主函数：自动切换至延迟最低的节点。"""
    if not API_KEY:
        print("错误：未设置 KUAILIAN_API_KEY 环境变量。")
        return

    servers = fetch_server_list()
    if not servers:
        print("未获取到可用节点。")
        return

    print(f"共获取到 {len(servers)} 个节点，正在测试延迟...")

    # 测试每个节点的延迟（为演示，仅测试前5个）
    tested_servers = []
    for server in servers[:5]:
        delay = ping_server(server['hostname'])
        server['measured_delay'] = delay
        tested_servers.append(server)
        print(f"  节点 {server['name']} ({server['country']}) 延迟: {delay:.2f} ms")

    # 按延迟排序，排除不可达节点
    available_servers = [s for s in tested_servers if s['measured_delay'] < float('inf')]
    if not available_servers:
        print("所有测试节点均不可达。")
        return

    best_server = min(available_servers, key=lambda x: x['measured_delay'])
    print(f"\n最佳节点: {best_server['name']}，延迟 {best_server['measured_delay']:.2f} ms")

    # 执行切换
    if connect_to_server(best_server['id']):
        print("节点切换指令已发送。请稍后在客户端或通过状态接口确认。")
    else:
        print("节点切换失败。")

if __name__ == "__main__":
    auto_switch_to_best_node()

#!/bin/bash

# 配置
API_BASE="https://api.kuailianvpn.com/v1"
API_KEY="YOUR_API_KEY_HERE" # 实际使用时建议从文件或环境变量读取
TARGET_COUNTRY="JP" # 目标国家代码，例如日本

# 获取访问令牌（假设需要）
AUTH_RESPONSE=$(curl -s -X POST "${API_BASE}/auth/token" \
  -H "Content-Type: application/json" \
  -d "{\"api_key\": \"${API_KEY}\"}")
ACCESS_TOKEN=$(echo $AUTH_RESPONSE | jq -r '.access_token')

if [ -z "$ACCESS_TOKEN" ] || [ "$ACCESS_TOKEN" = "null" ]; then
  echo "认证失败！"
  exit 1
fi

# 获取节点列表并筛选出目标国家的第一个节点
SERVER_LIST_RESPONSE=$(curl -s -X GET "${API_BASE}/servers" \
  -H "Authorization: Bearer ${ACCESS_TOKEN}")
TARGET_SERVER_ID=$(echo $SERVER_LIST_RESPONSE | jq -r ".data[] | select(.country_code == \"${TARGET_COUNTRY}\") | .id" | head -1)

if [ -z "$TARGET_SERVER_ID" ]; then
  echo "未找到 ${TARGET_COUNTRY} 的可用节点。"
  exit 1
fi

echo "找到节点 ID: ${TARGET_SERVER_ID}，正在尝试连接..."

# 发起连接请求
CONNECT_RESPONSE=$(curl -s -X POST "${API_BASE}/connection/connect" \
  -H "Authorization: Bearer ${ACCESS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d "{\"server_id\": \"${TARGET_SERVER_ID}\"}")

echo "连接请求响应：$CONNECT_RESPONSE"

import requests
import smtplib
import json
from email.mime.text import MIMEText
from datetime import datetime
import os

API_BASE_URL = "https://api.kuailianvpn.com/v1"
API_KEY = os.getenv("KUAILIAN_API_KEY")
CHECK_INTERVAL = 300  # 检查间隔，单位秒（5分钟）

# 告警配置（邮件示例）
ALERT_EMAIL = {
    'smtp_server': 'smtp.gmail.com',
    'smtp_port': 587,
    'sender': 'your-email@gmail.com',
    'password': os.getenv("EMAIL_PASSWORD"),
    'receiver': 'admin@yourcompany.com'
}

def get_connection_status():
    """获取当前VPN连接状态。"""
    headers = {"Authorization": f"Bearer {API_KEY}"}
    try:
        response = requests.get(f"{API_BASE_URL}/connection/status", headers=headers, timeout=10)
        if response.status_code == 200:
            return response.json()  # 假设返回包含 connected, server, ip, uptime 等字段
        else:
            print(f"状态查询失败，HTTP {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"状态查询请求异常: {e}")
        return None

def send_alert(subject, body):
    """发送邮件告警。"""
    msg = MIMEText(body, 'plain', 'utf-8')
    msg['Subject'] = subject
    msg['From'] = ALERT_EMAIL['sender']
    msg['To'] = ALERT_EMAIL['receiver']

    try:
        server = smtplib.SMTP(ALERT_EMAIL['smtp_server'], ALERT_EMAIL['smtp_port'])
        server.starttls()
        server.login(ALERT_EMAIL['sender'], ALERT_EMAIL['password'])
        server.send_message(msg)
        server.quit()
        print(f"告警邮件已发送: {subject}")
    except Exception as e:
        print(f"发送告警邮件失败: {e}")

def monitor_vpn_connection():
    """主监控循环。"""
    last_status = None
    while True:
        print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始检查VPN状态...")
        status = get_connection_status()

        if status is None:
            # API请求失败，网络或服务异常
            alert_subject = "[紧急] VPN状态监控API请求失败"
            alert_body = f"在 {datetime.now()} 无法获取快连VPN连接状态，请检查网络或API服务。"
            send_alert(alert_subject, alert_body)
        else:
            is_connected = status.get('connected', False)
            current_server = status.get('server_name', '未知')
            current_ip = status.get('ip_address', '未知')

            print(f"  状态: {'已连接' if is_connected else '未连接'}")
            print(f"  节点: {current_server}")
            print(f"  IP: {current_ip}")

            # 状态发生改变时告警
            if last_status is not None:
                last_connected = last_status.get('connected', False)
                if last_connected and not is_connected:
                    alert_subject = "[警告] VPN连接已断开"
                    alert_body = f"快连VPN连接于 {datetime.now()} 断开。\n前一节点: {last_status.get('server_name')}"
                    send_alert(alert_subject, alert_body)
                elif not last_connected and is_connected:
                    print("  VPN连接已恢复。")

            last_status = status

        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    monitor_vpn_connection()

import logging
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    handlers=[logging.FileHandler('vpn_manager.log'), logging.StreamHandler()])
logger = logging.getLogger(__name__)
logger.info("开始执行节点切换任务。")