AIstudioProxyAPI / simple_launcher.py
peijun1's picture
Deploy AI Studio Proxy API to Hugging Face Spaces
a5784e9
Raw
History Blame Contribute Delete
41.3 kB
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI Studio Proxy API - Simple GUI Launcher
Easy to use, modern interface launcher.
Features:
- Modern dark theme
- GNOME system tray support
- Account management (add, delete)
- Proxy settings
- Port configuration
- API test button
- Log saving
Usage:
poetry run python simple_launcher.py
"""
import json
import os
import platform
import signal
import socket
import subprocess
import sys
import threading
import time
import tkinter as tk
import webbrowser
from datetime import datetime
from pathlib import Path
from tkinter import messagebox, scrolledtext, simpledialog, ttk
from typing import Any, Dict, List, Optional
# Project directories
SCRIPT_DIR = Path(__file__).parent.absolute()
AUTH_PROFILES_DIR = SCRIPT_DIR / "auth_profiles"
SAVED_AUTH_DIR = AUTH_PROFILES_DIR / "saved"
ACTIVE_AUTH_DIR = AUTH_PROFILES_DIR / "active"
LAUNCH_SCRIPT = SCRIPT_DIR / "launch_camoufox.py"
CONFIG_FILE = SCRIPT_DIR / "simple_launcher_config.json"
LOG_FILE = SCRIPT_DIR / "logs" / "simple_launcher.log"
# Default settings
DEFAULT_CONFIG = {
"fastapi_port": 2048,
"camoufox_port": 9222,
"stream_port": 3120,
"proxy_address": "",
"proxy_enabled": False,
"last_account": "",
"dark_mode": True,
"minimize_to_tray": True,
}
# Modern Color Palette (Dark Theme)
COLORS = {
"bg_dark": "#1a1a2e",
"bg_medium": "#16213e",
"bg_light": "#0f3460",
"accent": "#e94560",
"accent_hover": "#ff6b6b",
"success": "#00d26a",
"warning": "#ffc107",
"error": "#dc3545",
"text_primary": "#ffffff",
"text_secondary": "#a0a0a0",
"border": "#2d2d44",
}
class ModernStyle:
"""Modern style manager"""
@staticmethod
def apply(root):
"""Apply dark theme styles"""
style = ttk.Style()
# Set theme
style.theme_use("clam")
# Frame styles
style.configure("TFrame", background=COLORS["bg_dark"])
style.configure("Card.TFrame", background=COLORS["bg_medium"])
# Label styles
style.configure(
"TLabel",
background=COLORS["bg_dark"],
foreground=COLORS["text_primary"],
font=("Segoe UI", 10),
)
style.configure(
"Header.TLabel",
background=COLORS["bg_dark"],
foreground=COLORS["text_primary"],
font=("Segoe UI", 14, "bold"),
)
style.configure(
"Status.TLabel",
background=COLORS["bg_dark"],
foreground=COLORS["success"],
font=("Segoe UI", 11, "bold"),
)
# LabelFrame styles
style.configure(
"TLabelframe",
background=COLORS["bg_medium"],
foreground=COLORS["text_primary"],
)
style.configure(
"TLabelframe.Label",
background=COLORS["bg_medium"],
foreground=COLORS["accent"],
font=("Segoe UI", 11, "bold"),
)
# Button styles
style.configure(
"TButton",
background=COLORS["bg_light"],
foreground=COLORS["text_primary"],
font=("Segoe UI", 10),
padding=(10, 5),
)
style.map(
"TButton",
background=[
("active", COLORS["accent"]),
("pressed", COLORS["accent_hover"]),
],
)
style.configure(
"Accent.TButton",
background=COLORS["accent"],
foreground=COLORS["text_primary"],
font=("Segoe UI", 10, "bold"),
)
# Entry styles
style.configure(
"TEntry",
fieldbackground=COLORS["bg_light"],
foreground=COLORS["text_primary"],
insertcolor=COLORS["text_primary"],
)
# Combobox styles
style.configure(
"TCombobox",
fieldbackground=COLORS["bg_light"],
background=COLORS["bg_light"],
foreground=COLORS["text_primary"],
arrowcolor=COLORS["text_primary"],
)
style.map(
"TCombobox",
fieldbackground=[("readonly", COLORS["bg_light"])],
selectbackground=[("readonly", COLORS["accent"])],
)
# Radiobutton styles - larger and more visible
style.configure(
"TRadiobutton",
background=COLORS["bg_medium"],
foreground=COLORS["text_primary"],
font=("Segoe UI", 11),
indicatorsize=20,
)
style.map(
"TRadiobutton",
indicatorcolor=[
("selected", COLORS["accent"]),
("!selected", COLORS["bg_light"]),
],
background=[("active", COLORS["bg_light"])],
)
# Checkbutton styles - larger and more visible
style.configure(
"TCheckbutton",
background=COLORS["bg_medium"],
foreground=COLORS["text_primary"],
font=("Segoe UI", 11),
indicatorsize=18,
)
style.map(
"TCheckbutton",
indicatorcolor=[
("selected", COLORS["accent"]),
("!selected", COLORS["bg_light"]),
],
background=[("active", COLORS["bg_light"])],
)
# Notebook styles
style.configure("TNotebook", background=COLORS["bg_dark"], borderwidth=0)
style.configure(
"TNotebook.Tab",
background=COLORS["bg_medium"],
foreground=COLORS["text_primary"],
padding=(15, 8),
font=("Segoe UI", 10),
)
style.map(
"TNotebook.Tab",
background=[("selected", COLORS["accent"])],
foreground=[("selected", COLORS["text_primary"])],
)
# Root widget background
root.configure(bg=COLORS["bg_dark"])
class TrayIcon:
"""GNOME system tray support - AppIndicator3 (Wayland) or pystray (X11)"""
def __init__(self, app):
self.app = app
self.indicator = None
self.supported = False
self.backend = None # "appindicator" or "pystray"
def create_icon(self):
"""Create tray icon - try AppIndicator3 first, then pystray"""
# 1. Try GNOME AppIndicator3 first (for Wayland)
if self._try_appindicator():
return
# 2. Then try pystray (for X11)
if self._try_pystray():
return
print("⚠️ System tray support not found. Tray disabled.")
def _try_appindicator(self) -> bool:
"""Try to create tray with AppIndicator3"""
try:
import gi
gi.require_version("Gtk", "3.0")
gi.require_version("AppIndicator3", "0.1")
from gi.repository import AppIndicator3, GLib, Gtk
# Create menu
menu = Gtk.Menu()
# Menu items
item_show = Gtk.MenuItem(label="📂 Show Window")
item_show.connect(
"activate", lambda w: GLib.idle_add(self.app._show_window)
)
menu.append(item_show)
menu.append(Gtk.SeparatorMenuItem())
item_start = Gtk.MenuItem(label="▶️ Start")
item_start.connect("activate", lambda w: GLib.idle_add(self.app._start))
menu.append(item_start)
item_stop = Gtk.MenuItem(label="⏹️ Stop")
item_stop.connect("activate", lambda w: GLib.idle_add(self.app._stop))
menu.append(item_stop)
menu.append(Gtk.SeparatorMenuItem())
item_test = Gtk.MenuItem(label="🔍 API Test")
item_test.connect("activate", lambda w: GLib.idle_add(self.app._api_test))
menu.append(item_test)
menu.append(Gtk.SeparatorMenuItem())
item_quit = Gtk.MenuItem(label="❌ Exit")
item_quit.connect(
"activate", lambda w: GLib.idle_add(self.app._close_completely)
)
menu.append(item_quit)
menu.show_all()
# Create AppIndicator
self.indicator = AppIndicator3.Indicator.new(
"aistudio-proxy",
"network-server", # Use system icon
AppIndicator3.IndicatorCategory.APPLICATION_STATUS,
)
self.indicator.set_status(AppIndicator3.IndicatorStatus.ACTIVE)
self.indicator.set_menu(menu)
self.indicator.set_title("AI Studio Proxy")
# Run GTK main loop in separate thread
def gtk_main():
try:
Gtk.main()
except Exception:
pass
threading.Thread(target=gtk_main, daemon=True).start()
self.supported = True
self.backend = "appindicator"
print("✅ GNOME AppIndicator3 tray started (Wayland compatible)")
return True
except Exception as e:
print(f"⚠️ AppIndicator3 could not be started: {e}")
return False
def _try_pystray(self) -> bool:
"""Try to create tray with pystray"""
try:
import pystray
from PIL import Image, ImageDraw
# Create a simple icon
size = 64
image = Image.new("RGBA", (size, size), (0, 0, 0, 0))
draw = ImageDraw.Draw(image)
draw.ellipse([4, 4, size - 4, size - 4], fill=COLORS["accent"])
draw.ellipse([16, 16, size - 16, size - 16], fill=COLORS["bg_dark"])
# Create menu
menu = pystray.Menu(
pystray.MenuItem("📂 Show Window", self._pystray_show),
pystray.Menu.SEPARATOR,
pystray.MenuItem("▶️ Start", self._pystray_start),
pystray.MenuItem("⏹️ Stop", self._pystray_stop),
pystray.Menu.SEPARATOR,
pystray.MenuItem("🔍 API Test", self._pystray_test),
pystray.Menu.SEPARATOR,
pystray.MenuItem("❌ Exit", self._pystray_quit),
)
self.indicator = pystray.Icon(
"AI Studio Proxy", image, "AI Studio Proxy", menu
)
threading.Thread(target=self.indicator.run, daemon=True).start()
self.supported = True
self.backend = "pystray"
print("✅ pystray tray started (X11)")
return True
except Exception as e:
print(f"⚠️ pystray could not be started: {e}")
return False
def _pystray_show(self, icon=None, item=None):
self.app.root.after(0, self.app._show_window)
def _pystray_start(self, icon=None, item=None):
self.app.root.after(0, self.app._start)
def _pystray_stop(self, icon=None, item=None):
self.app.root.after(0, self.app._stop)
def _pystray_test(self, icon=None, item=None):
self.app.root.after(0, self.app._api_test)
def _pystray_quit(self, icon=None, item=None):
self.app.root.after(0, self.app._close_completely)
def update_status(self, running: bool):
"""Update tray icon status"""
if not self.supported:
return
# Status update - can be extended later
def stop(self):
"""Stop tray icon"""
try:
if self.backend == "appindicator":
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
Gtk.main_quit()
elif self.backend == "pystray" and self.indicator:
self.indicator.stop()
except Exception:
pass
class SimpleGUILauncher:
"""Simple GUI Launcher"""
def __init__(self):
self.root = tk.Tk()
self.root.title("🚀 AI Studio Proxy API")
self.root.geometry("1050x700")
self.root.minsize(800, 500)
# Apply modern style
ModernStyle.apply(self.root)
# Load configuration
self.config = self._load_config()
# Variables
self.selected_account = tk.StringVar(value=self.config.get("last_account", ""))
self.run_mode = tk.StringVar(value="headless")
self.status = tk.StringVar(value="⚪ Ready")
self.fastapi_port = tk.StringVar(
value=str(self.config.get("fastapi_port", 2048))
)
self.stream_port = tk.StringVar(value=str(self.config.get("stream_port", 3120)))
self.proxy_address = tk.StringVar(value=self.config.get("proxy_address", ""))
self.proxy_enabled = tk.BooleanVar(
value=self.config.get("proxy_enabled", False)
)
self.process: Optional[subprocess.Popen] = None
self.log_thread: Optional[threading.Thread] = None
self.running = False
# Tray icon
self.tray = TrayIcon(self)
# Create log directory
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
# Create interface
self._create_interface()
# Load accounts
self._load_accounts()
# Start tray icon
if self.config.get("minimize_to_tray", True):
self.tray.create_icon()
# Close handler
self.root.protocol("WM_DELETE_WINDOW", self._minimize_to_tray)
def _load_config(self) -> Dict[str, Any]:
"""Load configuration"""
try:
if CONFIG_FILE.exists():
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
return {**DEFAULT_CONFIG, **json.load(f)}
except Exception as e:
print(f"⚠️ Configuration could not be loaded: {e}")
return DEFAULT_CONFIG.copy()
def _save_config(self):
"""Save configuration"""
try:
config = {
"fastapi_port": int(self.fastapi_port.get() or 2048),
"stream_port": int(self.stream_port.get() or 3120),
"proxy_address": self.proxy_address.get(),
"proxy_enabled": self.proxy_enabled.get(),
"last_account": self.selected_account.get(),
"dark_mode": True,
"minimize_to_tray": True,
}
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2, ensure_ascii=False)
except Exception as e:
self._log(f"⚠️ Configuration could not be saved: {e}")
def _create_interface(self):
"""Create modern interface"""
# Main container
main_container = ttk.Frame(self.root, padding="15")
main_container.pack(fill=tk.BOTH, expand=True)
# === HEADER ===
header_frame = ttk.Frame(main_container)
header_frame.pack(fill=tk.X, pady=(0, 15))
ttk.Label(
header_frame, text="🚀 AI Studio Proxy API", style="Header.TLabel"
).pack(side=tk.LEFT)
# Status indicator (top right)
self.status_label = ttk.Label(
header_frame, textvariable=self.status, style="Status.TLabel"
)
self.status_label.pack(side=tk.RIGHT)
# === NOTEBOOK (Tabs) ===
notebook = ttk.Notebook(main_container)
notebook.pack(fill=tk.BOTH, expand=True)
# Tab 1: Main Control
self.main_tab = ttk.Frame(notebook, padding="10")
notebook.add(self.main_tab, text="🎮 Control")
self._create_main_tab()
# Tab 2: Account Management
self.account_tab = ttk.Frame(notebook, padding="10")
notebook.add(self.account_tab, text="👤 Accounts")
self._create_account_tab()
# Tab 3: Settings
self.settings_tab = ttk.Frame(notebook, padding="10")
notebook.add(self.settings_tab, text="⚙️ Settings")
self._create_settings_tab()
# Tab 4: Logs
self.log_tab = ttk.Frame(notebook, padding="10")
notebook.add(self.log_tab, text="📋 Logs")
self._create_log_tab()
def _create_main_tab(self):
"""Main control tab"""
# Account selection (quick access)
quick_frame = ttk.LabelFrame(self.main_tab, text="⚡ Quick Start", padding="15")
quick_frame.pack(fill=tk.X, pady=(0, 15))
row1 = ttk.Frame(quick_frame)
row1.pack(fill=tk.X, pady=(0, 10))
ttk.Label(row1, text="Account:").pack(side=tk.LEFT)
self.account_combo = ttk.Combobox(
row1, textvariable=self.selected_account, state="readonly", width=30
)
self.account_combo.pack(side=tk.LEFT, padx=(10, 20))
ttk.Label(row1, text="Mode:").pack(side=tk.LEFT)
ttk.Radiobutton(
row1, text="Headless", variable=self.run_mode, value="headless"
).pack(side=tk.LEFT, padx=(10, 5))
ttk.Radiobutton(
row1, text="Visible", variable=self.run_mode, value="debug"
).pack(side=tk.LEFT)
# Large buttons
btn_frame = ttk.Frame(quick_frame)
btn_frame.pack(fill=tk.X, pady=(10, 0))
self.start_btn = tk.Button(
btn_frame,
text="▶️ START",
command=self._start,
bg=COLORS["success"],
fg="white",
font=("Segoe UI", 14, "bold"),
height=2,
width=15,
cursor="hand2",
activebackground=COLORS["accent"],
)
self.start_btn.pack(side=tk.LEFT, padx=(0, 10), expand=True, fill=tk.X)
self.stop_btn = tk.Button(
btn_frame,
text="⏹️ STOP",
command=self._stop,
bg=COLORS["error"],
fg="white",
font=("Segoe UI", 14, "bold"),
height=2,
width=15,
cursor="hand2",
state=tk.DISABLED,
activebackground=COLORS["accent_hover"],
)
self.stop_btn.pack(side=tk.LEFT, expand=True, fill=tk.X)
# Info cards
info_frame = ttk.Frame(self.main_tab)
info_frame.pack(fill=tk.X, pady=(15, 0))
# API Info
api_card = ttk.LabelFrame(info_frame, text="🌐 API Info", padding="10")
api_card.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0, 10))
self.api_url_label = ttk.Label(
api_card, text=f"http://127.0.0.1:{self.fastapi_port.get()}"
)
self.api_url_label.pack(anchor=tk.W)
api_btn_frame = ttk.Frame(api_card)
api_btn_frame.pack(fill=tk.X, pady=(10, 0))
ttk.Button(api_btn_frame, text="🔍 Test", command=self._api_test).pack(
side=tk.LEFT, padx=(0, 5)
)
ttk.Button(
api_btn_frame, text="🌐 Open in Browser", command=self._open_in_browser
).pack(side=tk.LEFT)
# Status card
status_card = ttk.LabelFrame(info_frame, text="📊 Status", padding="10")
status_card.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
self.status_details = ttk.Label(status_card, text="Service is stopped")
self.status_details.pack(anchor=tk.W)
self.pid_label = ttk.Label(status_card, text="PID: -")
self.pid_label.pack(anchor=tk.W)
def _create_account_tab(self):
"""Account management tab"""
# Account list
list_frame = ttk.LabelFrame(
self.account_tab, text="📋 Saved Accounts", padding="10"
)
list_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
# Listbox
self.account_listbox = tk.Listbox(
list_frame,
bg=COLORS["bg_light"],
fg=COLORS["text_primary"],
selectbackground=COLORS["accent"],
font=("Consolas", 11),
height=10,
)
self.account_listbox.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
# Account buttons
btn_frame = ttk.Frame(list_frame)
btn_frame.pack(fill=tk.X)
ttk.Button(
btn_frame, text="➕ Add New Account", command=self._add_new_account
).pack(side=tk.LEFT, padx=(0, 5))
ttk.Button(
btn_frame, text="🗑️ Delete Selected", command=self._delete_account
).pack(side=tk.LEFT, padx=(0, 5))
ttk.Button(btn_frame, text="🔄 Refresh", command=self._load_accounts).pack(
side=tk.LEFT
)
# Account details
detail_frame = ttk.LabelFrame(
self.account_tab, text="ℹ️ Account Details", padding="10"
)
detail_frame.pack(fill=tk.X)
self.account_detail = ttk.Label(
detail_frame, text="Select an account to see details"
)
self.account_detail.pack(anchor=tk.W)
# Listbox selection event
self.account_listbox.bind("<<ListboxSelect>>", self._account_selected)
def _create_settings_tab(self):
"""Settings tab"""
# Port settings
port_frame = ttk.LabelFrame(
self.settings_tab, text="🔌 Port Settings", padding="10"
)
port_frame.pack(fill=tk.X, pady=(0, 10))
row1 = ttk.Frame(port_frame)
row1.pack(fill=tk.X, pady=(0, 5))
ttk.Label(row1, text="FastAPI Port:").pack(side=tk.LEFT)
ttk.Entry(row1, textvariable=self.fastapi_port, width=10).pack(
side=tk.LEFT, padx=(10, 20)
)
ttk.Label(row1, text="Stream Port:").pack(side=tk.LEFT)
ttk.Entry(row1, textvariable=self.stream_port, width=10).pack(
side=tk.LEFT, padx=(10, 0)
)
# Proxy settings
proxy_frame = ttk.LabelFrame(
self.settings_tab, text="🌍 Proxy Settings", padding="10"
)
proxy_frame.pack(fill=tk.X, pady=(0, 10))
ttk.Checkbutton(
proxy_frame, text="Use Proxy", variable=self.proxy_enabled
).pack(anchor=tk.W)
proxy_row = ttk.Frame(proxy_frame)
proxy_row.pack(fill=tk.X, pady=(5, 0))
ttk.Label(proxy_row, text="Address:").pack(side=tk.LEFT)
ttk.Entry(proxy_row, textvariable=self.proxy_address, width=40).pack(
side=tk.LEFT, padx=(10, 0)
)
ttk.Label(
proxy_frame,
text="Example: http://127.0.0.1:7890",
foreground=COLORS["text_secondary"],
).pack(anchor=tk.W, pady=(5, 0))
# Save button
save_frame = ttk.Frame(self.settings_tab)
save_frame.pack(fill=tk.X, pady=(20, 0))
ttk.Button(
save_frame, text="💾 Save Settings", command=self._save_and_notify
).pack(side=tk.LEFT)
ttk.Button(
save_frame, text="🔄 Reset to Default", command=self._reset_config
).pack(side=tk.LEFT, padx=(10, 0))
def _create_log_tab(self):
"""Log tab"""
# Log area
self.log_area = scrolledtext.ScrolledText(
self.log_tab,
height=20,
state=tk.DISABLED,
font=("Consolas", 9),
bg=COLORS["bg_light"],
fg=COLORS["text_primary"],
insertbackground=COLORS["text_primary"],
)
self.log_area.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
# Log buttons
btn_frame = ttk.Frame(self.log_tab)
btn_frame.pack(fill=tk.X)
ttk.Button(btn_frame, text="🗑️ Clear", command=self._clear_logs).pack(
side=tk.LEFT, padx=(0, 5)
)
ttk.Button(btn_frame, text="💾 Save to File", command=self._save_logs).pack(
side=tk.LEFT, padx=(0, 5)
)
ttk.Button(
btn_frame, text="📂 Open Log File", command=self._open_log_file
).pack(side=tk.LEFT)
def _log(self, message: str, save_to_file: bool = True):
"""Add message to log area"""
timestamp = datetime.now().strftime("%H:%M:%S")
formatted = f"[{timestamp}] {message}"
self.log_area.config(state=tk.NORMAL)
self.log_area.insert(tk.END, f"{formatted}\n")
self.log_area.see(tk.END)
self.log_area.config(state=tk.DISABLED)
if save_to_file:
try:
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(f"{formatted}\n")
except Exception:
pass
def _clear_logs(self):
"""Clear log area"""
self.log_area.config(state=tk.NORMAL)
self.log_area.delete(1.0, tk.END)
self.log_area.config(state=tk.DISABLED)
def _save_logs(self):
"""Save logs to file"""
try:
log_content = self.log_area.get(1.0, tk.END)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
save_path = SCRIPT_DIR / "logs" / f"session_{timestamp}.log"
save_path.parent.mkdir(parents=True, exist_ok=True)
with open(save_path, "w", encoding="utf-8") as f:
f.write(log_content)
self._log(f"✅ Logs saved: {save_path.name}")
messagebox.showinfo("Success", f"Logs saved:\n{save_path}")
except Exception as e:
messagebox.showerror("Error", f"Logs could not be saved:\n{e}")
def _open_log_file(self):
"""Open log file"""
try:
if platform.system() == "Linux":
subprocess.Popen(["xdg-open", str(LOG_FILE.parent)])
elif platform.system() == "Darwin":
subprocess.Popen(["open", str(LOG_FILE.parent)])
else:
os.startfile(str(LOG_FILE.parent))
except Exception as e:
messagebox.showerror("Error", f"Folder could not be opened:\n{e}")
def _load_accounts(self):
"""Load saved accounts"""
accounts = []
if SAVED_AUTH_DIR.exists():
for file in sorted(SAVED_AUTH_DIR.glob("*.json")):
if file.name != ".gitkeep":
accounts.append(file.stem)
# Update combobox
self.account_combo["values"] = accounts
# Update listbox
self.account_listbox.delete(0, tk.END)
for acc in accounts:
self.account_listbox.insert(tk.END, f" 📧 {acc}")
# Select last account
last = self.config.get("last_account", "")
if last and last in accounts:
self.selected_account.set(last)
try:
idx = accounts.index(last)
self.account_listbox.selection_set(idx)
except Exception:
pass
elif accounts:
self.selected_account.set(accounts[0])
if accounts:
self._log(f"✅ {len(accounts)} account(s) loaded")
else:
self._log("⚠️ No saved accounts found")
def _account_selected(self, event):
"""Show details when account is selected"""
selection = self.account_listbox.curselection()
if not selection:
return
idx = selection[0]
account_name = self.account_listbox.get(idx).replace(" 📧 ", "")
# Read file info
auth_file = SAVED_AUTH_DIR / f"{account_name}.json"
if auth_file.exists():
stat = auth_file.stat()
mod_time = datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M")
size_kb = stat.st_size / 1024
self.account_detail.config(
text=f"📁 File: {account_name}.json\n"
f"📅 Last modified: {mod_time}\n"
f"📊 Size: {size_kb:.1f} KB"
)
# Update combobox as well
self.selected_account.set(account_name)
def _add_new_account(self):
"""Add new account"""
file_name = simpledialog.askstring(
"New Account",
"Enter a name for the account\n(e.g.: my_gmail_account):",
parent=self.root,
)
if not file_name:
return
if not file_name.replace("_", "").replace("-", "").isalnum():
messagebox.showerror("Error", "Only letters, numbers, - and _ are allowed!")
return
self._log(f"🔐 Adding new account: {file_name}")
self._log("📌 Browser will open, log in to your Google account")
self._log("📌 After logging in, account will be saved automatically")
self._start_internal(mode="debug", save_auth_as=file_name, exit_on_save=True)
def _delete_account(self):
"""Delete selected account"""
selection = self.account_listbox.curselection()
if not selection:
messagebox.showwarning("Warning", "Please select an account to delete")
return
idx = selection[0]
account_name = self.account_listbox.get(idx).replace(" 📧 ", "")
if not messagebox.askyesno(
"Confirm", f"Are you sure you want to delete '{account_name}'?"
):
return
try:
auth_file = SAVED_AUTH_DIR / f"{account_name}.json"
if auth_file.exists():
auth_file.unlink()
# Also delete from active
active_file = ACTIVE_AUTH_DIR / f"{account_name}.json"
if active_file.exists():
active_file.unlink()
self._log(f"🗑️ Account deleted: {account_name}")
self._load_accounts()
except Exception as e:
messagebox.showerror("Error", f"Account could not be deleted:\n{e}")
def _api_test(self):
"""Test the API"""
port = self.fastapi_port.get()
url = f"http://127.0.0.1:{port}/health"
self._log(f"🔍 Testing API: {url}")
try:
import urllib.request
with urllib.request.urlopen(url, timeout=5) as response:
if response.status == 200:
self._log("✅ API is running!")
self.status_details.config(text="✅ API is active and responding")
messagebox.showinfo("Success", "API is running! ✅")
else:
self._log(f"⚠️ API responded but status code: {response.status}")
except urllib.error.URLError:
self._log("❌ Could not connect to API. Service may not be running.")
self.status_details.config(text="❌ API is not responding")
messagebox.showwarning(
"Warning", "Could not connect to API.\nIs the service running?"
)
except Exception as e:
self._log(f"❌ API test error: {e}")
def _open_in_browser(self):
"""Open API in browser"""
port = self.fastapi_port.get()
webbrowser.open(f"http://127.0.0.1:{port}")
def _save_and_notify(self):
"""Save settings and notify"""
self._save_config()
self._log("💾 Settings saved")
messagebox.showinfo("Success", "Settings saved!")
# Update API URL
self.api_url_label.config(text=f"http://127.0.0.1:{self.fastapi_port.get()}")
def _reset_config(self):
"""Reset to default settings"""
if messagebox.askyesno(
"Confirm", "All settings will be reset to default. Continue?"
):
self.fastapi_port.set(str(DEFAULT_CONFIG["fastapi_port"]))
self.stream_port.set(str(DEFAULT_CONFIG["stream_port"]))
self.proxy_address.set("")
self.proxy_enabled.set(False)
self._save_config()
self._log("🔄 Settings reset to default")
def _is_port_in_use(self, port: int) -> bool:
"""Check if port is in use"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
try:
s.bind(("127.0.0.1", port))
return False
except OSError:
return True
def _find_port_pids(self, port: int) -> List[int]:
"""Find PIDs using the port"""
pids = []
try:
if platform.system() == "Linux":
result = subprocess.run(
["lsof", "-t", "-i", f":{port}"],
capture_output=True,
text=True,
timeout=5,
)
if result.stdout.strip():
pids = [int(p) for p in result.stdout.strip().split("\n") if p]
elif platform.system() == "Windows":
result = subprocess.run(
["netstat", "-ano", "-p", "TCP"],
capture_output=True,
text=True,
timeout=5,
)
for line in result.stdout.split("\n"):
if f":{port}" in line and "LISTENING" in line:
parts = line.split()
if parts:
try:
pids.append(int(parts[-1]))
except ValueError:
pass
except Exception as e:
self._log(f"⚠️ Could not find port PID: {e}")
return list(set(pids))
def _clean_ports(self) -> bool:
"""Clean ports"""
ports = [
int(self.fastapi_port.get() or 2048),
9222, # Camoufox
int(self.stream_port.get() or 3120),
]
cleaned = True
for port in ports:
if self._is_port_in_use(port):
self._log(f"🔍 Port {port} is in use, cleaning...")
pids = self._find_port_pids(port)
for pid in pids:
try:
if platform.system() == "Windows":
subprocess.run(
["taskkill", "/F", "/PID", str(pid)],
capture_output=True,
timeout=5,
)
else:
os.kill(pid, signal.SIGTERM)
time.sleep(0.5)
try:
os.kill(pid, 0)
os.kill(pid, signal.SIGKILL)
except ProcessLookupError:
pass
self._log(f" ✅ PID {pid} terminated")
except Exception as e:
self._log(f" ❌ PID {pid}: {e}")
cleaned = False
time.sleep(1)
if self._is_port_in_use(port):
self._log(f" ❌ Port {port} is still in use!")
cleaned = False
return cleaned
def _start(self):
"""Start the service"""
if self.running:
messagebox.showwarning("Warning", "Service is already running!")
return
account = self.selected_account.get()
if not account:
messagebox.showerror("Error", "Please select an account!")
return
# Save last account
self.config["last_account"] = account
self._save_config()
mode = self.run_mode.get()
self._start_internal(mode=mode, account=account)
def _start_internal(
self,
mode: str,
account: str = None,
save_auth_as: str = None,
exit_on_save: bool = False,
):
"""Internal start function"""
self._log("🔍 Checking ports...")
if not self._clean_ports():
if not messagebox.askyesno(
"Warning", "Some ports could not be cleaned. Continue?"
):
self._log("❌ User cancelled")
return
# Build command
cmd = [sys.executable, str(LAUNCH_SCRIPT)]
if mode == "headless":
cmd.append("--headless")
elif mode == "debug":
cmd.append("--debug")
# Port settings
cmd.extend(["--server-port", self.fastapi_port.get()])
cmd.extend(["--stream-port", self.stream_port.get()])
# Account
if account:
auth_file = SAVED_AUTH_DIR / f"{account}.json"
if auth_file.exists():
cmd.extend(["--active-auth-json", str(auth_file)])
# Save
if save_auth_as:
cmd.extend(["--save-auth-as", save_auth_as])
cmd.append("--auto-save-auth")
if exit_on_save:
cmd.append("--exit-on-auth-save")
# Proxy
if self.proxy_enabled.get() and self.proxy_address.get():
cmd.extend(["--internal-camoufox-proxy", self.proxy_address.get()])
self._log(f"🚀 Starting: {mode} mode")
# Environment variables
env = os.environ.copy()
env["DIRECT_LAUNCH"] = "true"
# When saving auth (new account), skip manual Enter prompt after login
# Login completion will be detected automatically by URL change
if save_auth_as:
env["SUPPRESS_LOGIN_WAIT"] = "true"
try:
self.process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
stdin=subprocess.PIPE if save_auth_as else None,
text=True,
bufsize=1,
cwd=str(SCRIPT_DIR),
env=env,
)
self.running = True
self.status.set("🟢 Running")
self.start_btn.config(state=tk.DISABLED)
self.stop_btn.config(state=tk.NORMAL)
self.status_details.config(text="Service started")
self.pid_label.config(text=f"PID: {self.process.pid}")
self.tray.update_status(True)
self.log_thread = threading.Thread(target=self._read_logs, daemon=True)
self.log_thread.start()
self._log(f"✅ Service started (PID: {self.process.pid})")
except Exception as e:
self._log(f"❌ Start error: {e}")
messagebox.showerror("Error", f"Could not start:\n{e}")
self.running = False
def _read_logs(self):
"""Log reading thread"""
try:
while self.process and self.process.poll() is None:
line = self.process.stdout.readline()
if line:
self.root.after(
0,
lambda log_line=line.strip(): self._log(
log_line, save_to_file=False
),
)
exit_code = self.process.returncode if self.process else -1
self.root.after(0, lambda: self._service_ended(exit_code))
except Exception as e:
self.root.after(0, lambda err=e: self._log(f"❌ Log error: {err}"))
def _service_ended(self, exit_code: int):
"""When service ends"""
self.running = False
self.process = None
if exit_code == 0:
self.status.set("⚪ Stopped")
self._log("✅ Service ended normally")
else:
self.status.set(f"🔴 Error ({exit_code})")
self._log(f"⚠️ Service stopped with error: {exit_code}")
self.start_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
self.status_details.config(text="Service is stopped")
self.pid_label.config(text="PID: -")
self.tray.update_status(False)
# Refresh account list (important after adding new account)
self._load_accounts()
self._log("🔄 Account list refreshed")
def _stop(self):
"""Stop the service"""
if not self.running or not self.process:
return
self._log("🛑 Stopping...")
self.status.set("🟡 Stopping...")
try:
if platform.system() == "Windows":
self.process.terminate()
else:
self.process.send_signal(signal.SIGINT)
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self._log("⚠️ Force closing...")
self.process.kill()
self.process.wait(timeout=3)
self._log("✅ Stopped")
except Exception as e:
self._log(f"❌ Stop error: {e}")
self._service_ended(0)
def _show_window(self):
"""Show window"""
self.root.deiconify()
self.root.lift()
self.root.focus_force()
def _minimize_to_tray(self):
"""Minimize to tray"""
if self.tray.supported and self.running:
self.root.withdraw()
self._log("📌 Minimized to system tray")
else:
self._close_completely()
def _close_completely(self):
"""Close completely"""
if self.running:
if messagebox.askyesno("Confirm", "Service is running. Stop and exit?"):
self._stop()
else:
return
self._save_config()
self.tray.stop()
self.root.destroy()
def run(self):
"""Run the application"""
self._log("🚀 AI Studio Proxy Simple Launcher ready")
self.root.mainloop()
def main():
"""Main function"""
SAVED_AUTH_DIR.mkdir(parents=True, exist_ok=True)
ACTIVE_AUTH_DIR.mkdir(parents=True, exist_ok=True)
app = SimpleGUILauncher()
app.run()
if __name__ == "__main__":
main()