# backend_singleton.py # 共享后端单连接客户端实例和WebSocket连接管理 from typing import Optional, Callable import threading import asyncio from threading import Thread from PyQt5.QtCore import QObject, pyqtSignal # 创建新的后端客户端 from WebSocket.BackendClient import BackendClient from Utils.JD.JdUtils import JDListenerForGUI as JDListenerForGUI_WS from Utils.Dy.DyUtils import DouYinListenerForGUI as DYListenerForGUI_WS from Utils.Pdd.PddUtils import PddListenerForGUI as PDDListenerForGUI_WS from Utils.QianNiu.QianNiuUtils import QianNiuListenerForGUI as QNListenerForGUI_WS class PlatformConnectionSignals(QObject): """平台连接信号（线程安全）""" platform_connected = pyqtSignal(str, str) # (platform_name, store_id) _backend_client = None def set_backend_client(client: BackendClient) -> None: global _backend_client _backend_client = client def get_backend_client() -> Optional[BackendClient]: return _backend_client class WebSocketManager: """WebSocket连接管理器，统一处理后端连接和平台监听""" def __init__(self): # 后端客户端 self.backend_client = None self.is_connected = False # 版本检查器 self.version_checker = None self.gui_update_callback = None self.platform_listeners = {} # 存储各平台的监听器 self.connected_platforms = [] # 存储已连接的平台列表 # <- 新增 # 平台连接信号（线程安全） self.platform_signals = PlatformConnectionSignals() self.platform_signals.platform_connected.connect(self._on_platform_signal_received) # 🔥 会话超时信号（用于拼多多会话过期重试） self.session_timeout_signal = None self.callbacks = { 'log': None, 'success': None, 'error': None, 'platform_connected': None, 'token_error': None, 'disconnect': None, # 新增：被踢下线回调 } def set_callbacks(self, log: Callable = None, success: Callable = None, error: Callable = None, platform_connected: Callable = None, platform_disconnected: Callable = None, token_error: Callable = None, disconnect: Callable = None, balance_insufficient: Callable = None): """设置回调函数""" if log: self.callbacks['log'] = log if success: self.callbacks['success'] = success if error: 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: self.callbacks['disconnect'] = disconnect if balance_insufficient: self.callbacks['balance_insufficient'] = balance_insufficient def _log(self, message: str, level: str = "INFO"): """内部日志方法""" if self.callbacks['log']: self.callbacks['log'](message, level) else: print(f"[{level}] {message}") def _on_platform_signal_received(self, platform_name: str, store_id: str): """接收平台连接信号（在主线程中执行）""" try: self._log(f"📡 收到平台连接信号: {platform_name} (店铺:{store_id})", "INFO") self._notify_platform_connected(platform_name) except Exception as e: self._log(f"处理平台连接信号失败: {e}", "ERROR") def _notify_platform_connected(self, platform_name: str): """通知GUI平台连接成功（仅在主线程中调用）""" try: if platform_name not in self.connected_platforms: self.connected_platforms.append(platform_name) if self.callbacks['platform_connected']: self.callbacks['platform_connected'](platform_name, self.connected_platforms.copy()) self._log(f"已通知GUI平台连接: {platform_name}", "INFO") 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 disconnect_all_async(self): """异步断开所有连接（余额不足时调用）- 不阻塞当前线程""" try: self._log("🔴 收到余额不足通知，开始断开所有连接...", "ERROR") # 1. 断开所有平台连接 if self.platform_listeners: platform_count = len(self.platform_listeners) self._log(f"正在断开 {platform_count} 个平台连接...", "INFO") # 复制键列表，避免遍历时修改字典 platform_keys = list(self.platform_listeners.keys()) for key in platform_keys: try: listener_info = self.platform_listeners.get(key) if listener_info: platform_type = listener_info.get('platform', '') self._log(f"断开 {platform_type} 平台连接: {key}", "INFO") # 移除连接记录 self.platform_listeners.pop(key, None) except Exception as e: self._log(f"断开平台连接失败: {e}", "ERROR") self._log(f"✅ 已断开所有 {platform_count} 个平台连接", "INFO") # 清空连接列表 self.connected_platforms.clear() # 2. 延迟断开后端连接（在新线程中执行，避免阻塞） import threading def _delayed_backend_disconnect(): try: import time time.sleep(0.5) # 延迟0.5秒，确保上面的操作完成 if self.backend_client: self.backend_client.should_stop = True self.backend_client.is_connected = False self._log("✅ 已标记后端连接为断开状态", "INFO") except Exception as e: self._log(f"断开后端连接失败: {e}", "ERROR") disconnect_thread = threading.Thread(target=_delayed_backend_disconnect, daemon=True) disconnect_thread.start() except Exception as e: self._log(f"断开所有连接失败: {e}", "ERROR") def connect_backend(self, token: str) -> bool: """连接后端WebSocket""" try: # 1 保存token到配置 try: from config import set_saved_token set_saved_token(token) except Exception: pass # 2 获取或创建后端客户端 backend = get_backend_client() if backend: # 检查现有客户端是否因token错误而停止 if backend.should_stop: self._log("检测到客户端因token错误已停止，创建新的客户端", "INFO") # 断开旧客户端 backend.disconnect() # 清除旧客户端引用 set_backend_client(None) backend = None else: # 3 如果有客户端更新token并重连 backend.set_token(token) # 设置回调函数 def _on_backend_success(): try: self._log("连接服务成功", "SUCCESS") if self.callbacks['success']: self.callbacks['success']() # 启动版本检查器 self._start_version_checker() except Exception as e: self._log(f"成功回调执行失败: {e}", "ERROR") 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_display}, cookies_len={len(cookies) if cookies else 0}", "INFO") 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") if self.callbacks['token_error']: self.callbacks['token_error'](error_content) def _on_disconnect(disconnect_msg: str): self._log(f"被踢下线: {disconnect_msg}", "WARNING") if self.callbacks['disconnect']: self.callbacks['disconnect'](disconnect_msg) def _on_balance_insufficient(balance_msg: str): """余额不足回调""" if self.callbacks['balance_insufficient']: self.callbacks['balance_insufficient'](balance_msg) def _on_log(message: str, level: str = "INFO"): """Backend client log callback""" self._log(message, level) backend.set_callbacks(success=_on_backend_success, login=_on_backend_login, token_error=_on_token_error, disconnect=_on_disconnect, balance_insufficient=_on_balance_insufficient, log=_on_log) if not backend.is_connected: backend.connect() self.backend_client = backend self._log("令牌已提交，已连接后端。等待后端下发平台cookies后自动连接平台...", "SUCCESS") return True # 如果没有现有客户端或客户端被重置，创建新的 if not backend: backend = BackendClient.from_exe_token(token) 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_display}, cookies_len={len(cookies) if cookies else 0}", "INFO") self._handle_platform_login(platform_name, store_id, cookies, store_name) def _on_backend_success(): try: self._log("连接服务成功", "SUCCESS") if self.callbacks['success']: self.callbacks['success']() # 启动版本检查器 self._start_version_checker() except Exception as e: self._log(f"成功回调执行失败: {e}", "ERROR") def _on_token_error(error_content: str): self._log(f"Token验证失败: {error_content}", "ERROR") if self.callbacks['token_error']: self.callbacks['token_error'](error_content) def _on_disconnect(disconnect_msg: str): self._log(f"被踢下线: {disconnect_msg}", "ERROR") if self.callbacks['disconnect']: self.callbacks['disconnect'](disconnect_msg) def _on_balance_insufficient(balance_msg: str): """余额不足回调""" if self.callbacks['balance_insufficient']: self.callbacks['balance_insufficient'](balance_msg) def _on_log(message: str, level: str = "INFO"): """Backend client log callback""" self._log(message, level) backend.set_callbacks(login=_on_backend_login, success=_on_backend_success, token_error=_on_token_error, disconnect=_on_disconnect, balance_insufficient=_on_balance_insufficient, log=_on_log) backend.connect() set_backend_client(backend) self.backend_client = backend self._log("已创建后端客户端并连接。等待后端下发平台cookies...", "SUCCESS") return True except Exception as e: self._log(f"连接后端失败: {e}", "ERROR") if self.callbacks['error']: self.callbacks['error'](str(e)) return False def _handle_platform_login(self, platform_name: str, store_id: str, cookies: str, store_name: str = ""): """处理平台登录请求""" try: # 🔥 检查并断开当前店铺的旧连接（策略B：先断开旧连接，再建立新连接） store_key_pattern = f":{store_id}" # 匹配 "平台名:store_id" 格式 keys_to_remove = [key for key in self.platform_listeners.keys() if key.endswith(store_key_pattern)] if keys_to_remove: self._log(f"🔄 检测到店铺 {store_id} 重复登录，断开 {len(keys_to_remove)} 个旧连接", "INFO") for key in keys_to_remove: listener_info = self.platform_listeners.get(key) if listener_info: platform_type = listener_info.get('platform', '') # 从各平台的 WebsocketManager 中获取连接并关闭WebSocket try: if platform_type == "京东": from Utils.JD.JdUtils import WebsocketManager as JDWSManager jd_mgr = JDWSManager() conn_info = jd_mgr.get_connection(key) if conn_info and conn_info.get('platform'): ws = conn_info['platform'].get('ws') if ws and hasattr(ws, 'close'): try: import asyncio loop = conn_info['platform'].get('loop') if loop and not loop.is_closed(): asyncio.run_coroutine_threadsafe(ws.close(), loop) self._log(f"✅ 已关闭京东WebSocket连接: {key}", "DEBUG") except Exception: pass jd_mgr.remove_connection(key) self._log(f"✅ 已从京东管理器移除连接: {key}", "DEBUG") elif platform_type == "抖音": from Utils.Dy.DyUtils import DouYinWebsocketManager as DYWSManager dy_mgr = DYWSManager() conn_info = dy_mgr.get_connection(key) if conn_info and conn_info.get('platform'): ws = conn_info['platform'].get('ws') if ws and hasattr(ws, 'close'): try: import asyncio loop = conn_info['platform'].get('loop') if loop and not loop.is_closed(): asyncio.run_coroutine_threadsafe(ws.close(), loop) self._log(f"✅ 已关闭抖音WebSocket连接: {key}", "DEBUG") except Exception: pass dy_mgr.remove_connection(key) self._log(f"✅ 已从抖音管理器移除连接: {key}", "DEBUG") elif platform_type == "千牛": from Utils.QianNiu.QianNiuUtils import QianNiuWebsocketManager as QNWSManager qn_mgr = QNWSManager() qn_mgr.remove_connection(key) self._log(f"✅ 已从千牛管理器移除连接: {key}", "DEBUG") elif platform_type == "拼多多": from Utils.Pdd.PddUtils import WebsocketManager as PDDWSManager pdd_mgr = PDDWSManager() conn_info = pdd_mgr.get_connection(key) # 清理旧listener的定时器（避免遗留定时器触发） old_listener = listener_info.get('listener') if old_listener and hasattr(old_listener, 'pdd_bot') and old_listener.pdd_bot: try: old_listener.pdd_bot._cancel_session_timeout_check() self._log(f"✅ [PDD] 旧监听器的定时器已清理", "DEBUG") except Exception: pass if conn_info and conn_info.get('platform'): # 关闭WebSocket连接 ws = conn_info['platform'].get('ws') if ws and hasattr(ws, 'close'): try: import asyncio loop = conn_info['platform'].get('loop') if loop and not loop.is_closed(): asyncio.run_coroutine_threadsafe(ws.close(), loop) self._log(f"✅ 已关闭拼多多WebSocket连接: {key}", "DEBUG") except Exception as ws_e: self._log(f"⚠️ 关闭WebSocket时出错: {ws_e}", "DEBUG") pdd_mgr.remove_connection(key) self._log(f"✅ 已从拼多多管理器移除连接: {key}", "DEBUG") except Exception as e: self._log(f"⚠️ 移除{platform_type}连接时出错: {e}", "WARNING") # 从监听器字典中移除 self.platform_listeners.pop(key, None) # 给WebSocket一点时间完全关闭 import time time.sleep(0.5) self._log(f"✅ 旧连接已全部断开，准备建立新连接", "INFO") # 平台名称映射 platform_map = { "淘宝": "千牛", "QIANNIU": "千牛", "京东": "京东", "JD": "京东", "抖音": "抖音", "DY": "抖音", "DOUYIN": "抖音", "拼多多": "拼多多", "PDD": "拼多多", "PINDUODUO": "拼多多" } # 标准化平台名称 normalized_platform = platform_map.get(platform_name.upper(), platform_name) self._log(f"处理平台登录: {platform_name} -> {normalized_platform}, 店铺={store_id}", "INFO") # 🔧 关键修改：确保每个平台都能独立处理自己的登录请求 if normalized_platform == "千牛": if cookies == "login_success": self._log("⚠️ 千牛平台收到空cookies，但允许启动监听器", "WARNING") cookies = "" # 清空cookies，千牛不需要真实cookies self._start_qianniu_listener(store_id, cookies, store_name) elif normalized_platform == "京东": self._start_jd_listener(store_id, cookies, store_name) elif normalized_platform == "抖音": self._start_douyin_listener(store_id, cookies, store_name) elif normalized_platform == "拼多多": self._start_pdd_listener(store_id, cookies, store_name) else: self._log(f"❌ 不支持的平台: {platform_name}", "ERROR") except Exception as e: self._log(f"处理平台登录失败: {e}", "ERROR") import traceback self._log(f"错误详情: {traceback.format_exc()}", "DEBUG") def _start_version_checker(self): """启动版本检查器""" try: from version_checker import VersionChecker self.version_checker = VersionChecker( backend_client=self.backend_client, update_callback=self._on_update_available ) self.backend_client.version_callback = self.version_checker.handle_version_response self.version_checker.start() self._log("✅ 版本检查器已启动，将每10分钟检查一次更新", "SUCCESS") except Exception as e: self._log(f"❌ 启动版本检查器失败: {e}", "ERROR") def _on_update_available(self, latest_version, download_url): """发现新版本时的处理（在子线程中调用）""" self._log(f"🔔 发现新版本 {latest_version}", "INFO") # 通知主GUI显示更新提醒（通过 Qt 信号机制，线程安全） if hasattr(self, 'gui_update_callback') and self.gui_update_callback: try: # 直接调用回调（回调内部使用信号机制调度到主线程） self.gui_update_callback(latest_version, download_url) self._log(f"✅ 已调用更新回调", "DEBUG") except Exception as e: self._log(f"❌ 调用更新回调失败: {e}", "ERROR") import traceback self._log(f"详细错误: {traceback.format_exc()}", "ERROR") def _start_jd_listener(self, store_id: str, cookies: str, store_name: str = ""): """启动京东平台监听""" try: def _runner(): try: listener = JDListenerForGUI_WS() asyncio.run(listener.start_with_cookies(store_id, cookies)) except Exception as e: self._log(f"京东监听器运行异常: {e}", "ERROR") # 在新线程中启动监听器 thread = Thread(target=_runner, daemon=True) thread.start() # 保存监听器引用 self.platform_listeners[f"京东:{store_id}"] = { 'thread': thread, 'platform': '京东', 'store_id': store_id, 'store_name': store_name # 保存店铺名称 } # 上报连接状态给后端 if self.backend_client: try: self.backend_client.send_message({ "type": "connect_message", "store_id": store_id, "status": True }) except Exception as e: self._log(f"上报连接状态失败: {e}", "WARNING") self._log("已启动京东平台监听", "SUCCESS") self._notify_platform_connected("京东") # ← 新增 except Exception as e: self._log(f"启动京东平台监听失败: {e}", "ERROR") # 确保失败时也上报状态 if self.backend_client: try: self.backend_client.send_message({ "type": "connect_message", "store_id": store_id, "status": False }) except Exception as send_e: self._log(f"失败状态下报连接状态也失败: {send_e}", "ERROR") def _start_douyin_listener(self, store_id: str, cookies: str, store_name: str = ""): """启动抖音平台监听""" try: def _runner(): try: import json self._log("🚀 开始创建抖音监听器实例", "DEBUG") listener = DYListenerForGUI_WS() self._log("✅ 抖音监听器实例创建成功", "DEBUG") # 🔥 检查是否为登录参数模式（与拼多多保持一致） if cookies and ('"login_flow"' in cookies or '"phone_number"' in cookies): # 使用登录参数模式 self._log("📋 使用登录参数启动抖音监听器", "INFO") self._log("🔄 开始执行 start_with_login_params", "DEBUG") result = asyncio.run(listener.start_with_login_params(store_id=store_id, login_params=cookies)) self._log(f"📊 start_with_login_params 执行结果: {result}", "DEBUG") # 详细的结果分析（仅日志记录，GUI 已在主线程中通知） if result == "need_verification_code": self._log("✅ [DY] 登录流程正常，已发送验证码需求通知给后端", "SUCCESS") elif result == "verification_code_error": self._log("⚠️ [DY] 验证码错误，已发送错误通知给后端", "WARNING") elif result: self._log("✅ [DY] 登录成功，平台连接已建立", "SUCCESS") else: self._log("❌ [DY] 登录失败", "ERROR") else: # 传统cookie模式（保留兼容性） self._log("🍪 使用Cookie启动抖音监听器", "INFO") self._log("🔄 开始执行 start_with_cookies", "DEBUG") try: # 🔥 修复：尝试JSON解析，失败时用ast.literal_eval解析Python字典字符串 if isinstance(cookies, str): try: cookie_dict = json.loads(cookies) except json.JSONDecodeError: # 后端发送的是Python字典字符串格式，使用ast.literal_eval import ast cookie_dict = ast.literal_eval(cookies) self._log("✅ 使用ast.literal_eval成功解析cookies", "DEBUG") else: cookie_dict = cookies except (json.JSONDecodeError, ValueError, SyntaxError) as e: self._log(f"❌ Cookie解析失败: {e}", "ERROR") return False result = asyncio.run(listener.start_with_cookies(store_id=store_id, cookie_dict=cookie_dict)) self._log(f"📊 start_with_cookies 执行结果: {result}", "DEBUG") # Cookie启动成功时记录日志（GUI 已在主线程中通知） if result: self._log("✅ [DY] Cookie启动成功，平台连接已建立", "SUCCESS") # 🔥 移除：不再在backend_singleton中发送connect_message # 抖音的连接状态报告应该在DyUtils中的DyLogin类中发送，与拼多多保持一致 # 所有特殊状态通知都已经在DyLogin中发送过了，这里不需要重复发送 return result except Exception as e: self._log(f"抖音监听器运行异常: {e}", "ERROR") return False # 在新线程中启动监听器 thread = Thread(target=_runner, daemon=True) thread.start() # 保存监听器引用 self.platform_listeners[f"抖音:{store_id}"] = { 'thread': thread, 'platform': '抖音', 'store_id': store_id, 'store_name': store_name # 保存店铺名称 } # 更新监听器状态 if f"抖音:{store_id}" in self.platform_listeners: self.platform_listeners[f"抖音:{store_id}"]['status'] = 'success' # ✅ 临时方案：启动后立即通知（因为 DY 监听器也会阻塞） # DY 内部会处理验证码流程，失败时会向后端上报相应状态 self._log("已启动抖音平台监听", "SUCCESS") self._notify_platform_connected("抖音") # ← 立即通知 except Exception as e: self._log(f"启动抖音平台监听失败: {e}", "ERROR") # 🔥 移除：确保失败时也不在这里上报状态 # 失败状态应该在DyLogin中处理，与拼多多保持一致 def _start_qianniu_listener(self, store_id: str, cookies: str, store_name: str = ""): """启动千牛平台监听（单连接多店铺架构）""" try: # 获取用户token（从后端客户端获取） exe_token = None if self.backend_client: exe_token = getattr(self.backend_client, 'token', None) if not exe_token: self._log("❌ 缺少exe_token，无法启动千牛监听器", "ERROR") return def _runner(): try: listener = QNListenerForGUI_WS() # 千牛平台使用新架构：传递store_id和exe_token asyncio.run(listener.start_listening_with_store(store_id, exe_token)) except Exception as e: self._log(f"千牛监听器运行异常: {e}", "ERROR") # 在新线程中启动监听器 thread = Thread(target=_runner, daemon=True) thread.start() # 保存监听器引用 self.platform_listeners[f"千牛:{store_id}"] = { 'thread': thread, 'platform': '千牛', 'store_id': store_id, 'store_name': store_name, # 保存店铺名称 'exe_token': exe_token } # 上报连接状态给后端 if self.backend_client: try: self.backend_client.send_message({ "type": "connect_message", "store_id": store_id, "status": True }) except Exception as e: self._log(f"上报连接状态失败: {e}", "WARNING") self._log("已启动千牛平台监听（单连接多店铺架构）", "SUCCESS") self._notify_platform_connected("千牛") # ← 新增 except Exception as e: self._log(f"启动千牛平台监听失败: {e}", "ERROR") # 确保失败时也上报状态 if self.backend_client: try: self.backend_client.send_message({ "type": "connect_message", "store_id": store_id, "status": False }) except Exception as send_e: self._log(f"失败状态下报千牛平台连接状态也失败: {send_e}", "ERROR") def _start_pdd_listener(self, store_id: str, data: str, store_name: str = ""): """启动拼多多平台监听""" try: # 在创建新实例前，清理旧实例的定时器（避免遗留定时器触发） shop_key = f"拼多多:{store_id}" old_listener_info = self.platform_listeners.get(shop_key) if old_listener_info: try: old_listener = old_listener_info.get('listener') if old_listener and hasattr(old_listener, 'pdd_bot') and old_listener.pdd_bot: old_listener.pdd_bot._cancel_session_timeout_check() self._log(f"✅ 已清理旧监听器定时器", "DEBUG") except Exception: pass def _runner(): try: # 传递session_timeout_signal（用于会话过期重试） listener = PDDListenerForGUI_WS( log_callback=self._log, session_timeout_signal=self.session_timeout_signal ) # 立即保存listener引用到platform_listeners（用于后续清理定时器） shop_key = f"拼多多:{store_id}" if shop_key in self.platform_listeners: self.platform_listeners[shop_key]['listener'] = listener # 判断是登录参数还是Cookie if self._is_pdd_login_params(data): # 使用登录参数启动 result = asyncio.run(listener.start_with_login_params(store_id=store_id, login_params=data)) else: # 使用Cookie启动（兼容旧方式） result = asyncio.run(listener.start_with_cookies(store_id=store_id, cookies=data)) # 根据实际登录结果上报状态给后端 if self.backend_client and result not in ["need_verification_code", "verification_code_error", "login_failure", "cookie_expired"]: # 🔥 如果是特殊状态（包括cookie_expired），说明通知已经发送，不需要重复发送 try: message = { "type": "connect_message", "store_id": store_id, "status": bool(result) } self.backend_client.send_message(message) status_text = "成功" if result else "失败" self._log(f"上报拼多多平台连接状态{status_text}: {message}", "SUCCESS" if result else "ERROR") except Exception as send_e: self._log(f"上报拼多多平台连接状态失败: {send_e}", "ERROR") elif result == "need_verification_code": self._log("需要验证码，验证码通知已由PddLogin发送，等待后端重新下发登录参数", "INFO") elif result == "verification_code_error": self._log("验证码错误，错误通知已由PddLogin发送，等待后端处理", "INFO") elif result == "login_failure": self._log("登录失败，失败通知已由PddLogin发送，等待后端处理", "INFO") elif result == "cookie_expired": self._log("🔄 Cookie已过期，过期通知已发送，等待后端重新下发Cookie", "INFO") return result except Exception as e: self._log(f"拼多多监听器运行异常: {e}", "ERROR") import traceback self._log(f"异常详情: {traceback.format_exc()}", "DEBUG") # 异常情况下上报失败状态 if self.backend_client: try: # 截取异常信息前100个字符，避免消息过长 error_msg = str(e)[:100] if len(str(e)) > 100 else str(e) message = { "type": "connect_message", "store_id": store_id, "status": False, "content": f"登录异常: {error_msg}" } self.backend_client.send_message(message) self._log(f"异常情况下上报拼多多平台连接失败状态: {message}", "ERROR") except Exception as send_e: self._log(f"异常情况下上报状态也失败: {send_e}", "ERROR") return False # 在新线程中启动监听器 thread = Thread(target=_runner, daemon=True) thread.start() # 保存监听器引用 self.platform_listeners[f"拼多多:{store_id}"] = { 'thread': thread, 'platform': '拼多多', 'store_id': store_id, 'store_name': store_name # 保存店铺名称 } # ✅ 临时方案：启动后立即通知（因为 PDD 监听器会阻塞，无法通过返回值判断） # PDD 内部会处理验证码流程，失败时会向后端上报 status=False self._log("已启动拼多多平台监听", "SUCCESS") self._notify_platform_connected("拼多多") # ← 立即通知 except Exception as e: self._log(f"启动拼多多平台监听失败: {e}", "ERROR") # 确保失败时也上报状态 if self.backend_client: try: self.backend_client.send_message({ "type": "connect_message", "store_id": store_id, "status": False }) self._log(f"启动失败情况下上报拼多多平台连接状态失败", "ERROR") except Exception as send_e: self._log(f"启动失败情况下上报状态也失败: {send_e}", "ERROR") def _is_pdd_login_params(self, data: str) -> bool: """判断是否为拼多多登录参数""" try: import json parsed_data = json.loads(data) login_params = parsed_data.get("data", {}).get("login_params", {}) if not login_params: return False # 检查必需的登录参数字段 required_fields = ["username", "password", "anti_content", "risk_sign", "timestamp"] has_all_fields = all(field in login_params for field in required_fields) return has_all_fields except Exception: return False def send_message(self, message: dict): """发送消息到后端""" if self.backend_client and self.backend_client.is_connected: return self.backend_client.send_message(message) else: raise Exception("后端未连接") def disconnect_all(self): """断开所有连接""" try: # 断开后端连接 if self.backend_client: self.backend_client.disconnect() self.backend_client = None # 清理平台监听器 self.platform_listeners.clear() self._log("所有连接已断开", "INFO") except Exception as e: self._log(f"断开连接时发生错误: {e}", "ERROR") # 全局WebSocket管理器实例 _websocket_manager = None def get_websocket_manager() -> WebSocketManager: """获取全局WebSocket管理器实例""" global _websocket_manager if _websocket_manager is None: _websocket_manager = WebSocketManager() return _websocket_manager