refactor(gateway/platforms): compact qqbot adapter — fold reconnect/ready/ws-close helpers, dedupe token/gateway fetch + http-client guard, collapse dm/group ACL ladders, magic-byte table, tighter layouts

2026-09-02 18:47:51 -07:00
parent a166aa73bb
commit 7b6ec94ec1
8 changed files with 380 additions and 994 deletions
--- 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,
 )

--- a/gateway/platforms/qqbot/adapter.py
+++ b/gateway/platforms/qqbot/adapter.py
--- 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'``)."""
--- 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
--- 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()


--- 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:<session_key>:<decision>      # decision = allow-once|allow-always|deny
-    update_prompt:<answer>                # 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:<session_key>:<decision>``
+(decision = allow-once|allow-always|deny) and ``update_prompt:<answer>`` (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
--- 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):
--- 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]: