autoTools：Kookeey 代理网关管理工具

项目文稿说明：本文按现有笔记整理，代码是节选或设计示意；仓库实现、线上地址和测试结果尚未逐项核验。后续计划不代表已经完成。

autoTools：我用 Python 写了一个代理网关管理工具

博客：菜牙爱吃菜 | 作者：菜牙 | 日期：2025-11-22

最近花了大概两周的业余时间，写了这个小工具 autoTools。起因很简单：我在管理一堆 Kookeey 代理账号的时候，手动切换 IP、看流量、查状态太烦了，就想着不如自己写一个 GUI 工具统一管理。

这篇文章就当作是给这个项目做个总结，顺便记录一下开发过程中踩过的坑和学到的东西。

一、这玩意儿是干嘛的？

简单说，autoTools 是一个 Kookeey 代理网关管理工具，主要功能包括：

代理网关服务：在本地起一个 HTTP 代理服务器，把流量转发到 Kookeey 平台
多实例支持：同时管理多个代理实例，每个实例独立配置
IP 切换：一键切换出口 IP，不用手动去 Kookeey 后台操作
流量监控：实时查看每个代理的流量消耗
账号管理：管理 Kookeey 账号，状态一目了然

用 PyQt6 写的桌面 GUI，数据存储用 SQLite，浏览器自动化部分用 Selenium。

二、技术栈

层面	选型	理由
GUI 框架	PyQt6	比 Tkinter 好看，比 PySide6 资料多，信号槽机制适合这种多模块通信的场景
数据库	SQLite	单文件、零配置、Python 内置支持，对个人工具来说足够了
代理转发	自研 HTTP Proxy	基于 `http.server` 和 `requests` 封装，可控性高
浏览器自动化	Selenium	用于自动登录 Kookeey 后台、提取 cookie/token
账号状态管理	状态机模式	账号有明确的生命周期，用状态机管理很合适

三、架构设计

整体架构大概长这样：

autoTools/
├── main_gui.py            # GUI 入口，主窗口
├── gateway_service.py     # 代理网关核心服务
├── proxy_manager.py       # 代理实例管理
├── proxy_utils.py         # 代理相关工具函数
├── database_manager.py    # SQLite 数据库操作
├── kookeey_data.db        # 数据库文件
└── chrome_extension/      # Chrome 代理插件生成模板

核心模块说明

gateway_service.py 是整个工具的核心。它启动一个本地 HTTP 代理服务器，浏览器的流量先走到这里，然后由它转发到 Kookeey 的上游代理。

## gateway_service.py 简化示意
import http.server
import requests

class ProxyHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        # 从当前激活的代理实例获取配置
        config = self.server.proxy_manager.get_active()

        proxies = {
            "http": f"http://{config['username']}:{config['password']}@"
                    f"{config['host']}:{config['port']}",
            "https": f"http://{config['username']}:{config['password']}@"
                     f"{config['host']}:{config['port']}",
        }

        try:
            resp = requests.get(
                self.path, proxies=proxies, timeout=30
            )
            self.send_response(resp.status_code)
            self.end_headers()
            self.wfile.write(resp.content)
            # 记录流量
            self.server.proxy_manager.record_traffic(
                config['id'], len(resp.content)
            )
        except Exception as e:
            self.send_error(502, str(e))

database_manager.py 负责所有数据持久化。账号状态管理用了简单的状态机：

## database_manager.py 简化示意
import sqlite3
from enum import Enum

class AccountState(Enum):
    NEW = "new"           # 新添加，未验证
    ACTIVE = "active"     # 正常使用中
    GATEWAY = "gateway"   # 正在作为网关运行
    EXHAUSTED = "exhausted"  # 流量已用完
    ERROR = "error"       # 连接异常

