shuidrop_gui/WebSocket/backend_singleton.py

# 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:

            # 🔥 根据配置决定是否保存token
            # 生产模式：保存token，方便用户下次自动加载
            # 测试模式：不保存，避免多实例冲突
            import config as cfg
            if not cfg.is_multi_instance_mode():
                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 获取或创建后端客户端
            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