#!/usr/bin/env python3 import re import tkinter as tk from tkinter import ttk, messagebox, simpledialog, scrolledtext import subprocess import os import sys import platform import threading import time import socket import signal from typing import List, Dict, Any, Optional, Tuple from urllib.parse import urlparse import shlex import logging import json import requests # 新增导入 from dotenv import load_dotenv # 加载 .env 文件 load_dotenv() # --- Configuration & Globals --- PYTHON_EXECUTABLE = sys.executable SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) LAUNCH_CAMOUFOX_PY = os.path.join(SCRIPT_DIR, "launch_camoufox.py") SERVER_PY_FILENAME = "server.py" # For context AUTH_PROFILES_DIR = os.path.join(SCRIPT_DIR, "auth_profiles") # 确保这些目录存在 ACTIVE_AUTH_DIR = os.path.join(AUTH_PROFILES_DIR, "active") SAVED_AUTH_DIR = os.path.join(AUTH_PROFILES_DIR, "saved") DEFAULT_FASTAPI_PORT = int(os.environ.get('DEFAULT_FASTAPI_PORT', '2048')) DEFAULT_CAMOUFOX_PORT_GUI = int(os.environ.get('DEFAULT_CAMOUFOX_PORT', '9222')) # 与 launch_camoufox.py 中的 DEFAULT_CAMOUFOX_PORT 一致 managed_process_info: Dict[str, Any] = { "popen": None, "service_name_key": None, "monitor_thread": None, "stdout_thread": None, "stderr_thread": None, "output_area": None, "fully_detached": False # 新增：标记进程是否完全独立 } # 添加按钮防抖机制 button_debounce_info: Dict[str, float] = {} def debounce_button(func_name: str, delay_seconds: float = 2.0): """ 按钮防抖装饰器，防止在指定时间内重复执行同一函数 """ def decorator(func): def wrapper(*args, **kwargs): import time current_time = time.time() last_call_time = button_debounce_info.get(func_name, 0) if current_time - last_call_time < delay_seconds: logger.info(f"按钮防抖：忽略 {func_name} 的重复调用") return button_debounce_info[func_name] = current_time return func(*args, **kwargs) return wrapper return decorator # 添加全局logger定义 logger = logging.getLogger("GUILauncher") logger.setLevel(logging.INFO) console_handler = logging.StreamHandler() console_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) logger.addHandler(console_handler) os.makedirs(os.path.join(SCRIPT_DIR, "logs"), exist_ok=True) file_handler = logging.FileHandler(os.path.join(SCRIPT_DIR, "logs", "gui_launcher.log"), encoding='utf-8') file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) logger.addHandler(file_handler) # 在LANG_TEXTS声明之前定义长文本 service_closing_guide_message_zh = """由于服务在独立终端中运行，您可以通过以下方式关闭服务： 1. 使用端口管理功能: - 点击"查询端口进程"按钮 - 选择相关的Python进程 - 点击"停止选中进程" 2. 手动终止进程: - Windows: 使用任务管理器 - macOS: 使用活动监视器或terminal - Linux: 使用kill命令 3. 直接关闭服务运行的终端窗口""" service_closing_guide_message_en = """Since the service runs in an independent terminal, you can close it using these methods: 1. Using port management in GUI: - Click "Query Port Processes" button - Select the relevant Python process - Click "Stop Selected Process" 2. Manually terminate process: - Windows: Use Task Manager - macOS: Use Activity Monitor or terminal - Linux: Use kill command 3. Directly close the terminal window running the service""" # --- Internationalization (i18n) --- LANG_TEXTS = { "title": {"zh": "AI Studio Proxy API Launcher GUI", "en": "AI Studio Proxy API Launcher GUI"}, "status_idle": {"zh": "空闲，请选择操作。", "en": "Idle. Select an action."}, "port_section_label": {"zh": "服务端口配置", "en": "Service Port Configuration"}, "port_input_description_lbl": {"zh": "提示: 启动时将使用下方指定的FastAPI服务端口和Camoufox调试端口。", "en": "Note: The FastAPI service port and Camoufox debug port specified below will be used for launch."}, "fastapi_port_label": {"zh": "FastAPI 服务端口:", "en": "FastAPI Port:"}, "camoufox_debug_port_label": {"zh": "Camoufox 调试端口:", "en": "Camoufox Debug Port:"}, "query_pids_btn": {"zh": "查询端口进程", "en": "Query Port Processes"}, "stop_selected_pid_btn": {"zh": "停止选中进程", "en": "Stop Selected Process"}, "pids_on_port_label": {"zh": "端口占用情况 (PID - 名称):", "en": "Processes on Port (PID - Name):"}, # Static version for initialization "pids_on_port_label_dynamic": {"zh": "端口 {port} 占用情况 (PID - 名称):", "en": "Processes on Port {port} (PID - Name):"}, # Dynamic version "no_pids_found": {"zh": "未找到占用该端口的进程。", "en": "No processes found on this port."}, "static_pid_list_title": {"zh": "启动所需端口占用情况 (PID - 名称)", "en": "Required Ports Usage (PID - Name)"}, # 新增标题 "launch_options_label": {"zh": "启动选项", "en": "Launch Options"}, "launch_options_note_revised": {"zh": "提示：有头/无头模式均会在新的独立终端窗口中启动服务。\n有头模式用于调试和认证。无头模式需预先认证。\n关闭此GUI不会停止已独立启动的服务。", "en": "Tip: Headed/Headless modes will launch the service in a new independent terminal window.\nHeaded mode is for debug and auth. Headless mode requires pre-auth.\nClosing this GUI will NOT stop independently launched services."}, "launch_headed_interactive_btn": {"zh": "启动有头模式 (新终端)", "en": "Launch Headed Mode (New Terminal)"}, "launch_headless_btn": {"zh": "启动无头模式 (新终端)", "en": "Launch Headless Mode (New Terminal)"}, "launch_virtual_display_btn": {"zh": "启动虚拟显示模式 (Linux)", "en": "Launch Virtual Display (Linux)"}, "stop_gui_service_btn": {"zh": "停止当前GUI管理的服务", "en": "Stop Current GUI-Managed Service"}, "status_label": {"zh": "状态", "en": "Status"}, "output_label": {"zh": "输出日志", "en": "Output Log"}, "menu_language_fixed": {"zh": "Language", "en": "Language"}, "menu_lang_zh_option": {"zh": "中文 (Chinese)", "en": "中文 (Chinese)"}, "menu_lang_en_option": {"zh": "英文 (English)", "en": "英文 (English)"}, "confirm_quit_title": {"zh": "确认退出", "en": "Confirm Quit"}, "confirm_quit_message": {"zh": "服务可能仍在独立终端中运行。确认退出GUI吗?", "en": "Services may still be running in independent terminals. Confirm quit GUI?"}, "confirm_quit_message_independent": {"zh": "独立后台服务 '{service_name}' 可能仍在运行。直接退出GUI吗 (服务将继续运行)?", "en": "Independent background service '{service_name}' may still be running. Quit GUI (service will continue to run)?"}, "error_title": {"zh": "错误", "en": "Error"}, "info_title": {"zh": "信息", "en": "Info"}, "warning_title": {"zh": "警告", "en": "Warning"}, "service_already_running": {"zh": "服务 ({service_name}) 已在运行。", "en": "A service ({service_name}) is already running."}, "proxy_config_title": {"zh": "代理配置", "en": "Proxy Configuration"}, "proxy_config_message_generic": {"zh": "是否为此启动启用 HTTP/HTTPS 代理?", "en": "Enable HTTP/HTTPS proxy for this launch?"}, "proxy_address_title": {"zh": "代理地址", "en": "Proxy Address"}, "proxy_address_prompt": {"zh": "输入代理地址 (例如 http://host:port)\n默认: {default_proxy}", "en": "Enter proxy address (e.g., http://host:port)\nDefault: {default_proxy}"}, "proxy_configured_status": {"zh": "代理已配置: {proxy_addr}", "en": "Proxy configured: {proxy_addr}"}, "proxy_skip_status": {"zh": "用户跳过代理设置。", "en": "Proxy setup skipped by user."}, "script_not_found_error_msgbox": {"zh": "启动失败: 未找到 Python 执行文件或脚本。\n命令: {cmd}", "en": "Failed to start: Python executable or script not found.\nCommand: {cmd}"}, "startup_error_title": {"zh": "启动错误", "en": "Startup Error"}, "startup_script_not_found_msgbox": {"zh": "必需的脚本 '{script}' 在当前目录未找到。\n请将此GUI启动器与 launch_camoufox.py 和 server.py 放在同一目录。", "en": "Required script '{script}' not found in the current directory.\nPlace this GUI launcher in the same directory as launch_camoufox.py and server.py."}, "service_starting_status": {"zh": "{service_name} 启动中... PID: {pid}", "en": "{service_name} starting... PID: {pid}"}, "service_stopped_gracefully_status": {"zh": "{service_name} 已平稳停止。", "en": "{service_name} stopped gracefully."}, "service_stopped_exit_code_status": {"zh": "{service_name} 已停止。退出码: {code}", "en": "{service_name} stopped. Exit code: {code}"}, "service_stop_fail_status": {"zh": "{service_name} (PID: {pid}) 未能平稳终止。正在强制停止...", "en": "{service_name} (PID: {pid}) did not terminate gracefully. Forcing kill..."}, "service_killed_status": {"zh": "{service_name} (PID: {pid}) 已被强制停止。", "en": "{service_name} (PID: {pid}) killed."}, "error_stopping_service_msgbox": {"zh": "停止 {service_name} (PID: {pid}) 时出错: {e}", "en": "Error stopping {service_name} (PID: {pid}): {e}"}, "no_service_running_status": {"zh": "当前没有GUI管理的服务在运行。", "en": "No GUI-managed service is currently running."}, "stopping_initiated_status": {"zh": "{service_name} (PID: {pid}) 停止已启动。最终状态待定。", "en": "{service_name} (PID: {pid}) stopping initiated. Final status pending."}, "service_name_headed_interactive": {"zh": "有头交互服务", "en": "Headed Interactive Service"}, "service_name_headless": {"zh": "无头服务", "en": "Headless Service"}, # Key 修改 "service_name_virtual_display": {"zh": "虚拟显示无头服务", "en": "Virtual Display Headless Service"}, "status_headed_launch": {"zh": "有头模式：启动中，请关注新控制台的提示...", "en": "Headed Mode: Launching, check new console for prompts..."}, "status_headless_launch": {"zh": "无头服务：启动中...新的独立终端将打开。", "en": "Headless Service: Launching... A new independent terminal will open."}, "status_virtual_display_launch": {"zh": "虚拟显示模式启动中...", "en": "Virtual Display Mode launching..."}, "info_service_is_independent": {"zh": "当前服务为独立后台进程，关闭GUI不会停止它。请使用系统工具或端口管理手动停止此服务。", "en": "The current service is an independent background process. Closing the GUI will not stop it. Please manage this service manually using system tools or port management."}, "info_service_new_terminal": {"zh": "服务已在新的独立终端启动。关闭此GUI不会影响该服务。", "en": "Service has been launched in a new independent terminal. Closing this GUI will not affect the service."}, "warn_cannot_stop_independent_service": {"zh": "通过此GUI启动的服务在独立终端中运行，无法通过此按钮停止。请直接管理其终端或使用系统工具。", "en": "Services launched via this GUI run in independent terminals and cannot be stopped by this button. Please manage their terminals directly or use system tools."}, "enter_valid_port_warn": {"zh": "请输入有效的端口号 (1024-65535)。", "en": "Please enter a valid port number (1024-65535)."}, "pid_list_empty_for_stop_warn": {"zh": "进程列表为空或未选择进程。", "en": "PID list is empty or no process selected."}, "confirm_stop_pid_title": {"zh": "确认停止进程", "en": "Confirm Stop Process"}, "confirm_stop_pid_message": {"zh": "确定要尝试停止 PID {pid} ({name}) 吗?", "en": "Are you sure you want to attempt to stop PID {pid} ({name})?"}, "confirm_stop_pid_admin_title": {"zh": "以管理员权限停止进程", "en": "Stop Process with Admin Privileges"}, "confirm_stop_pid_admin_message": {"zh": "以普通权限停止 PID {pid} ({name}) 可能失败。是否尝试使用管理员权限停止?", "en": "Stopping PID {pid} ({name}) with normal privileges may fail. Try with admin privileges?"}, "admin_stop_success": {"zh": "已成功使用管理员权限停止 PID {pid}", "en": "Successfully stopped PID {pid} with admin privileges"}, "admin_stop_failure": {"zh": "使用管理员权限停止 PID {pid} 失败: {error}", "en": "Failed to stop PID {pid} with admin privileges: {error}"}, "status_error_starting": {"zh": "启动 {service_name} 失败。", "en": "Error starting {service_name}"}, "status_script_not_found": {"zh": "错误: 未找到 {service_name} 的可执行文件/脚本。", "en": "Error: Executable/script not found for {service_name}."}, "error_getting_process_name": {"zh": "获取 PID {pid} 的进程名失败。", "en": "Failed to get process name for PID {pid}."}, "pid_info_format": {"zh": "PID: {pid} (端口: {port}) - 名称: {name}", "en": "PID: {pid} (Port: {port}) - Name: {name}"}, "status_stopping_service": {"zh": "正在停止 {service_name} (PID: {pid})...", "en": "Stopping {service_name} (PID: {pid})..."}, "error_title_invalid_selection": {"zh": "无效的选择格式: {selection}", "en": "Invalid selection format: {selection}"}, "error_parsing_pid": {"zh": "无法从 '{selection}' 解析PID。", "en": "Could not parse PID from '{selection}'."}, "terminate_request_sent": {"zh": "终止请求已发送。", "en": "Termination request sent."}, "terminate_attempt_failed": {"zh": "尝试终止 PID {pid} ({name}) 可能失败。", "en": "Attempt to terminate PID {pid} ({name}) may have failed."}, "unknown_process_name_placeholder": {"zh": "未知进程名", "en": "Unknown Process Name"}, "kill_custom_pid_label": {"zh": "或输入PID终止:", "en": "Or Enter PID to Kill:"}, "kill_custom_pid_btn": {"zh": "终止指定PID", "en": "Kill Specified PID"}, "pid_input_empty_warn": {"zh": "请输入要终止的PID。", "en": "Please enter a PID to kill."}, "pid_input_invalid_warn": {"zh": "输入的PID无效，请输入纯数字。", "en": "Invalid PID entered. Please enter numbers only."}, "confirm_kill_custom_pid_title": {"zh": "确认终止PID", "en": "Confirm Kill PID"}, "status_sending_sigint": {"zh": "正在向 {service_name} (PID: {pid}) 发送 SIGINT...", "en": "Sending SIGINT to {service_name} (PID: {pid})..."}, "status_waiting_after_sigint": {"zh": "{service_name} (PID: {pid})：SIGINT 已发送，等待 {timeout} 秒优雅退出...", "en": "{service_name} (PID: {pid}): SIGINT sent, waiting {timeout}s for graceful exit..."}, "status_sigint_effective": {"zh": "{service_name} (PID: {pid}) 已响应 SIGINT 并停止。", "en": "{service_name} (PID: {pid}) responded to SIGINT and stopped."}, "status_sending_sigterm": {"zh": "{service_name} (PID: {pid})：未在规定时间内响应 SIGINT，正在发送 SIGTERM...", "en": "{service_name} (PID: {pid}): Did not respond to SIGINT in time, sending SIGTERM..."}, "status_waiting_after_sigterm": {"zh": "{service_name} (PID: {pid})：SIGTERM 已发送，等待 {timeout} 秒优雅退出...", "en": "{service_name} (PID: {pid}): SIGTERM sent, waiting {timeout}s for graceful exit..."}, "status_sigterm_effective": {"zh": "{service_name} (PID: {pid}) 已响应 SIGTERM 并停止。", "en": "{service_name} (PID: {pid}) responded to SIGTERM and stopped."}, "status_forcing_kill": {"zh": "{service_name} (PID: {pid})：未在规定时间内响应 SIGTERM，正在强制终止 (SIGKILL)...", "en": "{service_name} (PID: {pid}): Did not respond to SIGTERM in time, forcing kill (SIGKILL)..."}, "enable_stream_proxy_label": {"zh": "启用流式代理服务", "en": "Enable Stream Proxy Service"}, "stream_proxy_port_label": {"zh": "流式代理端口:", "en": "Stream Proxy Port:"}, "enable_helper_label": {"zh": "启用外部Helper服务", "en": "Enable External Helper Service"}, "helper_endpoint_label": {"zh": "Helper端点URL:", "en": "Helper Endpoint URL:"}, "auth_manager_title": {"zh": "认证文件管理", "en": "Authentication File Manager"}, "saved_auth_files_label": {"zh": "已保存的认证文件:", "en": "Saved Authentication Files:"}, "no_file_selected": {"zh": "请选择一个认证文件", "en": "Please select an authentication file"}, "auth_file_activated": {"zh": "认证文件 '{file}' 已成功激活", "en": "Authentication file '{file}' has been activated successfully"}, "error_activating_file": {"zh": "激活文件 '{file}' 时出错: {error}", "en": "Error activating file '{file}': {error}"}, "activate_selected_btn": {"zh": "激活选中的文件", "en": "Activate Selected File"}, "cancel_btn": {"zh": "取消", "en": "Cancel"}, "auth_files_management": {"zh": "认证文件管理", "en": "Auth Files Management"}, "manage_auth_files_btn": {"zh": "管理认证文件", "en": "Manage Auth Files"}, "no_saved_auth_files": {"zh": "保存目录中没有认证文件", "en": "No authentication files in saved directory"}, "auth_dirs_missing": {"zh": "认证目录不存在，请确保目录结构正确", "en": "Authentication directories missing, please ensure correct directory structure"}, "confirm_kill_port_title": {"zh": "确认清理端口", "en": "Confirm Port Cleanup"}, "confirm_kill_port_message": {"zh": "端口 {port} 被以下PID占用: {pids}。是否尝试终止这些进程?", "en": "Port {port} is in use by PID(s): {pids}. Try to terminate them?"}, "port_cleared_success": {"zh": "端口 {port} 已成功清理", "en": "Port {port} has been cleared successfully"}, "port_still_in_use": {"zh": "端口 {port} 仍被占用，请手动处理", "en": "Port {port} is still in use, please handle manually"}, "port_in_use_no_pids": {"zh": "端口 {port} 被占用，但无法识别进程", "en": "Port {port} is in use, but processes cannot be identified"}, "error_removing_file": {"zh": "删除文件 '{file}' 时出错: {error}", "en": "Error removing file '{file}': {error}"}, "stream_port_out_of_range": {"zh": "流式代理端口必须为0(禁用)或1024-65535之间的值", "en": "Stream proxy port must be 0 (disabled) or a value between 1024-65535"}, "port_auto_check": {"zh": "启动前自动检查端口", "en": "Auto-check port before launch"}, "auto_port_check_enabled": {"zh": "已启用端口自动检查", "en": "Port auto-check enabled"}, "port_check_running": {"zh": "正在检查端口 {port}...", "en": "Checking port {port}..."}, "port_name_fastapi": {"zh": "FastAPI服务", "en": "FastAPI Service"}, "port_name_camoufox_debug": {"zh": "Camoufox调试", "en": "Camoufox Debug"}, "port_name_stream_proxy": {"zh": "流式代理", "en": "Stream Proxy"}, "checking_port_with_name": {"zh": "正在检查{port_name}端口 {port}...", "en": "Checking {port_name} port {port}..."}, "port_check_all_completed": {"zh": "所有端口检查完成", "en": "All port checks completed"}, "port_check_failed": {"zh": "{port_name}端口 {port} 检查失败，启动已中止", "en": "{port_name} port {port} check failed, launch aborted"}, "port_name_helper_service": {"zh": "Helper服务", "en": "Helper Service"}, "confirm_kill_multiple_ports_title": {"zh": "确认清理多个端口", "en": "Confirm Multiple Ports Cleanup"}, "confirm_kill_multiple_ports_message": {"zh": "以下端口被占用:\n{occupied_ports_details}\n是否尝试终止这些进程?", "en": "The following ports are in use:\n{occupied_ports_details}\nAttempt to terminate these processes?"}, "all_ports_cleared_success": {"zh": "所有选定端口已成功清理。", "en": "All selected ports have been cleared successfully."}, "some_ports_still_in_use": {"zh": "部分端口在清理后仍被占用，请手动处理。启动已中止。", "en": "Some ports are still in use after cleanup attempt. Please handle manually. Launch aborted."}, "port_check_user_declined_cleanup": {"zh": "用户选择不清理占用的端口，启动已中止。", "en": "User chose not to clean up occupied ports. Launch aborted."}, "reset_button": {"zh": "重置为默认设置", "en": "Reset to Defaults"}, "confirm_reset_title": {"zh": "确认重置", "en": "Confirm Reset"}, "confirm_reset_message": {"zh": "确定要重置所有设置为默认值吗？", "en": "Are you sure you want to reset all settings to default values?"}, "reset_success": {"zh": "已重置为默认设置", "en": "Reset to default settings successfully"}, "proxy_config_last_used": {"zh": "使用上次的代理: {proxy}", "en": "Using last proxy: {proxy}"}, "proxy_config_other": {"zh": "使用其他代理地址", "en": "Use a different proxy address"}, "service_closing_guide": {"zh": "关闭服务指南", "en": "Service Closing Guide"}, "service_closing_guide_btn": {"zh": "如何关闭服务?", "en": "How to Close Service?"}, "service_closing_guide_message": {"zh": service_closing_guide_message_zh, "en": service_closing_guide_message_en}, "enable_proxy_label": {"zh": "启用浏览器代理", "en": "Enable Browser Proxy"}, "proxy_address_label": {"zh": "代理地址:", "en": "Proxy Address:"}, "current_auth_file_display_label": {"zh": "当前认证: ", "en": "Current Auth: "}, "current_auth_file_none": {"zh": "无", "en": "None"}, "current_auth_file_selected_format": {"zh": "{file}", "en": "{file}"}, "test_proxy_btn": {"zh": "测试", "en": "Test"}, "proxy_section_label": {"zh": "代理配置", "en": "Proxy Configuration"}, "proxy_test_url_default": "http://httpbin.org/get", # 默认测试URL "proxy_test_url_backup": "http://www.google.com", # 备用测试URL "proxy_not_enabled_warn": {"zh": "代理未启用或地址为空，请先配置。", "en": "Proxy not enabled or address is empty. Please configure first."}, "proxy_test_success": {"zh": "代理连接成功 ({url})", "en": "Proxy connection successful ({url})"}, "proxy_test_failure": {"zh": "代理连接失败 ({url}):\n{error}", "en": "Proxy connection failed ({url}):\n{error}"}, "proxy_testing_status": {"zh": "正在测试代理 {proxy_addr}...", "en": "Testing proxy {proxy_addr}..."}, "proxy_test_success_status": {"zh": "代理测试成功 ({url})", "en": "Proxy test successful ({url})"}, "proxy_test_failure_status": {"zh": "代理测试失败: {error}", "en": "Proxy test failed: {error}"}, "proxy_test_retrying": {"zh": "代理测试失败，正在重试 ({attempt}/{max_attempts})...", "en": "Proxy test failed, retrying ({attempt}/{max_attempts})..."}, "proxy_test_backup_url": {"zh": "主测试URL失败，尝试备用URL...", "en": "Primary test URL failed, trying backup URL..."}, "proxy_test_all_failed": {"zh": "所有代理测试尝试均失败", "en": "All proxy test attempts failed"}, "querying_ports_status": {"zh": "正在查询端口: {ports_desc}...", "en": "Querying ports: {ports_desc}..."}, "port_query_result_format": {"zh": "[{port_type} - {port_num}] {pid_info}", "en": "[{port_type} - {port_num}] {pid_info}"}, "port_not_in_use_format": {"zh": "[{port_type} - {port_num}] 未被占用", "en": "[{port_type} - {port_num}] Not in use"}, "pids_on_multiple_ports_label": {"zh": "多端口占用情况:", "en": "Multi-Port Usage:"}, "launch_llm_service_btn": {"zh": "启动本地LLM模拟服务", "en": "Launch Local LLM Mock Service"}, "stop_llm_service_btn": {"zh": "停止本地LLM模拟服务", "en": "Stop Local LLM Mock Service"}, "llm_service_name_key": {"zh": "本地LLM模拟服务", "en": "Local LLM Mock Service"}, "status_llm_starting": {"zh": "本地LLM模拟服务启动中 (PID: {pid})...", "en": "Local LLM Mock Service starting (PID: {pid})..."}, "status_llm_stopped": {"zh": "本地LLM模拟服务已停止。", "en": "Local LLM Mock Service stopped."}, "status_llm_stop_error": {"zh": "停止本地LLM模拟服务时出错。", "en": "Error stopping Local LLM Mock Service."}, "status_llm_already_running": {"zh": "本地LLM模拟服务已在运行 (PID: {pid})。", "en": "Local LLM Mock Service is already running (PID: {pid})."}, "status_llm_not_running": {"zh": "本地LLM模拟服务未在运行。", "en": "Local LLM Mock Service is not running."}, "status_llm_backend_check": {"zh": "正在检查LLM后端服务 ...", "en": "Checking LLM backend service ..."}, "status_llm_backend_ok_starting": {"zh": "LLM后端服务 (localhost:{port}) 正常，正在启动模拟服务...", "en": "LLM backend service (localhost:{port}) OK, starting mock service..."}, "status_llm_backend_fail": {"zh": "LLM后端服务 (localhost:{port}) 未响应，无法启动模拟服务。", "en": "LLM backend service (localhost:{port}) not responding, cannot start mock service."}, "confirm_stop_llm_title": {"zh": "确认停止LLM服务", "en": "Confirm Stop LLM Service"}, "confirm_stop_llm_message": {"zh": "确定要停止本地LLM模拟服务吗?", "en": "Are you sure you want to stop the Local LLM Mock Service?"} } # 删除重复的定义 current_language = 'zh' root_widget: Optional[tk.Tk] = None process_status_text_var: Optional[tk.StringVar] = None port_entry_var: Optional[tk.StringVar] = None # 将用于 FastAPI 端口 camoufox_debug_port_var: Optional[tk.StringVar] = None pid_listbox_widget: Optional[tk.Listbox] = None custom_pid_entry_var: Optional[tk.StringVar] = None widgets_to_translate: List[Dict[str, Any]] = [] proxy_address_var: Optional[tk.StringVar] = None # 添加变量存储代理地址 proxy_enabled_var: Optional[tk.BooleanVar] = None # 添加变量标记代理是否启用 active_auth_file_display_var: Optional[tk.StringVar] = None # 用于显示当前认证文件 LLM_PY_FILENAME = "llm.py" llm_service_process_info: Dict[str, Any] = { "popen": None, "monitor_thread": None, "stdout_thread": None, "stderr_thread": None, "service_name_key": "llm_service_name_key" # Corresponds to a LANG_TEXTS key } # 将所有辅助函数定义移到 build_gui 之前 def get_text(key: str, **kwargs) -> str: try: text_template = LANG_TEXTS[key][current_language] except KeyError: text_template = LANG_TEXTS[key].get('en', f"<{key}_MISSING_{current_language}>") return text_template.format(**kwargs) if kwargs else text_template def update_status_bar(message_key: str, **kwargs): message = get_text(message_key, **kwargs) def _perform_gui_updates(): # Update the status bar label's text variable if process_status_text_var: process_status_text_var.set(message) # Update the main log text area (if it exists) if managed_process_info.get("output_area"): # The 'message' variable is captured from the outer scope (closure) if root_widget: # Ensure root_widget is still valid output_area_widget = managed_process_info["output_area"] output_area_widget.config(state=tk.NORMAL) output_area_widget.insert(tk.END, f"[STATUS] {message}\n") output_area_widget.see(tk.END) output_area_widget.config(state=tk.DISABLED) if root_widget: root_widget.after_idle(_perform_gui_updates) def is_port_in_use(port: int) -> bool: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("0.0.0.0", port)) return False except OSError: return True except Exception: return True def get_process_name_by_pid(pid: int) -> str: system = platform.system() name = get_text("unknown_process_name_placeholder") cmd_args = [] try: if system == "Windows": cmd_args = ["tasklist", "/NH", "/FO", "CSV", "/FI", f"PID eq {pid}"] process = subprocess.run(cmd_args, capture_output=True, text=True, check=True, timeout=3, creationflags=subprocess.CREATE_NO_WINDOW) if process.stdout.strip(): parts = process.stdout.strip().split('","') if len(parts) > 0: name = parts[0].strip('"') elif system == "Linux": cmd_args = ["ps", "-p", str(pid), "-o", "comm="] process = subprocess.run(cmd_args, capture_output=True, text=True, check=True, timeout=3) if process.stdout.strip(): name = process.stdout.strip() elif system == "Darwin": cmd_args = ["ps", "-p", str(pid), "-o", "comm="] process = subprocess.run(cmd_args, capture_output=True, text=True, check=True, timeout=3) raw_path = process.stdout.strip() if process.stdout.strip() else "" cmd_args = ["ps", "-p", str(pid), "-o", "command="] process = subprocess.run(cmd_args, capture_output=True, text=True, check=True, timeout=3) if raw_path: base_name = os.path.basename(raw_path) name = f"{base_name} ({raw_path})" except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError): pass except Exception: pass return name def find_processes_on_port(port: int) -> List[Dict[str, Any]]: process_details = [] pids_only: List[int] = [] system = platform.system() command_pid = "" try: if system == "Linux" or system == "Darwin": command_pid = f"lsof -ti tcp:{port} -sTCP:LISTEN" process = subprocess.Popen(command_pid, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, universal_newlines=True, close_fds=True) stdout_pid, _ = process.communicate(timeout=5) if process.returncode == 0 and stdout_pid: pids_only = [int(p) for p in stdout_pid.strip().splitlines() if p.isdigit()] elif system == "Windows": command_pid = 'netstat -ano -p TCP' process = subprocess.Popen(command_pid, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, universal_newlines=True, creationflags=subprocess.CREATE_NO_WINDOW) stdout_pid, _ = process.communicate(timeout=10) if process.returncode == 0 and stdout_pid: for line in stdout_pid.strip().splitlines(): parts = line.split() if len(parts) >= 5 and parts[0].upper() == 'TCP': if parts[3].upper() != 'LISTENING': continue local_address_full = parts[1] try: last_colon_idx = local_address_full.rfind(':') if last_colon_idx == -1: continue extracted_port_str = local_address_full[last_colon_idx+1:] if extracted_port_str.isdigit() and int(extracted_port_str) == port: pid_str = parts[4] if pid_str.isdigit(): pids_only.append(int(pid_str)) except (ValueError, IndexError): continue pids_only = list(set(pids_only)) except Exception: pass for pid_val in pids_only: name = get_process_name_by_pid(pid_val) process_details.append({"pid": pid_val, "name": name}) return process_details def kill_process_pid(pid: int) -> bool: system = platform.system() success = False logger.info(f"Attempting to kill PID {pid} with normal privileges on {system}") try: if system == "Linux" or system == "Darwin": # 1. Attempt SIGTERM (best effort) logger.debug(f"Sending SIGTERM to PID {pid}") subprocess.run(["kill", "-TERM", str(pid)], capture_output=True, text=True, timeout=3) # check=False time.sleep(0.5) # 2. Check if process is gone (or if we lack permission to check) try: logger.debug(f"Checking PID {pid} with kill -0 after SIGTERM attempt") # This will raise CalledProcessError if process is gone OR user lacks permission for kill -0 subprocess.run(["kill", "-0", str(pid)], check=True, capture_output=True, text=True, timeout=1) # If kill -0 succeeded, process is still alive and we have permission to signal it. # 3. Attempt SIGKILL logger.info(f"PID {pid} still alive after SIGTERM attempt (kill -0 succeeded). Sending SIGKILL.") subprocess.run(["kill", "-KILL", str(pid)], check=True, capture_output=True, text=True, timeout=3) # Raises on perm error for SIGKILL # 4. Verify with kill -0 again that it's gone time.sleep(0.1) logger.debug(f"Verifying PID {pid} with kill -0 after SIGKILL attempt") try: subprocess.run(["kill", "-0", str(pid)], check=True, capture_output=True, text=True, timeout=1) # If kill -0 still succeeds, SIGKILL failed to terminate it or it's unkillable logger.warning(f"PID {pid} still alive even after SIGKILL was sent and did not error.") success = False except subprocess.CalledProcessError as e_final_check: # kill -0 failed, means process is gone. Check stderr for "No such process". if e_final_check.stderr and "no such process" in e_final_check.stderr.lower(): logger.info(f"PID {pid} successfully terminated with SIGKILL (confirmed by final kill -0).") success = True else: # kill -0 failed for other reason (e.g. perms, though unlikely if SIGKILL 'succeeded') logger.warning(f"Final kill -0 check for PID {pid} failed unexpectedly. Stderr: {e_final_check.stderr}") success = False # Unsure, so treat as failure for normal kill except subprocess.CalledProcessError as e: # This block is reached if initial `kill -0` fails, or `kill -KILL` fails. # `e` is the error from the *first* command that failed with check=True in the try block. if e.stderr and "no such process" in e.stderr.lower(): logger.info(f"Process {pid} is gone (kill -0 or kill -KILL reported 'No such process'). SIGTERM might have worked or it was already gone.") success = True else: # Failure was likely due to permissions (e.g., "Operation not permitted") or other reasons. # This means normal kill attempt failed. logger.warning(f"Normal kill attempt for PID {pid} failed or encountered permission issue. Stderr from failing cmd: {e.stderr}") success = False elif system == "Windows": logger.debug(f"Using taskkill for PID {pid} on Windows.") result = subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, text=True, check=False, timeout=5, creationflags=subprocess.CREATE_NO_WINDOW) if result.returncode == 0: logger.info(f"Taskkill for PID {pid} succeeded (rc=0).") success = True else: # Check if process was not found output_lower = (result.stdout + result.stderr).lower() if "pid" in output_lower and ("not found" in output_lower or "no running instance" in output_lower or ("could not be terminated" in output_lower and "reason: there is no running instance" in output_lower)) : logger.info(f"Taskkill for PID {pid} reported process not found or already terminated.") success = True else: logger.warning(f"Taskkill for PID {pid} failed. RC: {result.returncode}. Output: {output_lower}") success = False except Exception as e_outer: # Catch any other unexpected exceptions logger.error(f"Outer exception in kill_process_pid for PID {pid}: {e_outer}", exc_info=True) success = False logger.info(f"kill_process_pid for PID {pid} final result: {success}") return success def enhanced_port_check(port, port_name_key=""): port_display_name = get_text(f"port_name_{port_name_key}") if port_name_key else "" update_status_bar("checking_port_with_name", port_name=port_display_name, port=port) if is_port_in_use(port): pids_data = find_processes_on_port(port) if pids_data: pids_info_str_list = [] for proc_info in pids_data: pids_info_str_list.append(f"{proc_info['pid']} ({proc_info['name']})") return {"port": port, "name_key": port_name_key, "pids_data": pids_data, "pids_str": ", ".join(pids_info_str_list)} else: return {"port": port, "name_key": port_name_key, "pids_data": [], "pids_str": get_text("unknown_process_name_placeholder")} return None def check_all_required_ports(ports_to_check: List[Tuple[int, str]]) -> bool: occupied_ports_info = [] for port, port_name_key in ports_to_check: result = enhanced_port_check(port, port_name_key) if result: occupied_ports_info.append(result) if not occupied_ports_info: update_status_bar("port_check_all_completed") return True occupied_ports_details_for_msg = [] for info in occupied_ports_info: port_display_name = get_text(f"port_name_{info['name_key']}") if info['name_key'] else "" occupied_ports_details_for_msg.append(f" - {port_display_name} (端口 {info['port']}): 被 PID(s) {info['pids_str']} 占用") details_str = "\n".join(occupied_ports_details_for_msg) if messagebox.askyesno( get_text("confirm_kill_multiple_ports_title"), get_text("confirm_kill_multiple_ports_message", occupied_ports_details=details_str), parent=root_widget ): pids_processed_this_cycle = set() # Tracks PIDs for which kill attempts (normal or admin) have been made in this call for info in occupied_ports_info: if info['pids_data']: for p_data in info['pids_data']: pid = p_data['pid'] name = p_data['name'] if pid in pids_processed_this_cycle: continue # Avoid reprocessing a PID if it appeared for multiple ports logger.info(f"Port Check Cleanup: Attempting normal kill for PID {pid} ({name}) on port {info['port']}") normal_kill_ok = kill_process_pid(pid) if normal_kill_ok: logger.info(f"Port Check Cleanup: Normal kill succeeded for PID {pid} ({name})") pids_processed_this_cycle.add(pid) else: logger.warning(f"Port Check Cleanup: Normal kill FAILED for PID {pid} ({name}). Prompting for admin kill.") if messagebox.askyesno( get_text("confirm_stop_pid_admin_title"), get_text("confirm_stop_pid_admin_message", pid=pid, name=name), parent=root_widget ): logger.info(f"Port Check Cleanup: User approved admin kill for PID {pid} ({name}). Attempting.") admin_kill_initiated = kill_process_pid_admin(pid) # Optimistic for macOS if admin_kill_initiated: logger.info(f"Port Check Cleanup: Admin kill attempt for PID {pid} ({name}) initiated (result optimistic: {admin_kill_initiated}).") # We still rely on the final port check, so no success message here. else: logger.warning(f"Port Check Cleanup: Admin kill attempt for PID {pid} ({name}) failed to initiate or was denied by user at OS level.") else: logger.info(f"Port Check Cleanup: User declined admin kill for PID {pid} ({name}).") pids_processed_this_cycle.add(pid) # Mark as processed even if admin declined/failed, to avoid re-prompting in this cycle logger.info("Port Check Cleanup: Waiting for 2 seconds for processes to terminate...") time.sleep(2) still_occupied_after_cleanup = False for info in occupied_ports_info: # Re-check all originally occupied ports if is_port_in_use(info['port']): port_display_name = get_text(f"port_name_{info['name_key']}") if info['name_key'] else str(info['port']) logger.warning(f"Port Check Cleanup: Port {port_display_name} ({info['port']}) is still in use after cleanup attempts.") still_occupied_after_cleanup = True break if not still_occupied_after_cleanup: messagebox.showinfo(get_text("info_title"), get_text("all_ports_cleared_success"), parent=root_widget) update_status_bar("port_check_all_completed") return True else: messagebox.showwarning(get_text("warning_title"), get_text("some_ports_still_in_use"), parent=root_widget) return False else: update_status_bar("port_check_user_declined_cleanup") return False def _update_active_auth_display(): """更新GUI中显示的当前活动认证文件""" if not active_auth_file_display_var or not root_widget: return active_files = [f for f in os.listdir(ACTIVE_AUTH_DIR) if f.lower().endswith('.json')] if active_files: # 通常 active 目录只有一个文件，但以防万一，取第一个 active_file_name = sorted(active_files)[0] active_auth_file_display_var.set(get_text("current_auth_file_selected_format", file=active_file_name)) else: active_auth_file_display_var.set(get_text("current_auth_file_none")) def manage_auth_files_gui(): if not os.path.exists(AUTH_PROFILES_DIR): # 检查根目录 messagebox.showerror(get_text("error_title"), get_text("auth_dirs_missing"), parent=root_widget) return # 确保 active 和 saved 目录存在，如果不存在则创建 os.makedirs(ACTIVE_AUTH_DIR, exist_ok=True) os.makedirs(SAVED_AUTH_DIR, exist_ok=True) all_auth_files = set() # 扫描 auth_profiles/ 根目录 if os.path.exists(AUTH_PROFILES_DIR): for f in os.listdir(AUTH_PROFILES_DIR): if f.lower().endswith('.json') and os.path.isfile(os.path.join(AUTH_PROFILES_DIR, f)): all_auth_files.add(f) # 扫描 auth_profiles/active/ 子目录 if os.path.exists(ACTIVE_AUTH_DIR): for f in os.listdir(ACTIVE_AUTH_DIR): if f.lower().endswith('.json') and os.path.isfile(os.path.join(ACTIVE_AUTH_DIR, f)): all_auth_files.add(f) # 通常 active 目录的文件也应该在 saved 或根目录有副本 # 扫描 auth_profiles/saved/ 子目录 if os.path.exists(SAVED_AUTH_DIR): for f in os.listdir(SAVED_AUTH_DIR): if f.lower().endswith('.json') and os.path.isfile(os.path.join(SAVED_AUTH_DIR, f)): all_auth_files.add(f) sorted_auth_files = sorted(list(all_auth_files)) if not sorted_auth_files: messagebox.showinfo(get_text("info_title"), get_text("no_saved_auth_files"), parent=root_widget) # 可以复用这个文本 return auth_window = tk.Toplevel(root_widget) auth_window.title(get_text("auth_manager_title")) auth_window.geometry("400x300") auth_window.resizable(True, True) ttk.Label(auth_window, text=get_text("saved_auth_files_label")).pack(pady=5) files_frame = ttk.Frame(auth_window) files_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) files_listbox = tk.Listbox(files_frame, selectmode=tk.SINGLE) files_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) files_scrollbar = ttk.Scrollbar(files_frame, command=files_listbox.yview) files_scrollbar.pack(side=tk.RIGHT, fill=tk.Y) files_listbox.config(yscrollcommand=files_scrollbar.set) for file_name in sorted_auth_files: files_listbox.insert(tk.END, file_name) def activate_selected_file(): if not files_listbox.curselection(): messagebox.showwarning(get_text("warning_title"), get_text("no_file_selected"), parent=auth_window) return selected_file_name = files_listbox.get(files_listbox.curselection()[0]) # 确定源文件路径，优先从 saved, 然后是根目录 source_path = None potential_saved_path = os.path.join(SAVED_AUTH_DIR, selected_file_name) potential_root_path = os.path.join(AUTH_PROFILES_DIR, selected_file_name) if os.path.exists(potential_saved_path): source_path = potential_saved_path elif os.path.exists(potential_root_path): source_path = potential_root_path else: # 如果文件在 active 目录但不在 saved 或根目录，也允许激活（理论上不应发生，但作为容错） potential_active_path = os.path.join(ACTIVE_AUTH_DIR, selected_file_name) if os.path.exists(potential_active_path): source_path = potential_active_path else: messagebox.showerror(get_text("error_title"), f"源文件 {selected_file_name} 未找到!", parent=auth_window) return # 清空 active 目录 for existing_file in os.listdir(ACTIVE_AUTH_DIR): if existing_file.lower().endswith('.json'): try: os.remove(os.path.join(ACTIVE_AUTH_DIR, existing_file)) except OSError as e: messagebox.showerror(get_text("error_title"), get_text("error_removing_file", file=existing_file, error=str(e)), parent=auth_window) return try: import shutil dest_path = os.path.join(ACTIVE_AUTH_DIR, selected_file_name) shutil.copy2(source_path, dest_path) # 将选中的文件复制到 active 目录 messagebox.showinfo(get_text("info_title"), get_text("auth_file_activated", file=selected_file_name), parent=auth_window) _update_active_auth_display() # 更新主界面显示 auth_window.destroy() except Exception as e: messagebox.showerror(get_text("error_title"), get_text("error_activating_file", file=selected_file_name, error=str(e)), parent=auth_window) _update_active_auth_display() # 即使失败也尝试更新显示 buttons_frame = ttk.Frame(auth_window) buttons_frame.pack(fill=tk.X, padx=10, pady=10) ttk.Button(buttons_frame, text=get_text("activate_selected_btn"), command=activate_selected_file).pack(side=tk.LEFT, padx=5) ttk.Button(buttons_frame, text=get_text("cancel_btn"), command=auth_window.destroy).pack(side=tk.RIGHT, padx=5) def get_active_auth_json_path_for_launch() -> Optional[str]: """获取用于启动命令的 --active-auth-json 参数值""" active_files = [f for f in os.listdir(ACTIVE_AUTH_DIR) if f.lower().endswith('.json') and os.path.isfile(os.path.join(ACTIVE_AUTH_DIR, f))] if active_files: # 如果 active 目录有文件，总是使用它（按名称排序的第一个） return os.path.join(ACTIVE_AUTH_DIR, sorted(active_files)[0]) return None def build_launch_command(mode, fastapi_port, camoufox_debug_port, stream_port_enabled, stream_port, helper_enabled, helper_endpoint): cmd = [PYTHON_EXECUTABLE, LAUNCH_CAMOUFOX_PY, f"--{mode}", "--server-port", str(fastapi_port), "--camoufox-debug-port", str(camoufox_debug_port)] active_auth_path = get_active_auth_json_path_for_launch() if active_auth_path: cmd.extend(["--active-auth-json", active_auth_path]) logger.info(f"将使用认证文件: {active_auth_path}") else: logger.info("未找到活动的认证文件，不传递 --active-auth-json 参数。") if stream_port_enabled: cmd.extend(["--stream-port", str(stream_port)]) else: cmd.extend(["--stream-port", "0"]) # 显式传递0表示禁用 if helper_enabled and helper_endpoint: cmd.extend(["--helper", helper_endpoint]) else: cmd.extend(["--helper", ""]) # 显式传递空字符串表示禁用 # 修复：添加统一代理配置参数传递 # 使用 --internal-camoufox-proxy 参数确保最高优先级，而不是仅依赖环境变量 if proxy_enabled_var.get(): proxy_addr = proxy_address_var.get().strip() if proxy_addr: cmd.extend(["--internal-camoufox-proxy", proxy_addr]) logger.info(f"将使用GUI配置的代理: {proxy_addr}") else: cmd.extend(["--internal-camoufox-proxy", ""]) logger.info("GUI代理已启用但地址为空，明确禁用代理") else: cmd.extend(["--internal-camoufox-proxy", ""]) logger.info("GUI代理未启用，明确禁用代理") return cmd # --- GUI构建与主逻辑区段的函数定义 --- # (这些函数调用上面定义的辅助函数，所以它们的定义顺序很重要) def enqueue_stream_output(stream, stream_name_prefix): try: for line_bytes in iter(stream.readline, b''): if not line_bytes: break line = line_bytes.decode(sys.stdout.encoding or 'utf-8', errors='replace') if managed_process_info.get("output_area") and root_widget: def _update_stream_output(line_to_insert): current_line = line_to_insert if managed_process_info.get("output_area"): managed_process_info["output_area"].config(state=tk.NORMAL) managed_process_info["output_area"].insert(tk.END, current_line) managed_process_info["output_area"].see(tk.END) managed_process_info["output_area"].config(state=tk.DISABLED) root_widget.after_idle(_update_stream_output, f"[{stream_name_prefix}] {line}") else: print(f"[{stream_name_prefix}] {line.strip()}", flush=True) except ValueError: pass except Exception: pass finally: if hasattr(stream, 'close') and not stream.closed: stream.close() def is_service_running(): return managed_process_info.get("popen") and \ managed_process_info["popen"].poll() is None and \ not managed_process_info.get("fully_detached", False) def is_any_service_known(): return managed_process_info.get("popen") is not None def monitor_process_thread_target(): popen = managed_process_info.get("popen") service_name_key = managed_process_info.get("service_name_key") is_detached = managed_process_info.get("fully_detached", False) if not popen or not service_name_key: return stdout_thread = None; stderr_thread = None if popen.stdout: stdout_thread = threading.Thread(target=enqueue_stream_output, args=(popen.stdout, "stdout"), daemon=True) managed_process_info["stdout_thread"] = stdout_thread stdout_thread.start() if popen.stderr: stderr_thread = threading.Thread(target=enqueue_stream_output, args=(popen.stderr, "stderr"), daemon=True) managed_process_info["stderr_thread"] = stderr_thread stderr_thread.start() popen.wait() exit_code = popen.returncode if stdout_thread and stdout_thread.is_alive(): stdout_thread.join(timeout=1) if stderr_thread and stderr_thread.is_alive(): stderr_thread.join(timeout=1) if managed_process_info.get("service_name_key") == service_name_key: service_name = get_text(service_name_key) if not is_detached: if exit_code == 0: update_status_bar("service_stopped_gracefully_status", service_name=service_name) else: update_status_bar("service_stopped_exit_code_status", service_name=service_name, code=exit_code) managed_process_info["popen"] = None managed_process_info["service_name_key"] = None managed_process_info["fully_detached"] = False def get_fastapi_port_from_gui() -> int: try: port_str = port_entry_var.get() if not port_str: messagebox.showwarning(get_text("warning_title"), get_text("enter_valid_port_warn")); return DEFAULT_FASTAPI_PORT port = int(port_str) if not (1024 <= port <= 65535): raise ValueError("Port out of range") return port except ValueError: messagebox.showwarning(get_text("warning_title"), get_text("enter_valid_port_warn")) port_entry_var.set(str(DEFAULT_FASTAPI_PORT)) return DEFAULT_FASTAPI_PORT def get_camoufox_debug_port_from_gui() -> int: try: port_str = camoufox_debug_port_var.get() if not port_str: camoufox_debug_port_var.set(str(DEFAULT_CAMOUFOX_PORT_GUI)) return DEFAULT_CAMOUFOX_PORT_GUI port = int(port_str) if not (1024 <= port <= 65535): raise ValueError("Port out of range") return port except ValueError: messagebox.showwarning(get_text("warning_title"), get_text("enter_valid_port_warn")) camoufox_debug_port_var.set(str(DEFAULT_CAMOUFOX_PORT_GUI)) return DEFAULT_CAMOUFOX_PORT_GUI # 配置文件路径 CONFIG_FILE_PATH = os.path.join(SCRIPT_DIR, "gui_config.json") # 默认配置 - 从环境变量读取，如果没有则使用硬编码默认值 DEFAULT_CONFIG = { "fastapi_port": DEFAULT_FASTAPI_PORT, "camoufox_debug_port": DEFAULT_CAMOUFOX_PORT_GUI, "stream_port": int(os.environ.get('GUI_DEFAULT_STREAM_PORT', '3120')), "stream_port_enabled": True, "helper_endpoint": os.environ.get('GUI_DEFAULT_HELPER_ENDPOINT', ''), "helper_enabled": False, "proxy_address": os.environ.get('GUI_DEFAULT_PROXY_ADDRESS', 'http://127.0.0.1:7890'), "proxy_enabled": False } # 加载配置 def load_config(): if os.path.exists(CONFIG_FILE_PATH): try: with open(CONFIG_FILE_PATH, 'r', encoding='utf-8') as f: config = json.load(f) logger.info(f"成功加载配置文件: {CONFIG_FILE_PATH}") return config except Exception as e: logger.error(f"加载配置文件失败: {e}") logger.info(f"使用默认配置") return DEFAULT_CONFIG.copy() # 保存配置 def save_config(): config = { "fastapi_port": port_entry_var.get(), "camoufox_debug_port": camoufox_debug_port_var.get(), "stream_port": stream_port_var.get(), "stream_port_enabled": stream_port_enabled_var.get(), "helper_endpoint": helper_endpoint_var.get(), "helper_enabled": helper_enabled_var.get(), "proxy_address": proxy_address_var.get(), "proxy_enabled": proxy_enabled_var.get() } try: with open(CONFIG_FILE_PATH, 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) logger.info(f"成功保存配置到: {CONFIG_FILE_PATH}") except Exception as e: logger.error(f"保存配置失败: {e}") # 重置为默认配置，包含代理设置 def reset_to_defaults(): if messagebox.askyesno(get_text("confirm_reset_title"), get_text("confirm_reset_message"), parent=root_widget): port_entry_var.set(str(DEFAULT_FASTAPI_PORT)) camoufox_debug_port_var.set(str(DEFAULT_CAMOUFOX_PORT_GUI)) stream_port_var.set("3120") stream_port_enabled_var.set(True) helper_endpoint_var.set("") helper_enabled_var.set(False) proxy_address_var.set("http://127.0.0.1:7890") proxy_enabled_var.set(False) messagebox.showinfo(get_text("info_title"), get_text("reset_success"), parent=root_widget) def _configure_proxy_env_vars() -> Dict[str, str]: """ 配置代理环境变量（已弃用，现在主要通过 --internal-camoufox-proxy 参数传递） 保留此函数以维持向后兼容性，但现在主要用于状态显示 """ proxy_env = {} if proxy_enabled_var.get(): proxy_addr = proxy_address_var.get().strip() if proxy_addr: # 注意：现在主要通过 --internal-camoufox-proxy 参数传递代理配置 # 环境变量作为备用方案，但优先级较低 update_status_bar("proxy_configured_status", proxy_addr=proxy_addr) else: update_status_bar("proxy_skip_status") else: update_status_bar("proxy_skip_status") return proxy_env def _launch_process_gui(cmd: List[str], service_name_key: str, env_vars: Optional[Dict[str, str]] = None): global managed_process_info # managed_process_info is now informational for these launches service_name = get_text(service_name_key) # Clear previous output area for GUI messages, actual process output will be in the new terminal if managed_process_info.get("output_area"): managed_process_info["output_area"].config(state=tk.NORMAL) managed_process_info["output_area"].delete('1.0', tk.END) managed_process_info["output_area"].insert(tk.END, f"[INFO] Preparing to launch {service_name} in a new terminal...\\n") managed_process_info["output_area"].config(state=tk.DISABLED) effective_env = os.environ.copy() if env_vars: effective_env.update(env_vars) effective_env['PYTHONIOENCODING'] = 'utf-8' popen_kwargs: Dict[str, Any] = {"env": effective_env} system = platform.system() launch_cmd_for_terminal: Optional[List[str]] = None # Prepare command string for terminals that take a single command string # Ensure correct quoting for arguments with spaces cmd_parts_for_string = [] for part in cmd: if " " in part and not (part.startswith('"') and part.endswith('"')): cmd_parts_for_string.append(f'"{part}"') else: cmd_parts_for_string.append(part) cmd_str_for_terminal_execution = " ".join(cmd_parts_for_string) if system == "Windows": # CREATE_NEW_CONSOLE opens a new window. # The new process will be a child of this GUI initially, but if python.exe # itself handles its lifecycle well, closing GUI might not kill it. # To be more robust for independence, one might use 'start' cmd, # but simple CREATE_NEW_CONSOLE often works for python scripts. # For true independence and GUI not waiting, Popen should be on python.exe directly. popen_kwargs["creationflags"] = subprocess.CREATE_NEW_CONSOLE launch_cmd_for_terminal = cmd # Direct command elif system == "Darwin": # macOS # import shlex # Ensure shlex is imported (should be at top of file) # Build the shell command string with proper quoting for each argument. # The command will first change to SCRIPT_DIR, then execute the python script. script_dir_quoted = shlex.quote(SCRIPT_DIR) python_executable_quoted = shlex.quote(cmd[0]) script_path_quoted = shlex.quote(cmd[1]) args_for_script_quoted = [shlex.quote(arg) for arg in cmd[2:]] # 构建环境变量设置字符串 env_prefix_parts = [] if env_vars: # env_vars 应该是从 _configure_proxy_env_vars() 来的 proxy_env for key, value in env_vars.items(): if value is not None: # 确保值存在且不为空字符串 env_prefix_parts.append(f"{shlex.quote(key)}={shlex.quote(str(value))}") env_prefix_str = " ".join(env_prefix_parts) # Construct the full shell command to be executed in the new terminal shell_command_parts = [ f"cd {script_dir_quoted}", "&&" # Ensure command separation ] if env_prefix_str: shell_command_parts.append(env_prefix_str) shell_command_parts.extend([ python_executable_quoted, script_path_quoted ]) shell_command_parts.extend(args_for_script_quoted) shell_command_str = " ".join(shell_command_parts) # Now, escape this shell_command_str for embedding within an AppleScript double-quoted string. # In AppleScript strings, backslash `\\` and double quote `\"` are special and need to be escaped. applescript_arg_escaped = shell_command_str.replace('\\\\', '\\\\\\\\').replace('\"', '\\\\\"') # Construct the AppleScript command # 修复：使用简化的AppleScript命令避免AppleEvent处理程序失败 # 直接创建新窗口并执行命令，避免复杂的条件判断 applescript_command = f''' tell application "Terminal" do script "{applescript_arg_escaped}" activate end tell ''' launch_cmd_for_terminal = ["osascript", "-e", applescript_command.strip()] elif system == "Linux": import shutil terminal_emulator = shutil.which("x-terminal-emulator") or shutil.which("gnome-terminal") or shutil.which("konsole") or shutil.which("xfce4-terminal") or shutil.which("xterm") if terminal_emulator: # Construct command ensuring SCRIPT_DIR is CWD for the launched script # Some terminals might need `sh -c "cd ... && python ..."` # For simplicity, let's try to pass the command directly if possible or via sh -c cd_command = f"cd '{SCRIPT_DIR}' && " full_command_to_run = cd_command + cmd_str_for_terminal_execution if "gnome-terminal" in terminal_emulator or "mate-terminal" in terminal_emulator: launch_cmd_for_terminal = [terminal_emulator, "--", "bash", "-c", full_command_to_run + "; exec bash"] elif "konsole" in terminal_emulator or "xfce4-terminal" in terminal_emulator or "lxterminal" in terminal_emulator: launch_cmd_for_terminal = [terminal_emulator, "-e", f"bash -c '{full_command_to_run}; exec bash'"] elif "xterm" in terminal_emulator: # xterm might need careful quoting launch_cmd_for_terminal = [terminal_emulator, "-hold", "-e", "bash", "-c", f"{full_command_to_run}"] else: # Generic x-terminal-emulator launch_cmd_for_terminal = [terminal_emulator, "-e", f"bash -c '{full_command_to_run}; exec bash'"] else: messagebox.showerror(get_text("error_title"), "未找到兼容的Linux终端模拟器 (如 x-terminal-emulator, gnome-terminal, xterm)。无法在新终端中启动服务。") update_status_bar("status_error_starting", service_name=service_name) return else: # Fallback for other OS or if specific terminal launch fails messagebox.showerror(get_text("error_title"), f"不支持为操作系统 {system} 在新终端中启动。") update_status_bar("status_error_starting", service_name=service_name) return if not launch_cmd_for_terminal: # Should not happen if logic above is correct messagebox.showerror(get_text("error_title"), f"无法为 {system} 构建终端启动命令。") update_status_bar("status_error_starting", service_name=service_name) return try: # Launch the terminal command. This Popen object is for the terminal launcher. # The actual Python script is a child of that new terminal. logger.info(f"Launching in new terminal with command: {' '.join(launch_cmd_for_terminal)}") logger.info(f"Effective environment for new terminal: {effective_env}") # For non-Windows, where we launch `osascript` or a terminal emulator, # these Popen objects complete quickly. # For Windows, `CREATE_NEW_CONSOLE` means the Popen object is for the new python process. # However, we are treating all as fire-and-forget for the GUI. process = subprocess.Popen(launch_cmd_for_terminal, **popen_kwargs) # We no longer store this popen object in managed_process_info for direct GUI management # as the process is meant to be independent. # managed_process_info["popen"] = process # managed_process_info["service_name_key"] = service_name_key # managed_process_info["fully_detached"] = True # No monitoring threads from GUI for these independent processes. # managed_process_info["monitor_thread"] = None # managed_process_info["stdout_thread"] = None # managed_process_info["stderr_thread"] = None update_status_bar("info_service_new_terminal") if managed_process_info.get("output_area"): managed_process_info["output_area"].config(state=tk.NORMAL) managed_process_info["output_area"].insert(tk.END, f"[INFO] {get_text('info_service_new_terminal')}\\n") managed_process_info["output_area"].insert(tk.END, f"[INFO] {service_name} (PID: {process.pid if system == 'Windows' else 'N/A for terminal launcher'}) should be running in a new window.\\n") managed_process_info["output_area"].see(tk.END) managed_process_info["output_area"].config(state=tk.DISABLED) if root_widget: # Query ports after a delay, as service might take time to start root_widget.after(3500, query_port_and_display_pids_gui) except FileNotFoundError: messagebox.showerror(get_text("error_title"), get_text("script_not_found_error_msgbox", cmd=' '.join(cmd))) update_status_bar("status_script_not_found", service_name=service_name) except Exception as e: messagebox.showerror(get_text("error_title"), f"{service_name} - {get_text('error_title')}: {e}") update_status_bar("status_error_starting", service_name=service_name) logger.error(f"Error in _launch_process_gui for {service_name}: {e}", exc_info=True) @debounce_button("start_headed_interactive", 3.0) def start_headed_interactive_gui(): launch_params = _get_launch_parameters() if not launch_params: return if port_auto_check_var.get(): ports_to_check = [ (launch_params["fastapi_port"], "fastapi"), (launch_params["camoufox_debug_port"], "camoufox_debug") ] if launch_params["stream_port_enabled"] and launch_params["stream_port"] != 0: ports_to_check.append((launch_params["stream_port"], "stream_proxy")) if launch_params["helper_enabled"] and launch_params["helper_endpoint"]: try: pu = urlparse(launch_params["helper_endpoint"]) if pu.hostname in ("localhost", "127.0.0.1") and pu.port: ports_to_check.append((pu.port, "helper_service")) except Exception as e: print(f"解析Helper URL失败(有头模式): {e}") if not check_all_required_ports(ports_to_check): return proxy_env = _configure_proxy_env_vars() cmd = build_launch_command( "debug", launch_params["fastapi_port"], launch_params["camoufox_debug_port"], launch_params["stream_port_enabled"], launch_params["stream_port"], launch_params["helper_enabled"], launch_params["helper_endpoint"] ) update_status_bar("status_headed_launch") _launch_process_gui(cmd, "service_name_headed_interactive", env_vars=proxy_env) @debounce_button("start_headless", 3.0) def start_headless_gui(): launch_params = _get_launch_parameters() if not launch_params: return if port_auto_check_var.get(): ports_to_check = [ (launch_params["fastapi_port"], "fastapi"), (launch_params["camoufox_debug_port"], "camoufox_debug") ] if launch_params["stream_port_enabled"] and launch_params["stream_port"] != 0: ports_to_check.append((launch_params["stream_port"], "stream_proxy")) if launch_params["helper_enabled"] and launch_params["helper_endpoint"]: try: pu = urlparse(launch_params["helper_endpoint"]) if pu.hostname in ("localhost", "127.0.0.1") and pu.port: ports_to_check.append((pu.port, "helper_service")) except Exception as e: print(f"解析Helper URL失败(无头模式): {e}") if not check_all_required_ports(ports_to_check): return proxy_env = _configure_proxy_env_vars() cmd = build_launch_command( "headless", launch_params["fastapi_port"], launch_params["camoufox_debug_port"], launch_params["stream_port_enabled"], launch_params["stream_port"], launch_params["helper_enabled"], launch_params["helper_endpoint"] ) update_status_bar("status_headless_launch") _launch_process_gui(cmd, "service_name_headless", env_vars=proxy_env) @debounce_button("start_virtual_display", 3.0) def start_virtual_display_gui(): if platform.system() != "Linux": messagebox.showwarning(get_text("warning_title"), "虚拟显示模式仅在Linux上受支持。") return launch_params = _get_launch_parameters() if not launch_params: return if port_auto_check_var.get(): ports_to_check = [ (launch_params["fastapi_port"], "fastapi"), (launch_params["camoufox_debug_port"], "camoufox_debug") ] if launch_params["stream_port_enabled"] and launch_params["stream_port"] != 0: ports_to_check.append((launch_params["stream_port"], "stream_proxy")) if launch_params["helper_enabled"] and launch_params["helper_endpoint"]: try: pu = urlparse(launch_params["helper_endpoint"]) if pu.hostname in ("localhost", "127.0.0.1") and pu.port: ports_to_check.append((pu.port, "helper_service")) except Exception as e: print(f"解析Helper URL失败(虚拟显示模式): {e}") if not check_all_required_ports(ports_to_check): return proxy_env = _configure_proxy_env_vars() cmd = build_launch_command( "virtual-display", launch_params["fastapi_port"], launch_params["camoufox_debug_port"], launch_params["stream_port_enabled"], launch_params["stream_port"], launch_params["helper_enabled"], launch_params["helper_endpoint"] ) update_status_bar("status_virtual_display_launch") _launch_process_gui(cmd, "service_name_virtual_display", env_vars=proxy_env) # --- LLM Mock Service Management --- def is_llm_service_running() -> bool: """检查本地LLM模拟服务是否正在运行""" return llm_service_process_info.get("popen") and \ llm_service_process_info["popen"].poll() is None def monitor_llm_process_thread_target(): """监控LLM服务进程，捕获输出并更新状态""" popen = llm_service_process_info.get("popen") service_name_key = llm_service_process_info.get("service_name_key") # "llm_service_name_key" output_area = managed_process_info.get("output_area") # Use the main output area if not popen or not service_name_key or not output_area: logger.error("LLM monitor thread: Popen, service_name_key, or output_area is None.") return service_name = get_text(service_name_key) logger.info(f"Starting monitor thread for {service_name} (PID: {popen.pid})") # stdout/stderr redirection if popen.stdout: llm_service_process_info["stdout_thread"] = threading.Thread( target=enqueue_stream_output, args=(popen.stdout, f"{service_name}-stdout"), daemon=True ) llm_service_process_info["stdout_thread"].start() if popen.stderr: llm_service_process_info["stderr_thread"] = threading.Thread( target=enqueue_stream_output, args=(popen.stderr, f"{service_name}-stderr"), daemon=True ) llm_service_process_info["stderr_thread"].start() popen.wait() # Wait for the process to terminate exit_code = popen.returncode logger.info(f"{service_name} (PID: {popen.pid}) terminated with exit code {exit_code}.") if llm_service_process_info.get("stdout_thread") and llm_service_process_info["stdout_thread"].is_alive(): llm_service_process_info["stdout_thread"].join(timeout=1) if llm_service_process_info.get("stderr_thread") and llm_service_process_info["stderr_thread"].is_alive(): llm_service_process_info["stderr_thread"].join(timeout=1) # Update status only if this was the process we were tracking if llm_service_process_info.get("popen") == popen: update_status_bar("status_llm_stopped") llm_service_process_info["popen"] = None llm_service_process_info["monitor_thread"] = None llm_service_process_info["stdout_thread"] = None llm_service_process_info["stderr_thread"] = None def _actually_launch_llm_service(): """实际启动 llm.py 脚本""" global llm_service_process_info service_name_key = "llm_service_name_key" service_name = get_text(service_name_key) output_area = managed_process_info.get("output_area") if not output_area: logger.error("Cannot launch LLM service: Main output area is not available.") update_status_bar("status_error_starting", service_name=service_name) return llm_script_path = os.path.join(SCRIPT_DIR, LLM_PY_FILENAME) if not os.path.exists(llm_script_path): messagebox.showerror(get_text("error_title"), get_text("startup_script_not_found_msgbox", script=LLM_PY_FILENAME)) update_status_bar("status_script_not_found", service_name=service_name) return # Get the main server port from GUI to pass to llm.py main_server_port = get_fastapi_port_from_gui() # Ensure this function is available and returns the correct port cmd = [PYTHON_EXECUTABLE, llm_script_path, f"--main-server-port={main_server_port}"] logger.info(f"Attempting to launch LLM service with command: {' '.join(cmd)}") try: # Clear previous LLM service output if any, or add a header output_area.config(state=tk.NORMAL) output_area.insert(tk.END, f"--- Starting {service_name} ---\n") output_area.config(state=tk.DISABLED) effective_env = os.environ.copy() effective_env['PYTHONUNBUFFERED'] = '1' # Ensure unbuffered output for real-time logging effective_env['PYTHONIOENCODING'] = 'utf-8' popen = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False, # Read as bytes for enqueue_stream_output cwd=SCRIPT_DIR, env=effective_env, creationflags=subprocess.CREATE_NO_WINDOW if platform.system() == "Windows" else 0 ) llm_service_process_info["popen"] = popen llm_service_process_info["service_name_key"] = service_name_key update_status_bar("status_llm_starting", pid=popen.pid) logger.info(f"{service_name} started with PID: {popen.pid}") # Start monitoring thread monitor_thread = threading.Thread(target=monitor_llm_process_thread_target, daemon=True) llm_service_process_info["monitor_thread"] = monitor_thread monitor_thread.start() except FileNotFoundError: messagebox.showerror(get_text("error_title"), get_text("script_not_found_error_msgbox", cmd=' '.join(cmd))) update_status_bar("status_script_not_found", service_name=service_name) logger.error(f"FileNotFoundError when trying to launch LLM service: {cmd}") except Exception as e: messagebox.showerror(get_text("error_title"), f"{service_name} - {get_text('error_title')}: {e}") update_status_bar("status_error_starting", service_name=service_name) logger.error(f"Exception when launching LLM service: {e}", exc_info=True) llm_service_process_info["popen"] = None # Ensure it's cleared on failure def _check_llm_backend_and_launch_thread(): """检查LLM后端服务 (动态端口) 并在成功后启动llm.py""" # Get the current FastAPI port from the GUI # This needs to be called within this thread, right before the check, # as port_entry_var might be accessed from a different thread if called outside. # However, Tkinter GUI updates should ideally be done from the main thread. # For reading a StringVar, it's generally safe. current_fastapi_port = get_fastapi_port_from_gui() # Update status bar and logger with the dynamic port # For status bar updates from a thread, it's better to use root_widget.after or a queue, # but for simplicity in this context, direct update_status_bar call is used. # Ensure update_status_bar is thread-safe or schedules GUI updates. # The existing update_status_bar uses root_widget.after_idle, which is good. # Dynamically create the message keys for status bar to include the port backend_check_msg_key = "status_llm_backend_check" # Original key backend_ok_msg_key = "status_llm_backend_ok_starting" backend_fail_msg_key = "status_llm_backend_fail" # It's better to pass the port as a parameter to get_text if the LANG_TEXTS are updated # For now, we'll just log the dynamic port separately. update_status_bar(backend_check_msg_key) # Still uses the generic message logger.info(f"Checking LLM backend service at localhost:{current_fastapi_port}...") backend_ok = False try: with socket.create_connection(("localhost", current_fastapi_port), timeout=3) as sock: backend_ok = True logger.info(f"LLM backend service (localhost:{current_fastapi_port}) is responsive.") except (socket.timeout, ConnectionRefusedError, OSError) as e: logger.warning(f"LLM backend service (localhost:{current_fastapi_port}) not responding: {e}") backend_ok = False if root_widget: # Ensure GUI is still there if backend_ok: update_status_bar(backend_ok_msg_key, port=current_fastapi_port) # Pass port to fill placeholder _actually_launch_llm_service() # This already gets the port via get_fastapi_port_from_gui() else: # Update status bar with the dynamic port for failure message update_status_bar(backend_fail_msg_key, port=current_fastapi_port) # Show warning messagebox with the dynamic port # The status bar is already updated by update_status_bar, # so no need to manually set process_status_text_var or write to output_area here again for the same message. # The update_status_bar function handles writing to the output_area if configured. messagebox.showwarning( get_text("warning_title"), get_text(backend_fail_msg_key, port=current_fastapi_port), # Use get_text with port for the messagebox parent=root_widget ) def start_llm_service_gui(): """GUI命令：启动本地LLM模拟服务""" if is_llm_service_running(): pid = llm_service_process_info["popen"].pid update_status_bar("status_llm_already_running", pid=pid) messagebox.showinfo(get_text("info_title"), get_text("status_llm_already_running", pid=pid), parent=root_widget) return # Run the check and actual launch in a new thread to keep GUI responsive # The check itself can take a few seconds if the port is unresponsive. threading.Thread(target=_check_llm_backend_and_launch_thread, daemon=True).start() def stop_llm_service_gui(): """GUI命令：停止本地LLM模拟服务""" service_name = get_text(llm_service_process_info.get("service_name_key", "llm_service_name_key")) popen = llm_service_process_info.get("popen") if not popen or popen.poll() is not None: update_status_bar("status_llm_not_running") # messagebox.showinfo(get_text("info_title"), get_text("status_llm_not_running"), parent=root_widget) return if messagebox.askyesno(get_text("confirm_stop_llm_title"), get_text("confirm_stop_llm_message"), parent=root_widget): logger.info(f"Attempting to stop {service_name} (PID: {popen.pid})") update_status_bar("status_stopping_service", service_name=service_name, pid=popen.pid) try: # Attempt graceful termination first if platform.system() == "Windows": # On Windows, sending SIGINT to a Popen object created with CREATE_NO_WINDOW # might not work as expected for Flask apps. taskkill is more reliable. # We can try to send Ctrl+C to the console if it had one, but llm.py is simple. # For Flask, direct popen.terminate() or popen.kill() is often used. logger.info(f"Sending SIGTERM/terminate to {service_name} (PID: {popen.pid}) on Windows.") popen.terminate() # Sends SIGTERM on Unix, TerminateProcess on Windows else: # Linux/macOS logger.info(f"Sending SIGINT to {service_name} (PID: {popen.pid}) on {platform.system()}.") popen.send_signal(signal.SIGINT) # Wait for a short period for graceful shutdown try: popen.wait(timeout=5) # Wait up to 5 seconds logger.info(f"{service_name} (PID: {popen.pid}) terminated gracefully after signal.") update_status_bar("status_llm_stopped") except subprocess.TimeoutExpired: logger.warning(f"{service_name} (PID: {popen.pid}) did not terminate after signal. Forcing kill.") popen.kill() # Force kill popen.wait(timeout=2) # Wait for kill to take effect update_status_bar("status_llm_stopped") # Assume killed logger.info(f"{service_name} (PID: {popen.pid}) was force-killed.") except Exception as e: logger.error(f"Error stopping {service_name} (PID: {popen.pid}): {e}", exc_info=True) update_status_bar("status_llm_stop_error") messagebox.showerror(get_text("error_title"), f"Error stopping {service_name}: {e}", parent=root_widget) finally: # Ensure threads are joined and resources cleaned up, even if already done by monitor if llm_service_process_info.get("stdout_thread") and llm_service_process_info["stdout_thread"].is_alive(): llm_service_process_info["stdout_thread"].join(timeout=0.5) if llm_service_process_info.get("stderr_thread") and llm_service_process_info["stderr_thread"].is_alive(): llm_service_process_info["stderr_thread"].join(timeout=0.5) llm_service_process_info["popen"] = None llm_service_process_info["monitor_thread"] = None llm_service_process_info["stdout_thread"] = None llm_service_process_info["stderr_thread"] = None # Clear related output from the main log area or add a "stopped" message output_area = managed_process_info.get("output_area") if output_area: output_area.config(state=tk.NORMAL) output_area.insert(tk.END, f"--- {service_name} stopped ---\n") output_area.see(tk.END) output_area.config(state=tk.DISABLED) else: logger.info(f"User cancelled stopping {service_name}.") # --- End LLM Mock Service Management --- def query_port_and_display_pids_gui(): ports_to_query_info = [] ports_desc_list = [] # 1. FastAPI Port fastapi_port = get_fastapi_port_from_gui() ports_to_query_info.append({"port": fastapi_port, "type_key": "port_name_fastapi", "type_name": get_text("port_name_fastapi")}) ports_desc_list.append(f"{get_text('port_name_fastapi')}:{fastapi_port}") # 2. Camoufox Debug Port camoufox_port = get_camoufox_debug_port_from_gui() ports_to_query_info.append({"port": camoufox_port, "type_key": "port_name_camoufox_debug", "type_name": get_text("port_name_camoufox_debug")}) ports_desc_list.append(f"{get_text('port_name_camoufox_debug')}:{camoufox_port}") # 3. Stream Proxy Port (if enabled) if stream_port_enabled_var.get(): try: stream_p_val_str = stream_port_var.get().strip() stream_p = int(stream_p_val_str) if stream_p_val_str else 0 # Default to 0 if empty, meaning disabled if stream_p != 0 and not (1024 <= stream_p <= 65535): messagebox.showwarning(get_text("warning_title"), get_text("stream_port_out_of_range"), parent=root_widget) # Optionally, do not query this port or handle as error elif stream_p != 0 : # Only query if valid and non-zero ports_to_query_info.append({"port": stream_p, "type_key": "port_name_stream_proxy", "type_name": get_text("port_name_stream_proxy")}) ports_desc_list.append(f"{get_text('port_name_stream_proxy')}:{stream_p}") except ValueError: messagebox.showwarning(get_text("warning_title"), get_text("stream_port_out_of_range") + " (非数字)", parent=root_widget) update_status_bar("querying_ports_status", ports_desc=", ".join(ports_desc_list)) if pid_listbox_widget and pid_list_lbl_frame_ref: pid_listbox_widget.delete(0, tk.END) pid_list_lbl_frame_ref.config(text=get_text("pids_on_multiple_ports_label")) # Update title found_any_process = False for port_info in ports_to_query_info: current_port = port_info["port"] port_type_name = port_info["type_name"] processes_on_current_port = find_processes_on_port(current_port) if processes_on_current_port: found_any_process = True for proc_info in processes_on_current_port: pid_display_info = f"{proc_info['pid']} - {proc_info['name']}" display_text = get_text("port_query_result_format", port_type=port_type_name, port_num=current_port, pid_info=pid_display_info) pid_listbox_widget.insert(tk.END, display_text) else: display_text = get_text("port_not_in_use_format", port_type=port_type_name, port_num=current_port) pid_listbox_widget.insert(tk.END, display_text) if not found_any_process and not any(find_processes_on_port(p["port"]) for p in ports_to_query_info): # Recheck if all are empty # If after checking all, still no processes, we can add a general "no pids found on queried ports" # but the per-port "not in use" message is usually clearer. pass # Individual messages already cover this. else: logger.error("pid_listbox_widget or pid_list_lbl_frame_ref is None in query_port_and_display_pids_gui") def _perform_proxy_test_single(proxy_address: str, test_url: str, timeout: int = 15) -> Tuple[bool, str, int]: """ 单次代理测试尝试 Returns (success_status, message_or_error_string, status_code). """ proxies = { "http": proxy_address, "https": proxy_address, } try: logger.info(f"Testing proxy {proxy_address} with URL {test_url} (timeout: {timeout}s)") response = requests.get(test_url, proxies=proxies, timeout=timeout, allow_redirects=True) status_code = response.status_code # 检查HTTP状态码 if 200 <= status_code < 300: logger.info(f"Proxy test to {test_url} via {proxy_address} successful. Status: {status_code}") return True, get_text("proxy_test_success", url=test_url), status_code elif status_code == 503: # 503 Service Unavailable - 可能是临时性问题 logger.warning(f"Proxy test got 503 Service Unavailable from {test_url} via {proxy_address}") return False, f"HTTP {status_code}: Service Temporarily Unavailable", status_code elif 400 <= status_code < 500: # 4xx 客户端错误 logger.warning(f"Proxy test got client error {status_code} from {test_url} via {proxy_address}") return False, f"HTTP {status_code}: Client Error", status_code elif 500 <= status_code < 600: # 5xx 服务器错误 logger.warning(f"Proxy test got server error {status_code} from {test_url} via {proxy_address}") return False, f"HTTP {status_code}: Server Error", status_code else: logger.warning(f"Proxy test got unexpected status {status_code} from {test_url} via {proxy_address}") return False, f"HTTP {status_code}: Unexpected Status", status_code except requests.exceptions.ProxyError as e: logger.error(f"ProxyError connecting to {test_url} via {proxy_address}: {e}") return False, f"Proxy Error: {e}", 0 except requests.exceptions.ConnectTimeout as e: logger.error(f"ConnectTimeout connecting to {test_url} via {proxy_address}: {e}") return False, f"Connection Timeout: {e}", 0 except requests.exceptions.ReadTimeout as e: logger.error(f"ReadTimeout from {test_url} via {proxy_address}: {e}") return False, f"Read Timeout: {e}", 0 except requests.exceptions.SSLError as e: logger.error(f"SSLError connecting to {test_url} via {proxy_address}: {e}") return False, f"SSL Error: {e}", 0 except requests.exceptions.RequestException as e: logger.error(f"RequestException connecting to {test_url} via {proxy_address}: {e}") return False, str(e), 0 except Exception as e: # Catch any other unexpected errors logger.error(f"Unexpected error during proxy test to {test_url} via {proxy_address}: {e}", exc_info=True) return False, f"Unexpected error: {e}", 0 def _perform_proxy_test(proxy_address: str, test_url: str) -> Tuple[bool, str]: """ 增强的代理测试函数，包含重试机制和备用URL Returns (success_status, message_or_error_string). """ max_attempts = 3 backup_url = LANG_TEXTS["proxy_test_url_backup"] urls_to_try = [test_url] # 如果主URL不是备用URL，则添加备用URL if test_url != backup_url: urls_to_try.append(backup_url) for url_index, current_url in enumerate(urls_to_try): if url_index > 0: logger.info(f"Trying backup URL: {current_url}") update_status_bar("proxy_test_backup_url") for attempt in range(1, max_attempts + 1): if attempt > 1: logger.info(f"Retrying proxy test (attempt {attempt}/{max_attempts})") update_status_bar("proxy_test_retrying", attempt=attempt, max_attempts=max_attempts) time.sleep(2) # 重试前等待2秒 success, error_msg, status_code = _perform_proxy_test_single(proxy_address, current_url) if success: return True, get_text("proxy_test_success", url=current_url) # 如果是503错误或超时，值得重试 should_retry = ( status_code == 503 or "timeout" in error_msg.lower() or "temporarily unavailable" in error_msg.lower() ) if not should_retry: # 对于非临时性错误，不重试，直接尝试下一个URL logger.info(f"Non-retryable error for {current_url}: {error_msg}") break if attempt == max_attempts: logger.warning(f"All {max_attempts} attempts failed for {current_url}: {error_msg}") # 所有URL和重试都失败了 return False, get_text("proxy_test_all_failed") def _proxy_test_thread(proxy_addr: str, test_url: str): """在后台线程中执行代理测试""" try: success, message = _perform_proxy_test(proxy_addr, test_url) # 在主线程中更新GUI def update_gui(): if success: messagebox.showinfo(get_text("info_title"), message, parent=root_widget) update_status_bar("proxy_test_success_status", url=test_url) else: messagebox.showerror(get_text("error_title"), get_text("proxy_test_failure", url=test_url, error=message), parent=root_widget) update_status_bar("proxy_test_failure_status", error=message) if root_widget: root_widget.after_idle(update_gui) except Exception as e: logger.error(f"Proxy test thread error: {e}", exc_info=True) def show_error(): messagebox.showerror(get_text("error_title"), f"代理测试过程中发生错误: {e}", parent=root_widget) update_status_bar("proxy_test_failure_status", error=str(e)) if root_widget: root_widget.after_idle(show_error) def test_proxy_connectivity_gui(): if not proxy_enabled_var.get() or not proxy_address_var.get().strip(): messagebox.showwarning(get_text("warning_title"), get_text("proxy_not_enabled_warn"), parent=root_widget) return proxy_addr_to_test = proxy_address_var.get().strip() test_url = LANG_TEXTS["proxy_test_url_default"] # Use the default from LANG_TEXTS # 显示测试开始状态 update_status_bar("proxy_testing_status", proxy_addr=proxy_addr_to_test) # 在后台线程中执行测试，避免阻塞GUI test_thread = threading.Thread( target=_proxy_test_thread, args=(proxy_addr_to_test, test_url), daemon=True ) test_thread.start() def stop_selected_pid_from_list_gui(): if not pid_listbox_widget: return selected_indices = pid_listbox_widget.curselection() if not selected_indices: messagebox.showwarning(get_text("warning_title"), get_text("pid_list_empty_for_stop_warn"), parent=root_widget) return selected_text = pid_listbox_widget.get(selected_indices[0]).strip() pid_to_stop = -1 process_name_to_stop = get_text("unknown_process_name_placeholder") try: # Check for "no process" entry first, as it's a known non-PID format no_process_indicator_zh = get_text("port_not_in_use_format", port_type="_", port_num="_").split("] ")[-1].strip() no_process_indicator_en = LANG_TEXTS["port_not_in_use_format"]["en"].split("] ")[-1].strip() general_no_pids_msg_zh = get_text("no_pids_found") general_no_pids_msg_en = LANG_TEXTS["no_pids_found"]["en"] is_no_process_entry = (no_process_indicator_zh in selected_text or \ no_process_indicator_en in selected_text or \ selected_text == general_no_pids_msg_zh or \ selected_text == general_no_pids_msg_en) if is_no_process_entry: logger.info(f"Selected item is a 'no process' entry: {selected_text}") return # Silently return for "no process" entries # Try to parse the format: "[Type - Port] PID - Name (Path)" or "PID - Name (Path)" # This regex will match either the detailed format or the simple "PID - Name" format # It's flexible enough to handle the optional leading "[...]" part match = re.match(r"^(?:\[[^\]]+\]\s*)?(\d+)\s*-\s*(.*)$", selected_text) if match: pid_to_stop = int(match.group(1)) process_name_to_stop = match.group(2).strip() elif selected_text.isdigit(): # Handles if the listbox item is just a PID pid_to_stop = int(selected_text) # process_name_to_stop remains the default unknown else: # Genuine parsing error for an unexpected format messagebox.showerror(get_text("error_title"), get_text("error_parsing_pid", selection=selected_text), parent=root_widget) return except ValueError: # Catches int() conversion errors messagebox.showerror(get_text("error_title"), get_text("error_parsing_pid", selection=selected_text), parent=root_widget) return # If pid_to_stop is still -1 at this point, it means an unhandled case or logic error in parsing. # The returns above should prevent reaching here with pid_to_stop == -1 if it's an error or "no process". if pid_to_stop == -1: # This path implies a non-parsable string that wasn't identified as a "no process" message and didn't raise ValueError. logger.warning(f"PID parsing resulted in -1 for non-'no process' entry: {selected_text}. This indicates an unexpected format or logic gap.") messagebox.showerror(get_text("error_title"), get_text("error_parsing_pid", selection=selected_text), parent=root_widget) return if messagebox.askyesno(get_text("confirm_stop_pid_title"), get_text("confirm_stop_pid_message", pid=pid_to_stop, name=process_name_to_stop), parent=root_widget): normal_kill_success = kill_process_pid(pid_to_stop) if normal_kill_success: messagebox.showinfo(get_text("info_title"), get_text("terminate_request_sent", pid=pid_to_stop, name=process_name_to_stop), parent=root_widget) else: # 普通权限停止失败，询问是否尝试管理员权限 if messagebox.askyesno(get_text("confirm_stop_pid_admin_title"), get_text("confirm_stop_pid_admin_message", pid=pid_to_stop, name=process_name_to_stop), parent=root_widget): admin_kill_success = kill_process_pid_admin(pid_to_stop) if admin_kill_success: messagebox.showinfo(get_text("info_title"), get_text("admin_stop_success", pid=pid_to_stop), parent=root_widget) else: messagebox.showwarning(get_text("warning_title"), get_text("admin_stop_failure", pid=pid_to_stop, error="未知错误"), parent=root_widget) else: messagebox.showwarning(get_text("warning_title"), get_text("terminate_attempt_failed", pid=pid_to_stop, name=process_name_to_stop), parent=root_widget) query_port_and_display_pids_gui() def kill_process_pid_admin(pid: int) -> bool: """使用管理员权限尝试终止进程。""" system = platform.system() success = False logger.info(f"尝试以管理员权限终止进程 PID: {pid} (系统: {system})") try: if system == "Windows": # 在Windows上使用PowerShell以管理员权限运行taskkill import ctypes if ctypes.windll.shell32.IsUserAnAdmin() == 0: # 如果当前不是管理员，则尝试用管理员权限启动新进程 # 准备 PowerShell 命令 logger.info(f"当前非管理员权限，使用PowerShell提升权限") ps_cmd = f"Start-Process -Verb RunAs taskkill -ArgumentList '/PID {pid} /F /T'" logger.debug(f"执行PowerShell命令: {ps_cmd}") result = subprocess.run(["powershell", "-Command", ps_cmd], capture_output=True, text=True, creationflags=subprocess.CREATE_NO_WINDOW) logger.info(f"PowerShell命令结果: 返回码={result.returncode}, 输出={result.stdout}, 错误={result.stderr}") success = result.returncode == 0 else: # 如果已经是管理员，则直接运行taskkill logger.info(f"当前已是管理员权限，直接执行taskkill") result = subprocess.run(["taskkill", "/PID", str(pid), "/F", "/T"], capture_output=True, text=True, creationflags=subprocess.CREATE_NO_WINDOW) logger.info(f"Taskkill命令结果: 返回码={result.returncode}, 输出={result.stdout}, 错误={result.stderr}") success = result.returncode == 0 elif system in ["Linux", "Darwin"]: # Linux或macOS # 使用sudo尝试终止进程 logger.info(f"使用sudo在新终端中终止进程") cmd = ["sudo", "kill", "-9", str(pid)] # 对于GUI程序，我们需要让用户在终端输入密码，所以使用新终端窗口 if system == "Darwin": # macOS logger.info(f"在macOS上使用AppleScript打开Terminal并执行sudo命令") applescript = f'tell application "Terminal" to do script "sudo kill -9 {pid}"' result = subprocess.run(["osascript", "-e", applescript], capture_output=True, text=True) logger.info(f"AppleScript结果: 返回码={result.returncode}, 输出={result.stdout}, 错误={result.stderr}") success = result.returncode == 0 else: # Linux # 查找可用的终端模拟器 import shutil logger.info(f"在Linux上查找可用的终端模拟器") terminal_emulator = shutil.which("x-terminal-emulator") or shutil.which("gnome-terminal") or \ shutil.which("konsole") or shutil.which("xfce4-terminal") or shutil.which("xterm") if terminal_emulator: logger.info(f"使用终端模拟器: {terminal_emulator}") if "gnome-terminal" in terminal_emulator: logger.info(f"针对gnome-terminal的特殊处理") result = subprocess.run([terminal_emulator, "--", "sudo", "kill", "-9", str(pid)]) else: logger.info(f"使用通用终端启动命令") result = subprocess.run([terminal_emulator, "-e", f"sudo kill -9 {pid}"]) logger.info(f"终端命令结果: 返回码={result.returncode}") success = result.returncode == 0 else: # 如果找不到终端模拟器，尝试直接使用sudo logger.warning(f"未找到终端模拟器，尝试直接使用sudo (可能需要当前进程已有sudo权限)") result = subprocess.run(["sudo", "kill", "-9", str(pid)], capture_output=True, text=True) logger.info(f"直接sudo命令结果: 返回码={result.returncode}, 输出={result.stdout}, 错误={result.stderr}") success = result.returncode == 0 except Exception as e: logger.error(f"使用管理员权限终止PID {pid}时出错: {e}", exc_info=True) success = False logger.info(f"管理员权限终止进程 PID: {pid} 结果: {'成功' if success else '失败'}") return success def kill_custom_pid_gui(): if not custom_pid_entry_var or not root_widget: return pid_str = custom_pid_entry_var.get() if not pid_str: messagebox.showwarning(get_text("warning_title"), get_text("pid_input_empty_warn"), parent=root_widget) return if not pid_str.isdigit(): messagebox.showwarning(get_text("warning_title"), get_text("pid_input_invalid_warn"), parent=root_widget) return pid_to_kill = int(pid_str) process_name_to_kill = get_process_name_by_pid(pid_to_kill) confirm_msg = get_text("confirm_stop_pid_message", pid=pid_to_kill, name=process_name_to_kill) if messagebox.askyesno(get_text("confirm_kill_custom_pid_title"), confirm_msg, parent=root_widget): normal_kill_success = kill_process_pid(pid_to_kill) if normal_kill_success: messagebox.showinfo(get_text("info_title"), get_text("terminate_request_sent", pid=pid_to_kill, name=process_name_to_kill), parent=root_widget) else: # 普通权限停止失败，询问是否尝试管理员权限 if messagebox.askyesno(get_text("confirm_stop_pid_admin_title"), get_text("confirm_stop_pid_admin_message", pid=pid_to_kill, name=process_name_to_kill), parent=root_widget): admin_kill_success = kill_process_pid_admin(pid_to_kill) if admin_kill_success: messagebox.showinfo(get_text("info_title"), get_text("admin_stop_success", pid=pid_to_kill), parent=root_widget) else: messagebox.showwarning(get_text("warning_title"), get_text("admin_stop_failure", pid=pid_to_kill, error="未知错误"), parent=root_widget) else: messagebox.showwarning(get_text("warning_title"), get_text("terminate_attempt_failed", pid=pid_to_kill, name=process_name_to_kill), parent=root_widget) custom_pid_entry_var.set("") query_port_and_display_pids_gui() menu_bar_ref: Optional[tk.Menu] = None def update_all_ui_texts_gui(): if not root_widget: return root_widget.title(get_text("title")) for item in widgets_to_translate: widget = item["widget"] key = item["key"] prop = item.get("property", "text") text_val = get_text(key, **item.get("kwargs", {})) if hasattr(widget, 'config'): try: widget.config(**{prop: text_val}) except tk.TclError: pass current_status_text = process_status_text_var.get() if process_status_text_var else "" is_idle_status = any(current_status_text == LANG_TEXTS["status_idle"].get(lang_code, "") for lang_code in LANG_TEXTS["status_idle"]) if is_idle_status: update_status_bar("status_idle") def switch_language_gui(lang_code: str): global current_language if lang_code in LANG_TEXTS["title"]: current_language = lang_code update_all_ui_texts_gui() def build_gui(root: tk.Tk): global process_status_text_var, port_entry_var, camoufox_debug_port_var, pid_listbox_widget, widgets_to_translate, managed_process_info, root_widget, menu_bar_ref, custom_pid_entry_var global stream_port_enabled_var, stream_port_var, helper_enabled_var, helper_endpoint_var, port_auto_check_var, proxy_address_var, proxy_enabled_var global active_auth_file_display_var # 添加新的全局变量 global pid_list_lbl_frame_ref # 确保全局变量在此处声明 root_widget = root root.title(get_text("title")) root.minsize(950, 600) # 加载保存的配置 config = load_config() s = ttk.Style() s.configure('TButton', padding=3) s.configure('TLabelFrame.Label', font=('Default', 10, 'bold')) s.configure('TLabelFrame', padding=4) try: os.makedirs(ACTIVE_AUTH_DIR, exist_ok=True) os.makedirs(SAVED_AUTH_DIR, exist_ok=True) except OSError as e: messagebox.showerror(get_text("error_title"), f"无法创建认证目录: {e}") process_status_text_var = tk.StringVar(value=get_text("status_idle")) port_entry_var = tk.StringVar(value=str(config.get("fastapi_port", DEFAULT_FASTAPI_PORT))) camoufox_debug_port_var = tk.StringVar(value=str(config.get("camoufox_debug_port", DEFAULT_CAMOUFOX_PORT_GUI))) custom_pid_entry_var = tk.StringVar() stream_port_enabled_var = tk.BooleanVar(value=config.get("stream_port_enabled", True)) stream_port_var = tk.StringVar(value=str(config.get("stream_port", "3120"))) helper_enabled_var = tk.BooleanVar(value=config.get("helper_enabled", False)) helper_endpoint_var = tk.StringVar(value=config.get("helper_endpoint", "")) port_auto_check_var = tk.BooleanVar(value=True) proxy_address_var = tk.StringVar(value=config.get("proxy_address", "http://127.0.0.1:7890")) proxy_enabled_var = tk.BooleanVar(value=config.get("proxy_enabled", False)) active_auth_file_display_var = tk.StringVar() # 初始化为空，后续由 _update_active_auth_display 更新 # 联动逻辑：移除强制启用代理的逻辑，现在代理配置更加灵活 # 用户可以根据需要独立配置流式代理和浏览器代理 def on_stream_proxy_toggle(*args): # 不再强制启用代理，用户可以自由选择 pass stream_port_enabled_var.trace_add("write", on_stream_proxy_toggle) menu_bar_ref = tk.Menu(root) lang_menu = tk.Menu(menu_bar_ref, tearoff=0) lang_menu.add_command(label="中文 (Chinese)", command=lambda: switch_language_gui('zh')) lang_menu.add_command(label="English", command=lambda: switch_language_gui('en')) menu_bar_ref.add_cascade(label="Language", menu=lang_menu) root.config(menu=menu_bar_ref) # --- 主 PanedWindow 实现三栏 --- main_paned_window = ttk.PanedWindow(root, orient=tk.HORIZONTAL) main_paned_window.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) # --- 左栏 Frame --- left_frame_container = ttk.Frame(main_paned_window, padding="5") main_paned_window.add(left_frame_container, weight=3) # 增大左栏初始权重 left_frame_container.columnconfigure(0, weight=1) # 配置行权重，使得launch_options_frame和auth_section之间可以有空白，或者让它们紧凑排列 # 假设 port_section, launch_options_frame, auth_section 依次排列 left_frame_container.rowconfigure(0, weight=0) # port_section left_frame_container.rowconfigure(1, weight=0) # launch_options_frame left_frame_container.rowconfigure(2, weight=0) # auth_section (移到此处后) left_frame_container.rowconfigure(3, weight=1) # 添加一个占位符Frame，使其填充剩余空间 left_current_row = 0 # 端口配置部分 port_section = ttk.LabelFrame(left_frame_container, text="") port_section.grid(row=left_current_row, column=0, sticky="ew", padx=2, pady=(2,10)) widgets_to_translate.append({"widget": port_section, "key": "port_section_label", "property": "text"}) left_current_row += 1 # 添加重置按钮和服务关闭指南按钮 port_controls_frame = ttk.Frame(port_section) port_controls_frame.pack(fill=tk.X, padx=5, pady=3) btn_reset = ttk.Button(port_controls_frame, text="", command=reset_to_defaults) btn_reset.pack(side=tk.LEFT, padx=(0,5)) widgets_to_translate.append({"widget": btn_reset, "key": "reset_button"}) btn_closing_guide = ttk.Button(port_controls_frame, text="", command=show_service_closing_guide) btn_closing_guide.pack(side=tk.RIGHT, padx=(5,0)) widgets_to_translate.append({"widget": btn_closing_guide, "key": "service_closing_guide_btn"}) # (内部控件保持在port_section中，使用pack使其紧凑) # FastAPI Port fastapi_frame = ttk.Frame(port_section) fastapi_frame.pack(fill=tk.X, padx=5, pady=3) lbl_port = ttk.Label(fastapi_frame, text="") lbl_port.pack(side=tk.LEFT, padx=(0,5)) widgets_to_translate.append({"widget": lbl_port, "key": "fastapi_port_label"}) entry_port = ttk.Entry(fastapi_frame, textvariable=port_entry_var, width=12) entry_port.pack(side=tk.LEFT, expand=True, fill=tk.X) # Camoufox Debug Port camoufox_frame = ttk.Frame(port_section) camoufox_frame.pack(fill=tk.X, padx=5, pady=3) lbl_camoufox_debug_port = ttk.Label(camoufox_frame, text="") lbl_camoufox_debug_port.pack(side=tk.LEFT, padx=(0,5)) widgets_to_translate.append({"widget": lbl_camoufox_debug_port, "key": "camoufox_debug_port_label"}) entry_camoufox_debug_port = ttk.Entry(camoufox_frame, textvariable=camoufox_debug_port_var, width=12) entry_camoufox_debug_port.pack(side=tk.LEFT, expand=True, fill=tk.X) # Stream Proxy Port stream_port_frame_outer = ttk.Frame(port_section) stream_port_frame_outer.pack(fill=tk.X, padx=5, pady=3) stream_port_checkbox = ttk.Checkbutton(stream_port_frame_outer, variable=stream_port_enabled_var, text="") stream_port_checkbox.pack(side=tk.LEFT, padx=(0,2)) widgets_to_translate.append({"widget": stream_port_checkbox, "key": "enable_stream_proxy_label", "property": "text"}) stream_port_details_frame = ttk.Frame(stream_port_frame_outer) stream_port_details_frame.pack(side=tk.LEFT, fill=tk.X, expand=True) lbl_stream_port = ttk.Label(stream_port_details_frame, text="") lbl_stream_port.pack(side=tk.LEFT, padx=(0,5)) widgets_to_translate.append({"widget": lbl_stream_port, "key": "stream_proxy_port_label"}) entry_stream_port = ttk.Entry(stream_port_details_frame, textvariable=stream_port_var, width=10) entry_stream_port.pack(side=tk.LEFT, expand=True, fill=tk.X) # Helper Service helper_frame_outer = ttk.Frame(port_section) helper_frame_outer.pack(fill=tk.X, padx=5, pady=3) helper_checkbox = ttk.Checkbutton(helper_frame_outer, variable=helper_enabled_var, text="") helper_checkbox.pack(side=tk.LEFT, padx=(0,2)) widgets_to_translate.append({"widget": helper_checkbox, "key": "enable_helper_label", "property": "text"}) helper_details_frame = ttk.Frame(helper_frame_outer) helper_details_frame.pack(side=tk.LEFT, fill=tk.X, expand=True) lbl_helper_endpoint = ttk.Label(helper_details_frame, text="") lbl_helper_endpoint.pack(side=tk.LEFT, padx=(0,5)) widgets_to_translate.append({"widget": lbl_helper_endpoint, "key": "helper_endpoint_label"}) entry_helper_endpoint = ttk.Entry(helper_details_frame, textvariable=helper_endpoint_var) entry_helper_endpoint.pack(side=tk.LEFT, fill=tk.X, expand=True) # 添加分隔符 ttk.Separator(port_section, orient=tk.HORIZONTAL).pack(fill=tk.X, padx=5, pady=(8,5)) # 代理配置部分 - 独立的LabelFrame proxy_section = ttk.LabelFrame(port_section, text="") proxy_section.pack(fill=tk.X, padx=5, pady=(5,8)) widgets_to_translate.append({"widget": proxy_section, "key": "proxy_section_label", "property": "text"}) # 代理启用复选框 proxy_enable_frame = ttk.Frame(proxy_section) proxy_enable_frame.pack(fill=tk.X, padx=5, pady=(5,3)) proxy_checkbox = ttk.Checkbutton(proxy_enable_frame, variable=proxy_enabled_var, text="") proxy_checkbox.pack(side=tk.LEFT) widgets_to_translate.append({"widget": proxy_checkbox, "key": "enable_proxy_label", "property": "text"}) # 代理地址输入 proxy_address_frame = ttk.Frame(proxy_section) proxy_address_frame.pack(fill=tk.X, padx=5, pady=(0,5)) lbl_proxy_address = ttk.Label(proxy_address_frame, text="") lbl_proxy_address.pack(side=tk.LEFT, padx=(0,5)) widgets_to_translate.append({"widget": lbl_proxy_address, "key": "proxy_address_label"}) entry_proxy_address = ttk.Entry(proxy_address_frame, textvariable=proxy_address_var) entry_proxy_address.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=(0,5)) # 代理测试按钮 btn_test_proxy_inline = ttk.Button(proxy_address_frame, text="", command=test_proxy_connectivity_gui, width=8) btn_test_proxy_inline.pack(side=tk.RIGHT) widgets_to_translate.append({"widget": btn_test_proxy_inline, "key": "test_proxy_btn"}) # Port auto check port_auto_check_frame = ttk.Frame(port_section) port_auto_check_frame.pack(fill=tk.X, padx=5, pady=3) port_auto_check_btn = ttk.Checkbutton(port_auto_check_frame, variable=port_auto_check_var, text="") port_auto_check_btn.pack(side=tk.LEFT) widgets_to_translate.append({"widget": port_auto_check_btn, "key": "port_auto_check", "property": "text"}) # 启动选项部分 launch_options_frame = ttk.LabelFrame(left_frame_container, text="") launch_options_frame.grid(row=left_current_row, column=0, sticky="ew", padx=2, pady=5) widgets_to_translate.append({"widget": launch_options_frame, "key": "launch_options_label", "property": "text"}) left_current_row += 1 lbl_launch_options_note = ttk.Label(launch_options_frame, text="", wraplength=240) # 调整wraplength lbl_launch_options_note.pack(fill=tk.X, padx=5, pady=(5, 8)) widgets_to_translate.append({"widget": lbl_launch_options_note, "key": "launch_options_note_revised"}) # (启动按钮) btn_headed = ttk.Button(launch_options_frame, text="", command=start_headed_interactive_gui) btn_headed.pack(fill=tk.X, padx=5, pady=3) widgets_to_translate.append({"widget": btn_headed, "key": "launch_headed_interactive_btn"}) btn_headless = ttk.Button(launch_options_frame, text="", command=start_headless_gui) # command 和 key 修改 btn_headless.pack(fill=tk.X, padx=5, pady=3) widgets_to_translate.append({"widget": btn_headless, "key": "launch_headless_btn"}) # key 修改 btn_virtual_display = ttk.Button(launch_options_frame, text="", command=start_virtual_display_gui) btn_virtual_display.pack(fill=tk.X, padx=5, pady=3) widgets_to_translate.append({"widget": btn_virtual_display, "key": "launch_virtual_display_btn"}) if platform.system() != "Linux": btn_virtual_display.state(['disabled']) # Separator for LLM service buttons ttk.Separator(launch_options_frame, orient=tk.HORIZONTAL).pack(fill=tk.X, padx=5, pady=(8,5)) # LLM Service Buttons btn_start_llm_service = ttk.Button(launch_options_frame, text="", command=start_llm_service_gui) btn_start_llm_service.pack(fill=tk.X, padx=5, pady=3) widgets_to_translate.append({"widget": btn_start_llm_service, "key": "launch_llm_service_btn"}) btn_stop_llm_service = ttk.Button(launch_options_frame, text="", command=stop_llm_service_gui) btn_stop_llm_service.pack(fill=tk.X, padx=5, pady=3) widgets_to_translate.append({"widget": btn_stop_llm_service, "key": "stop_llm_service_btn"}) # 移除不再有用的"停止当前GUI管理的服务"按钮 # btn_stop_service = ttk.Button(launch_options_frame, text="", command=stop_managed_service_gui) # btn_stop_service.pack(fill=tk.X, padx=5, pady=3) # widgets_to_translate.append({"widget": btn_stop_service, "key": "stop_gui_service_btn"}) # 添加一个占位符Frame以推高左侧内容 (如果需要消除底部所有空白) spacer_frame_left = ttk.Frame(left_frame_container) spacer_frame_left.grid(row=left_current_row, column=0, sticky="nsew") left_frame_container.rowconfigure(left_current_row, weight=1) # 让这个spacer扩展 # --- 中栏 Frame --- middle_frame_container = ttk.Frame(main_paned_window, padding="5") main_paned_window.add(middle_frame_container, weight=2) # 调整中栏初始权重 middle_frame_container.columnconfigure(0, weight=1) middle_frame_container.rowconfigure(0, weight=1) middle_frame_container.rowconfigure(1, weight=0) middle_frame_container.rowconfigure(2, weight=0) # 认证管理现在在中栏 middle_current_row = 0 pid_section_frame = ttk.Frame(middle_frame_container) pid_section_frame.grid(row=middle_current_row, column=0, sticky="nsew", padx=2, pady=2) pid_section_frame.columnconfigure(0, weight=1) pid_section_frame.rowconfigure(0, weight=1) middle_current_row +=1 global pid_list_lbl_frame_ref pid_list_lbl_frame_ref = ttk.LabelFrame(pid_section_frame, text=get_text("static_pid_list_title")) # 使用新的固定标题 pid_list_lbl_frame_ref.grid(row=0, column=0, columnspan=2, sticky="nsew", padx=2, pady=2) pid_list_lbl_frame_ref.columnconfigure(0, weight=1) pid_list_lbl_frame_ref.rowconfigure(0, weight=1) pid_listbox_widget = tk.Listbox(pid_list_lbl_frame_ref, height=4, exportselection=False) pid_listbox_widget.grid(row=0, column=0, padx=5, pady=5, sticky="nsew") scrollbar = ttk.Scrollbar(pid_list_lbl_frame_ref, orient="vertical", command=pid_listbox_widget.yview) scrollbar.grid(row=0, column=1, sticky="ns", padx=(0,5), pady=5) pid_listbox_widget.config(yscrollcommand=scrollbar.set) pid_buttons_frame = ttk.Frame(pid_section_frame) pid_buttons_frame.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(5,2)) pid_buttons_frame.columnconfigure(0, weight=1) pid_buttons_frame.columnconfigure(1, weight=1) btn_query = ttk.Button(pid_buttons_frame, text="", command=query_port_and_display_pids_gui) btn_query.grid(row=0, column=0, sticky="ew", padx=(0,2)) widgets_to_translate.append({"widget": btn_query, "key": "query_pids_btn"}) btn_stop_pid = ttk.Button(pid_buttons_frame, text="", command=stop_selected_pid_from_list_gui) btn_stop_pid.grid(row=0, column=1, sticky="ew", padx=(2,0)) widgets_to_translate.append({"widget": btn_stop_pid, "key": "stop_selected_pid_btn"}) # 代理测试按钮已移至代理配置部分，此处不再重复 kill_custom_frame = ttk.LabelFrame(middle_frame_container, text="") kill_custom_frame.grid(row=middle_current_row, column=0, sticky="ew", padx=2, pady=5) widgets_to_translate.append({"widget": kill_custom_frame, "key": "kill_custom_pid_label", "property":"text"}) middle_current_row += 1 kill_custom_frame.columnconfigure(0, weight=1) entry_custom_pid = ttk.Entry(kill_custom_frame, textvariable=custom_pid_entry_var, width=10) entry_custom_pid.pack(side=tk.LEFT, padx=5, pady=5, fill=tk.X, expand=True) btn_kill_custom_pid = ttk.Button(kill_custom_frame, text="", command=kill_custom_pid_gui) btn_kill_custom_pid.pack(side=tk.LEFT, padx=5, pady=5) widgets_to_translate.append({"widget": btn_kill_custom_pid, "key": "kill_custom_pid_btn"}) # 认证文件管理 (移到中栏PID终止功能下方) auth_section_middle = ttk.LabelFrame(middle_frame_container, text="") auth_section_middle.grid(row=middle_current_row, column=0, sticky="ew", padx=2, pady=5) widgets_to_translate.append({"widget": auth_section_middle, "key": "auth_files_management", "property": "text"}) middle_current_row += 1 btn_manage_auth_middle = ttk.Button(auth_section_middle, text="", command=manage_auth_files_gui) btn_manage_auth_middle.pack(fill=tk.X, padx=5, pady=5) widgets_to_translate.append({"widget": btn_manage_auth_middle, "key": "manage_auth_files_btn"}) # 显示当前认证文件 auth_display_frame = ttk.Frame(auth_section_middle) auth_display_frame.pack(fill=tk.X, padx=5, pady=(0,5)) lbl_current_auth_static = ttk.Label(auth_display_frame, text="") lbl_current_auth_static.pack(side=tk.LEFT) widgets_to_translate.append({"widget": lbl_current_auth_static, "key": "current_auth_file_display_label"}) lbl_current_auth_dynamic = ttk.Label(auth_display_frame, textvariable=active_auth_file_display_var, wraplength=180) lbl_current_auth_dynamic.pack(side=tk.LEFT, fill=tk.X, expand=True) # --- 右栏 Frame --- right_frame_container = ttk.Frame(main_paned_window, padding="5") main_paned_window.add(right_frame_container, weight=2) # 调整右栏初始权重，使其相对小一些 right_frame_container.columnconfigure(0, weight=1) right_frame_container.rowconfigure(1, weight=1) right_current_row = 0 status_area_frame = ttk.LabelFrame(right_frame_container, text="") status_area_frame.grid(row=right_current_row, column=0, padx=2, pady=2, sticky="ew") widgets_to_translate.append({"widget": status_area_frame, "key": "status_label", "property": "text"}) right_current_row += 1 lbl_status_val = ttk.Label(status_area_frame, textvariable=process_status_text_var, wraplength=280) lbl_status_val.pack(fill=tk.X, padx=5, pady=5) def rewrap_status_label(event=None): if root_widget and lbl_status_val.winfo_exists(): new_width = status_area_frame.winfo_width() - 20 if new_width > 100: lbl_status_val.config(wraplength=new_width) status_area_frame.bind("", rewrap_status_label) output_log_area_frame = ttk.LabelFrame(right_frame_container, text="") output_log_area_frame.grid(row=right_current_row, column=0, padx=2, pady=2, sticky="nsew") widgets_to_translate.append({"widget": output_log_area_frame, "key": "output_label", "property": "text"}) output_log_area_frame.columnconfigure(0, weight=1) output_log_area_frame.rowconfigure(0, weight=1) output_scrolled_text = scrolledtext.ScrolledText(output_log_area_frame, height=10, width=35, wrap=tk.WORD, state=tk.DISABLED) # 调整宽度 output_scrolled_text.grid(row=0, column=0, padx=5, pady=5, sticky="nsew") managed_process_info["output_area"] = output_scrolled_text update_all_ui_texts_gui() query_port_and_display_pids_gui() # 初始化时查询一次FastAPI端口 _update_active_auth_display() # 初始化时更新认证文件显示 root.protocol("WM_DELETE_WINDOW", on_app_close_main) pid_list_lbl_frame_ref: Optional[ttk.LabelFrame] = None # 新增辅助函数用于获取和验证启动参数 def _get_launch_parameters() -> Optional[Dict[str, Any]]: """从GUI收集并验证启动参数。如果无效则返回None。""" params = {} try: params["fastapi_port"] = get_fastapi_port_from_gui() params["camoufox_debug_port"] = get_camoufox_debug_port_from_gui() params["stream_port_enabled"] = stream_port_enabled_var.get() sp_val_str = stream_port_var.get().strip() if params["stream_port_enabled"]: params["stream_port"] = int(sp_val_str) if sp_val_str else 3120 if not (params["stream_port"] == 0 or 1024 <= params["stream_port"] <= 65535): messagebox.showwarning(get_text("warning_title"), get_text("stream_port_out_of_range")) return None else: params["stream_port"] = 0 # 如果未启用，则端口视为0（禁用） params["helper_enabled"] = helper_enabled_var.get() params["helper_endpoint"] = helper_endpoint_var.get().strip() if params["helper_enabled"] else "" return params except ValueError: # 通常来自 int() 转换失败 messagebox.showwarning(get_text("warning_title"), get_text("enter_valid_port_warn")) # 或者更具体的错误 return None except Exception as e: messagebox.showerror(get_text("error_title"), f"获取启动参数时出错: {e}") return None # 更新on_app_close_main函数，反映服务独立性 def on_app_close_main(): # 保存当前配置 save_config() # Attempt to stop LLM service if it's running if is_llm_service_running(): logger.info("LLM service is running. Attempting to stop it before exiting GUI.") # We can call stop_llm_service_gui directly, but it shows a confirmation. # For closing, we might want a more direct stop or a specific "closing" stop. # For now, let's try a direct stop without user confirmation for this specific path. popen = llm_service_process_info.get("popen") service_name = get_text(llm_service_process_info.get("service_name_key", "llm_service_name_key")) if popen: try: logger.info(f"Sending SIGINT to {service_name} (PID: {popen.pid}) during app close.") if platform.system() == "Windows": popen.terminate() # TerminateProcess on Windows else: popen.send_signal(signal.SIGINT) # Give it a very short time to exit, don't block GUI closing for too long popen.wait(timeout=1.5) logger.info(f"{service_name} (PID: {popen.pid}) hopefully stopped during app close.") except subprocess.TimeoutExpired: logger.warning(f"{service_name} (PID: {popen.pid}) did not stop quickly during app close. May need manual cleanup.") popen.kill() # Force kill if it didn't stop except Exception as e: logger.error(f"Error stopping {service_name} during app close: {e}") finally: llm_service_process_info["popen"] = None # Clear it # 服务都是在独立终端中启动的，所以只需确认用户是否想关闭GUI if messagebox.askyesno(get_text("confirm_quit_title"), get_text("confirm_quit_message"), parent=root_widget): if root_widget: root_widget.destroy() def show_service_closing_guide(): messagebox.showinfo( get_text("service_closing_guide"), get_text("service_closing_guide_message"), parent=root_widget ) if __name__ == "__main__": if not os.path.exists(LAUNCH_CAMOUFOX_PY) or not os.path.exists(os.path.join(SCRIPT_DIR, SERVER_PY_FILENAME)): err_lang = current_language err_title_key = "startup_error_title" err_msg_key = "startup_script_not_found_msgbox" err_title = LANG_TEXTS[err_title_key].get(err_lang, LANG_TEXTS[err_title_key]['en']) err_msg_template = LANG_TEXTS[err_msg_key].get(err_lang, LANG_TEXTS[err_msg_key]['en']) err_msg = err_msg_template.format(script=f"{os.path.basename(LAUNCH_CAMOUFOX_PY)} or {SERVER_PY_FILENAME}") try: root_err = tk.Tk(); root_err.withdraw() messagebox.showerror(err_title, err_msg, parent=None) root_err.destroy() except tk.TclError: print(f"ERROR: {err_msg}", file=sys.stderr) sys.exit(1) app_root = tk.Tk() build_gui(app_root) app_root.mainloop()