diff --git a/BOT/lux_apis.py b/BOT/lux_apis.py index 650ed91..5809f34 100644 --- a/BOT/lux_apis.py +++ b/BOT/lux_apis.py @@ -15,10 +15,11 @@ from typing import Any, Awaitable, Callable, Dict, List, Optional # Third-Party Library Imports import aiohttp -import zstandard as zstd from dotenv import load_dotenv from websockets.asyncio.client import connect as wsconnect +from spectra_ws_payload import SpectraPayloadError, decode_spectra_ws_payload + # Load environment variables load_dotenv() @@ -74,7 +75,6 @@ async def ws_replay_listener(callback: Callable[[List[Dict[str, Any]]], Awaitabl """ auth_value = API_KEY if API_KEY.startswith("Bearer ") else f"Bearer {API_KEY}" headers = {"Authorization": auth_value} - decompressor = zstd.ZstdDecompressor() reconnect_delay = 1 # Start queue processor as background task @@ -86,23 +86,12 @@ async def ws_replay_listener(callback: Callable[[List[Dict[str, Any]]], Awaitabl logger.info(f"WebSocket connected to {label}") async for message in ws: try: - if isinstance(message, str): - raw = message.encode("utf-8") - else: - raw = bytes(message) - - text: str - try: - text = decompressor.decompress(raw, max_output_size=64 * 1024 * 1024).decode("utf-8") - except zstd.ZstdError: - text = raw.decode("utf-8") - - data = json.loads(text) + data = decode_spectra_ws_payload(message) replays = normalize_ws_message(data) if replays: await _replay_queue.put(replays) - except json.JSONDecodeError: - logger.warning("Invalid JSON from WS frame") + except SpectraPayloadError as e: + logger.warning(f"Invalid Spectra WS frame: {e}") except Exception as e: logger.error(f"Error processing WS message: {e}")