快连VPN API接口实战应用：自动化切换节点与批量账号管理

pip install requests

import requests
import json

def get_server_list(api_key):
    """
    获取快连VPN可用服务器列表
    """
    url = "https://api.kuailian.com/v1/servers"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status() # 检查HTTP错误
        servers = response.json()
        return servers.get('data', []) # 假设返回数据在‘data’字段中
    except requests.exceptions.RequestException as e:
        print(f"获取节点列表失败: {e}")
        return []

# 使用您的API Key
API_KEY = "YOUR_ACTUAL_API_KEY"
server_list = get_server_list(API_KEY)

# 打印前几个节点信息示例
for server in server_list[:3]:
    print(f"ID: {server['id']}, 地区: {server['region']}, 城市: {server['city']}, 负载: {server['load']}%, 推荐: {server['recommend']}")

def connect_to_server(api_key, server_id):
    """
    连接到指定ID的服务器
    """
    url = "https://api.kuailian.com/v1/connection/connect"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "server_id": server_id
        # 可能还有其他参数，如协议选择‘protocol’: ‘wireguard’
    }
    
    try:
        response = requests.post(url, json=payload, headers=headers, timeout=15)
        response.raise_for_status()
        result = response.json()
        if result.get('status') == 'success':
            print(f"成功连接到服务器 {server_id}")
            return True
        else:
            print(f"连接失败: {result.get('message')}")
            return False
    except requests.exceptions.RequestException as e:
        print(f"连接请求异常: {e}")
        return False

def disconnect_vpn(api_key):
    """
    断开当前VPN连接
    """
    url = "https://api.kuailian.com/v1/connection/disconnect"
    headers = {"Authorization": f"Bearer {api_key}"}
    
    try:
        response = requests.post(url, headers=headers, timeout=10)
        response.raise_for_status()
        print("VPN连接已断开")
        return True
    except requests.exceptions.RequestException as e:
        print(f"断开连接请求异常: {e}")
        return False

import time
import schedule

def auto_switch_task():
    print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 开始执行自动切换任务...")
    # 1. 获取节点列表
    servers = get_server_list(API_KEY)
    if not servers:
        print("获取节点列表为空，任务终止。")
        return
    
    # 2. 选择节点（策略：最低负载）
    selected_server = select_by_lowest_load(servers)
    if not selected_server:
        print("未能选择到合适节点。")
        return
    
    print(f"选中最优节点: ID={selected_server['id']}, 地区={selected_server['region']}, 负载={selected_server['load']}%")
    
    # 3. 先断开现有连接（可选，取决于API是否要求）
    # disconnect_vpn(API_KEY)
    # time.sleep(2) # 等待短暂间隔
    
    # 4. 连接到新节点
    connect_to_server(API_KEY, selected_server['id'])

# 安排每30分钟执行一次
schedule.every(30).minutes.do(auto_switch_task)

print("自动化节点切换脚本已启动，按 Ctrl+C 退出。")
while True:
    schedule.run_pending()
    time.sleep(1)

def get_accounts_summary(api_key):
    """
    获取账号列表摘要
    """
    url = "https://api.kuailian.com/v1/accounts"
    headers = {"Authorization": f"Bearer {api_key}"}
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        accounts = response.json().get('data', [])
        return accounts
    except requests.exceptions.RequestException as e:
        print(f"获取账号列表失败: {e}")
        return []

def get_account_detail(api_key, account_id):
    """
    获取单个账号的详细信息，包括流量、有效期等
    """
    url = f"https://api.kuailian.com/v1/accounts/{account_id}"
    headers = {"Authorization": f"Bearer {api_key}"}
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        return response.json().get('data', {})
    except requests.exceptions.RequestException as e:
        print(f"获取账号 {account_id} 详情失败: {e}")
        return {}

# 示例：检查所有账号状态
accounts = get_accounts_summary(API_KEY)
print(f"共管理 {len(accounts)} 个账号")
print("-" * 50)

for acc in accounts:
    acc_id = acc['id']
    detail = get_account_detail(API_KEY, acc_id)
    print(f"账号: {acc.get('email', acc_id)}")
    print(f"  状态: {detail.get('status', 'N/A')}")
    print(f"  套餐: {detail.get('plan_name', 'N/A')}")
    print(f"  流量: {detail.get('used_traffic', 0)} / {detail.get('total_traffic', 'Unlimited')}")
    print(f"  过期时间: {detail.get('expire_date', 'N/A')}")
    print()

import smtplib
from email.mime.text import MIMEText
from datetime import datetime, timedelta

def check_and_alert(api_key, alert_days_before=7, traffic_threshold=0.9):
    """
    检查账号并发送告警
    traffic_threshold: 流量使用超过此比例（如90%）时告警
    """
    accounts = get_accounts_summary(api_key)
    alerts = []
    
    for acc in accounts:
        acc_id = acc['id']
        detail = get_account_detail(API_KEY, acc_id)
        email = acc.get('email', '未知邮箱')
        
        # 检查过期时间
        expire_str = detail.get('expire_date')
        if expire_str:
            try:
                expire_date = datetime.fromisoformat(expire_str.replace('Z', '+00:00'))
                days_left = (expire_date - datetime.utcnow()).days
                if 0 < days_left <= alert_days_before:
                    alerts.append(f"账号 {email} 将在 {days_left} 天后 ({expire_date.date()}) 过期。")
            except ValueError:
                pass
        
        # 检查流量使用
        used = detail.get('used_traffic', 0)
        total = detail.get('total_traffic')
        if total and isinstance(total, (int, float)) and total > 0:
            usage_ratio = used / total
            if usage_ratio >= traffic_threshold:
                alerts.append(f"账号 {email} 流量使用已超过 {traffic_threshold*100:.0f}% ({used:.2f} GB / {total} GB)。")
    
    # 如果有告警，发送邮件（此处为示例，需配置SMTP）
    if alerts:
        alert_message = "\n".join(alerts)
        print("发现告警：\n", alert_message)
        # send_email_alert(alert_message) # 需要实现send_email_alert函数
    else:
        print("所有账号状态正常。")

# 可以将其加入定时任务
schedule.every().day.at("09:00").do(lambda: check_and_alert(API_KEY))

def robust_api_call(request_func, max_retries=3, delay=5):
    """
    带重试机制的API调用包装器
    """
    for attempt in range(max_retries):
        try:
            response = request_func()
            # 可以检查response.status_code，对特定错误码（如429速率限制）进行特殊处理
            if response.status_code == 429:
                wait_time = int(response.headers.get('Retry-After', delay*2))
                print(f"触发速率限制，等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
                continue
            response.raise_for_status()
            return response
        except requests.exceptions.HTTPError as e:
            if attempt == max_retries - 1:
                raise e
            print(f"HTTP错误 ({e})，第 {attempt+1} 次重试...")
        except requests.exceptions.ConnectionError as e:
            print(f"连接错误 ({e})，第 {attempt+1} 次重试...")
        except requests.exceptions.Timeout as e:
            print(f"请求超时 ({e})，第 {attempt+1} 次重试...")
        time.sleep(delay * (attempt + 1)) # 指数退避
    return None # 所有重试失败