diff --git a/utils/gscore_client.py b/utils/gscore_client.py index 70b3fff..87cea4a 100644 --- a/utils/gscore_client.py +++ b/utils/gscore_client.py @@ -10,13 +10,16 @@ class GsCoreClient: def __init__(self): self._ws = None self._url: str = "" + self._token: str = "" # 新增 token 存储 self._recv_task: Optional[asyncio.Task] = None self._message_handler: Optional[Callable[[dict], Awaitable[None]]] = None - self._is_closing = False # 用于区分是“意外断开”还是“主动关闭” + self._is_closing = False - def configure(self, url: str, handler: Callable[[dict], Awaitable[None]]): + def configure(self, url: str, handler: Callable[[dict], Awaitable[None]], token: str = "liuwei"): + """配置连接信息,默认 token 为 liuwei""" self._url = url self._message_handler = handler + self._token = token async def connect(self): if not self._url: @@ -26,24 +29,33 @@ class GsCoreClient: return True try: logger.info(f"[GsCoreClient] 正在连接早柚核心: {self._url}") - self._ws = await websockets.connect(self._url, max_size=10 ** 7) + + # 在 Header 中携带 Token 进行鉴权 + headers = { + "Authorization": f"Bearer {self._token}", + "Token": self._token # 兼容某些直接校验 Token 字段的服务端 + } + + self._ws = await websockets.connect( + self._url, + max_size=10 ** 7, + extra_headers=headers + ) self._is_closing = False - # 确保旧任务已清理 if self._recv_task and not self._recv_task.done(): self._recv_task.cancel() loop = asyncio.get_running_loop() self._recv_task = loop.create_task(self._recv_loop()) - logger.success("[GsCoreClient] 连接成功") + logger.success("[GsCoreClient] 连接成功且鉴权通过") return True except Exception as e: - logger.error(f"[GsCoreClient] 连接早柚核心失败: {e}") + logger.error(f"[GsCoreClient] 连接或鉴权失败: {e}") self._ws = None return False async def _recv_loop(self): - """接收消息循环,包含自动重连触发""" while not self._is_closing: try: if not self._ws: @@ -57,12 +69,12 @@ class GsCoreClient: if self._message_handler: await self._message_handler(payload) except Exception as e: - logger.exception(f"[GsCoreClient] 解析或处理核心返回失败: {e}") + logger.exception(f"[GsCoreClient] 解析核心返回失败: {e}") except websockets.exceptions.ConnectionClosed as e: if not self._is_closing: - logger.warning(f"[GsCoreClient] WebSocket连接意外断开 ({e.code}),5秒后尝试重连...") - break # 跳出循环,触发下方的重连 + logger.warning(f"[GsCoreClient] WebSocket连接断开 ({e.code}),5秒后重连...") + break else: break except Exception as e: @@ -70,14 +82,12 @@ class GsCoreClient: break await asyncio.sleep(0) - # 退出循环后的重连逻辑 if not self._is_closing: self._ws = None await asyncio.sleep(5) await self.connect() async def send(self, message: str) -> bool: - """发送消息,支持失败重连一次""" if not self._ws: if not await self.connect(): return False @@ -85,24 +95,20 @@ class GsCoreClient: await self._ws.send(message.encode("utf-8")) return True except (websockets.exceptions.ConnectionClosed, Exception) as e: - logger.error(f"[GsCoreClient] 发送失败 ({e}),尝试重连并重发...") - # 立即清理并重连 + logger.error(f"[GsCoreClient] 发送失败,尝试重连: {e}") if await self.reconnect(): try: - # 重连成功后重试发送 await self._ws.send(message.encode("utf-8")) return True - except Exception as re_e: - logger.error(f"[GsCoreClient] 重连后发送依然失败: {re_e}") + except Exception: + pass return False async def reconnect(self): - logger.info("[GsCoreClient] 正在执行重连流程...") - await self.close(manual=False) # 标记为非手动关闭,允许 recv_loop 重启 + await self.close(manual=False) return await self.connect() async def close(self, manual: bool = True): - """关闭连接。manual=True 表示人为停止,不再自动重连""" self._is_closing = manual if self._recv_task: self._recv_task.cancel() @@ -113,8 +119,6 @@ class GsCoreClient: except Exception: pass self._ws = None - if manual: - logger.info("[GsCoreClient] 连接已手动关闭") gs_core_client = GsCoreClient() \ No newline at end of file