class DatabaseManager:
    def __init__(self, db_path="kookeey_data.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_tables()

    def _init_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS accounts (
                id INTEGER PRIMARY KEY,
                username TEXT UNIQUE,
                password TEXT,
                api_key TEXT,
                state TEXT DEFAULT 'new',
                total_traffic INTEGER DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.conn.commit()

    def update_state(self, account_id: int, new_state: AccountState):
        self.conn.execute(
            "UPDATE accounts SET state = ? WHERE id = ?",
            (new_state.value, account_id)
        )
        self.conn.commit()

main_gui.py 是 PyQt6 的主界面，用了 QTabWidget 做分页：

## main_gui.py 简化示意
from PyQt6.QtWidgets import (
    QMainWindow, QTabWidget, QWidget, QVBoxLayout,
    QPushButton, QTableWidget, QLabel
)

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("autoTools - Kookeey 代理网关管理")
        self.resize(900, 600)

        self.tabs = QTabWidget()
        self.setCentralWidget(self.tabs)

        # 四个主页面
        self.tabs.addTab(AccountTab(), "账号管理")
        self.tabs.addTab(GatewayTab(), "网关服务")
        self.tabs.addTab(TrafficTab(), "流量监控")
        self.tabs.addTab(SettingsTab(), "设置")

四、核心功能

4.1 多实例 + 代理网关

这个是我花时间最多的部分。每个代理实例就是一个独立的 Kookeey 账号，可以单独启动/停止。GUI 上可以一目了然看到每个实例的状态：

## proxy_manager.py 简化示意
import threading

class ProxyInstance:
    def __init__(self, account_id, config):
        self.account_id = account_id
        self.config = config
        self.server = None
        self.thread = None
        self.running = False

    def start(self):
        self.server = HTTPServer(
            ('127.0.0.1', self.config['local_port']),
            ProxyHandler
        )
        self.server.proxy_manager = self.manager
        self.thread = threading.Thread(target=self.server.serve_forever)
        self.thread.daemon = True
        self.thread.start()
        self.running = True

    def stop(self):
        if self.server:
            self.server.shutdown()
        self.running = False

class ProxyManager:
    def __init__(self, db_manager):
        self.db = db_manager
        self.instances = {}  # account_id -> ProxyInstance

    def switch_gateway(self, account_id):
        """切换当前活跃的网关实例"""
        # 先停掉旧的
        for inst in self.instances.values():
            if inst.running:
                inst.stop()
                self.db.update_state(inst.account_id, AccountState.ACTIVE)

        # 启动新的
        config = self.db.get_account(account_id)
        inst = ProxyInstance(account_id, config)
        inst.manager = self
        inst.start()
        self.instances[account_id] = inst
        self.db.update_state(account_id, AccountState.GATEWAY)

4.2 Chrome 代理插件生成

这个功能算是个小惊喜——与其让用户手动配置系统代理，不如自动生成一个 Chrome 插件，装了就能用：

def generate_proxy_extension(username, password, host, port, local_port):
    """生成 Chrome PAC 代理插件"""
    manifest = {
        "manifest_version": 3,
        "name": "autoTools Proxy",
        "version": "1.0",
        "permissions": ["proxy"],
        "background": {
            "service_worker": "background.js"
        },
        "action": {
            "default_title": "autoTools Proxy"
        }
    }

    background_js = """
    chrome.proxy.settings.set({
        value: {
            mode: "fixed_servers",
            rules: {
                singleProxy: {
                    host: "%s",
                    port: %s
                }
            }
        },
        scope: "regular"
    }, function() {});

    chrome.webRequest.onAuthRequired.addListener(
        function(details, callbackFn) {
            callbackFn({
                authCredentials: {
                    username: "%s",
                    password: "%s"
                }
            });
        },
        {urls: ["<all_urls>"]},
        ["asyncBlocking"]
    );
    """ % (host, port, username, password)

    return manifest, background_js

4.3 流量监控

在 GUI 里加了一个实时流量面板，用 QTimer 定时从数据库拉数据刷新：

class TrafficTab(QWidget):
    def __init__(self, db_manager):
        super().__init__()
        self.db = db_manager
        self.timer = QTimer()
        self.timer.timeout.connect(self.refresh)
        self.timer.start(2000)  # 每 2 秒刷新

    def refresh(self):
        stats = self.db.get_traffic_summary()
        for row in stats:
            self.table.setItem(row['id'], 1,
                QTableWidgetItem(f"{row['total']/1024/1024:.1f} MB"))
            # 根据流量使用率变色
            usage = row['total'] / row['quota']
            if usage > 0.9:
                self.table.item(row['id'], 1).setBackground(
                    QColor(255, 200, 200)
                )

五、踩过的坑

坑 1：PyQt6 的信号槽和线程

代理网关跑在后台线程里，想更新 GUI 状态的时候直接调 setText() 会崩。PyQt6 里子线程不能直接操作 GUI 组件，必须用信号槽：

## 正确姿势：用信号跨线程通信
class GatewaySignals(QObject):
    state_changed = pyqtSignal(int, str)  # account_id, new_state
    traffic_updated = pyqtSignal(int, int)  # account_id, bytes

## 在 gateway_service.py 里发信号
def _notify_traffic(self, account_id, bytes_count):
    self.signals.traffic_updated.emit(account_id, bytes_count)

## 在 GUI 里接收
self.signals.traffic_updated.connect(self.on_traffic_updated)

坑 2：SQLite 并发写

多实例同时运行的时候，多个线程同时写数据库会报 database is locked。解决办法是写操作加锁 + 设置 busy_timeout：

self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute("PRAGMA busy_timeout = 5000")
self._write_lock = threading.Lock()

def safe_write(self, sql, params=()):
    with self._write_lock:
        self.conn.execute(sql, params)
        self.conn.commit()

坑 3：Kookeey 的认证方式

原稿对 Kookeey 认证方式的描述待核对，不能概括为所有套餐通用规则，而是要先调 API 拿 token，然后用 token 做连接。试了好几种姿势才跑通：

## proxy_utils.py
import requests

def get_kookeey_token(username, password):
    """Kookeey 需要先获取 token 才能使用代理"""
    resp = requests.post(
        "https://api.kookeey.com/v1/auth",
        json={"username": username, "password": password},
        timeout=10
    )
    data = resp.json()
    return data["token"]

def test_proxy(token, host, port):
    """测试代理连通性"""
    proxies = {
        "http": f"http://{token}@{host}:{port}",
        "https": f"http://{token}@{host}:{port}",
    }
    try:
        resp = requests.get(
            "https://httpbin.org/ip", proxies=proxies, timeout=15
        )
        return resp.json()["origin"]
    except Exception as e:
        return str(e)

坑 4：PyQt6 打包

用 PyInstaller 打包的时候，PyQt6 的插件经常漏掉，导致运行时崩。最后是在 .spec 文件里手动加了：

## autoTools.spec
a = Analysis(
    ['main_gui.py'],
    pathex=[],
    binaries=[],
    datas=[
        ('chrome_extension/*', 'chrome_extension'),
    ],
    hiddenimports=[
        'PyQt6.QtCore',
        'PyQt6.QtWidgets',
        'PyQt6.QtGui',
    ],
)

六、一些感想

状态机真的好用

账号从 new → active → gateway → exhausted/error 的状态流转，用一个简单的状态机管理之后，逻辑清晰了很多。以前用一堆 if-else 判断账号状态，改需求的时候改到崩溃。

SQLite 对个人工具来说真的够了

一开始还想过去用 PostgreSQL，后来想想——这工具就我一个人用，数据量也不大，SQLite 完全够。PRAGMA journal_mode=WAL 打开之后，读写性能也还行。

GUI 工具的测试很头疼

PyQt6 的 GUI 测试一直是个痛点，最后是用了 pytest-qt 做了一些基本的自动化测试，但覆盖度有限。核心逻辑（gateway_service.py、proxy_manager.py）都抽出来做了单元测试，保证业务逻辑的正确性。

七、后续计划

这个项目目前算是 v1.0，该有的功能都有了。后续打算加：

定时自动切换 IP：根据设定的时间间隔自动轮换出口 IP
流量告警：接近配额上限时发通知
多平台支持：除了 Kookeey，支持更多代理服务商
Docker 部署：方便在服务器上跑

八、结语

写这个工具的过程中，最大的感受是：先跑起来，再优化。最初版本就是一堆 print 调通代理转发逻辑，然后逐步加 GUI、加数据库、加状态管理。如果一开始就想做"完美架构"，可能到现在还没写完。

另外，状态机、信号槽、线程安全这些"老生常谈"的东西，真正写的时候还是会踩坑。理论归理论，实践归实践。

代码放在私有仓库里，如果有人在用 Kookeey 并且有类似的需求，欢迎交流。

以上是 autoTools 项目的开发总结。我是菜牙，一个喜欢写代码的菜鸡，下次见。

代码节选边界

仅实现 do_GET 的处理器不是完整 HTTP/HTTPS 代理，缺少 CONNECT、请求体与头部处理。扩展使用的 webRequest 认证 API 需要与 Manifest 权限及版本一致，不能只声明 proxy 权限。SQLite 的跨线程连接仍需序列化访问，WAL 不代表允许多个并发写事务。第三方 API 路径与流量计数口径需核实。