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

[patch] 修改适中心跳机制 JD的平台登录抢占 提示弹框优化 并下发连接断开通知 托盘右击提示小标题显示精炼 增加store_name入回调方法做店铺名称定位

Browse Source
This commit is contained in:
haosicheng
2025-10-17 15:21:03 +08:00
parent 90bf763bde
commit c32d326c69
5 changed files with 282 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
WebSocket/BackendClient.py
View File

@@ -130,6 +130,10 @@ class BackendClient:
self._log("已禁用心跳机制", "WARNING")
self.is_connected = True
# 🔥 在重置之前记录是否是重连（用于后续上报平台状态）
was_reconnecting = self.reconnect_attempts > 0
self.reconnect_attempts = 0 # 重置重连计数
self.is_reconnecting = False
self._log("后端WebSocket连接成功", "SUCCESS")
@@ -140,7 +144,7 @@ class BackendClient:
# 发送连接状态通知给后端
self._notify_connection_status(True)
self.on_connected()
self.on_connected(was_reconnecting)
# 消息循环
async for message in self.websocket:
@@ -308,53 +312,68 @@ class BackendClient:
if log:
self.log_callback = log
def on_connected(self):
def on_connected(self, was_reconnecting: bool = False):
"""连接成功时的处理"""
if self.reconnect_attempts > 0:
self._log(f"后端WebSocket重连成功！(第{self.reconnect_attempts}次尝试)", "SUCCESS")
else:
self._log("后端WebSocket连接成功", "SUCCESS")
# 重连成功后可选择上报状态给后端
if self.reconnect_attempts > 0:
if was_reconnecting:
self._log("后端WebSocket重连成功！", "SUCCESS")
# 重连成功后上报平台状态给后端
self._report_reconnect_status()
else:
self._log("后端WebSocket首次连接成功", "SUCCESS")
# 不再主动请求 get_store，避免与后端不兼容导致协程未完成
def _report_reconnect_status(self):
"""重连成功后上报当前状态（可选）"""
"""重连成功后上报当前状态"""
try:
# 获取当前已连接的平台列表
from WebSocket.backend_singleton import get_websocket_manager
manager = get_websocket_manager()
if hasattr(manager, 'platform_listeners') and manager.platform_listeners:
platform_count = len(manager.platform_listeners)
self._log(f"🔄 检测到 {platform_count} 个活跃平台连接，准备重新上报状态", "INFO")
# 延迟1秒，确保后端完全准备好
import time
time.sleep(1.0)
for platform_key, listener_info in manager.platform_listeners.items():
store_id = listener_info.get('store_id')
platform = listener_info.get('platform')
store_name = listener_info.get('store_name', '')
if store_id and platform:
# 上报平台仍在连接状态
try:
store_display = store_name or store_id[:8] + "..."
reconnect_message = {
"type": "connect_message",
"store_id": store_id,
"status": True,
"content": f"GUI重连成功，{platform}平台状态正常"
"cookies": "" # 重连时无需再次发送cookies
}
# 异步发送，不阻塞连接过程
asyncio.run_coroutine_threadsafe(
# 同步发送，确保发送成功
future = asyncio.run_coroutine_threadsafe(
self._send_to_backend(reconnect_message),
self.loop
)
print(f"[重连] 已上报{platform}平台状态")
# 等待发送完成（最多2秒）
future.result(timeout=2)
self._log(f"✅ 已重新上报 {platform} 平台状态: {store_display}", "SUCCESS")
except Exception as e:
print(f"[重连] 上报{platform}平台状态失败: {e}")
self._log(f"❌ 上报 {platform} 平台状态失败: {e}", "ERROR")
import traceback
self._log(f"详细错误: {traceback.format_exc()}", "DEBUG")
else:
print("[重连] 当前无活跃平台连接，跳过状态上报")
self._log("当前无活跃平台连接，跳过状态上报", "INFO")
except Exception as e:
print(f"[重连] 状态上报过程异常: {e}")
self._log(f"状态上报过程异常: {e}", "ERROR")
import traceback
self._log(f"详细错误: {traceback.format_exc()}", "DEBUG")
def _notify_connection_status(self, connected: bool):
"""通知后端连接状态变化"""
@@ -365,30 +384,21 @@ class BackendClient:
# 获取当前活跃平台的store_id
active_store_id = None
try:
from Utils.JD.JdUtils import WebsocketManager as JdManager
from Utils.Dy.DyUtils import DouYinWebsocketManager as DyManager
from Utils.Pdd.PddUtils import WebsocketManager as PddManager
# 检查各平台是否有活跃连接
for mgr_class, platform_name in [(JdManager, "京东"), (DyManager, "抖音"), (PddManager, "拼多多")]:
try:
mgr = mgr_class()
if hasattr(mgr, '_store') and mgr._store:
for shop_key, entry in mgr._store.items():
if entry and entry.get('platform'):
# 从shop_key中提取store_id（格式：平台:store_id）
if ':' in shop_key:
_, store_id = shop_key.split(':', 1)
active_store_id = store_id
print(f"[状态] 检测到活跃{platform_name}平台: {store_id}")
break
if active_store_id:
break
except Exception as e:
print(f"[状态] 检查{platform_name}平台失败: {e}")
continue
from WebSocket.backend_singleton import get_websocket_manager
manager = get_websocket_manager()
# 从 platform_listeners 获取活跃平台
if hasattr(manager, 'platform_listeners') and manager.platform_listeners:
# 获取第一个活跃平台的 store_id
for platform_key, listener_info in manager.platform_listeners.items():
store_id = listener_info.get('store_id')
platform = listener_info.get('platform')
if store_id and platform:
active_store_id = store_id
self._log(f"检测到活跃{platform}平台: {store_id}", "DEBUG")
break
except Exception as e:
print(f"[状态] 获取活跃平台信息失败: {e}")
self._log(f"获取活跃平台信息失败: {e}", "DEBUG")
status_message = {
"type": "connection_status",
@@ -400,9 +410,9 @@ class BackendClient:
# 如果有活跃平台，添加store_id
if active_store_id:
status_message["store_id"] = active_store_id
print(f"[状态] 添加store_id到状态消息: {active_store_id}")
self._log(f"添加store_id到状态消息: {active_store_id}", "DEBUG")
else:
print(f"[状态] 未检测到活跃平台，不添加store_id")
self._log("未检测到活跃平台，不添加store_id", "DEBUG")
# 异步发送状态通知
asyncio.run_coroutine_threadsafe(
@@ -411,12 +421,12 @@ class BackendClient:
)
status_text = "连接" if connected else "断开"
print(f"[状态] 已通知后端GUI客户端{status_text}")
self._log(f"已通知后端GUI客户端{status_text}", "DEBUG")
except Exception as e:
print(f"[状态] 发送状态通知失败: {e}")
self._log(f"发送状态通知失败: {e}", "ERROR")
import traceback
print(f"[状态] 详细错误: {traceback.format_exc()}")
self._log(f"详细错误: {traceback.format_exc()}", "DEBUG")
def on_message_received(self, message: Dict[str, Any]):
"""处理接收到的消息 - 根据WebSocket文档v2更新"""
@@ -1172,10 +1182,11 @@ class BackendClient:
self.get_store()
def _handle_login(self, message: Dict[str, Any]):
"""处理平台登录消息（新版：type=login, cookies/login_params, store_id, platform_name）"""
"""处理平台登录消息（新版：type=login, cookies/login_params, store_id, platform_name, store_name）"""
cookies = message.get('cookies', '')
store_id = message.get('store_id', '')
platform_name = message.get('platform_name', '')
store_name = message.get('store_name', '') # 新增：获取店铺名称
content = message.get('content', '')
data = message.get('data', {})
@@ -1184,17 +1195,17 @@ class BackendClient:
(("拼多多登录" in content and data.get('login_params')) or
("抖音登录" in content and data.get('login_flow')))):
# 登录参数模式 - 传递完整的消息JSON给处理器
print(f"收到{platform_name}登录参数: 平台={platform_name}, 店铺={store_id}, 类型={content}")
print(f"收到{platform_name}登录参数: 平台={platform_name}, 店铺={store_name or store_id}, 类型={content}")
if self.login_callback:
# 传递完整的JSON消息，让处理器来解析登录参数
import json
full_message = json.dumps(message)
self.login_callback(platform_name, store_id, full_message)
self.login_callback(platform_name, store_id, full_message, store_name)
else:
# 普通Cookie模式
print(f"收到登录指令: 平台={platform_name}, 店铺={store_id}, cookies_len={len(cookies) if cookies else 0}")
print(f"收到登录指令: 平台={platform_name}, 店铺={store_name or store_id}, cookies_len={len(cookies) if cookies else 0}")
if self.login_callback:
self.login_callback(platform_name, store_id, cookies)
self.login_callback(platform_name, store_id, cookies, store_name)
def _handle_error_message(self, message: Dict[str, Any]):
"""处理错误消息"""
@@ -1253,7 +1264,7 @@ class BackendClient:
def _handle_disconnect(self, message: Dict[str, Any]):
"""处理被踢下线消息"""
disconnect_message = message.get('message', '您的账号在其他设备登录，当前连接已断开')
disconnect_message = message.get('content', '您的账号在其他设备登录，当前连接已断开')
print(f"[断开] 收到后端断开通知: {disconnect_message}")

72
WebSocket/backend_singleton.py
View File

@@ -59,7 +59,8 @@ class WebSocketManager:
}
def set_callbacks(self, log: Callable = None, success: Callable = None, error: Callable = None,
platform_connected: Callable = None, token_error: Callable = None, disconnect: Callable = None):
platform_connected: Callable = None, platform_disconnected: Callable = None,
token_error: Callable = None, disconnect: Callable = None):
"""设置回调函数"""
if log:
self.callbacks['log'] = log
@@ -69,6 +70,8 @@ class WebSocketManager:
self.callbacks['error'] = error
if platform_connected: # ← 新增
self.callbacks['platform_connected'] = platform_connected
if platform_disconnected:
self.callbacks['platform_disconnected'] = platform_disconnected
if token_error:
self.callbacks['token_error'] = token_error
if disconnect:
@@ -101,6 +104,35 @@ class WebSocketManager:
except Exception as e:
self._log(f"通知平台连接失败: {e}", "ERROR")
def notify_platform_kicked(self, platform_name: str, store_name: str, reason: str = "账号在其他设备登录", store_id: str = None):
"""通知GUI平台被踢下线（供平台监听器调用）"""
try:
self._log(f"⚠️ 平台被踢下线: {platform_name} - {store_name}, 原因: {reason}", "WARNING")
# 从连接列表中移除
if platform_name in self.connected_platforms:
self.connected_platforms.remove(platform_name)
# 🔥 发送平台断开消息给后端（status=false）
if store_id and self.backend_client:
try:
disconnect_message = {
"type": "connect_message",
"store_id": store_id,
"status": False,
"cookies": ""
}
self.backend_client.send_message(disconnect_message)
self._log(f"✅ 已通知后端 {platform_name} 平台断开: {store_id}", "INFO")
except Exception as send_error:
self._log(f"❌ 发送平台断开消息失败: {send_error}", "ERROR")
# 通知GUI显示弹窗
if self.callbacks['platform_disconnected']:
self.callbacks['platform_disconnected'](platform_name, store_name, reason)
except Exception as e:
self._log(f"通知平台断开失败: {e}", "ERROR")
def connect_backend(self, token: str) -> bool:
"""连接后端WebSocket"""
try:
@@ -139,11 +171,12 @@ class WebSocketManager:
except Exception as e:
self._log(f"成功回调执行失败: {e}", "ERROR")
def _on_backend_login(platform_name: str, store_id: str, cookies: str):
def _on_backend_login(platform_name: str, store_id: str, cookies: str, store_name: str = ""):
store_display = store_name or store_id
self._log(
f"收到后端登录指令: 平台={platform_name}, 店铺={store_id}, cookies_len={len(cookies) if cookies else 0}",
f"收到后端登录指令: 平台={platform_name}, 店铺={store_display}, cookies_len={len(cookies) if cookies else 0}",
"INFO")
self._handle_platform_login(platform_name, store_id, cookies)
self._handle_platform_login(platform_name, store_id, cookies, store_name)
def _on_token_error(error_content: str):
self._log(f"Token验证失败: {error_content}", "ERROR")
@@ -175,11 +208,12 @@ class WebSocketManager:
backend = BackendClient.from_exe_token(token)
def _on_backend_login(platform_name: str, store_id: str, cookies: str):
def _on_backend_login(platform_name: str, store_id: str, cookies: str, store_name: str = ""):
store_display = store_name or store_id
self._log(
f"收到后端登录指令: 平台={platform_name}, 店铺={store_id}, cookies_len={len(cookies) if cookies else 0}",
f"收到后端登录指令: 平台={platform_name}, 店铺={store_display}, cookies_len={len(cookies) if cookies else 0}",
"INFO")
self._handle_platform_login(platform_name, store_id, cookies)
self._handle_platform_login(platform_name, store_id, cookies, store_name)
def _on_backend_success():
try:
@@ -221,7 +255,7 @@ class WebSocketManager:
self.callbacks['error'](str(e))
return False
def _handle_platform_login(self, platform_name: str, store_id: str, cookies: str):
def _handle_platform_login(self, platform_name: str, store_id: str, cookies: str, store_name: str = ""):
"""处理平台登录请求"""
try:
# 🔥 检查并断开当前店铺的旧连接（策略B：先断开旧连接，再建立新连接）
@@ -335,16 +369,16 @@ class WebSocketManager:
if cookies == "login_success":
self._log("⚠️ 千牛平台收到空cookies，但允许启动监听器", "WARNING")
cookies = "" # 清空cookies，千牛不需要真实cookies
self._start_qianniu_listener(store_id, cookies)
self._start_qianniu_listener(store_id, cookies, store_name)
elif normalized_platform == "京东":
self._start_jd_listener(store_id, cookies)
self._start_jd_listener(store_id, cookies, store_name)
elif normalized_platform == "抖音":
self._start_douyin_listener(store_id, cookies)
self._start_douyin_listener(store_id, cookies, store_name)
elif normalized_platform == "拼多多":
self._start_pdd_listener(store_id, cookies)
self._start_pdd_listener(store_id, cookies, store_name)
else:
self._log(f"❌ 不支持的平台: {platform_name}", "ERROR")
@@ -382,7 +416,7 @@ class WebSocketManager:
import traceback
self._log(f"详细错误: {traceback.format_exc()}", "ERROR")
def _start_jd_listener(self, store_id: str, cookies: str):
def _start_jd_listener(self, store_id: str, cookies: str, store_name: str = ""):
"""启动京东平台监听"""
try:
def _runner():
@@ -400,7 +434,8 @@ class WebSocketManager:
self.platform_listeners[f"京东:{store_id}"] = {
'thread': thread,
'platform': '京东',
'store_id': store_id
'store_id': store_id,
'store_name': store_name # 保存店铺名称
}
# 上报连接状态给后端
@@ -431,7 +466,7 @@ class WebSocketManager:
except Exception as send_e:
self._log(f"失败状态下报连接状态也失败: {send_e}", "ERROR")
def _start_douyin_listener(self, store_id: str, cookies: str):
def _start_douyin_listener(self, store_id: str, cookies: str, store_name: str = ""):
"""启动抖音平台监听"""
try:
def _runner():
@@ -502,6 +537,7 @@ class WebSocketManager:
'thread': thread,
'platform': '抖音',
'store_id': store_id,
'store_name': store_name # 保存店铺名称
}
# 更新监听器状态
@@ -519,7 +555,7 @@ class WebSocketManager:
# 🔥 移除：确保失败时也不在这里上报状态
# 失败状态应该在DyLogin中处理，与拼多多保持一致
def _start_qianniu_listener(self, store_id: str, cookies: str):
def _start_qianniu_listener(self, store_id: str, cookies: str, store_name: str = ""):
"""启动千牛平台监听（单连接多店铺架构）"""
try:
# 获取用户token（从后端客户端获取）
@@ -548,6 +584,7 @@ class WebSocketManager:
'thread': thread,
'platform': '千牛',
'store_id': store_id,
'store_name': store_name, # 保存店铺名称
'exe_token': exe_token
}
@@ -579,7 +616,7 @@ class WebSocketManager:
except Exception as send_e:
self._log(f"失败状态下报千牛平台连接状态也失败: {send_e}", "ERROR")
def _start_pdd_listener(self, store_id: str, data: str):
def _start_pdd_listener(self, store_id: str, data: str, store_name: str = ""):
"""启动拼多多平台监听"""
try:
def _runner():
@@ -671,6 +708,7 @@ class WebSocketManager:
'thread': thread,
'platform': '拼多多',
'store_id': store_id,
'store_name': store_name # 保存店铺名称
}
# ✅ 临时方案：启动后立即通知（因为 PDD 监听器会阻塞，无法通过返回值判断）