From 7b6ec94ec153c8287ad69d8ccba2c1152dee4f77 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:47:51 -0700 Subject: [PATCH] =?UTF-8?q?refactor(gateway/platforms):=20compact=20qqbot?= =?UTF-8?q?=20adapter=20=E2=80=94=20fold=20reconnect/ready/ws-close=20help?= =?UTF-8?q?ers,=20dedupe=20token/gateway=20fetch=20+=20http-client=20guard?= =?UTF-8?q?,=20collapse=20dm/group=20ACL=20ladders,=20magic-byte=20table,?= =?UTF-8?q?=20tighter=20layouts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gateway/platforms/qqbot/__init__.py | 36 +- gateway/platforms/qqbot/adapter.py | 1048 ++++++--------------- gateway/platforms/qqbot/chunked_upload.py | 176 +--- gateway/platforms/qqbot/constants.py | 11 +- gateway/platforms/qqbot/crypto.py | 8 +- gateway/platforms/qqbot/keyboards.py | 54 +- gateway/platforms/qqbot/onboard.py | 28 +- gateway/platforms/qqbot/utils.py | 13 +- 8 files changed, 380 insertions(+), 994 deletions(-) diff --git a/gateway/platforms/qqbot/__init__.py b/gateway/platforms/qqbot/__init__.py index 395ba630f1..9d93995dac 100644 --- a/gateway/platforms/qqbot/__init__.py +++ b/gateway/platforms/qqbot/__init__.py @@ -1,37 +1,15 @@ -"""QQBot platform package. +"""QQBot platform package. Re-exports adapter symbols so existing import paths +(``from gateway.platforms.qqbot import QQAdapter, check_qq_requirements``) keep working. +Sub-modules: constants, utils, crypto (AES-256-GCM), onboard (QR), chunked_upload, keyboards.""" -Re-exports the adapter symbols from ``adapter.py`` (the original ``qqbot.py``) -so all existing import paths remain unchanged, e.g. -``from gateway.platforms.qqbot import QQAdapter, check_qq_requirements``. - -Sub-modules: ``constants``, ``utils`` (User-Agent, config helpers), ``crypto`` -(AES-256-GCM), ``onboard`` (QR scan-to-configure), ``chunked_upload``, ``keyboards``. -""" - -from .adapter import ( # noqa: F401 - QQAdapter, - QQCloseError, - check_qq_requirements, - _coerce_list, - _ssrf_redirect_guard, -) +from .adapter import QQAdapter, QQCloseError, check_qq_requirements, _coerce_list, _ssrf_redirect_guard # noqa: F401 from .onboard import BindStatus, build_connect_url, qr_register # noqa: F401 from .crypto import decrypt_secret, generate_bind_key # noqa: F401 from .utils import build_user_agent, get_api_headers, coerce_list # noqa: F401 -from .chunked_upload import ( # noqa: F401 - ChunkedUploader, - UploadDailyLimitExceededError, - UploadFileTooLargeError, -) +from .chunked_upload import ChunkedUploader, UploadDailyLimitExceededError, UploadFileTooLargeError # noqa: F401 from .keyboards import ( # noqa: F401 - ApprovalRequest, - InlineKeyboard, - InteractionEvent, - build_approval_keyboard, - build_approval_text, - build_update_prompt_keyboard, - parse_approval_button_data, - parse_interaction_event, + ApprovalRequest, InlineKeyboard, InteractionEvent, build_approval_keyboard, build_approval_text, + build_update_prompt_keyboard, parse_approval_button_data, parse_interaction_event, parse_update_prompt_button_data, ) diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py index a78a97e96e..072bb8a13a 100644 --- a/gateway/platforms/qqbot/adapter.py +++ b/gateway/platforms/qqbot/adapter.py @@ -1,32 +1,11 @@ -""" -QQ Bot platform adapter using the Official QQ Bot API (v2). +"""QQ Bot platform adapter (Official QQ Bot API v2): WebSocket gateway for inbound +events, REST (``api.sgroup.qq.com``) for outbound messages and media uploads. -Connects to the QQ Bot WebSocket Gateway for inbound events and uses the -REST API (``api.sgroup.qq.com``) for outbound messages and media uploads. - -Configuration in config.yaml: - platforms: - qq: - enabled: true - extra: - app_id: "your-app-id" # or QQ_APP_ID env var - client_secret: "your-secret" # or QQ_CLIENT_SECRET env var - markdown_support: true # enable QQ markdown (msg_type 2) - dm_policy: "pairing" # open | allowlist | disabled | pairing - allow_from: ["openid_1"] - group_policy: "pairing" # open | allowlist | disabled | pairing - group_allow_from: ["group_openid_1"] - stt: # Voice-to-text config (optional) - provider: "zai" # zai (GLM-ASR), openai (Whisper), etc. - baseUrl: "https://open.bigmodel.cn/api/coding/paas/v4" - apiKey: "your-stt-api-key" # or set QQ_STT_API_KEY env var - model: "glm-asr" # glm-asr, whisper-1, etc. - - Voice transcription priority: - 1. QQ's built-in ``asr_refer_text`` (Tencent ASR — free, always tried first) - 2. Configured STT provider via ``stt`` config or ``QQ_STT_*`` env vars - -Reference: https://bot.q.qq.com/wiki/develop/api-v2/ +config.yaml ``platforms.qq.extra``: app_id / client_secret (or QQ_APP_ID / +QQ_CLIENT_SECRET), markdown_support, dm_policy + allow_from, group_policy + +group_allow_from (open | allowlist | disabled | pairing), and optional ``stt`` +{provider, baseUrl, apiKey, model} (or QQ_STT_* env vars). Voice transcription +tries QQ's free ``asr_refer_text`` first, then the configured STT provider. """ from __future__ import annotations @@ -45,7 +24,6 @@ from urllib.parse import urlparse try: import aiohttp - AIOHTTP_AVAILABLE = True except ImportError: AIOHTTP_AVAILABLE = False @@ -53,7 +31,6 @@ except ImportError: try: import httpx - HTTPX_AVAILABLE = True except ImportError: HTTPX_AVAILABLE = False @@ -61,14 +38,8 @@ except ImportError: from gateway.config import Platform, PlatformConfig from gateway.platforms.base import ( - gateway_trust_env, - BasePlatformAdapter, - MessageEvent, - MessageType, - SendResult, - _ssrf_redirect_guard, - cache_document_from_bytes, - cache_image_from_bytes, + gateway_trust_env, BasePlatformAdapter, MessageEvent, MessageType, SendResult, + _ssrf_redirect_guard, cache_document_from_bytes, cache_image_from_bytes, ) from gateway.platforms.helpers import strip_markdown from gateway.platforms.media_cache import ext_for_mime @@ -86,43 +57,19 @@ class QQCloseError(Exception): from gateway.platforms.qqbot.constants import ( - API_BASE, - TOKEN_URL, - GATEWAY_URL_PATH, - DEFAULT_API_TIMEOUT, - FILE_UPLOAD_TIMEOUT, - CONNECT_TIMEOUT_SECONDS, - RECONNECT_BACKOFF, - MAX_RECONNECT_ATTEMPTS, - RATE_LIMIT_DELAY, - QUICK_DISCONNECT_THRESHOLD, - MAX_QUICK_DISCONNECT_COUNT, - MAX_MESSAGE_LENGTH, - DEDUP_WINDOW_SECONDS, - DEDUP_MAX_SIZE, - MSG_TYPE_TEXT, - MSG_TYPE_MARKDOWN, - MSG_TYPE_MEDIA, - MSG_TYPE_INPUT_NOTIFY, - MEDIA_TYPE_IMAGE, - MEDIA_TYPE_VIDEO, - MEDIA_TYPE_VOICE, - MEDIA_TYPE_FILE, + API_BASE, TOKEN_URL, GATEWAY_URL_PATH, DEFAULT_API_TIMEOUT, FILE_UPLOAD_TIMEOUT, + CONNECT_TIMEOUT_SECONDS, RECONNECT_BACKOFF, MAX_RECONNECT_ATTEMPTS, RATE_LIMIT_DELAY, + QUICK_DISCONNECT_THRESHOLD, MAX_QUICK_DISCONNECT_COUNT, MAX_MESSAGE_LENGTH, + DEDUP_WINDOW_SECONDS, DEDUP_MAX_SIZE, MSG_TYPE_TEXT, MSG_TYPE_MARKDOWN, MSG_TYPE_MEDIA, + MSG_TYPE_INPUT_NOTIFY, MEDIA_TYPE_IMAGE, MEDIA_TYPE_VIDEO, MEDIA_TYPE_VOICE, MEDIA_TYPE_FILE, ) from gateway.platforms.qqbot.utils import coerce_list as _coerce_list, build_user_agent from gateway.platforms.qqbot.chunked_upload import ( - ChunkedUploader, - UploadDailyLimitExceededError, - UploadFileTooLargeError, + ChunkedUploader, UploadDailyLimitExceededError, UploadFileTooLargeError, ) from gateway.platforms.qqbot.keyboards import ( - ApprovalRequest, - InlineKeyboard, - InteractionEvent, - build_approval_keyboard, - build_update_prompt_keyboard, - parse_approval_button_data, - parse_interaction_event, + ApprovalRequest, InlineKeyboard, InteractionEvent, build_approval_keyboard, + build_update_prompt_keyboard, parse_approval_button_data, parse_interaction_event, parse_update_prompt_button_data, ) from gateway.platforms._shared import get_scoped_secret as _resolve_qq_secret @@ -153,15 +100,9 @@ class QQAdapter(BasePlatformAdapter): # WS close codes that are unrecoverable → stop reconnecting. _FATAL_CLOSE_CODES = { - 4001: "invalid opcode", - 4002: "invalid payload", - 4010: "invalid shard", - 4011: "sharding required", - 4012: "invalid API version", - 4013: "invalid intent", - 4014: "intent not authorized", - 4914: "offline/sandbox-only", - 4915: "banned", + 4001: "invalid opcode", 4002: "invalid payload", 4010: "invalid shard", + 4011: "sharding required", 4012: "invalid API version", 4013: "invalid intent", + 4014: "intent not authorized", 4914: "offline/sandbox-only", 4915: "banned", } # WS close codes that invalidate the session → clear it and re-identify on # the next Hello. 4009 (connection timeout) is deliberately absent: it is @@ -209,9 +150,7 @@ class QQAdapter(BasePlatformAdapter): self._dm_policy = str(extra.get("dm_policy", "pairing")).strip().lower() self._allow_from = _coerce_list(extra.get("allow_from") or extra.get("allowFrom")) self._group_policy = str(extra.get("group_policy", "pairing")).strip().lower() - self._group_allow_from = _coerce_list( - extra.get("group_allow_from") or extra.get("groupAllowFrom") - ) + self._group_allow_from = _coerce_list(extra.get("group_allow_from") or extra.get("groupAllowFrom")) # Connection state self._session: Optional[aiohttp.ClientSession] = None @@ -224,14 +163,10 @@ class QQAdapter(BasePlatformAdapter): self._last_seq: Optional[int] = None self._chat_type_map: Dict[str, str] = {} # chat_id → "c2c"|"group"|"guild"|"dm" - # Request/response correlation - self._pending_responses: Dict[str, asyncio.Future] = {} + self._pending_responses: Dict[str, asyncio.Future] = {} # request/response correlation self._seen_messages: Dict[str, float] = {} - - # Last inbound message ID per chat — used by send_typing - self._last_msg_id: Dict[str, str] = {} - # Typing debounce: chat_id → last send_typing timestamp - self._typing_sent_at: Dict[str, float] = {} + self._last_msg_id: Dict[str, str] = {} # last inbound message ID per chat (send_typing) + self._typing_sent_at: Dict[str, float] = {} # typing debounce: chat_id → last send_typing ts # Token cache self._access_token: Optional[str] = None @@ -245,9 +180,7 @@ class QQAdapter(BasePlatformAdapter): Callable[[InteractionEvent], Awaitable[None]] ] = self._default_interaction_dispatch - # ------------------------------------------------------------------ - # Properties - # ------------------------------------------------------------------ + # ── Properties ── @property def name(self) -> str: @@ -258,20 +191,16 @@ class QQAdapter(BasePlatformAdapter): """QQBot gates DM/group access at intake via dm_policy/group_policy.""" return True - # ------------------------------------------------------------------ - # Connection lifecycle - # ------------------------------------------------------------------ + # ── Connection lifecycle ── async def connect(self, *, is_reconnect: bool = False) -> bool: - """ - Authenticate, obtain gateway URL, and open the WebSocket. - - ``is_reconnect`` is accepted for interface conformance only (QQBot has no - server-side update queue). - """ + """Authenticate, obtain gateway URL, and open the WebSocket. ``is_reconnect`` + is accepted for interface conformance only (QQBot has no server-side update queue).""" for ok, code, message, hint in ( - (AIOHTTP_AVAILABLE, "qq_missing_dependency", "QQ startup failed: aiohttp not installed", ". Run: pip install aiohttp"), - (HTTPX_AVAILABLE, "qq_missing_dependency", "QQ startup failed: httpx not installed", ". Run: pip install httpx"), + (AIOHTTP_AVAILABLE, "qq_missing_dependency", "QQ startup failed: aiohttp not installed", + ". Run: pip install aiohttp"), + (HTTPX_AVAILABLE, "qq_missing_dependency", "QQ startup failed: httpx not installed", + ". Run: pip install httpx"), (self._app_id and self._client_secret, "qq_missing_credentials", "QQ startup failed: QQ_APP_ID and QQ_CLIENT_SECRET are required", ""), ): @@ -352,35 +281,34 @@ class QQAdapter(BasePlatformAdapter): self._http_client = None self._fail_pending("Disconnected") - # ------------------------------------------------------------------ - # Token management - # ------------------------------------------------------------------ + # ── Token management ── + + async def _fetch_json(self, what: str, request: Callable[[], Awaitable[Any]]) -> Dict[str, Any]: + """Run an httpx request and return its JSON; any failure → RuntimeError.""" + try: + resp = await request() + resp.raise_for_status() + return resp.json() + except Exception as exc: + raise RuntimeError(f"Failed to get QQ Bot {what}: {exc}") from exc + + def _token_fresh(self) -> bool: + return bool(self._access_token) and time.time() < self._token_expires_at - 60 async def _ensure_token(self) -> str: """Return a valid access token, refreshing if needed (with singleflight).""" - if self._access_token and time.time() < self._token_expires_at - 60: + if self._token_fresh(): return self._access_token - async with self._token_lock: - # Double-check after acquiring lock - if self._access_token and time.time() < self._token_expires_at - 60: + if self._token_fresh(): # double-check after acquiring lock return self._access_token - - try: - resp = await self._http_client.post( - TOKEN_URL, - json={"appId": self._app_id, "clientSecret": self._client_secret}, - timeout=DEFAULT_API_TIMEOUT, - ) - resp.raise_for_status() - data = resp.json() - except Exception as exc: - raise RuntimeError(f"Failed to get QQ Bot access token: {exc}") from exc - + data = await self._fetch_json("access token", lambda: self._http_client.post( + TOKEN_URL, json={"appId": self._app_id, "clientSecret": self._client_secret}, + timeout=DEFAULT_API_TIMEOUT, + )) token = data.get("access_token") if not token: raise RuntimeError(f"QQ Bot token response missing access_token: {data}") - expires_in = int(data.get("expires_in", 7200)) self._access_token = token self._token_expires_at = time.time() + expires_in @@ -390,81 +318,64 @@ class QQAdapter(BasePlatformAdapter): async def _get_gateway_url(self) -> str: """Fetch the WebSocket gateway URL from the REST API.""" token = await self._ensure_token() - try: - resp = await self._http_client.get( - f"{API_BASE}{GATEWAY_URL_PATH}", - headers={"Authorization": f"QQBot {token}", "User-Agent": build_user_agent()}, - timeout=DEFAULT_API_TIMEOUT, - ) - resp.raise_for_status() - data = resp.json() - except Exception as exc: - raise RuntimeError(f"Failed to get QQ Bot gateway URL: {exc}") from exc - + data = await self._fetch_json("gateway URL", lambda: self._http_client.get( + f"{API_BASE}{GATEWAY_URL_PATH}", + headers={"Authorization": f"QQBot {token}", "User-Agent": build_user_agent()}, + timeout=DEFAULT_API_TIMEOUT, + )) url = data.get("url") if not url: raise RuntimeError(f"QQ Bot gateway response missing url: {data}") return url - # ------------------------------------------------------------------ - # WebSocket lifecycle - # ------------------------------------------------------------------ + # ── WebSocket lifecycle ── async def _open_ws(self, gateway_url: str) -> None: """Open a WebSocket connection to the QQ Bot gateway.""" await self._close_ws() # Honor proxy env vars for the WebSocket (WSL setups need this). self._session = aiohttp.ClientSession(trust_env=gateway_trust_env()) - ws_proxy = ( - os.getenv("WSS_PROXY") - or os.getenv("wss_proxy") - or os.getenv("HTTPS_PROXY") - or os.getenv("https_proxy") - or os.getenv("ALL_PROXY") - or os.getenv("all_proxy") - ) + proxy_vars = ("WSS_PROXY", "wss_proxy", "HTTPS_PROXY", "https_proxy", "ALL_PROXY", "all_proxy") + ws_proxy = next((v for v in map(os.getenv, proxy_vars) if v), None) self._ws = await self._session.ws_connect( - gateway_url, - headers={"User-Agent": build_user_agent()}, - timeout=CONNECT_TIMEOUT_SECONDS, - proxy=ws_proxy, + gateway_url, headers={"User-Agent": build_user_agent()}, timeout=CONNECT_TIMEOUT_SECONDS, proxy=ws_proxy, ) logger.info("[%s] WebSocket connected to %s", self._log_tag, gateway_url) async def _listen_loop(self) -> None: - """Read WebSocket events and reconnect on errors. - - Close codes: 4004 → refresh token; 4006/4007/49xx → clear session and - re-identify; 4008 → rate limited, back off; _FATAL_CLOSE_CODES → stop. - """ + """Read WebSocket events and reconnect on errors. Close codes: 4004 → refresh + token; 4006/4007/49xx → clear session and re-identify; 4008 → rate limited, + back off; _FATAL_CLOSE_CODES → stop.""" backoff_idx = 0 connect_time = 0.0 quick_disconnect_count = 0 + async def reconnect() -> None: + nonlocal backoff_idx, quick_disconnect_count + if await self._reconnect(backoff_idx): + backoff_idx = quick_disconnect_count = 0 + else: + backoff_idx += 1 + while self._running: try: connect_time = time.monotonic() await self._read_events() - backoff_idx = 0 - quick_disconnect_count = 0 + backoff_idx = quick_disconnect_count = 0 except asyncio.CancelledError: return except QQCloseError as exc: if not self._running: return - code = exc.code - logger.warning( - "[%s] WebSocket closed: code=%s reason=%s", self._log_tag, code, exc.reason - ) + logger.warning("[%s] WebSocket closed: code=%s reason=%s", self._log_tag, code, exc.reason) # Quick disconnect detection (permission issues, misconfiguration) duration = time.monotonic() - connect_time if duration < QUICK_DISCONNECT_THRESHOLD and connect_time > 0: quick_disconnect_count += 1 logger.info( - "[%s] Quick disconnect (%.1fs), count: %d", - self._log_tag, duration, quick_disconnect_count, + "[%s] Quick disconnect (%.1fs), count: %d", self._log_tag, duration, quick_disconnect_count ) if quick_disconnect_count >= MAX_QUICK_DISCONNECT_COUNT: logger.error( @@ -473,9 +384,7 @@ class QQAdapter(BasePlatformAdapter): self._log_tag, ) self._set_fatal_error( - "qq_quick_disconnect", - "Too many quick disconnects — check bot permissions", - retryable=True, + "qq_quick_disconnect", "Too many quick disconnects — check bot permissions", retryable=True ) return else: @@ -491,44 +400,29 @@ class QQAdapter(BasePlatformAdapter): return if code == 4008: - logger.info( - "[%s] Rate limited (4008), waiting %ds", self._log_tag, RATE_LIMIT_DELAY - ) + logger.info("[%s] Rate limited (4008), waiting %ds", self._log_tag, RATE_LIMIT_DELAY) if backoff_idx >= MAX_RECONNECT_ATTEMPTS: self._mark_disconnected() return await asyncio.sleep(RATE_LIMIT_DELAY) - if await self._reconnect(backoff_idx): - backoff_idx = 0 - quick_disconnect_count = 0 - else: - backoff_idx += 1 + await reconnect() continue if code == 4004: - logger.info( - "[%s] Invalid token (4004), will refresh and reconnect", self._log_tag - ) + logger.info("[%s] Invalid token (4004), will refresh and reconnect", self._log_tag) self._access_token = None self._token_expires_at = 0.0 if code in self._SESSION_INVALID_CLOSE_CODES: - logger.info( - "[%s] Session error (%d), clearing session for re-identify", - self._log_tag, code, - ) + logger.info("[%s] Session error (%d), clearing session for re-identify", self._log_tag, code) self._session_id = None self._last_seq = None - if await self._reconnect(backoff_idx): - backoff_idx = 0 - quick_disconnect_count = 0 - else: - backoff_idx += 1 - if backoff_idx >= MAX_RECONNECT_ATTEMPTS: - logger.error("[%s] Max reconnect attempts reached (QQCloseError)", self._log_tag) - self._mark_disconnected() - return + await reconnect() + if backoff_idx >= MAX_RECONNECT_ATTEMPTS: + logger.error("[%s] Max reconnect attempts reached (QQCloseError)", self._log_tag) + self._mark_disconnected() + return except Exception as exc: if not self._running: @@ -541,12 +435,7 @@ class QQAdapter(BasePlatformAdapter): logger.error("[%s] Max reconnect attempts reached", self._log_tag) self._mark_disconnected() return - - if await self._reconnect(backoff_idx): - backoff_idx = 0 - quick_disconnect_count = 0 - else: - backoff_idx += 1 + await reconnect() async def _reconnect(self, backoff_idx: int) -> bool: """Attempt to reconnect the WebSocket. Returns True on success.""" @@ -614,29 +503,21 @@ class QQAdapter(BasePlatformAdapter): return True async def _send_identify(self) -> None: - """Send op 2 Identify (reply to Hello); server answers with READY. - - Intents: C2C_GROUP_AT_MESSAGES | PUBLIC_GUILD_MESSAGES | DIRECT_MESSAGE | INTERACTION. - """ + """Send op 2 Identify (reply to Hello); server answers with READY. Intents: + C2C_GROUP_AT_MESSAGES | PUBLIC_GUILD_MESSAGES | DIRECT_MESSAGE | INTERACTION.""" token = await self._ensure_token() - payload = { - "op": 2, - "d": { - "token": f"QQBot {token}", - "intents": (1 << 25) | (1 << 30) | (1 << 12) | (1 << 26), - "shard": [0, 1], - "properties": {"$os": "macOS", "$browser": "hermes-agent", "$device": "hermes-agent"}, - }, - } + payload = {"op": 2, "d": { + "token": f"QQBot {token}", + "intents": (1 << 25) | (1 << 30) | (1 << 12) | (1 << 26), + "shard": [0, 1], + "properties": {"$os": "macOS", "$browser": "hermes-agent", "$device": "hermes-agent"}, + }} await self._send_ws_auth("Identify", payload, "Identify sent") async def _send_resume(self) -> None: """Send op 6 Resume after a reconnect; on failure clear session → Identify next Hello.""" token = await self._ensure_token() - payload = { - "op": 6, - "d": {"token": f"QQBot {token}", "session_id": self._session_id, "seq": self._last_seq}, - } + payload = {"op": 6, "d": {"token": f"QQBot {token}", "session_id": self._session_id, "seq": self._last_seq}} if not await self._send_ws_auth( "Resume", payload, "Resume sent (session_id=%s, seq=%s)", self._session_id, self._last_seq ): @@ -652,32 +533,31 @@ class QQAdapter(BasePlatformAdapter): except RuntimeError: return None + def _close_ws_soon(self) -> None: + """Close the WS so _read_events raises and _listen_loop reconnects (with Resume).""" + if self._ws and not self._ws.closed: + self._create_task(self._ws.close()) + def _dispatch_payload(self, payload: Dict[str, Any]) -> None: """Route inbound WebSocket payloads (dispatch synchronously, spawn async handlers).""" - op = payload.get("op") - t = payload.get("t") - s = payload.get("s") - d = payload.get("d") + op, t, s, d = payload.get("op"), payload.get("t"), payload.get("s"), payload.get("d") if isinstance(s, int) and (self._last_seq is None or s > self._last_seq): self._last_seq = s if op == 10: # Hello — reply with Resume (have session) or Identify - d_data = d if isinstance(d, dict) else {} - interval_ms = d_data.get("heartbeat_interval", 30000) + interval_ms = (d if isinstance(d, dict) else {}).get("heartbeat_interval", 30000) self._heartbeat_interval = interval_ms / 1000.0 * 0.8 # 80% of server interval logger.debug( "[%s] Hello received, heartbeat_interval=%dms (sending every %.1fs)", self._log_tag, interval_ms, self._heartbeat_interval, ) - if self._session_id and self._last_seq is not None: - self._create_task(self._send_resume()) - else: - self._create_task(self._send_identify()) - return - - if op == 0 and t: # Dispatch + resume = self._session_id and self._last_seq is not None + self._create_task(self._send_resume() if resume else self._send_identify()) + elif op == 0 and t: # Dispatch if t == "READY": - self._handle_ready(d) + if isinstance(d, dict): # store session_id for resume + self._session_id = d.get("session_id") + logger.info("[%s] Ready, session_id=%s", self._log_tag, self._session_id) elif t == "RESUMED": logger.info("[%s] Session resumed", self._log_tag) elif t in self._INBOUND_HANDLERS: @@ -686,41 +566,23 @@ class QQAdapter(BasePlatformAdapter): self._create_task(self._on_interaction(d)) else: logger.debug("[%s] Unhandled dispatch: %s", self._log_tag, t) - return - - if op == 11: # Heartbeat ACK - return - - if op == 7: # Server Reconnect — close so _read_events raises and we reconnect w/ Resume + elif op == 11: # Heartbeat ACK + pass + elif op == 7: # Server Reconnect logger.info("[%s] Server requested reconnect (op 7)", self._log_tag) - if self._ws and not self._ws.closed: - self._create_task(self._ws.close()) - return - - if op == 9: # Invalid Session — d=True resumable, d=False re-identify from scratch + self._close_ws_soon() + elif op == 9: # Invalid Session — d=True resumable, d=False re-identify from scratch if d is not None and bool(d): logger.info("[%s] Invalid session (op 9, resumable)", self._log_tag) else: - logger.info( - "[%s] Invalid session (op 9, not resumable), clearing session", self._log_tag - ) + logger.info("[%s] Invalid session (op 9, not resumable), clearing session", self._log_tag) self._session_id = None self._last_seq = None - if self._ws and not self._ws.closed: - self._create_task(self._ws.close()) - return + self._close_ws_soon() + else: + logger.debug("[%s] Unknown op: %s", self._log_tag, op) - logger.debug("[%s] Unknown op: %s", self._log_tag, op) - - def _handle_ready(self, d: Any) -> None: - """Handle the READY event — store session_id for resume.""" - if isinstance(d, dict): - self._session_id = d.get("session_id") - logger.info("[%s] Ready, session_id=%s", self._log_tag, self._session_id) - - # ------------------------------------------------------------------ - # JSON helpers - # ------------------------------------------------------------------ + # ── JSON helpers ── @staticmethod def _parse_json(raw: Any) -> Optional[Dict[str, Any]]: @@ -738,9 +600,7 @@ class QQAdapter(BasePlatformAdapter): rand = int(uuid.uuid4().hex[:4], 16) return (time_part ^ rand) % 65536 - # ------------------------------------------------------------------ - # Inbound message handling - # ------------------------------------------------------------------ + # ── Inbound message handling ── async def handle_message(self, event: MessageEvent) -> None: """Cache the last message ID per chat, then delegate to base.""" @@ -765,14 +625,9 @@ class QQAdapter(BasePlatformAdapter): if handler: await getattr(self, handler)(d, msg_id, content, author, timestamp) - # ------------------------------------------------------------------ - # Inline-keyboard interactions (INTERACTION_CREATE) - # ------------------------------------------------------------------ + # ── Inline-keyboard interactions (INTERACTION_CREATE) ── - def set_interaction_callback( - self, - callback: Optional[Callable[[InteractionEvent], Awaitable[None]]], - ) -> None: + def set_interaction_callback(self, callback: Optional[Callable[[InteractionEvent], Awaitable[None]]]) -> None: """Register (or clear) the callback invoked per ACKed INTERACTION_CREATE.""" self._interaction_callback = callback @@ -802,8 +657,7 @@ class QQAdapter(BasePlatformAdapter): callback = self._interaction_callback if callback is None: logger.debug( - "[%s] No interaction callback registered; dropping button click %r", - self._log_tag, event.button_data, + "[%s] No interaction callback registered; dropping button click %r", self._log_tag, event.button_data ) return try: @@ -813,13 +667,9 @@ class QQAdapter(BasePlatformAdapter): async def _acknowledge_interaction(self, interaction_id: str, code: int = 0) -> None: """ACK a button interaction via ``PUT /interactions/{id}`` (code 0 = success).""" - if not self._http_client: - raise RuntimeError("HTTP client not initialized — not connected?") - resp = await self._http_client.put( + resp = await self._require_http_client().put( f"{API_BASE}/interactions/{interaction_id}", - headers=await self._auth_headers(), - json={"code": code}, - timeout=DEFAULT_API_TIMEOUT, + headers=await self._auth_headers(), json={"code": code}, timeout=DEFAULT_API_TIMEOUT, ) if resp.status_code >= 400: raise RuntimeError(f"Interaction ACK failed [{resp.status_code}]: {resp.text[:200]}") @@ -839,11 +689,7 @@ class QQAdapter(BasePlatformAdapter): parsed["user_id"] = parts[5] return parsed - def _is_authorized_interaction_for_session( - self, - event: InteractionEvent, - session_key: str, - ) -> bool: + def _is_authorized_interaction_for_session(self, event: InteractionEvent, session_key: str) -> bool: """Authorize approval/update interactions against session + operator.""" parsed = self._parse_gateway_session_key(session_key) operator = str(event.operator_openid or "").strip() @@ -854,20 +700,15 @@ class QQAdapter(BasePlatformAdapter): chat_id = parsed.get("chat_id", "") if chat_type == "c2c": return bool(chat_id) and operator == chat_id - if chat_type in {"group", "guild"}: event_chat = str(event.group_openid or event.guild_id or "").strip() if not event_chat or event_chat != chat_id: return False session_user = str(parsed.get("user_id", "")).strip() return bool(session_user) and operator == session_user - return False - async def _default_interaction_dispatch( - self, - event: InteractionEvent, - ) -> None: + async def _default_interaction_dispatch(self, event: InteractionEvent) -> None: """Default interaction callback: ``approve::`` → tools.approval.resolve_gateway_approval; ``update_prompt:`` → ``~/.hermes/.update_response``; anything else is ignored at DEBUG.""" @@ -880,15 +721,11 @@ class QQAdapter(BasePlatformAdapter): session_key, decision = approval choice = self._APPROVAL_BUTTON_TO_CHOICE.get(decision) if choice is None: - logger.warning( - "[%s] Unknown approval decision %r (session=%s)", - self._log_tag, decision, session_key, - ) + logger.warning("[%s] Unknown approval decision %r (session=%s)", self._log_tag, decision, session_key) return if not self._is_authorized_interaction_for_session(event, session_key): logger.warning( - "[%s] Rejected unauthorized approval click for session %s " - "(operator=%s)", + "[%s] Rejected unauthorized approval click for session %s (operator=%s)", self._log_tag, session_key, event.operator_openid, ) return @@ -900,28 +737,21 @@ class QQAdapter(BasePlatformAdapter): self._log_tag, count, session_key, choice, event.operator_openid, ) except Exception as exc: - logger.error( - "[%s] resolve_gateway_approval failed for session %s: %s", - self._log_tag, session_key, exc, - ) + logger.error("[%s] resolve_gateway_approval failed for session %s: %s", self._log_tag, session_key, exc) return update_answer = parse_update_prompt_button_data(button_data) if update_answer is not None: - update_session_key = f"agent:main:qqbot:{event.scene}:{event.group_openid or event.guild_id or event.user_openid}" - if not self._is_authorized_interaction_for_session(event, update_session_key): + chat = event.group_openid or event.guild_id or event.user_openid + if not self._is_authorized_interaction_for_session(event, f"agent:main:qqbot:{event.scene}:{chat}"): logger.warning( - "[%s] Rejected unauthorized update prompt click (operator=%s)", - self._log_tag, event.operator_openid, + "[%s] Rejected unauthorized update prompt click (operator=%s)", self._log_tag, event.operator_openid ) return self._write_update_response(update_answer, event.operator_openid) return - logger.debug( - "[%s] Unrecognised button_data %r from interaction %s", - self._log_tag, button_data, event.id, - ) + logger.debug("[%s] Unrecognised button_data %r from interaction %s", self._log_tag, button_data, event.id) @staticmethod def _write_update_response(answer: str, operator: str = "") -> None: @@ -937,144 +767,83 @@ class QQAdapter(BasePlatformAdapter): except Exception as exc: logger.error("Failed to write update response: %s", exc) - async def _handle_c2c_message( - self, - d: Dict[str, Any], - msg_id: str, - content: str, - author: Dict[str, Any], - timestamp: str, - ) -> None: + async def _handle_c2c_message(self, d, msg_id, content, author, timestamp) -> None: """Handle a C2C (private) message event.""" user_openid = str(author.get("user_openid", "")) - if not user_openid: - return - if not self._is_dm_intake_allowed(user_openid): + if not user_openid or not self._is_dm_intake_allowed(user_openid): return attachments_raw = d.get("attachments") logger.info( "[%s] C2C message: id=%s content=%r attachments=%s", - self._log_tag, - msg_id, - content[:50] if content else "", - ( - f"{len(attachments_raw) if isinstance(attachments_raw, list) else 0} items" - if attachments_raw - else "None" - ), + self._log_tag, msg_id, content[:50] if content else "", + f"{len(attachments_raw) if isinstance(attachments_raw, list) else 0} items" if attachments_raw else "None", ) if attachments_raw and isinstance(attachments_raw, list): for _i, _att in enumerate(attachments_raw): if isinstance(_att, dict): logger.info( "[%s] attachment[%d]: content_type=%s url=%s filename=%s", - self._log_tag, - _i, - _att.get("content_type", ""), - str(_att.get("url", ""))[:80], - _att.get("filename", ""), + self._log_tag, _i, _att.get("content_type", ""), + str(_att.get("url", ""))[:80], _att.get("filename", ""), ) - text, image_urls, image_media_types, n_voice = await self._absorb_attachments( - content, attachments_raw - ) - logger.info( - "[%s] After processing: images=%d, voice=%d", self._log_tag, len(image_urls), n_voice - ) + text, image_urls, image_media_types, n_voice = await self._absorb_attachments(content, attachments_raw) + logger.info("[%s] After processing: images=%d, voice=%d", self._log_tag, len(image_urls), n_voice) await self._emit_inbound( d, msg_id, timestamp, text, image_urls, image_media_types, chat_id=user_openid, qq_chat_type="c2c", user_id=user_openid, chat_type="dm", ) - async def _handle_group_message( - self, - d: Dict[str, Any], - msg_id: str, - content: str, - author: Dict[str, Any], - timestamp: str, - ) -> None: + async def _handle_group_message(self, d, msg_id, content, author, timestamp) -> None: """Handle a group @-message event.""" group_openid = str(d.get("group_openid", "")) - if not group_openid: + member = str(author.get("member_openid", "")) + if not group_openid or not self._is_group_allowed(group_openid, member): return - if not self._is_group_allowed(group_openid, str(author.get("member_openid", ""))): - return - text, image_urls, image_media_types, _ = await self._absorb_attachments( self._strip_at_mention(content), d.get("attachments") ) await self._emit_inbound( d, msg_id, timestamp, text, image_urls, image_media_types, - chat_id=group_openid, qq_chat_type="group", - user_id=str(author.get("member_openid", "")), chat_type="group", + chat_id=group_openid, qq_chat_type="group", user_id=member, chat_type="group", ) - async def _handle_guild_message( - self, - d: Dict[str, Any], - msg_id: str, - content: str, - author: Dict[str, Any], - timestamp: str, - ) -> None: + async def _handle_guild_message(self, d, msg_id, content, author, timestamp) -> None: """Handle a guild/channel message event.""" channel_id = str(d.get("channel_id", "")) if not channel_id: return - # group_policy ACL — guild channels are group-like; without it any guild # member could bypass the allowlist. guild_id = str(d.get("guild_id", "")) author_id = str(author.get("id", "")) if not self._is_group_allowed(guild_id or channel_id, author_id): - logger.debug( - "[%s] Guild message blocked by ACL: channel=%s user=%s", - self._log_tag, channel_id, author_id, - ) + logger.debug("[%s] Guild message blocked by ACL: channel=%s user=%s", self._log_tag, channel_id, author_id) return member = d.get("member") if isinstance(d.get("member"), dict) else {} nick = str(member.get("nick", "")) or str(author.get("username", "")) - - text, image_urls, image_media_types, _ = await self._absorb_attachments( - content, d.get("attachments") - ) + text, image_urls, image_media_types, _ = await self._absorb_attachments(content, d.get("attachments")) await self._emit_inbound( d, msg_id, timestamp, text, image_urls, image_media_types, - chat_id=channel_id, qq_chat_type="guild", - user_id=str(author.get("id", "")), user_name=nick or None, chat_type="group", + chat_id=channel_id, qq_chat_type="guild", user_id=author_id, user_name=nick or None, chat_type="group", ) - async def _handle_dm_message( - self, - d: Dict[str, Any], - msg_id: str, - content: str, - author: Dict[str, Any], - timestamp: str, - ) -> None: + async def _handle_dm_message(self, d, msg_id, content, author, timestamp) -> None: """Handle a guild DM message event.""" guild_id = str(d.get("guild_id", "")) if not guild_id: return - # dm_policy ACL — without it any guild member could bypass the allowlist via DM. author_id = str(author.get("id", "")) if not self._is_dm_intake_allowed(author_id): - logger.debug( - "[%s] Guild DM blocked by ACL: guild=%s user=%s", - self._log_tag, guild_id, author_id, - ) + logger.debug("[%s] Guild DM blocked by ACL: guild=%s user=%s", self._log_tag, guild_id, author_id) return - - text, image_urls, image_media_types, _ = await self._absorb_attachments( - content, d.get("attachments") - ) + text, image_urls, image_media_types, _ = await self._absorb_attachments(content, d.get("attachments")) await self._emit_inbound( d, msg_id, timestamp, text, image_urls, image_media_types, - chat_id=guild_id, qq_chat_type="dm", user_id=str(author.get("id", "")), chat_type="dm", + chat_id=guild_id, qq_chat_type="dm", user_id=author_id, chat_type="dm", ) _INBOUND_HANDLERS = { @@ -1085,18 +854,14 @@ class QQAdapter(BasePlatformAdapter): "DIRECT_MESSAGE_CREATE": "_handle_dm_message", } - # ------------------------------------------------------------------ - # Shared inbound pipeline (all four message kinds) - # ------------------------------------------------------------------ + # ── Shared inbound pipeline (all four message kinds) ── @staticmethod def _append_block(text: str, block: str) -> str: """Append *block* to *text* after a blank line (or return block alone if text is blank).""" return (text + "\n\n" + block).strip() if text.strip() else block - async def _absorb_attachments( - self, text: str, attachments: Any, - ) -> Tuple[str, List[str], List[str], int]: + async def _absorb_attachments(self, text: str, attachments: Any) -> Tuple[str, List[str], List[str], int]: """Run attachments through _process_attachments and fold transcripts/file info into *text*. Returns (text, image_urls, image_media_types, n_voice).""" att = await self._process_attachments(attachments) @@ -1108,17 +873,8 @@ class QQAdapter(BasePlatformAdapter): return text, att["image_urls"], att["image_media_types"], len(voice_transcripts) async def _emit_inbound( - self, - d: Dict[str, Any], - msg_id: str, - timestamp: str, - text: str, - image_urls: List[str], - image_media_types: List[str], - *, - chat_id: str, - qq_chat_type: str, - **source_kwargs: Any, + self, d: Dict[str, Any], msg_id: str, timestamp: str, text: str, image_urls: List[str], + image_media_types: List[str], *, chat_id: str, qq_chat_type: str, **source_kwargs: Any, ) -> None: """Merge quoted context, drop empty events, remember the QQ chat kind and dispatch.""" quoted = await self._process_quoted_context(d) @@ -1126,7 +882,6 @@ class QQAdapter(BasePlatformAdapter): if quoted["image_urls"]: image_urls = image_urls + quoted["image_urls"] image_media_types = image_media_types + quoted["image_media_types"] - if not text.strip() and not image_urls: return @@ -1143,24 +898,14 @@ class QQAdapter(BasePlatformAdapter): ) await self.handle_message(event) - # ------------------------------------------------------------------ - # Quoted-message handling - # ------------------------------------------------------------------ + # ── Quoted-message handling ── - async def _process_quoted_context( - self, - d: Dict[str, Any], - ) -> Dict[str, Any]: - """Process the quoted message a user is replying to. - - A quote-reply has ``message_type == 103`` with the referenced message's - content + attachments in ``msg_elements`` (normally just [0]). Quoted - attachments run through the same _process_attachments pipeline, so - quoted voice gets STT and quoted images are cached identically. - - Returns ``{"quote_block", "image_urls", "image_media_types"}``; - quote_block is "" when nothing is quoted. - """ + async def _process_quoted_context(self, d: Dict[str, Any]) -> Dict[str, Any]: + """Process the quoted message a user is replying to (``message_type == 103``; + referenced content + attachments live in ``msg_elements``). Quoted attachments + go through _process_attachments so quoted voice gets STT and quoted images are + cached identically. Returns ``{"quote_block", "image_urls", "image_media_types"}``; + quote_block is "" when nothing is quoted.""" empty = {"quote_block": "", "image_urls": [], "image_media_types": []} try: if int(d.get("message_type", 0) or 0) != 103: @@ -1182,9 +927,7 @@ class QQAdapter(BasePlatformAdapter): quoted_text_parts.append(etext) eatts = elem.get("attachments") if isinstance(eatts, list): - for a in eatts: - if isinstance(a, dict): - all_attachments.append(a) + all_attachments.extend(a for a in eatts if isinstance(a, dict)) att_result = await self._process_attachments(all_attachments) quoted_images = att_result.get("image_urls") or [] @@ -1199,11 +942,8 @@ class QQAdapter(BasePlatformAdapter): if not lines and not quoted_images: return empty # Images-only quote still gets a marker so the LLM knows context was referenced. - quote_block = ( - "[Quoted message]:\n" + "\n".join(lines) if lines else "[Quoted message]: (image)" - ) return { - "quote_block": quote_block, + "quote_block": "[Quoted message]:\n" + "\n".join(lines) if lines else "[Quoted message]: (image)", "image_urls": quoted_images, "image_media_types": att_result.get("image_media_types") or [], } @@ -1213,13 +953,9 @@ class QQAdapter(BasePlatformAdapter): """Prepend ``quote_block`` to *text*, separated by a blank line.""" if not quote_block: return text - if text.strip(): - return f"{quote_block}\n\n{text}".strip() - return quote_block + return f"{quote_block}\n\n{text}".strip() if text.strip() else quote_block - # ------------------------------------------------------------------ - # Attachment processing - # ------------------------------------------------------------------ + # ── Attachment processing ── @staticmethod def _detect_message_type(media_urls: list, media_types: list): @@ -1238,16 +974,10 @@ class QQAdapter(BasePlatformAdapter): logger.debug("Unknown media content_type '%s', defaulting to TEXT", first_type) return MessageType.TEXT - async def _process_attachments( - self, - attachments: Any, - ) -> Dict[str, Any]: - """Process inbound attachments uniformly (images, voice, other files). - - Returns ``{"image_urls", "image_media_types", "voice_transcripts", "attachment_info"}`` - (cached image paths + MIME types, "[Voice] ..." transcripts, and a text - description of non-image/non-voice files). - """ + async def _process_attachments(self, attachments: Any) -> Dict[str, Any]: + """Process inbound attachments uniformly. Returns ``{"image_urls", + "image_media_types", "voice_transcripts", "attachment_info"}`` (cached image + paths + MIME types, "[Voice] ..." transcripts, text description of other files).""" image_urls: List[str] = [] image_media_types: List[str] = [] voice_transcripts: List[str] = [] @@ -1256,7 +986,6 @@ class QQAdapter(BasePlatformAdapter): for att in attachments if isinstance(attachments, list) else (): if not isinstance(att, dict): continue - ct = str(att.get("content_type", "")).strip().lower() url = str(att.get("url", "")).strip() filename = str(att.get("filename", "")) @@ -1264,7 +993,6 @@ class QQAdapter(BasePlatformAdapter): continue if url.startswith("//"): url = f"https:{url}" - logger.debug( "[%s] Processing attachment: content_type=%s, url=%s, filename=%s", self._log_tag, ct, url[:80], filename, @@ -1274,9 +1002,7 @@ class QQAdapter(BasePlatformAdapter): asr_refer = att.get("asr_refer_text") voice_wav_url = att.get("voice_wav_url") transcript = await self._stt_voice_attachment( - url, - ct, - filename, + url, ct, filename, asr_refer_text=(asr_refer.strip() if isinstance(asr_refer, str) else "") or None, voice_wav_url=(voice_wav_url.strip() if isinstance(voice_wav_url, str) else "") or None, ) @@ -1293,9 +1019,7 @@ class QQAdapter(BasePlatformAdapter): image_urls.append(cached_path) image_media_types.append(ct or "image/jpeg") elif cached_path: - logger.warning( - "[%s] Cached image path does not exist: %s", self._log_tag, cached_path - ) + logger.warning("[%s] Cached image path does not exist: %s", self._log_tag, cached_path) except Exception as exc: logger.debug("[%s] Failed to cache image: %s", self._log_tag, exc) else: @@ -1314,9 +1038,7 @@ class QQAdapter(BasePlatformAdapter): "attachment_info": "\n".join(other_attachments), } - async def _download_and_cache( - self, url: str, content_type: str, original_name: str = "", - ) -> Optional[str]: + async def _download_and_cache(self, url: str, content_type: str, original_name: str = "") -> Optional[str]: """Download a URL and cache it locally (``original_name`` falls back to the URL basename).""" from tools.url_safety import is_safe_url @@ -1324,7 +1046,6 @@ class QQAdapter(BasePlatformAdapter): raise ValueError(f"Blocked unsafe URL: {url[:80]}") if not self._http_client: return None - try: resp = await self._http_client.get(url, timeout=30.0, headers=self._qq_media_headers()) resp.raise_for_status() @@ -1334,11 +1055,8 @@ class QQAdapter(BasePlatformAdapter): return None if content_type.startswith("image/"): - # Historical qqbot mapping: trust mimetypes' guess (never the shared - # table) and fall back to .jpg. - ext = ext_for_mime( - content_type, use_defaults=False, use_mimetypes=True, fallback=".jpg" - ) or ".jpg" + # Historical qqbot mapping: trust mimetypes' guess (never the shared table), fall back to .jpg. + ext = ext_for_mime(content_type, use_defaults=False, use_mimetypes=True, fallback=".jpg") or ".jpg" return cache_image_from_bytes(data, ext) if content_type == "voice" or content_type.startswith("audio/"): # QQ voice is usually .amr/.silk — convert to .wav for STT engines. @@ -1359,18 +1077,11 @@ class QQAdapter(BasePlatformAdapter): def _qq_media_headers(self) -> Dict[str, str]: """Authorization header for QQ multimedia CDN downloads (required, else non-200).""" - if self._access_token: - return {"Authorization": f"QQBot {self._access_token}"} - return {} + return {"Authorization": f"QQBot {self._access_token}"} if self._access_token else {} async def _stt_voice_attachment( - self, - url: str, - content_type: str, - filename: str, - *, - asr_refer_text: Optional[str] = None, - voice_wav_url: Optional[str] = None, + self, url: str, content_type: str, filename: str, *, + asr_refer_text: Optional[str] = None, voice_wav_url: Optional[str] = None, ) -> Optional[str]: """Transcribe a voice attachment. Priority: QQ's free ``asr_refer_text`` → STT on ``voice_wav_url`` (pre-converted WAV, no SILK decode) → STT on the @@ -1397,7 +1108,6 @@ class QQAdapter(BasePlatformAdapter): if not self._http_client: logger.warning("[%s] STT: no HTTP client", self._log_tag) return None - download_headers = self._qq_media_headers() # QQ CDN requires Authorization logger.debug( "[%s] STT: downloading voice from %s (pre_wav=%s, headers=%s)", @@ -1414,17 +1124,13 @@ class QQAdapter(BasePlatformAdapter): ) if len(audio_data) < 10: logger.warning( - "[%s] STT: downloaded data too small (%d bytes), skipping", - self._log_tag, len(audio_data), + "[%s] STT: downloaded data too small (%d bytes), skipping", self._log_tag, len(audio_data) ) return None if is_pre_wav: wav_path = self._write_temp(audio_data, ".wav") - logger.debug( - "[%s] STT: using pre-converted WAV directly (%d bytes)", - self._log_tag, len(audio_data), - ) + logger.debug("[%s] STT: using pre-converted WAV directly (%d bytes)", self._log_tag, len(audio_data)) else: logger.debug("[%s] STT: converting to wav, filename=%r", self._log_tag, filename) wav_path = await self._convert_audio_to_wav_file(audio_data, filename) @@ -1437,17 +1143,13 @@ class QQAdapter(BasePlatformAdapter): transcript = await self._call_stt(wav_path) finally: self._unlink_quiet(wav_path) - if transcript: logger.debug("[%s] STT success: %r", self._log_tag, transcript[:100]) else: logger.warning("[%s] STT: ASR returned empty transcript", self._log_tag) return transcript except (httpx.HTTPStatusError, httpx.TransportError, IOError) as exc: - logger.warning( - "[%s] STT failed for voice attachment: %s: %s", - self._log_tag, type(exc).__name__, exc, - ) + logger.warning("[%s] STT failed for voice attachment: %s: %s", self._log_tag, type(exc).__name__, exc) return None @staticmethod @@ -1471,9 +1173,14 @@ class QQAdapter(BasePlatformAdapter): """True when *wav_path* exists and holds more than a bare 44-byte header.""" return Path(wav_path).exists() and Path(wav_path).stat().st_size > 44 - async def _convert_audio_to_wav_file( - self, audio_data: bytes, filename: str - ) -> Optional[str]: + def _log_converted(self, tool: str, src_path: str, wav_path: str) -> str: + logger.debug( + "[%s] %s converted %s to wav (%d bytes)", + self._log_tag, tool, Path(src_path).name, Path(wav_path).stat().st_size, + ) + return wav_path + + async def _convert_audio_to_wav_file(self, audio_data: bytes, filename: str) -> Optional[str]: """Convert audio bytes to a temp .wav: pilk (SILK, which ffmpeg can't decode) → ffmpeg → raw-PCM last resort. Returns the wav path or None.""" ext = Path(filename).suffix.lower() or self._guess_ext_from_data(audio_data) @@ -1483,7 +1190,6 @@ class QQAdapter(BasePlatformAdapter): ) src_path = self._write_temp(audio_data, ext) wav_path = src_path.rsplit(".", 1)[0] + ".wav" - result = ( await self._convert_silk_to_wav(src_path, wav_path) or await self._convert_ffmpeg_to_wav(src_path, wav_path) @@ -1492,20 +1198,16 @@ class QQAdapter(BasePlatformAdapter): self._unlink_quiet(src_path) return result - @staticmethod - def _guess_ext_from_data(data: bytes) -> str: + _MAGIC_EXTS = ( + (b"#!SILK", ".silk"), (b"\x02!", ".silk"), (b"RIFF", ".wav"), (b"fLaC", ".flac"), + (b"\xff\xfb", ".mp3"), (b"\xff\xf3", ".mp3"), (b"\xff\xf2", ".mp3"), + (b"\x30\x26\xb2\x75", ".ogg"), (b"\x4f\x67\x67\x53", ".ogg"), + ) + + @classmethod + def _guess_ext_from_data(cls, data: bytes) -> str: """Guess file extension from magic bytes (unknown → .amr, QQ's most common).""" - if data[:6] == b"#!SILK" or data[:2] == b"\x02!": - return ".silk" - if data[:4] == b"RIFF": - return ".wav" - if data[:4] == b"fLaC": - return ".flac" - if data[:2] in {b"\xff\xfb", b"\xff\xf3", b"\xff\xf2"}: - return ".mp3" - if data[:4] == b"\x30\x26\xb2\x75" or data[:4] == b"\x4f\x67\x67\x53": - return ".ogg" - return ".amr" + return next((ext for magic, ext in cls._MAGIC_EXTS if data.startswith(magic)), ".amr") @staticmethod def _looks_like_silk(data: bytes) -> bool: @@ -1517,20 +1219,13 @@ class QQAdapter(BasePlatformAdapter): try: import pilk except ImportError: - logger.warning( - "[%s] pilk not installed — cannot decode SILK audio. Run: pip install pilk", - self._log_tag, - ) + logger.warning("[%s] pilk not installed — cannot decode SILK audio. Run: pip install pilk", self._log_tag) return None try: pilk.silk_to_wav(src_path, wav_path, rate=16000) if self._wav_ok(wav_path): - logger.debug( - "[%s] pilk converted %s to wav (%d bytes)", - self._log_tag, Path(src_path).name, Path(wav_path).stat().st_size, - ) - return wav_path + return self._log_converted("pilk", src_path, wav_path) except Exception as exc: logger.debug("[%s] pilk direct conversion failed: %s", self._log_tag, exc) @@ -1573,8 +1268,7 @@ class QQAdapter(BasePlatformAdapter): try: proc = await asyncio.create_subprocess_exec( "ffmpeg", "-y", "-i", src_path, "-ar", "16000", "-ac", "1", wav_path, - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE, ) await asyncio.wait_for(proc.wait(), timeout=30) if proc.returncode != 0: @@ -1589,15 +1283,9 @@ class QQAdapter(BasePlatformAdapter): return None if not self._wav_ok(wav_path): - logger.warning( - "[%s] ffmpeg produced no/small output for %s", self._log_tag, Path(src_path).name - ) + logger.warning("[%s] ffmpeg produced no/small output for %s", self._log_tag, Path(src_path).name) return None - logger.debug( - "[%s] ffmpeg converted %s to wav (%d bytes)", - self._log_tag, Path(src_path).name, Path(wav_path).stat().st_size, - ) - return wav_path + return self._log_converted("ffmpeg", src_path, wav_path) def _resolve_stt_config(self) -> Optional[Dict[str, str]]: """Resolve STT backend: ``extra["stt"]`` config first, then ``QQ_STT_*`` env @@ -1613,11 +1301,8 @@ class QQAdapter(BasePlatformAdapter): provider = stt_cfg.get("provider", "zai") base_url = _STT_PROVIDER_BASE_URLS.get(provider, "") if base_url: - return { - "base_url": base_url, - "api_key": api_key, - "model": model or ("glm-asr" if provider in {"zai", "glm"} else "whisper-1"), - } + default_model = "glm-asr" if provider in {"zai", "glm"} else "whisper-1" + return {"base_url": base_url, "api_key": api_key, "model": model or default_model} qq_stt_key = _resolve_qq_secret("QQ_STT_API_KEY", "") if qq_stt_key: @@ -1654,18 +1339,12 @@ class QQAdapter(BasePlatformAdapter): content = choices[0].get("message", {}).get("content", "") if content.strip(): return content.strip() - text = result.get("text", "") - return text.strip() or None + return result.get("text", "").strip() or None except (httpx.HTTPStatusError, IOError) as exc: - logger.warning( - "[%s] STT API call failed (model=%s, base=%s): %s", - self._log_tag, model, base_url[:50], exc, - ) + logger.warning("[%s] STT API call failed (model=%s, base=%s): %s", self._log_tag, model, base_url[:50], exc) return None - async def _convert_audio_to_wav( - self, audio_data: bytes, source_url: str - ) -> Optional[str]: + async def _convert_audio_to_wav(self, audio_data: bytes, source_url: str) -> Optional[str]: """Convert audio bytes to .wav (pilk for SILK, else ffmpeg) and cache the result; on conversion failure the original bytes are cached as ``qq_voice``.""" ext = Path(urlparse(source_url).path).suffix.lower() @@ -1680,10 +1359,7 @@ class QQAdapter(BasePlatformAdapter): else: result = await self._convert_ffmpeg_to_wav(src_path, wav_path) if not result: - logger.warning( - "[%s] audio conversion failed for %s (format=%s)", - self._log_tag, source_url[:60], ext, - ) + logger.warning("[%s] audio conversion failed for %s (format=%s)", self._log_tag, source_url[:60], ext) return cache_document_from_bytes(audio_data, f"qq_voice{ext}") except Exception: return cache_document_from_bytes(audio_data, f"qq_voice{ext}") @@ -1698,30 +1374,24 @@ class QQAdapter(BasePlatformAdapter): logger.debug("[%s] Failed to read converted wav: %s", self._log_tag, exc) return None - # ------------------------------------------------------------------ - # Outbound messaging — REST API - # ------------------------------------------------------------------ + # ── Outbound messaging — REST API ── - async def _api_request( - self, - method: str, - path: str, - body: Optional[Dict[str, Any]] = None, - timeout: float = DEFAULT_API_TIMEOUT, - ) -> Dict[str, Any]: - """Make an authenticated REST API request to QQ Bot API.""" + def _require_http_client(self) -> "httpx.AsyncClient": if not self._http_client: raise RuntimeError("HTTP client not initialized — not connected?") + return self._http_client + + async def _api_request( + self, method: str, path: str, body: Optional[Dict[str, Any]] = None, timeout: float = DEFAULT_API_TIMEOUT, + ) -> Dict[str, Any]: + """Make an authenticated REST API request to QQ Bot API.""" + client = self._require_http_client() headers = await self._auth_headers() try: - resp = await self._http_client.request( - method, f"{API_BASE}{path}", headers=headers, json=body, timeout=timeout - ) + resp = await client.request(method, f"{API_BASE}{path}", headers=headers, json=body, timeout=timeout) data = resp.json() if resp.status_code >= 400: - raise RuntimeError( - f"QQ Bot API error [{resp.status_code}] {path}: {data.get('message', data)}" - ) + raise RuntimeError(f"QQ Bot API error [{resp.status_code}] {path}: {data.get('message', data)}") return data except httpx.TimeoutException as exc: raise RuntimeError(f"QQ Bot API timeout [{path}]: {exc}") from exc @@ -1736,18 +1406,11 @@ class QQAdapter(BasePlatformAdapter): } async def _upload_media( - self, - target_type: str, - target_id: str, - file_type: int, - url: Optional[str] = None, - file_data: Optional[str] = None, - srv_send_msg: bool = False, - file_name: Optional[str] = None, + self, target_type: str, target_id: str, file_type: int, url: Optional[str] = None, + file_data: Optional[str] = None, srv_send_msg: bool = False, file_name: Optional[str] = None, ) -> Dict[str, Any]: """Upload media and return file_info.""" - kind = "users" if target_type == "c2c" else "groups" - path = f"/v2/{kind}/{target_id}/files" + path = self._rest_path(target_type, target_id, "files") body: Dict[str, Any] = {"file_type": file_type, "srv_send_msg": srv_send_msg} if url: body["url"] = url @@ -1761,23 +1424,19 @@ class QQAdapter(BasePlatformAdapter): return await self._api_request("POST", path, body, timeout=FILE_UPLOAD_TIMEOUT) except RuntimeError as exc: err_msg = str(exc) - if any(kw in err_msg for kw in ("400", "401", "Invalid", "timeout", "Timeout")): - raise - if attempt < 2: - await asyncio.sleep(1.5 * (attempt + 1)) - else: + if attempt == 2 or any(kw in err_msg for kw in ("400", "401", "Invalid", "timeout", "Timeout")): raise + await asyncio.sleep(1.5 * (attempt + 1)) - # Maximum time (seconds) to wait for reconnection before giving up on send. - _RECONNECT_WAIT_SECONDS = 15.0 - # How often (seconds) to poll is_connected while waiting. - _RECONNECT_POLL_INTERVAL = 0.5 + _RECONNECT_WAIT_SECONDS = 15.0 # max wait for reconnection before giving up on send + _RECONNECT_POLL_INTERVAL = 0.5 # is_connected poll interval while waiting async def _wait_for_reconnection(self) -> bool: """Poll is_connected for up to _RECONNECT_WAIT_SECONDS — covers the race where send() lands between a disconnect and _listen_loop's reconnect.""" - logger.info("[%s] Not connected — waiting for reconnection (up to %.0fs)", - self._log_tag, self._RECONNECT_WAIT_SECONDS) + logger.info( + "[%s] Not connected — waiting for reconnection (up to %.0fs)", self._log_tag, self._RECONNECT_WAIT_SECONDS + ) waited = 0.0 while waited < self._RECONNECT_WAIT_SECONDS: await asyncio.sleep(self._RECONNECT_POLL_INTERVAL) @@ -1797,11 +1456,7 @@ class QQAdapter(BasePlatformAdapter): return self.is_connected or await self._wait_for_reconnection() async def send( - self, - chat_id: str, - content: str, - reply_to: Optional[str] = None, - metadata: Optional[Dict[str, Any]] = None, + self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Send text/markdown: format, split via truncate_message(), retry transient failures.""" del metadata @@ -1810,78 +1465,64 @@ class QQAdapter(BasePlatformAdapter): if not content or not content.strip(): return SendResult(success=True) - formatted = self.format_message(content) - chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH) - + chunks = self.truncate_message(self.format_message(content), self.MAX_MESSAGE_LENGTH) last_result = SendResult(success=False, error="No chunks") for chunk in chunks: last_result = await self._send_chunk(chat_id, chunk, reply_to) if not last_result.success: return last_result - # Only reply_to the first chunk - reply_to = None + reply_to = None # only reply_to the first chunk return last_result - async def _send_chunk( - self, - chat_id: str, - content: str, - reply_to: Optional[str] = None, - ) -> SendResult: + _PERMANENT_SEND_ERRORS = ("invalid", "forbidden", "not found") + + async def _send_chunk(self, chat_id: str, content: str, reply_to: Optional[str] = None) -> SendResult: """Send a single chunk with retry + exponential backoff.""" last_exc: Optional[Exception] = None chat_type = self._guess_chat_type(chat_id) - for attempt in range(3): try: if chat_type == "c2c": return await self._send_c2c_text(chat_id, content, reply_to) - elif chat_type == "group": + if chat_type == "group": return await self._send_group_text(chat_id, content, reply_to) - elif chat_type == "guild": + if chat_type == "guild": return await self._send_guild_text(chat_id, content, reply_to) - else: - return SendResult(success=False, error=f"Unknown chat type for {chat_id}") + return SendResult(success=False, error=f"Unknown chat type for {chat_id}") except Exception as exc: last_exc = exc err = str(exc).lower() - if any(k in err for k in ("invalid", "forbidden", "not found", "bad request")): + if any(k in err for k in self._PERMANENT_SEND_ERRORS + ("bad request",)): break # permanent — don't retry if attempt < 2: delay = 1.0 * (2 ** attempt) - logger.warning( - "[%s] send retry %d/3 after %.1fs: %s", - self._log_tag, attempt + 1, delay, exc, - ) + logger.warning("[%s] send retry %d/3 after %.1fs: %s", self._log_tag, attempt + 1, delay, exc) await asyncio.sleep(delay) error_msg = (str(last_exc) or type(last_exc).__name__) if last_exc else "Unknown error" logger.error("[%s] Send failed: %s", self._log_tag, error_msg) - retryable = not any( - k in error_msg.lower() for k in ("invalid", "forbidden", "not found") - ) + retryable = not any(k in error_msg.lower() for k in self._PERMANENT_SEND_ERRORS) return SendResult(success=False, error=error_msg, retryable=retryable) @staticmethod - def _messages_path(chat_type: str, target_id: str) -> str: - """REST path for outbound messages to a c2c user or a group.""" + def _rest_path(chat_type: str, target_id: str, endpoint: str) -> str: + """``/v2/{users|groups}/{id}/{endpoint}`` for a c2c user or a group.""" kind = "users" if chat_type == "c2c" else "groups" - return f"/v2/{kind}/{target_id}/messages" + return f"/v2/{kind}/{target_id}/{endpoint}" + + @classmethod + def _messages_path(cls, chat_type: str, target_id: str) -> str: + """REST path for outbound messages to a c2c user or a group.""" + return cls._rest_path(chat_type, target_id, "messages") async def _post_message(self, path: str, body: Dict[str, Any]) -> SendResult: """POST a message body and wrap the response as a successful SendResult.""" data = await self._api_request("POST", path, body) - return SendResult( - success=True, message_id=str(data.get("id", uuid.uuid4().hex[:12])), raw_response=data - ) + return SendResult(success=True, message_id=str(data.get("id", uuid.uuid4().hex[:12])), raw_response=data) async def _send_text_to( - self, - chat_type: str, - target_id: str, - content: str, - reply_to: Optional[str] = None, - keyboard: Optional[InlineKeyboard] = None, + self, chat_type: str, target_id: str, content: str, reply_to: Optional[str] = None, + keyboard: Optional[InlineKeyboard] = None, ) -> SendResult: """Send text (optionally with an inline keyboard) to a c2c user or group.""" self._next_msg_seq(reply_to or target_id) @@ -1898,25 +1539,17 @@ class QQAdapter(BasePlatformAdapter): async def _send_group_text(self, group_openid, content, reply_to=None, keyboard=None) -> SendResult: return await self._send_text_to("group", group_openid, content, reply_to, keyboard) - async def _send_guild_text( - self, channel_id: str, content: str, reply_to: Optional[str] = None - ) -> SendResult: + async def _send_guild_text(self, channel_id: str, content: str, reply_to: Optional[str] = None) -> SendResult: """Send text to a guild channel via REST API.""" body: Dict[str, Any] = {"content": content[: self.MAX_MESSAGE_LENGTH]} if reply_to: body["msg_id"] = reply_to return await self._post_message(f"/channels/{channel_id}/messages", body) - # ------------------------------------------------------------------ - # Inline-keyboard outbound helpers (approval / update-prompt flows) - # ------------------------------------------------------------------ + # ── Inline-keyboard outbound helpers (approval / update-prompt flows) ── async def send_with_keyboard( - self, - chat_id: str, - content: str, - keyboard: InlineKeyboard, - reply_to: Optional[str] = None, + self, chat_id: str, content: str, keyboard: InlineKeyboard, reply_to: Optional[str] = None, ) -> SendResult: """Send ONE text message with an inline keyboard (no chunking — splitting would orphan the buttons; keep bodies short). Guild chats are unsupported.""" @@ -1930,44 +1563,29 @@ class QQAdapter(BasePlatformAdapter): if chat_type == "group": return await self._send_group_text(chat_id, truncated, reply_to, keyboard=keyboard) return SendResult( - success=False, - error=f"Inline keyboards not supported for chat_type {chat_type!r}", - retryable=False, + success=False, error=f"Inline keyboards not supported for chat_type {chat_type!r}", retryable=False ) except Exception as exc: logger.error("[%s] send_with_keyboard failed: %s", self._log_tag, exc) return SendResult(success=False, error=str(exc) or type(exc).__name__) async def send_approval_request( - self, - chat_id: str, - req: ApprovalRequest, - reply_to: Optional[str] = None, + self, chat_id: str, req: ApprovalRequest, reply_to: Optional[str] = None ) -> SendResult: """Send a 3-button approval request (allow-once / allow-always / deny); clicks come back as INTERACTION_CREATE decoded by parse_approval_button_data.""" from gateway.platforms.qqbot.keyboards import build_approval_text - return await self.send_with_keyboard( - chat_id, - build_approval_text(req), - build_approval_keyboard( - req.session_key, allow_permanent=getattr(req, "allow_permanent", True) - ), - reply_to=reply_to, - ) + keyboard = build_approval_keyboard(req.session_key, allow_permanent=getattr(req, "allow_permanent", True)) + return await self.send_with_keyboard(chat_id, build_approval_text(req), keyboard, reply_to=reply_to) # Cross-adapter gateway contract: gateway/run.py detects send_exec_approval / # send_update_prompt on the adapter class for button-based approval/update UX. + _APPROVAL_TIMEOUT_SECONDS = 300 # matches gateway's default gateway_timeout + async def send_exec_approval( - self, - chat_id: str, - command: str, - session_key: str, - description: str = "dangerous command", - metadata: Optional[Dict[str, Any]] = None, - allow_permanent: bool = True, - allow_session: bool = True, + self, chat_id: str, command: str, session_key: str, description: str = "dangerous command", + metadata: Optional[Dict[str, Any]] = None, allow_permanent: bool = True, allow_session: bool = True, smart_denied: bool = False, ) -> SendResult: """Button-based exec-approval prompt (called by gateway/run.py while the @@ -1976,7 +1594,6 @@ class QQAdapter(BasePlatformAdapter): del allow_session # QQ's 3-button keyboard has no session tier. if smart_denied: description += " Owner override applies to this one operation only." - req = ApprovalRequest( session_key=session_key, title="Execute this command?", @@ -1988,29 +1605,20 @@ class QQAdapter(BasePlatformAdapter): # QQ requires a msg_id for passive replies; the last inbound id is the natural one. return await self.send_approval_request(chat_id, req, reply_to=self._last_msg_id.get(chat_id)) - _APPROVAL_TIMEOUT_SECONDS = 300 # matches gateway's default gateway_timeout - async def send_update_prompt( - self, - chat_id: str, - prompt: str, - default: str = "", - session_key: str = "", - metadata: Optional[Dict[str, Any]] = None, + self, chat_id: str, prompt: str, default: str = "", session_key: str = "", + metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Yes/No update-confirmation prompt; button clicks (``update_prompt:y|n``) are written to ``~/.hermes/.update_response`` by the interaction callback.""" del session_key, metadata # present for contract parity only. - default_hint = f" (default: {default})" if default else "" content = f"⚕ **Update Needs Your Input**\n\n{prompt}{default_hint}" return await self.send_with_keyboard( chat_id, content, build_update_prompt_keyboard(), reply_to=self._last_msg_id.get(chat_id) ) - def _build_text_body( - self, content: str, reply_to: Optional[str] = None - ) -> Dict[str, Any]: + def _build_text_body(self, content: str, reply_to: Optional[str] = None) -> Dict[str, Any]: """Build the message body for C2C/group text sending.""" msg_seq = self._next_msg_seq(reply_to or "default") text = content[: self.MAX_MESSAGE_LENGTH] @@ -2021,26 +1629,18 @@ class QQAdapter(BasePlatformAdapter): body["message_reference"] = {"message_id": reply_to} return body - # ------------------------------------------------------------------ - # Native media sending - # ------------------------------------------------------------------ + # ── Native media sending ── async def send_image( - self, - chat_id: str, - image_url: str, - caption: Optional[str] = None, - reply_to: Optional[str] = None, - metadata: Optional[Dict[str, Any]] = None, + self, chat_id: str, image_url: str, caption: Optional[str] = None, reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Send an image natively via QQ Bot API upload; URL sources fall back to text.""" del metadata result = await self._send_media(chat_id, image_url, MEDIA_TYPE_IMAGE, "image", caption, reply_to) if result.success or not self._is_url(image_url): return result - logger.warning( - "[%s] Image send failed, falling back to text: %s", self._log_tag, result.error - ) + logger.warning("[%s] Image send failed, falling back to text: %s", self._log_tag, result.error) fallback = f"{caption}\n{image_url}" if caption else image_url return await self.send(chat_id=chat_id, content=fallback, reply_to=reply_to) @@ -2057,7 +1657,7 @@ class QQAdapter(BasePlatformAdapter): return await self._send_media(chat_id, video_path, MEDIA_TYPE_VIDEO, "video", caption, reply_to) async def send_document( - self, chat_id, file_path, caption=None, file_name=None, reply_to=None, **kwargs + self, chat_id, file_path, caption=None, file_name=None, reply_to=None, **kwargs ) -> SendResult: """Send a file/document natively.""" return await self._send_media( @@ -2065,21 +1665,12 @@ class QQAdapter(BasePlatformAdapter): ) async def _send_media( - self, - chat_id: str, - media_source: str, - file_type: int, - kind: str, - caption: Optional[str] = None, - reply_to: Optional[str] = None, - file_name: Optional[str] = None, + self, chat_id: str, media_source: str, file_type: int, kind: str, caption: Optional[str] = None, + reply_to: Optional[str] = None, file_name: Optional[str] = None, ) -> SendResult: - """Upload media and send as a native message. - - HTTP(S) URLs → single ``POST .../files`` with ``url=`` (QQ fetches it). - Local files → chunked upload (prepare / PUT parts / complete), up to the - platform's ~100 MB per-file limit. - """ + """Upload media and send as a native message. HTTP(S) URLs → single ``POST + .../files`` with ``url=`` (QQ fetches it). Local files → chunked upload + (prepare / PUT parts / complete), up to the platform's ~100 MB per-file limit.""" if not await self._ensure_connected(): return self._NOT_CONNECTED chat_type = self._guess_chat_type(chat_id) @@ -2090,11 +1681,7 @@ class QQAdapter(BasePlatformAdapter): if self._is_url(media_source): resolved_name = file_name or Path(urlparse(media_source).path).name or "media" upload = await self._upload_media( - chat_type, - chat_id, - file_type, - url=media_source, - srv_send_msg=False, + chat_type, chat_id, file_type, url=media_source, srv_send_msg=False, file_name=resolved_name if file_type == MEDIA_TYPE_FILE else None, ) else: @@ -2119,16 +1706,11 @@ class QQAdapter(BasePlatformAdapter): except UploadDailyLimitExceededError as exc: # Non-retryable quota hit; give the model actionable text. logger.warning( - "[%s] Daily upload limit exceeded for %s (%s)", - self._log_tag, exc.file_name, exc.file_size_human, + "[%s] Daily upload limit exceeded for %s (%s)", self._log_tag, exc.file_name, exc.file_size_human ) return SendResult( - success=False, - error=( - f"QQ daily upload limit exceeded for {exc.file_name!r} " - f"({exc.file_size_human}). Retry tomorrow." - ), - retryable=False, + success=False, retryable=False, + error=f"QQ daily upload limit exceeded for {exc.file_name!r} ({exc.file_size_human}). Retry tomorrow.", ) except UploadFileTooLargeError as exc: logger.warning( @@ -2136,59 +1718,41 @@ class QQAdapter(BasePlatformAdapter): self._log_tag, exc.file_name, exc.file_size_human, exc.limit_human, ) return SendResult( - success=False, + success=False, retryable=False, error=( f"{exc.file_name!r} ({exc.file_size_human}) exceeds the " f"QQ per-file upload limit ({exc.limit_human})." ), - retryable=False, ) except Exception as exc: logger.error("[%s] Media send failed: %s", self._log_tag, exc) return SendResult(success=False, error=str(exc) or type(exc).__name__) async def _upload_local_file( - self, - chat_type: str, - chat_id: str, - media_source: str, - file_type: int, - file_name: Optional[str], + self, chat_type: str, chat_id: str, media_source: str, file_type: int, file_name: Optional[str], ) -> Tuple[str, Dict[str, Any]]: """Chunked-upload a local file; returns ``(resolved_name, complete_response)`` - whose ``file_info`` goes into the RichMedia body. - - Raises UploadDailyLimitExceededError / UploadFileTooLargeError from the - uploader, ValueError for placeholder paths like ````, FileNotFoundError. - """ - if not self._http_client: - raise RuntimeError("HTTP client not initialized — not connected?") - + whose ``file_info`` goes into the RichMedia body. Raises UploadDailyLimitExceededError / + UploadFileTooLargeError from the uploader, ValueError for placeholder paths like + ````, FileNotFoundError.""" + client = self._require_http_client() local_path = Path(media_source).expanduser() if not local_path.is_absolute(): local_path = (Path.cwd() / local_path).resolve() - if not local_path.exists() or not local_path.is_file(): if media_source.startswith("<") or len(media_source) < 3: raise ValueError(f"Invalid media source (looks like a placeholder): {media_source!r}") raise FileNotFoundError(f"Media file not found: {local_path}") resolved_name = file_name or local_path.name - uploader = ChunkedUploader( - api_request=self._api_request, http_put=self._http_client.put, log_tag=self._log_tag - ) + uploader = ChunkedUploader(api_request=self._api_request, http_put=client.put, log_tag=self._log_tag) complete = await uploader.upload( - chat_type=chat_type, - target_id=chat_id, - file_path=str(local_path), - file_type=file_type, + chat_type=chat_type, target_id=chat_id, file_path=str(local_path), file_type=file_type, file_name=resolved_name, ) return resolved_name, complete - # ------------------------------------------------------------------ - # Typing indicator - # ------------------------------------------------------------------ + # ── Typing indicator ── async def send_typing(self, chat_id: str, metadata=None) -> None: """C2C-only input notify, debounced to ~50s (API shows a 60s indicator); @@ -2201,7 +1765,6 @@ class QQAdapter(BasePlatformAdapter): now = time.time() if now - self._typing_sent_at.get(chat_id, 0.0) < self._TYPING_DEBOUNCE_SECONDS: return - try: body = { "msg_type": MSG_TYPE_INPUT_NOTIFY, @@ -2214,29 +1777,16 @@ class QQAdapter(BasePlatformAdapter): except Exception as exc: logger.debug("[%s] send_typing failed: %s", self._log_tag, exc) - # ------------------------------------------------------------------ - # Format - # ------------------------------------------------------------------ + # ── Format / chat info / helpers ── def format_message(self, content: str) -> str: """Pass markdown through when supported, else strip it (as BlueBubbles/SMS do).""" return content if self._markdown_support else strip_markdown(content) - # ------------------------------------------------------------------ - # Chat info - # ------------------------------------------------------------------ - async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: """Return chat info based on chat type heuristics.""" chat_type = self._guess_chat_type(chat_id) - return { - "name": chat_id, - "type": "group" if chat_type in {"group", "guild"} else "dm", - } - - # ------------------------------------------------------------------ - # Helpers - # ------------------------------------------------------------------ + return {"name": chat_id, "type": "group" if chat_type in {"group", "guild"} else "dm"} @staticmethod def _is_url(source: str) -> bool: @@ -2257,8 +1807,6 @@ class QQAdapter(BasePlatformAdapter): return _resolve_qq_secret("QQ_ALLOW_ALL_USERS", "").lower() in {"true", "1", "yes"} def _is_dm_allowed(self, user_id: str) -> bool: - if self._dm_policy == "disabled": - return False if self._dm_policy == "allowlist": return self._entry_matches(self._allow_from, user_id) if self._dm_policy == "open": @@ -2269,23 +1817,13 @@ class QQAdapter(BasePlatformAdapter): principal = str(user_id or "").strip() if not principal: return False - if self._dm_policy == "disabled": - return False - if self._dm_policy == "allowlist": - return self._entry_matches(self._allow_from, principal) if self._dm_policy == "pairing": return True - if self._dm_policy == "open": - return self._open_dm_opted_in() - return False + return self._is_dm_allowed(principal) def _is_group_allowed(self, group_id: str, user_id: str) -> bool: - if self._group_policy == "disabled": - return False if self._group_policy == "allowlist": return self._entry_matches(self._group_allow_from, group_id) - if self._group_policy == "pairing": - return False return self._group_policy == "open" @staticmethod diff --git a/gateway/platforms/qqbot/chunked_upload.py b/gateway/platforms/qqbot/chunked_upload.py index 32d62b517e..5a933c16af 100644 --- a/gateway/platforms/qqbot/chunked_upload.py +++ b/gateway/platforms/qqbot/chunked_upload.py @@ -1,21 +1,10 @@ -"""QQ Bot chunked upload flow. - -The QQ v2 API caps inline base64 uploads (``file_data`` / ``url``) at ~10 MB. -Files between 10 MB and ~100 MB use the three-step chunked flow:: - - 1. POST /v2/{users|groups}/{id}/upload_prepare - → upload_id, block_size, and pre-signed COS part URLs. - 2. Per part: PUT bytes to the COS URL, then POST .../upload_part_finish. - 3. POST /v2/{users|groups}/{id}/files with {"upload_id": ...} - → ``file_info`` token used in a RichMedia message. - -Error codes (QQ Bot v2 spec): ``40093001`` — ``upload_part_finish`` retryable -until the server's ``retry_timeout`` (or a local cap) elapses; ``40093002`` — -daily upload quota exceeded, surfaced as :class:`UploadDailyLimitExceededError`. -Other API/I/O failures raise ``RuntimeError``. - -Ported from WideLee's qqbot-agent-sdk v1.2.2 (``media_loader.py::ChunkedUploader``). -Authorship preserved via Co-authored-by. +"""QQ Bot chunked upload flow (files 10 MB..~100 MB; inline uploads cap at ~10 MB): +1. POST .../upload_prepare → upload_id, block_size, pre-signed COS part URLs; +2. per part: PUT bytes to COS, then POST .../upload_part_finish; +3. POST .../files with {"upload_id"} → ``file_info`` for a RichMedia message. +biz_code 40093001 = part_finish retryable until ``retry_timeout``; 40093002 = daily +quota (UploadDailyLimitExceededError); other API/I/O failures raise RuntimeError. +Ported from WideLee's qqbot-agent-sdk v1.2.2 (authorship via Co-authored-by). """ from __future__ import annotations @@ -34,60 +23,55 @@ logger = logging.getLogger(__name__) _BIZ_CODE_DAILY_LIMIT = 40093002 # upload_prepare: daily cumulative limit _BIZ_CODE_PART_RETRYABLE = 40093001 # upload_part_finish: transient - _DEFAULT_CONCURRENT_PARTS = 1 _MAX_CONCURRENT_PARTS = 10 - _PART_UPLOAD_TIMEOUT = 300.0 # 5 minutes per COS PUT _PART_UPLOAD_MAX_RETRIES = 2 _PART_FINISH_RETRY_INTERVAL = 1.0 _PART_FINISH_DEFAULT_TIMEOUT = 120.0 _PART_FINISH_MAX_TIMEOUT = 600.0 - _COMPLETE_UPLOAD_MAX_RETRIES = 2 _COMPLETE_UPLOAD_BASE_DELAY = 2.0 - -# First 10,002,432 bytes used for the ``md5_10m`` hash (per QQ API spec). -_MD5_10M_SIZE = 10_002_432 +_MD5_10M_SIZE = 10_002_432 # first N bytes used for the ``md5_10m`` hash (per QQ API spec) -# ── Exceptions ─────────────────────────────────────────────────────── +# ── Exceptions ── -class UploadDailyLimitExceededError(Exception): +class _UploadError(Exception): + def __init__(self, file_name: str, file_size: int, message: str) -> None: + self.file_name = file_name + self.file_size = file_size + super().__init__(message) + + @property + def file_size_human(self) -> str: + return format_size(self.file_size) + + +class UploadDailyLimitExceededError(_UploadError): """Raised when ``upload_prepare`` returns biz_code 40093002 (daily quota hit).""" def __init__(self, file_name: str, file_size: int, message: str = "") -> None: - self.file_name = file_name - self.file_size = file_size - super().__init__(message or f"Daily upload limit exceeded for {file_name!r}") - - @property - def file_size_human(self) -> str: - return format_size(self.file_size) + super().__init__(file_name, file_size, message or f"Daily upload limit exceeded for {file_name!r}") -class UploadFileTooLargeError(Exception): +class UploadFileTooLargeError(_UploadError): """Raised when a file exceeds the platform per-file size limit.""" def __init__(self, file_name: str, file_size: int, limit_bytes: int = 0, message: str = "") -> None: - self.file_name = file_name - self.file_size = file_size self.limit_bytes = limit_bytes limit_str = f" ({format_size(limit_bytes)})" if limit_bytes else "" super().__init__( - message or f"File {file_name!r} ({format_size(file_size)}) exceeds platform limit{limit_str}" + file_name, file_size, + message or f"File {file_name!r} ({format_size(file_size)}) exceeds platform limit{limit_str}", ) - @property - def file_size_human(self) -> str: - return format_size(self.file_size) - @property def limit_human(self) -> str: return format_size(self.limit_bytes) if self.limit_bytes else "unknown" -# ── Prepare-response shape ─────────────────────────────────────────── +# ── Prepare-response shape ── @dataclass class _PreparePart: @@ -138,47 +122,26 @@ def _api_path(chat_type: str, target_id: str, endpoint: str) -> str: return f"{base}/{target_id}/{endpoint}" -# ── Chunked upload driver ──────────────────────────────────────────── +# ── Chunked upload driver ── class ChunkedUploader: - """Run the prepare → PUT parts → complete sequence. - - :param api_request: Bound ``_api_request(method, path, body=..., timeout=...)`` - coroutine from the adapter (passed in rather than imported to avoid - circular imports). Must raise ``RuntimeError`` with the biz_code in - the message on API errors. - :param http_put: Coroutine ``(url, data, headers) -> httpx-like response`` - for COS part uploads. - :param log_tag: Log prefix. - """ + """Run the prepare → PUT parts → complete sequence. ``api_request`` is the adapter's + bound ``_api_request(method, path, body=..., timeout=...)`` (injected to avoid a + circular import; must raise RuntimeError with the biz_code in the message); + ``http_put`` is ``(url, data, headers) -> httpx-like response`` for COS PUTs.""" def __init__( - self, - api_request: Callable[..., Awaitable[Dict[str, Any]]], - http_put: Callable[..., Awaitable[Any]], + self, api_request: Callable[..., Awaitable[Dict[str, Any]]], http_put: Callable[..., Awaitable[Any]], log_tag: str = "QQBot", ) -> None: self._api_request = api_request self._http_put = http_put self._log_tag = log_tag - async def upload( - self, - chat_type: str, - target_id: str, - file_path: str, - file_type: int, - file_name: str, - ) -> Dict[str, Any]: - """Run the full chunked upload and return the ``complete_upload`` response. - - :param chat_type: ``'c2c'`` or ``'group'``. - :param file_type: ``MEDIA_TYPE_*`` constant. - :returns: Raw ``complete_upload`` response dict (contains ``file_info``). - :raises UploadDailyLimitExceededError: On biz_code 40093002. - :raises UploadFileTooLargeError: When the file exceeds the platform limit. - :raises RuntimeError: On other API or I/O failures. - """ + async def upload(self, chat_type: str, target_id: str, file_path: str, file_type: int, file_name: str) -> Dict[str, Any]: + """Run the full chunked upload (``chat_type`` 'c2c'|'group', ``file_type`` MEDIA_TYPE_*) + and return the raw ``complete_upload`` response (contains ``file_info``). + Raises UploadDailyLimitExceededError (40093002), UploadFileTooLargeError, RuntimeError.""" if chat_type not in {"c2c", "group"}: raise ValueError(f"ChunkedUploader: unsupported chat_type {chat_type!r}") @@ -189,9 +152,7 @@ class ChunkedUploader: ) # Hashing is blocking I/O → executor. - hashes = await asyncio.get_running_loop().run_in_executor( - None, _compute_file_hashes, file_path, file_size - ) + hashes = await asyncio.get_running_loop().run_in_executor(None, _compute_file_hashes, file_path, file_size) prepare = await self._prepare(chat_type, target_id, file_type, file_name, file_size, hashes) max_concurrent = min(prepare.concurrency, _MAX_CONCURRENT_PARTS) retry_timeout = min( @@ -220,26 +181,15 @@ class ChunkedUploader: return await self._complete(chat_type, target_id, prepare.upload_id) async def _prepare( - self, - chat_type: str, - target_id: str, - file_type: int, - file_name: str, - file_size: int, - hashes: Dict[str, str], + self, chat_type: str, target_id: str, file_type: int, file_name: str, file_size: int, hashes: Dict[str, str], ) -> _PrepareResult: body = { - "file_type": file_type, - "file_name": file_name, - "file_size": file_size, - "md5": hashes["md5"], - "sha1": hashes["sha1"], - "md5_10m": hashes["md5_10m"], + "file_type": file_type, "file_name": file_name, "file_size": file_size, + "md5": hashes["md5"], "sha1": hashes["sha1"], "md5_10m": hashes["md5_10m"], } try: raw = await self._api_request( - "POST", _api_path(chat_type, target_id, "upload_prepare"), - body=body, timeout=FILE_UPLOAD_TIMEOUT, + "POST", _api_path(chat_type, target_id, "upload_prepare"), body=body, timeout=FILE_UPLOAD_TIMEOUT, ) except RuntimeError as exc: err_msg = str(exc) @@ -249,17 +199,8 @@ class ChunkedUploader: return _parse_prepare_response(raw) async def _upload_one_part( - self, - chat_type: str, - target_id: str, - file_path: str, - file_size: int, - upload_id: str, - rsp_block_size: int, - part: _PreparePart, - retry_timeout: float, - total_parts: int, - completed: List[int], + self, chat_type: str, target_id: str, file_path: str, file_size: int, upload_id: str, rsp_block_size: int, + part: _PreparePart, retry_timeout: float, total_parts: int, completed: List[int], ) -> None: """PUT one part to COS, then call ``upload_part_finish``.""" part_index = part.index @@ -268,9 +209,7 @@ class ChunkedUploader: offset = (part_index - 1) * rsp_block_size length = min(actual_block_size, file_size - offset) - data = await asyncio.get_running_loop().run_in_executor( - None, _read_file_chunk, file_path, offset, length - ) + data = await asyncio.get_running_loop().run_in_executor(None, _read_file_chunk, file_path, offset, length) md5_hex = hashlib.md5(data).hexdigest() logger.debug( "[%s] Part %d/%d: uploading %s (offset=%d md5=%s)", @@ -289,12 +228,7 @@ class ChunkedUploader: ) async def _with_retries( - self, - attempt_fn: Callable[[], Awaitable[Any]], - *, - max_retries: int, - base_delay: float, - label: str, + self, attempt_fn: Callable[[], Awaitable[Any]], *, max_retries: int, base_delay: float, label: str, failure_label: str, ) -> Any: """Run *attempt_fn* up to ``max_retries + 1`` times with exponential backoff.""" @@ -325,11 +259,10 @@ class ChunkedUploader: if 200 <= status < 300: logger.debug("[%s] PUT part %d/%d: %d OK", self._log_tag, part_index, total_parts, status) return - body_preview = "" try: body_preview = getattr(resp, "text", "")[:200] except Exception: # pragma: no cover — defensive - pass + body_preview = "" raise RuntimeError(f"COS PUT returned {status}: {body_preview}") await self._with_retries( @@ -339,13 +272,7 @@ class ChunkedUploader: ) async def _part_finish_with_retry( - self, - chat_type: str, - target_id: str, - upload_id: str, - part_index: int, - block_size: int, - md5: str, + self, chat_type: str, target_id: str, upload_id: str, part_index: int, block_size: int, md5: str, retry_timeout: float, ) -> None: """Call ``upload_part_finish``, retrying on biz_code 40093001 until *retry_timeout*.""" @@ -375,11 +302,8 @@ class ChunkedUploader: await asyncio.sleep(_PART_FINISH_RETRY_INTERVAL) async def _complete(self, chat_type: str, target_id: str, upload_id: str) -> Dict[str, Any]: - """Call ``complete_upload`` with retry. - - Reuses the ``/files`` endpoint (same as the simple URL-based upload) but - signals the chunked-completion path by sending only ``upload_id``. - """ + """Call ``complete_upload`` with retry — the ``/files`` endpoint (same as the simple + URL upload) selects the chunked-completion path when only ``upload_id`` is sent.""" path = _api_path(chat_type, target_id, "files") body = {"upload_id": upload_id} return await self._with_retries( @@ -389,7 +313,7 @@ class ChunkedUploader: ) -# ── Helpers (module-level for testability) ─────────────────────────── +# ── Helpers (module-level for testability) ── def format_size(size_bytes: int) -> str: """Return a human-readable file size string (e.g. ``'12.3 MB'``).""" diff --git a/gateway/platforms/qqbot/constants.py b/gateway/platforms/qqbot/constants.py index 9ceb09c8e9..e4fec356d7 100644 --- a/gateway/platforms/qqbot/constants.py +++ b/gateway/platforms/qqbot/constants.py @@ -7,45 +7,36 @@ import os # Bump on functional changes to the adapter package. QQBOT_VERSION = "1.1.0" -# ── API endpoints ── # Portal domain is overridable (QQ_PORTAL_HOST) for corporate proxies / test environments. PORTAL_HOST = os.getenv("QQ_PORTAL_HOST", "q.qq.com") API_BASE = "https://api.sgroup.qq.com" TOKEN_URL = "https://bots.qq.com/app/getAppAccessToken" GATEWAY_URL_PATH = "/gateway" - -# QR-code onboard endpoints (on the portal host) ONBOARD_CREATE_PATH = "/lite/create_bind_task" ONBOARD_POLL_PATH = "/lite/poll_bind_result" QR_URL_TEMPLATE = "https://q.qq.com/qqbot/openclaw/connect.html?task_id={task_id}&_wv=2&source=hermes" -# ── Timeouts & retry ── DEFAULT_API_TIMEOUT = 30.0 FILE_UPLOAD_TIMEOUT = 120.0 CONNECT_TIMEOUT_SECONDS = 20.0 - RECONNECT_BACKOFF = [2, 5, 10, 30, 60] MAX_RECONNECT_ATTEMPTS = 100 RATE_LIMIT_DELAY = 60 # seconds QUICK_DISCONNECT_THRESHOLD = 5.0 # seconds MAX_QUICK_DISCONNECT_COUNT = 3 - ONBOARD_POLL_INTERVAL = 2.0 # seconds between poll_bind_result calls ONBOARD_API_TIMEOUT = 10.0 -# ── Message limits ── MAX_MESSAGE_LENGTH = 4000 DEDUP_WINDOW_SECONDS = 300 DEDUP_MAX_SIZE = 1000 -# ── QQ Bot message types ── +# QQ Bot message types / file media types MSG_TYPE_TEXT = 0 MSG_TYPE_MARKDOWN = 2 MSG_TYPE_MEDIA = 7 MSG_TYPE_INPUT_NOTIFY = 6 - -# ── QQ Bot file media types ── MEDIA_TYPE_IMAGE = 1 MEDIA_TYPE_VIDEO = 2 MEDIA_TYPE_VOICE = 3 diff --git a/gateway/platforms/qqbot/crypto.py b/gateway/platforms/qqbot/crypto.py index 8eefc8ce1b..af3b5ff854 100644 --- a/gateway/platforms/qqbot/crypto.py +++ b/gateway/platforms/qqbot/crypto.py @@ -7,12 +7,8 @@ import os def generate_bind_key() -> str: - """Generate a random 256-bit AES key as base64. - - Passed to ``create_bind_task`` so the server encrypts the bot's - *client_secret*; only this CLI holds the key, so the secret never travels - in plaintext. - """ + """Random 256-bit AES key (base64) passed to ``create_bind_task`` so the server + encrypts the bot's *client_secret*; only this CLI holds the key.""" return base64.b64encode(os.urandom(32)).decode() diff --git a/gateway/platforms/qqbot/keyboards.py b/gateway/platforms/qqbot/keyboards.py index 4e8f65f28c..5ba995f377 100644 --- a/gateway/platforms/qqbot/keyboards.py +++ b/gateway/platforms/qqbot/keyboards.py @@ -1,17 +1,10 @@ """QQ Bot inline keyboards + approval / update-prompt helpers. -QQ Bot v2 attaches inline keyboards to outbound messages. A button click -dispatches an ``INTERACTION_CREATE`` gateway event carrying the button's -``data`` payload; the bot must ACK promptly via ``PUT /interactions/{id}`` or -the user sees an error indicator on the button. - -``button_data`` formats:: - - approve:: # decision = allow-once|allow-always|deny - update_prompt: # answer = y|n - -Ported from WideLee's qqbot-agent-sdk v1.2.2 (``approval.py`` + ``dto.py`` -keyboard types). Authorship preserved via Co-authored-by. +A button click dispatches an ``INTERACTION_CREATE`` event carrying the button's +``data``; the bot must ACK promptly via ``PUT /interactions/{id}`` or the user +sees an error indicator. ``button_data`` formats: ``approve::`` +(decision = allow-once|allow-always|deny) and ``update_prompt:`` (y|n). +Ported from WideLee's qqbot-agent-sdk v1.2.2 (authorship via Co-authored-by). """ from __future__ import annotations @@ -29,7 +22,7 @@ _APPROVAL_DATA_RE = re.compile(r"^approve:(.+):(allow-once|allow-always|deny)$") _UPDATE_PROMPT_RE = re.compile(r"^update_prompt:(y|n)$") -# ── Keyboard dataclasses ───────────────────────────────────────────── +# ── Keyboard dataclasses ── def _to_dict(value: Any) -> Any: """Serialize a dataclass tree in field-declaration order (the wire shape).""" @@ -53,9 +46,8 @@ class KeyboardButtonPermission(_Serializable): @dataclass class KeyboardButtonAction(_Serializable): - """Click behaviour: ``type`` 1 = Callback (INTERACTION_CREATE with ``data``), 2 = Link. - ``click_limit=1`` = single-use. - """ + """Click behaviour: ``type`` 1 = Callback (INTERACTION_CREATE with ``data``), 2 = Link; + ``click_limit=1`` = single-use.""" type: int data: str permission: KeyboardButtonPermission = field(default_factory=KeyboardButtonPermission) @@ -95,7 +87,7 @@ class InlineKeyboard(_Serializable): content: KeyboardContent = field(default_factory=KeyboardContent) -# ── INTERACTION_CREATE parsing ─────────────────────────────────────── +# ── INTERACTION_CREATE parsing ── def parse_approval_button_data(button_data: str) -> Optional[tuple[str, str]]: """Parse approval ``button_data`` into ``(session_key, decision)`` or ``None``.""" @@ -109,11 +101,9 @@ def parse_update_prompt_button_data(button_data: str) -> Optional[str]: return m.group(1) if m else None -# ── Keyboard builders ──────────────────────────────────────────────── +# ── Keyboard builders ── -def _make_callback_button( - btn_id: str, label: str, visited_label: str, data: str, style: int, group_id: str, -) -> KeyboardButton: +def _make_callback_button(btn_id: str, label: str, visited_label: str, data: str, style: int, group_id: str) -> KeyboardButton: return KeyboardButton( id=btn_id, render_data=KeyboardButtonRenderData(label=label, visited_label=visited_label, style=style), @@ -128,10 +118,8 @@ def _single_row_keyboard(buttons: List[KeyboardButton]) -> InlineKeyboard: def build_approval_keyboard(session_key: str, *, allow_permanent: bool = True) -> InlineKeyboard: """Build ``[✅ 允许一次] [⭐ 始终允许] [❌ 拒绝]`` (one group, so a click greys the rest). - - The ⭐ button is hidden when persistent scope is unavailable. *session_key* - is embedded in ``button_data`` so the decision routes to the right approval. - """ + ⭐ is hidden when persistent scope is unavailable; *session_key* is embedded in + ``button_data`` so the decision routes to the right approval.""" prefix = f"{APPROVAL_BUTTON_PREFIX}{session_key}" buttons = [_make_callback_button("allow", "✅ 允许一次", "已允许", f"{prefix}:allow-once", 1, "approval")] if allow_permanent: @@ -148,15 +136,12 @@ def build_update_prompt_keyboard() -> InlineKeyboard: ]) -# ── ApprovalRequest + text builder ─────────────────────────────────── +# ── ApprovalRequest + text builder ── @dataclass class ApprovalRequest: - """Approval-request display data. - - ``command_preview`` / ``cwd`` are set for exec approvals, ``tool_name`` for - plugin approvals; ``severity`` is ``'critical' | 'info' | ''``. - """ + """Approval-request display data. ``command_preview`` / ``cwd`` are set for exec + approvals, ``tool_name`` for plugin approvals; ``severity`` is ``'critical' | 'info' | ''``.""" session_key: str title: str description: str = "" @@ -193,14 +178,11 @@ def build_approval_text(req: ApprovalRequest) -> str: return "\n".join(lines) -# ── INTERACTION_CREATE event shape ─────────────────────────────────── +# ── INTERACTION_CREATE event shape ── @dataclass class InteractionEvent: - """Parsed ``INTERACTION_CREATE`` payload. - - See https://bot.q.qq.com/wiki/develop/api-v2/dev-prepare/interface-framework/event-emit.html - """ + """Parsed ``INTERACTION_CREATE`` payload (api-v2 event-emit docs).""" id: str = "" # required for the ``PUT /interactions/{id}`` ACK type: int = 0 # event type code (11 = message button) chat_type: int = 0 # 0 = guild, 1 = group, 2 = c2c diff --git a/gateway/platforms/qqbot/onboard.py b/gateway/platforms/qqbot/onboard.py index 7fc1041079..7a1754892a 100644 --- a/gateway/platforms/qqbot/onboard.py +++ b/gateway/platforms/qqbot/onboard.py @@ -1,11 +1,6 @@ -"""QQBot scan-to-configure (QR code onboard) module. - -Mirrors the Feishu onboarding pattern: synchronous HTTP + one public entry-point -``qr_register()`` (create task → display QR → poll → decrypt credentials) against -the ``q.qq.com`` ``create_bind_task`` / ``poll_bind_result`` APIs. - -Reference: https://bot.q.qq.com/wiki/develop/api-v2/ -""" +"""QQBot scan-to-configure (QR code onboard). Mirrors the Feishu pattern: synchronous +HTTP + one entry-point ``qr_register()`` (create task → display QR → poll → decrypt +credentials) against ``q.qq.com`` ``create_bind_task`` / ``poll_bind_result``.""" from __future__ import annotations @@ -16,12 +11,7 @@ from typing import Optional, Tuple from urllib.parse import quote from .constants import ( - ONBOARD_API_TIMEOUT, - ONBOARD_CREATE_PATH, - ONBOARD_POLL_INTERVAL, - ONBOARD_POLL_PATH, - PORTAL_HOST, - QR_URL_TEMPLATE, + ONBOARD_API_TIMEOUT, ONBOARD_CREATE_PATH, ONBOARD_POLL_INTERVAL, ONBOARD_POLL_PATH, PORTAL_HOST, QR_URL_TEMPLATE, ) from .crypto import decrypt_secret, generate_bind_key from .utils import get_api_headers @@ -31,7 +21,6 @@ logger = logging.getLogger(__name__) class BindStatus(IntEnum): """Status codes returned by ``_poll_bind_result``.""" - NONE = 0 PENDING = 1 COMPLETED = 2 @@ -103,13 +92,8 @@ _MAX_REFRESHES = 3 def qr_register(timeout_seconds: int = 600) -> Optional[dict]: - """Run the QQBot scan-to-configure QR registration flow. - - Unexpected errors propagate to the caller. - - :returns: ``{"app_id", "client_secret", "user_openid"}`` on success, or - ``None`` on failure / expiry / cancellation. - """ + """Run the QR registration flow; returns ``{"app_id", "client_secret", "user_openid"}`` + or None on failure / expiry / cancellation. Unexpected errors propagate.""" deadline = time.monotonic() + timeout_seconds for refresh_count in range(_MAX_REFRESHES + 1): diff --git a/gateway/platforms/qqbot/utils.py b/gateway/platforms/qqbot/utils.py index 156fbc68cc..a548d3f966 100644 --- a/gateway/platforms/qqbot/utils.py +++ b/gateway/platforms/qqbot/utils.py @@ -28,16 +28,9 @@ def build_user_agent() -> str: def get_api_headers() -> Dict[str, str]: - """Standard QQBot API headers. - - ``q.qq.com`` requires ``Accept: application/json`` — without it the server - returns a JavaScript anti-bot challenge page. - """ - return { - "Content-Type": "application/json", - "Accept": "application/json", - "User-Agent": build_user_agent(), - } + """Standard QQBot API headers. ``q.qq.com`` requires ``Accept: application/json`` + — without it the server returns a JavaScript anti-bot challenge page.""" + return {"Content-Type": "application/json", "Accept": "application/json", "User-Agent": build_user_agent()} def coerce_list(value: Any) -> List[str]: