haosicheng/shuidrop_gui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[patch] 新增多实例测试开关控制流 逻辑

Browse Source
This commit is contained in:
kris 郝
2025-11-03 11:31:20 +08:00
parent 2083516b8c
commit 8a5da5e906
3 changed files with 195 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
WebSocket/backend_singleton.py
View File

@@ -194,12 +194,19 @@ class WebSocketManager:
"""连接后端WebSocket""" """连接后端WebSocket"""
try: try:
# 1 保存token到配置 # 🔥 根据配置决定是否保存token
try: # 生产模式：保存token，方便用户下次自动加载
from config import set_saved_token # 测试模式：不保存，避免多实例冲突
set_saved_token(token) import config as cfg
except Exception: if not cfg.is_multi_instance_mode():
pass try:
from config import set_saved_token
set_saved_token(token)
self._log("生产模式：已保存token到配置文件", "INFO")
except Exception as e:
self._log(f"保存token失败: {e}", "WARNING")
else:
self._log("测试模式：不保存token（支持多实例运行）", "INFO")
# 2 获取或创建后端客户端 # 2 获取或创建后端客户端
backend = get_backend_client() backend = get_backend_client()

303
config.py
View File

@@ -1,136 +1,169 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
项目配置文件 项目配置文件
统一管理所有配置参数，避免硬编码 统一管理所有配置参数，避免硬编码
""" """
# 用户访问令牌（默认空字符串） # 用户访问令牌（默认空字符串）
import os # 用于路径与目录操作（写入用户配置目录） import os # 用于路径与目录操作（写入用户配置目录）
import json # 用于将令牌保存为 JSON 格式 import json # 用于将令牌保存为 JSON 格式
# 后端服务器配置 # 后端服务器配置
# BACKEND_HOST = "192.168.5.233" # BACKEND_HOST = "192.168.5.233"
# BACKEND_HOST = "192.168.5.106" # BACKEND_HOST = "192.168.5.106"
BACKEND_HOST = "192.168.5.12" BACKEND_HOST = "192.168.5.12"
# BACKEND_HOST = "shuidrop.com" # BACKEND_HOST = "shuidrop.com"
# BACKEND_HOST = "test.shuidrop.com" # BACKEND_HOST = "test.shuidrop.com"
BACKEND_PORT = "8000" BACKEND_PORT = "8000"
# BACKEND_PORT = "" # BACKEND_PORT = ""
BACKEND_WS_URL = f"ws://{BACKEND_HOST}:{BACKEND_PORT}" BACKEND_WS_URL = f"ws://{BACKEND_HOST}:{BACKEND_PORT}"
# BACKEND_WS_URL = f"wss://{BACKEND_HOST}" # BACKEND_WS_URL = f"wss://{BACKEND_HOST}"
# WebSocket配置 # WebSocket配置
WS_CONNECT_TIMEOUT = 16.0 WS_CONNECT_TIMEOUT = 16.0
WS_MESSAGE_TIMEOUT = 30.0 WS_MESSAGE_TIMEOUT = 30.0
WS_PING_INTERVAL = 15 # 10秒ping间隔（提高检测频率） WS_PING_INTERVAL = 15 # 10秒ping间隔（提高检测频率）
WS_PING_TIMEOUT = 10 # 5秒ping超时（更快检测断线） WS_PING_TIMEOUT = 10 # 5秒ping超时（更快检测断线）
WS_ENABLE_PING = True # 是否启用WebSocket原生ping心跳 WS_ENABLE_PING = True # 是否启用WebSocket原生ping心跳
WS_ENABLE_APP_PING = False # 禁用应用层ping心跳（避免重复） WS_ENABLE_APP_PING = False # 禁用应用层ping心跳（避免重复）
# AI处理超时配置 # AI处理超时配置
AI_PROCESS_TIMEOUT = 30 # AI处理超时时间（秒） AI_PROCESS_TIMEOUT = 30 # AI处理超时时间（秒）
AI_LONG_PROCESS_THRESHOLD = 10 # AI长时间处理阈值（秒） AI_LONG_PROCESS_THRESHOLD = 10 # AI长时间处理阈值（秒）
# 内存管理配置 # 内存管理配置
MAX_PENDING_REPLIES = 100 MAX_PENDING_REPLIES = 100
CLEANUP_INTERVAL = 300 # 5分钟 CLEANUP_INTERVAL = 300 # 5分钟
FUTURE_TIMEOUT = 300 # 5分钟 FUTURE_TIMEOUT = 300 # 5分钟
# 终端日志配置（简化） # 终端日志配置（简化）
LOG_LEVEL = "INFO" LOG_LEVEL = "INFO"
VERSION = "1.0" VERSION = "1.0"
# GUI配置 # GUI配置
WINDOW_TITLE = "AI回复连接入口-V1.0" WINDOW_TITLE = "AI回复连接入口-V1.0"
# 应用版本号（用于版本检查） # 应用版本号（用于版本检查）
APP_VERSION = "1.5.66" APP_VERSION = "1.5.66"
# 平台特定配置 # 🔥 多实例运行模式开关
PLATFORMS = { # - True: 测试模式（多实例，不保存token，避免冲突）
"JD": { # - False: 生产模式（单实例，保存token，自动加载）
"name": "京东", #
"ws_url": "wss://dongdong.jd.com/workbench/websocket" # 使用方法：
}, # 1. 修改此值：MULTI_INSTANCE_MODE = False # 改为生产模式
"DOUYIN": { # 2. 或设置环境变量：SHUIDROP_MULTI_INSTANCE=0 # 临时切换到生产模式
"name": "抖音", MULTI_INSTANCE_MODE = True # 默认：测试模式
"ws_url": None # 动态获取
}, def is_multi_instance_mode() -> bool:
"QIANNIU": { """
"name": "千牛(淘宝)", 检查是否为多实例模式（支持环境变量覆盖）
"ws_url": "ws://127.0.0.1:3030"
}, 优先级：
"PDD": { 1. 环境变量 SHUIDROP_MULTI_INSTANCE（0=生产，1=测试）
"name": "拼多多", 2. 配置文件 MULTI_INSTANCE_MODE
"ws_url": None # 动态获取
} Returns:
} bool: True=多实例模式，False=单实例模式
"""
def get_backend_ws_url(platform: str, store_id: str) -> str: # 检查环境变量
"""获取旧版后端WebSocket URL（按店铺建连接，保留兼容）""" env_value = os.getenv('SHUIDROP_MULTI_INSTANCE')
return f"{BACKEND_WS_URL}/ws/platform/{platform.lower()}/{store_id}/" if env_value is not None:
# 0, false, False, no, No → 生产模式
def get_gui_ws_url(exe_token: str) -> str: if env_value.lower() in ('0', 'false', 'no'):
"""获取新版单连接GUI专用WebSocket URL（按用户token建一条连接）""" return False
return f"{BACKEND_WS_URL}/ws/gui/{exe_token}/" # 1, true, True, yes, Yes → 测试模式
if env_value.lower() in ('1', 'true', 'yes'):
def get_config(): return True
"""获取所有配置"""
return { # 使用配置文件值 (如果不做设置我们可以直接用编码变量进行控制是否可以允许多实例的方式运行)
'backend_host': BACKEND_HOST, return MULTI_INSTANCE_MODE
'backend_port': BACKEND_PORT,
'backend_ws_url': BACKEND_WS_URL, # 平台特定配置
'ws_connect_timeout': WS_CONNECT_TIMEOUT, PLATFORMS = {
'ws_message_timeout': WS_MESSAGE_TIMEOUT, "JD": {
'max_pending_replies': MAX_PENDING_REPLIES, "name": "京东",
'cleanup_interval': CLEANUP_INTERVAL, "ws_url": "wss://dongdong.jd.com/workbench/websocket"
'platforms': PLATFORMS },
} "DOUYIN": {
"name": "抖音",
"ws_url": None # 动态获取
},
APP_NAME = "ShuidropGUI" # 应用名称（作为配置目录名） "QIANNIU": {
"name": "千牛(淘宝)",
API_TOKEN = 'sd_acF0TisgfFOtsBm4ytqb17MQbcxuX9Vp' # 默认回退令牌（仅当未找到外部配置时使用） "ws_url": "ws://127.0.0.1:3030"
},
def _get_config_paths(): "PDD": {
"""返回（配置目录, 配置文件路径），位于 %APPDATA%/ShuidropGUI/config.json""" "name": "拼多多",
base_dir = os.getenv('APPDATA') or os.path.expanduser('~') # 优先使用 APPDATA，其次使用用户主目录 "ws_url": None # 动态获取
cfg_dir = os.path.join(base_dir, APP_NAME) # 组合配置目录路径 }
cfg_file = os.path.join(cfg_dir, 'config.json') # 组合配置文件路径 }
return cfg_dir, cfg_file
def get_backend_ws_url(platform: str, store_id: str) -> str:
"""获取旧版后端WebSocket URL（按店铺建连接，保留兼容）"""
def get_saved_token() -> str: return f"{BACKEND_WS_URL}/ws/platform/{platform.lower()}/{store_id}/"
"""优先从外部 JSON 配置读取令牌，不存在时回退到内置 API_TOKEN"""
try: def get_gui_ws_url(exe_token: str) -> str:
cfg_dir, cfg_file = _get_config_paths() # 获取目录与文件路径 """获取新版单连接GUI专用WebSocket URL（按用户token建一条连接）"""
if os.path.exists(cfg_file): # 如果配置文件存在 return f"{BACKEND_WS_URL}/ws/gui/{exe_token}/"
with open(cfg_file, 'r', encoding='utf-8') as f: # 以 UTF-8 读取
data = json.load(f) # 解析 JSON 内容 def get_config():
token = data.get('token', '') # 读取 token 字段 """获取所有配置"""
if token: # 如果有效 return {
return token # 返回读取到的令牌 'backend_host': BACKEND_HOST,
except Exception: 'backend_port': BACKEND_PORT,
pass # 读取失败时静默回退 'backend_ws_url': BACKEND_WS_URL,
return API_TOKEN # 回退为内置的默认值 'ws_connect_timeout': WS_CONNECT_TIMEOUT,
'ws_message_timeout': WS_MESSAGE_TIMEOUT,
'max_pending_replies': MAX_PENDING_REPLIES,
def set_saved_token(new_token: str) -> bool: 'cleanup_interval': CLEANUP_INTERVAL,
"""将访问令牌写入外部 JSON 配置，并更新内存中的值 'platforms': PLATFORMS
- new_token: 新的访问令牌字符串 }
返回: True 表示写入成功，False 表示失败
"""
try:
cfg_dir, cfg_file = _get_config_paths() APP_NAME = "ShuidropGUI" # 应用名称（作为配置目录名）
os.makedirs(cfg_dir, exist_ok=True)
data = {'token': new_token} API_TOKEN = 'sd_acF0TisgfFOtsBm4ytqb17MQbcxuX9Vp' # 默认回退令牌（仅当未找到外部配置时使用）
with open(cfg_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2) def _get_config_paths():
# 同步更新内存变量，保证运行期可立即生效 """返回（配置目录, 配置文件路径），位于 %APPDATA%/ShuidropGUI/config.json"""
global API_TOKEN base_dir = os.getenv('APPDATA') or os.path.expanduser('~') # 优先使用 APPDATA，其次使用用户主目录
API_TOKEN = new_token cfg_dir = os.path.join(base_dir, APP_NAME) # 组合配置目录路径
return True cfg_file = os.path.join(cfg_dir, 'config.json') # 组合配置文件路径
except Exception as e: return cfg_dir, cfg_file
# 发生异常时打印提示并返回失败
print(f"写入令牌失败: {e}")
def get_saved_token() -> str:
"""优先从外部 JSON 配置读取令牌，不存在时回退到内置 API_TOKEN"""
try:
cfg_dir, cfg_file = _get_config_paths() # 获取目录与文件路径
if os.path.exists(cfg_file): # 如果配置文件存在
with open(cfg_file, 'r', encoding='utf-8') as f: # 以 UTF-8 读取
data = json.load(f) # 解析 JSON 内容
token = data.get('token', '') # 读取 token 字段
if token: # 如果有效
return token # 返回读取到的令牌
except Exception:
pass # 读取失败时静默回退
return API_TOKEN # 回退为内置的默认值
def set_saved_token(new_token: str) -> bool:
"""将访问令牌写入外部 JSON 配置，并更新内存中的值
- new_token: 新的访问令牌字符串
返回: True 表示写入成功，False 表示失败
"""
try:
cfg_dir, cfg_file = _get_config_paths()
os.makedirs(cfg_dir, exist_ok=True)
data = {'token': new_token}
with open(cfg_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 同步更新内存变量，保证运行期可立即生效
global API_TOKEN
API_TOKEN = new_token
return True
except Exception as e:
# 发生异常时打印提示并返回失败
print(f"写入令牌失败: {e}")
return False return False

22
main.py
View File

@@ -145,14 +145,20 @@ class LoginWindow(QMainWindow):
# noinspection PyUnresolvedReferences # noinspection PyUnresolvedReferences
self.token_input.returnPressed.connect(self.login) # 表示回车提交 self.token_input.returnPressed.connect(self.login) # 表示回车提交
self.token_input.setMinimumHeight(34) # 最小输入框高度 self.token_input.setMinimumHeight(34) # 最小输入框高度
# 预填已保存的令牌（如果存在） # 🔥 根据配置决定是否自动加载token
try: # 生产模式：自动加载保存的token（方便用户）
from config import get_saved_token # 测试模式：不加载，避免多实例冲突
saved = get_saved_token() if not config.is_multi_instance_mode():
if saved: try:
self.token_input.setText(saved) from config import get_saved_token
except Exception: saved = get_saved_token()
pass if saved:
self.token_input.setText(saved)
print("[INFO] 生产模式：已自动加载保存的token")
except Exception:
pass
else:
print("[INFO] 测试模式：不自动加载token，支持多实例运行")
token_layout.addWidget(token_label) token_layout.addWidget(token_label)
token_layout.addWidget(self.token_input) token_layout.addWidget(self.token_input)