diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py index 5b4a396ed2fd..5f4b58fd0c3e 100644 --- a/gateway/platforms/qqbot/adapter.py +++ b/gateway/platforms/qqbot/adapter.py @@ -41,7 +41,7 @@ import uuid from datetime import datetime, timezone from pathlib import Path -from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple +from typing import Any, ClassVar, Dict, List, Optional, Tuple from urllib.parse import urlparse try: @@ -119,21 +119,6 @@ def __init__(self, code, reason=""): coerce_list as _coerce_list_impl, build_user_agent, ) -from gateway.platforms.qqbot.chunked_upload import ( - 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, - parse_update_prompt_button_data, -) def check_qq_requirements() -> bool: @@ -175,27 +160,19 @@ def _fail_pending(self, reason: str) -> None: fut.set_exception(RuntimeError(reason)) self._pending_responses.clear() - def _mark_transport_disconnected(self) -> None: - """Mark QQ WS down without stopping the reconnect loop. + # -- Active instance registry (class-level singleton) ------------------- - BasePlatformAdapter uses _running for both process lifecycle and - connection status. QQBot needs to keep the listener task alive across - transient transport drops so it can continue reconnect attempts after a - short-lived gateway or network failure. - """ - if self.has_fatal_error: - return - self._write_runtime_status_safe( - "disconnected", - platform_state="disconnected", - error_code=None, - error_message=None, - ) + _active_instance: ClassVar[Optional["QQAdapter"]] = None - @property - def is_connected(self) -> bool: - """Return True only when the QQ WebSocket transport is usable.""" - return bool(self._running and self._ws and not self._ws.closed) + @classmethod + def get_active(cls) -> Optional["QQAdapter"]: + """Return the currently connected QQAdapter, or None.""" + return cls._active_instance + + @classmethod + def set_active(cls, adapter: Optional["QQAdapter"]) -> None: + """Register (or clear) the active adapter instance.""" + cls._active_instance = adapter def __init__(self, config: PlatformConfig): super().__init__(config, Platform.QQBOT) @@ -240,27 +217,11 @@ def __init__(self, config: PlatformConfig): # Token cache self._access_token: Optional[str] = None self._token_expires_at: float = 0.0 - self._token_lock = asyncio.Lock() + self._token_lock: Optional[asyncio.Lock] = None # created in connect() # Upload cache: content_hash -> {file_info, file_uuid, expires_at} self._upload_cache: Dict[str, Dict[str, Any]] = {} - # Inline-keyboard interaction routing. The callback (if set) is invoked - # for every INTERACTION_CREATE event after the adapter has already - # ACKed it. Callers (gateway wiring for approvals / update prompts) - # register via set_interaction_callback(). - self._interaction_callback: Optional[ - Callable[[InteractionEvent], Awaitable[None]] - ] = None - - # Default interaction dispatcher: routes approval-button clicks to - # tools.approval.resolve_gateway_approval() and update-prompt clicks - # to ~/.hermes/.update_response. Set here so the cross-adapter gateway - # contract (send_exec_approval / send_update_prompt) works out of the - # box; callers can override with set_interaction_callback(None) or - # register a custom handler. - self._interaction_callback = self._default_interaction_dispatch - # ------------------------------------------------------------------ # Properties # ------------------------------------------------------------------ @@ -269,11 +230,6 @@ def __init__(self, config: PlatformConfig): def name(self) -> str: return "QQBot" - @property - def enforces_own_access_policy(self) -> bool: - """QQBot gates DM/group access at intake via dm_policy/group_policy.""" - return True - # ------------------------------------------------------------------ # Connection lifecycle # ------------------------------------------------------------------ @@ -311,7 +267,8 @@ async def connect(self) -> bool: limits=platform_httpx_limits(), ) - # 1. Get access token + # 1. Get access token (create lock in the running event loop) + self._token_lock = asyncio.Lock() await self._ensure_token() # 2. Get WebSocket gateway URL @@ -325,6 +282,7 @@ async def connect(self) -> bool: self._listen_task = asyncio.create_task(self._listen_loop()) self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) self._mark_connected() + QQAdapter.set_active(self) logger.info("[%s] Connected", self._log_tag) return True except Exception as exc: @@ -339,6 +297,8 @@ async def disconnect(self) -> None: """Close all connections and stop listeners.""" self._running = False self._mark_disconnected() + if QQAdapter._active_instance is self: + QQAdapter.set_active(None) if self._listen_task: self._listen_task.cancel() @@ -455,24 +415,13 @@ async def _open_ws(self, gateway_url: str) -> None: await self._session.close() self._session = None - # Honor WSL proxy env for QQ WebSocket. Hermes upgrades overwrite this - # local patch, so QQ can regress to direct-connect timeouts after update. - self._session = aiohttp.ClientSession(trust_env=True) - 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") - ) + self._session = aiohttp.ClientSession() self._ws = await self._session.ws_connect( 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) @@ -535,33 +484,12 @@ async def _listen_loop(self) -> None: else: quick_disconnect_count = 0 - self._mark_transport_disconnected() + self._mark_disconnected() self._fail_pending("Connection closed") - # Stop reconnecting for fatal codes (unrecoverable errors) - if code in { - 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 - }: - fatal_descriptions = { - 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", - } - desc = fatal_descriptions.get(code, f"fatal error (code={code})") + # Stop reconnecting for fatal codes + if code in (4914, 4915): + desc = "offline/sandbox-only" if code == 4914 else "banned" logger.error( "[%s] Bot is %s. Check QQ Open Platform.", self._log_tag, desc ) @@ -578,7 +506,6 @@ async def _listen_loop(self) -> None: 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): @@ -598,11 +525,10 @@ async def _listen_loop(self) -> None: self._token_expires_at = 0.0 # Session invalid → clear session, will re-identify on next Hello - # Note: 4009 (connection timeout) is NOT included here — it is - # resumable per the QQ protocol and should preserve session state. - if code in { + if code in ( 4006, 4007, + 4009, 4900, 4901, 4902, @@ -617,7 +543,7 @@ async def _listen_loop(self) -> None: 4911, 4912, 4913, - }: + ): logger.info( "[%s] Session error (%d), clearing session for re-identify", self._log_tag, @@ -633,19 +559,17 @@ async def _listen_loop(self) -> None: backoff_idx += 1 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: return logger.warning("[%s] WebSocket error: %s", self._log_tag, exc) - self._mark_transport_disconnected() + self._mark_disconnected() self._fail_pending("Connection interrupted") if backoff_idx >= MAX_RECONNECT_ATTEMPTS: logger.error("[%s] Max reconnect attempts reached", self._log_tag) - self._mark_disconnected() return if await self._reconnect(backoff_idx): @@ -688,12 +612,12 @@ async def _read_events(self) -> None: payload = self._parse_json(msg.data) if payload: self._dispatch_payload(payload) - elif msg.type in {aiohttp.WSMsgType.PING,}: + elif msg.type in (aiohttp.WSMsgType.PING,): # aiohttp auto-replies with PONG pass elif msg.type == aiohttp.WSMsgType.CLOSE: raise QQCloseError(msg.data, msg.extra) - elif msg.type in {aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR}: + elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): raise RuntimeError("WebSocket closed") async def _heartbeat_loop(self) -> None: @@ -731,8 +655,9 @@ async def _send_identify(self) -> None: "token": f"QQBot {token}", "intents": (1 << 25) | (1 << 30) - | (1 << 12) - | (1 << 26), # C2C_GROUP_AT_MESSAGES + PUBLIC_GUILD_MESSAGES + DIRECT_MESSAGE + INTERACTION + | ( + 1 << 12 + ), # C2C_GROUP_AT_MESSAGES + PUBLIC_GUILD_MESSAGES + DIRECT_MESSAGE "shard": [0, 1], "properties": { "$os": "macOS", @@ -833,16 +758,14 @@ def _dispatch_payload(self, payload: Dict[str, Any]) -> None: self._handle_ready(d) elif t == "RESUMED": logger.info("[%s] Session resumed", self._log_tag) - elif t in { + elif t in ( "C2C_MESSAGE_CREATE", "GROUP_AT_MESSAGE_CREATE", "DIRECT_MESSAGE_CREATE", "GUILD_MESSAGE_CREATE", "GUILD_AT_MESSAGE_CREATE", - }: + ): asyncio.create_task(self._on_message(t, d)) - elif t == "INTERACTION_CREATE": - self._create_task(self._on_interaction(d)) else: logger.debug("[%s] Unhandled dispatch: %s", self._log_tag, t) return @@ -851,32 +774,6 @@ def _dispatch_payload(self, payload: Dict[str, Any]) -> None: if op == 11: return - # op 7 = Server Reconnect — server asks client to reconnect (e.g. - # load-balancing, maintenance). Close the WS so _read_events raises - # and the outer loop triggers a reconnect with Resume. - if op == 7: - 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 - - # op 9 = Invalid Session — d=True means session is resumable, - # d=False means we must re-identify from scratch. - if op == 9: - resumable = bool(d) if d is not None else False - if not resumable: - logger.info( - "[%s] Invalid session (op 9, not resumable), clearing session", - self._log_tag, - ) - self._session_id = None - self._last_seq = None - else: - logger.info("[%s] Invalid session (op 9, resumable)", self._log_tag) - if self._ws and not self._ws.closed: - self._create_task(self._ws.close()) - return - logger.debug("[%s] Unknown op: %s", self._log_tag, op) def _handle_ready(self, d: Any) -> None: @@ -935,267 +832,13 @@ async def _on_message(self, event_type: str, d: Any) -> None: # Route by event type if event_type == "C2C_MESSAGE_CREATE": await self._handle_c2c_message(d, msg_id, content, author, timestamp) - elif event_type in {"GROUP_AT_MESSAGE_CREATE",}: + elif event_type in ("GROUP_AT_MESSAGE_CREATE",): await self._handle_group_message(d, msg_id, content, author, timestamp) - elif event_type in {"GUILD_MESSAGE_CREATE", "GUILD_AT_MESSAGE_CREATE"}: + elif event_type in ("GUILD_MESSAGE_CREATE", "GUILD_AT_MESSAGE_CREATE"): await self._handle_guild_message(d, msg_id, content, author, timestamp) elif event_type == "DIRECT_MESSAGE_CREATE": await self._handle_dm_message(d, msg_id, content, author, timestamp) - # ------------------------------------------------------------------ - # Inline-keyboard interactions (INTERACTION_CREATE) - # ------------------------------------------------------------------ - - def set_interaction_callback( - self, - callback: Optional[Callable[[InteractionEvent], Awaitable[None]]], - ) -> None: - """Register (or clear) the interaction callback. - - Invoked once per ``INTERACTION_CREATE`` event *after* the adapter has - ACKed the interaction. The callback is responsible for routing the - button click to the right subsystem (approval resolver, update-prompt - resolver, etc.) based on the ``button_data`` payload. - """ - self._interaction_callback = callback - - async def _on_interaction(self, d: Any) -> None: - """Handle an ``INTERACTION_CREATE`` event. - - Responsibilities: - - 1. Parse the raw payload into an :class:`InteractionEvent`. - 2. ACK the interaction (``PUT /interactions/{id}``) so the client - stops showing a loading indicator on the button. - 3. Dispatch to the registered interaction callback, if any. - """ - if not isinstance(d, dict): - return - try: - event = parse_interaction_event(d) - except Exception as exc: - logger.warning( - "[%s] Failed to parse INTERACTION_CREATE: %s", self._log_tag, exc - ) - return - - if not event.id: - logger.warning( - "[%s] INTERACTION_CREATE missing id, skipping ACK", self._log_tag - ) - return - - # ACK the interaction promptly — per the QQ docs the client will show - # an error icon on the button if we don't respond quickly. - try: - await self._acknowledge_interaction(event.id) - except Exception as exc: - logger.warning( - "[%s] Failed to ACK interaction %s: %s", - self._log_tag, event.id, exc, - ) - - logger.info( - "[%s] Interaction: scene=%s button_data=%r operator=%s", - self._log_tag, event.scene, event.button_data, event.operator_openid, - ) - - 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, - ) - return - try: - await callback(event) - except Exception as exc: - logger.error( - "[%s] Interaction callback raised: %s", - self._log_tag, exc, exc_info=True, - ) - - async def _acknowledge_interaction( - self, - interaction_id: str, - code: int = 0, - ) -> None: - """ACK a button interaction via ``PUT /interactions/{id}``. - - :param interaction_id: The ``id`` field from the - ``INTERACTION_CREATE`` event. - :param code: Response code (``0`` = success). - """ - if not self._http_client: - raise RuntimeError("HTTP client not initialized — not connected?") - token = await self._ensure_token() - headers = { - "Authorization": f"QQBot {token}", - "Content-Type": "application/json", - "User-Agent": build_user_agent(), - } - resp = await self._http_client.put( - f"{API_BASE}/interactions/{interaction_id}", - headers=headers, - json={"code": code}, - timeout=DEFAULT_API_TIMEOUT, - ) - if resp.status_code >= 400: - raise RuntimeError( - f"Interaction ACK failed [{resp.status_code}]: " - f"{resp.text[:200]}" - ) - - # Mapping from QQ keyboard button decisions → the ``choice`` vocabulary - # accepted by ``tools.approval.resolve_gateway_approval``. QQ's 3-button - # layout (mobile-space constraint) collapses "session" and "always" into - # a single "always" button; users wanting session-only approval can fall - # back to the ``/approve session`` text command. - _APPROVAL_BUTTON_TO_CHOICE = { - "allow-once": "once", - "allow-always": "always", - "deny": "deny", - } - - @staticmethod - def _parse_gateway_session_key(session_key: str) -> Optional[Dict[str, str]]: - """Parse ``agent:main:::[:]``.""" - parts = str(session_key or "").split(":") - if len(parts) < 5 or parts[0] != "agent" or parts[1] != "main": - return None - parsed = { - "platform": parts[2], - "chat_type": parts[3], - "chat_id": parts[4], - } - if len(parts) > 5: - parsed["user_id"] = parts[5] - return parsed - - 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() - if not parsed or parsed.get("platform") != "qqbot" or not operator: - return False - - chat_type = parsed.get("chat_type", "") - 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: - """Route ``INTERACTION_CREATE`` button clicks to the right subsystem. - - - ``approve::`` → - :func:`tools.approval.resolve_gateway_approval` - (unblocks the agent thread waiting on a dangerous-command approval). - - ``update_prompt:`` → - writes the answer to ``~/.hermes/.update_response`` for the - detached ``hermes update --gateway`` process to consume. - - Anything else is logged at DEBUG and ignored. - - Installed as the adapter's default interaction callback in - ``__init__``. Callers can replace via - :meth:`set_interaction_callback` to route clicks elsewhere (or pass - ``None`` to drop them entirely). - """ - button_data = event.button_data - if not button_data: - return - - approval = parse_approval_button_data(button_data) - if approval is not None: - 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, - ) - return - if not self._is_authorized_interaction_for_session(event, session_key): - logger.warning( - "[%s] Rejected unauthorized approval click for session %s " - "(operator=%s)", - self._log_tag, session_key, event.operator_openid, - ) - return - try: - # Import lazily to keep the adapter importable in tests that - # don't exercise the approval subsystem. - from tools.approval import resolve_gateway_approval - count = resolve_gateway_approval(session_key, choice) - logger.info( - "[%s] Button resolved %d approval(s) for session %s " - "(choice=%s, operator=%s)", - 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, - ) - 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): - logger.warning( - "[%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, - ) - - @staticmethod - def _write_update_response(answer: str, operator: str = "") -> None: - """Atomically write the update-prompt answer to ``.update_response``. - - Mirrors the Discord / Telegram / Feishu adapters: the detached - ``hermes update --gateway`` watcher polls this file for a ``y``/``n`` - response to its interactive prompts (stash-restore, config migration). - Writes via ``tmp + rename`` so a partial write can't fool the reader. - """ - try: - from hermes_constants import get_hermes_home - home = get_hermes_home() - response_path = home / ".update_response" - tmp = response_path.with_suffix(".tmp") - tmp.write_text(answer) - tmp.replace(response_path) - logger.info( - "QQ update prompt answered %r by %s", - answer, operator or "(unknown)", - ) - except Exception as exc: - logger.error("Failed to write update response: %s", exc) - async def _handle_c2c_message( self, d: Dict[str, Any], @@ -1264,13 +907,6 @@ async def _handle_c2c_message( len(voice_transcripts), ) - # Merge any quoted-message context (message_type=103 → msg_elements[0]). - quoted = await self._process_quoted_context(d) - text = self._merge_quote_into(text, quoted["quote_block"]) - 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 @@ -1329,13 +965,6 @@ async def _handle_group_message( else attachment_info ) - # Merge any quoted-message context (message_type=103 → msg_elements[0]). - quoted = await self._process_quoted_context(d) - text = self._merge_quote_into(text, quoted["quote_block"]) - 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 @@ -1403,13 +1032,6 @@ async def _handle_guild_message( else attachment_info ) - # Merge any quoted-message context (message_type=103 → msg_elements[0]). - quoted = await self._process_quoted_context(d) - text = self._merge_quote_into(text, quoted["quote_block"]) - 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 @@ -1474,13 +1096,6 @@ async def _handle_dm_message( else attachment_info ) - # Merge any quoted-message context (message_type=103 → msg_elements[0]). - quoted = await self._process_quoted_context(d) - text = self._merge_quote_into(text, quoted["quote_block"]) - 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 @@ -1501,113 +1116,6 @@ async def _handle_dm_message( ) await self.handle_message(event) - # ------------------------------------------------------------------ - # 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. - - When a user replies while quoting another message, the platform sets - ``message_type = 103`` and pushes the referenced message's content and - attachments inside ``msg_elements[0]``. The old adapter ignored - ``msg_elements`` entirely, so: - - - Quoted text was surfaced only when the user typed something of - their own — bare quote-replies showed nothing. - - Quoted attachments (images, voice, files) were never downloaded - or described. - - Quoted voice messages specifically produced no transcript, so the - LLM had no way to see what the user was referring to. - - This method parses ``msg_elements`` and runs the quoted attachments - through the same :meth:`_process_attachments` pipeline as the main - message body, so quoted voice messages get STT transcripts and - quoted images are cached identically. - - :param d: Raw inbound message dict (from the WS dispatch payload). - :returns: Dict with keys: - - - ``quote_block``: string to prepend to the user's text body - (empty when there's nothing quoted). - - ``image_urls``: list of cached quoted-image paths. - - ``image_media_types``: parallel list of image MIME types. - """ - empty = { - "quote_block": "", - "image_urls": [], - "image_media_types": [], - } - # Short-circuit: only message_type 103 indicates a quote. - try: - if int(d.get("message_type", 0) or 0) != 103: - return empty - except (TypeError, ValueError): - return empty - - elements = d.get("msg_elements") - if not isinstance(elements, list) or not elements: - return empty - - # msg_elements[0] carries the referenced message. Additional elements - # (if any) are very rare in practice; we concatenate their text and - # union their attachments for completeness. - quoted_text_parts: List[str] = [] - all_attachments: List[Dict[str, Any]] = [] - for elem in elements: - if not isinstance(elem, dict): - continue - etext = str(elem.get("content", "")).strip() - if etext: - 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) - - att_result = await self._process_attachments(all_attachments) - quoted_voice = att_result.get("voice_transcripts") or [] - quoted_info = att_result.get("attachment_info") or "" - quoted_images = att_result.get("image_urls") or [] - quoted_image_types = att_result.get("image_media_types") or [] - - lines: List[str] = [] - if quoted_text_parts: - lines.append(" ".join(quoted_text_parts)) - for t in quoted_voice: - lines.append(t) - if quoted_info: - lines.append(quoted_info) - - if not lines and not quoted_images: - return empty - - if lines: - quote_block = "[Quoted message]:\n" + "\n".join(lines) - else: - # Images-only quote: give the LLM at least a marker so it knows - # context was referenced. - quote_block = "[Quoted message]: (image)" - - return { - "quote_block": quote_block, - "image_urls": quoted_images, - "image_media_types": quoted_image_types, - } - - @staticmethod - def _merge_quote_into(text: str, quote_block: str) -> str: - """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 - # ------------------------------------------------------------------ # Attachment processing # ------------------------------------------------------------------ @@ -1712,7 +1220,7 @@ async def _process_attachments( elif ct.startswith("image/"): # Image: download and cache locally. try: - cached_path = await self._download_and_cache(url, ct, filename) + cached_path = await self._download_and_cache(url, ct) if cached_path and os.path.isfile(cached_path): image_urls.append(cached_path) image_media_types.append(ct or "image/jpeg") @@ -1725,15 +1233,11 @@ async def _process_attachments( except Exception as exc: logger.debug("[%s] Failed to cache image: %s", self._log_tag, exc) else: - # Other attachments (video, file, etc.): download and record with path. + # Other attachments (video, file, etc.): record as text. try: - cached_path = await self._download_and_cache(url, ct, filename) + cached_path = await self._download_and_cache(url, ct) if cached_path: - name = filename or ct - if ct.startswith("video/"): - other_attachments.append(f"[video: {name} ({cached_path})]") - else: - other_attachments.append(f"[file: {name} ({cached_path})]") + other_attachments.append(f"[Attachment: {filename or ct}]") except Exception as exc: logger.debug("[%s] Failed to cache attachment: %s", self._log_tag, exc) @@ -1745,14 +1249,8 @@ async def _process_attachments( "attachment_info": attachment_info, } - async def _download_and_cache( - self, url: str, content_type: str, original_name: str = "", - ) -> Optional[str]: - """Download a URL and cache it locally. - - :param original_name: Preferred filename from attachment metadata. - Falls back to the URL path basename if empty. - """ + async def _download_and_cache(self, url: str, content_type: str) -> Optional[str]: + """Download a URL and cache it locally.""" from tools.url_safety import is_safe_url if not is_safe_url(url): @@ -1783,11 +1281,7 @@ async def _download_and_cache( # Convert to .wav using ffmpeg so STT engines can process it. return await self._convert_audio_to_wav(data, url) else: - filename = ( - original_name - or Path(urlparse(url).path).name - or "qq_attachment" - ) + filename = Path(urlparse(url).path).name or "qq_attachment" return cache_document_from_bytes(data, filename) @staticmethod @@ -2000,7 +1494,7 @@ async def _convert_audio_to_wav_file( @staticmethod def _guess_ext_from_data(data: bytes) -> str: """Guess file extension from magic bytes.""" - if data[:9] == b"#!SILK_V3" or data[:6] == b"#!SILK": + if data[:9] == b"#!SILK_V3" or data[:5] == b"#!SILK": return ".silk" if data[:2] == b"\x02!": return ".silk" @@ -2008,7 +1502,7 @@ def _guess_ext_from_data(data: bytes) -> str: return ".wav" if data[:4] == b"fLaC": return ".flac" - if data[:2] in {b"\xff\xfb", b"\xff\xf3", b"\xff\xf2"}: + 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" @@ -2020,7 +1514,7 @@ def _guess_ext_from_data(data: bytes) -> str: @staticmethod def _looks_like_silk(data: bytes) -> bool: """Check if bytes look like a SILK audio file.""" - return data[:6] == b"#!SILK" or data[:2] == b"\x02!" or data[:9] == b"#!SILK_V3" + return data[:4] == b"#!SILK" or data[:2] == b"\x02!" or data[:9] == b"#!SILK_V3" async def _convert_silk_to_wav(self, src_path: str, wav_path: str) -> Optional[str]: """Convert audio file to WAV using the pilk library. @@ -2177,7 +1671,7 @@ def _resolve_stt_config(self) -> Optional[Dict[str, str]]: "base_url": base_url, "api_key": api_key, "model": model - or ("glm-asr" if provider in {"zai", "glm"} else "whisper-1"), + or ("glm-asr" if provider in ("zai", "glm") else "whisper-1"), } # 2. QQ-specific env vars (set by `hermes setup gateway` / `hermes gateway`) @@ -2259,7 +1753,7 @@ async def _convert_audio_to_wav( if urlparse(source_url).path else "" ) - if not ext or ext not in { + if not ext or ext not in ( ".silk", ".amr", ".mp3", @@ -2268,7 +1762,7 @@ async def _convert_audio_to_wav( ".m4a", ".aac", ".flac", - }: + ): ext = self._guess_ext_from_data(audio_data) with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tmp_src: @@ -2432,6 +1926,7 @@ async def send( Applies format_message(), splits long messages via truncate_message(), and retries transient failures with exponential backoff. + Also extracts MEDIA: tags and delivers media files natively. """ del metadata @@ -2442,17 +1937,46 @@ async def send( 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) + # Extract MEDIA: tags from content + media_files, cleaned_content = self.extract_media(content) + + _AUDIO_EXTS = {".ogg", ".opus", ".mp3", ".wav", ".m4a", ".flac"} + _VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".3gp"} + _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"} + + # Deliver media files first + for media_path, is_voice in media_files: + try: + ext = Path(media_path).suffix.lower() + if is_voice or ext in _AUDIO_EXTS: + await self.send_voice(chat_id=chat_id, audio_path=media_path) + elif ext in _VIDEO_EXTS: + await self.send_video(chat_id=chat_id, video_path=media_path) + elif ext in _IMAGE_EXTS: + await self.send_image_file(chat_id=chat_id, image_path=media_path) + else: + await self.send_document(chat_id=chat_id, file_path=media_path) + except Exception as exc: + logger.warning( + "[%s] media delivery failed for %s: %s", + self._log_tag, media_path, exc, + ) + + # Deliver remaining text content + if cleaned_content and cleaned_content.strip(): + formatted = self.format_message(cleaned_content) + chunks = self.truncate_message(formatted, 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 - return last_result + 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 + return last_result + + return SendResult(success=True) async def _send_chunk( self, @@ -2505,44 +2029,26 @@ async def _send_chunk( return SendResult(success=False, error=error_msg, retryable=retryable) async def _send_c2c_text( - self, - openid: str, - content: str, - reply_to: Optional[str] = None, - keyboard: Optional[InlineKeyboard] = None, + self, openid: str, content: str, reply_to: Optional[str] = None ) -> SendResult: - """Send text to a C2C user via REST API. - - :param keyboard: Optional inline keyboard attached to the message. - """ + """Send text to a C2C user via REST API.""" self._next_msg_seq(reply_to or openid) body = self._build_text_body(content, reply_to) if reply_to: body["msg_id"] = reply_to - if keyboard is not None: - body["keyboard"] = keyboard.to_dict() data = await self._api_request("POST", f"/v2/users/{openid}/messages", body) msg_id = str(data.get("id", uuid.uuid4().hex[:12])) return SendResult(success=True, message_id=msg_id, raw_response=data) async def _send_group_text( - self, - group_openid: str, - content: str, - reply_to: Optional[str] = None, - keyboard: Optional[InlineKeyboard] = None, + self, group_openid: str, content: str, reply_to: Optional[str] = None ) -> SendResult: - """Send text to a group via REST API. - - :param keyboard: Optional inline keyboard attached to the message. - """ + """Send text to a group via REST API.""" self._next_msg_seq(reply_to or group_openid) body = self._build_text_body(content, reply_to) if reply_to: body["msg_id"] = reply_to - if keyboard is not None: - body["keyboard"] = keyboard.to_dict() data = await self._api_request( "POST", f"/v2/groups/{group_openid}/messages", body @@ -2562,156 +2068,6 @@ async def _send_guild_text( msg_id = str(data.get("id", uuid.uuid4().hex[:12])) return SendResult(success=True, message_id=msg_id, raw_response=data) - # ------------------------------------------------------------------ - # 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, - ) -> SendResult: - """Send a single text message with an inline keyboard attached. - - Unlike :meth:`send`, this does NOT split long content into chunks — - a keyboard message has exactly one interactive surface, and splitting - would orphan the buttons from the first chunk. Callers should keep - approval/update-prompt bodies short. - - Guild (channel) chats don't support inline keyboards; returns a - non-retryable failure for those. - """ - if not self.is_connected: - if not await self._wait_for_reconnection(): - return SendResult( - success=False, error="Not connected", retryable=True - ) - - chat_type = self._guess_chat_type(chat_id) - formatted = self.format_message(content) - truncated = formatted[: self.MAX_MESSAGE_LENGTH] - try: - if chat_type == "c2c": - return await self._send_c2c_text( - chat_id, truncated, reply_to, keyboard=keyboard, - ) - 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 " - f"{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)) - - async def send_approval_request( - self, - chat_id: str, - req: ApprovalRequest, - reply_to: Optional[str] = None, - ) -> SendResult: - """Send a 3-button approval request (``allow-once / allow-always / deny``). - - The rendered text comes from :func:`build_approval_text`; callers can - override by passing a custom :class:`ApprovalRequest`. - - Users click the button → ``INTERACTION_CREATE`` fires → the adapter's - registered :meth:`set_interaction_callback` handler decodes - ``button_data`` via :func:`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), - reply_to=reply_to, - ) - - # ------------------------------------------------------------------ - # Cross-adapter gateway contract — send_exec_approval + send_update_prompt - # ------------------------------------------------------------------ - # - # These mirror the signatures that gateway/run.py detects on the adapter - # class (e.g. type(adapter).send_exec_approval, type(adapter).send_update_prompt) - # for button-based approval / update-confirm UX. Discord, Telegram, Slack, - # Matrix, and Feishu already implement the same contract. - - async def send_exec_approval( - self, - chat_id: str, - command: str, - session_key: str, - description: str = "dangerous command", - metadata: Optional[Dict[str, Any]] = None, - ) -> SendResult: - """Send a button-based exec-approval prompt for a dangerous command. - - Called by ``gateway/run.py``'s ``_approval_notify_sync`` when the - agent is blocked waiting for approval. Button clicks resolve via - :func:`tools.approval.resolve_gateway_approval` — dispatched by the - adapter's interaction callback (:meth:`_default_interaction_dispatch`). - """ - del metadata # QQ doesn't have thread_id / DM targeting overrides. - - # Use the reply-to message for passive-message context when we have one. - # QQ requires a msg_id on outbound messages to a user we've never - # seen; the last inbound msg_id is the natural choice. - msg_id = self._last_msg_id.get(chat_id) - - req = ApprovalRequest( - session_key=session_key, - title=f"Execute this command?", - description=description, - command_preview=command, - timeout_sec=self._APPROVAL_TIMEOUT_SECONDS, - ) - return await self.send_approval_request( - chat_id, req, reply_to=msg_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, - ) -> SendResult: - """Send a Yes/No update-confirmation prompt with inline buttons. - - Matches the cross-adapter contract used by - ``gateway/run.py``'s ``hermes update --gateway`` watcher. Button - clicks surface as ``INTERACTION_CREATE`` with - ``button_data = 'update_prompt:y'`` or ``'update_prompt:n'``; - the adapter's interaction callback writes the answer to - ``~/.hermes/.update_response`` so the detached update process - can read it. - """ - 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}" - msg_id = self._last_msg_id.get(chat_id) - return await self.send_with_keyboard( - chat_id, - content, - build_update_prompt_keyboard(), - reply_to=msg_id, - ) - def _build_text_body( self, content: str, reply_to: Optional[str] = None ) -> Dict[str, Any]: @@ -2841,62 +2197,42 @@ async def _send_media( reply_to: Optional[str] = None, file_name: Optional[str] = None, ) -> SendResult: - """Upload media and send as a native message. - - Upload strategy: - - - **HTTP(S) URLs** → single ``POST /v2/{users|groups}/{id}/files`` - with ``url=...``. The QQ platform fetches the URL directly; fastest - path when the source is already hosted. - - **Local files** → three-step chunked upload (prepare / PUT parts / - complete). Handles files up to the platform's ~100 MB per-file - limit without the ~10 MB inline-base64 cap of the old adapter. - """ + """Upload media and send as a native message.""" if not self.is_connected: if not await self._wait_for_reconnection(): return SendResult(success=False, error="Not connected", retryable=True) - chat_type = self._guess_chat_type(chat_id) - if chat_type == "guild": - # Guild channels don't support native media upload in the same way. - return SendResult( - success=False, - error="Guild media send not supported via this path", + try: + # Resolve media source + data, content_type, resolved_name = await self._load_media( + media_source, file_name ) - try: - if self._is_url(media_source): - # URL upload — let the platform fetch it directly. - 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, - file_name=resolved_name if file_type == MEDIA_TYPE_FILE else None, - ) - else: - # Local file — chunked upload (prepare / PUT parts / complete). - resolved_name, upload = await self._upload_local_file( - chat_type, - chat_id, - media_source, - file_type, - file_name, + # Route + chat_type = self._guess_chat_type(chat_id) + + if chat_type == "guild": + # Guild channels don't support native media upload in the same way + # Send as URL fallback + return SendResult( + success=False, error="Guild media send not supported via this path" ) - file_info = upload.get("file_info") or ( - upload.get("data", {}) or {} - ).get("file_info") + # Upload + upload = await self._upload_media( + chat_type, + chat_id, + file_type, + file_data=data if not self._is_url(media_source) else None, + url=media_source if self._is_url(media_source) else None, + srv_send_msg=False, + file_name=resolved_name if file_type == MEDIA_TYPE_FILE else None, + ) + + file_info = upload.get("file_info") if not file_info: return SendResult( - success=False, - error=f"Upload returned no file_info: {upload}", + success=False, error=f"Upload returned no file_info: {upload}" ) # Send media message @@ -2925,86 +2261,10 @@ async def _send_media( message_id=str(send_data.get("id", uuid.uuid4().hex[:12])), raw_response=send_data, ) - except UploadDailyLimitExceededError as exc: - # Non-retryable: daily quota hit. Give the caller actionable text - # so the model can compose a helpful reply. - logger.warning( - "[%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, - ) - except UploadFileTooLargeError as exc: - logger.warning( - "[%s] File too large: %s (%s, platform limit %s)", - self._log_tag, exc.file_name, exc.file_size_human, exc.limit_human, - ) - return SendResult( - success=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)) - async def _upload_local_file( - 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 and return ``(resolved_name, complete_response)``. - - The returned ``complete_response`` contains the ``file_info`` token - that goes into the subsequent RichMedia message body. - - :raises UploadDailyLimitExceededError: On biz_code 40093002. - :raises UploadFileTooLargeError: When the file exceeds the platform limit. - :raises FileNotFoundError: If the path does not exist. - :raises ValueError: If the path looks like a placeholder (````). - :raises RuntimeError: If the HTTP client is not initialized. - """ - if not self._http_client: - raise RuntimeError("HTTP client not initialized — not connected?") - - 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, - ) - complete = await uploader.upload( - 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 - async def _load_media( self, source: str, file_name: Optional[str] = None ) -> Tuple[str, str, str]: @@ -3014,7 +2274,7 @@ async def _load_media( raise ValueError("Media source is required") parsed = urlparse(source) - if parsed.scheme in {"http", "https"}: + if parsed.scheme in ("http", "https"): # For URLs, pass through directly to the upload API content_type = mimetypes.guess_type(source)[0] or "application/octet-stream" resolved_name = file_name or Path(parsed.path).name or "media" @@ -3110,7 +2370,7 @@ async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: chat_type = self._guess_chat_type(chat_id) return { "name": chat_id, - "type": "group" if chat_type in {"group", "guild"} else "dm", + "type": "group" if chat_type in ("group", "guild") else "dm", } # ------------------------------------------------------------------ @@ -3119,7 +2379,7 @@ async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: @staticmethod def _is_url(source: str) -> bool: - return urlparse(str(source)).scheme in {"http", "https"} + return urlparse(str(source)).scheme in ("http", "https") def _guess_chat_type(self, chat_id: str) -> str: """Determine chat type from stored inbound metadata, fallback to 'c2c'.""" @@ -3188,3 +2448,8 @@ def _is_duplicate(self, msg_id: str) -> bool: return True self._seen_messages[msg_id] = now return False + + +def get_active_adapter() -> Optional["QQAdapter"]: + """Return the currently connected QQAdapter singleton, or None.""" + return QQAdapter.get_active() diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 9ea0b9af41b5..dac111eb2628 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -10,9 +10,9 @@ import logging import os import re +from typing import Dict, Optional import ssl import time -from email.utils import formatdate from agent.redact import redact_sensitive_text @@ -26,9 +26,7 @@ # because the API requires a conversation ID. To DM a user you must first call # conversations.open to obtain a D... ID. Without this gate, Slack IDs fall # through to channel-name resolution, which only matches by name and fails. -_SLACK_TARGET_RE = re.compile(r"^\s*([CGDU][A-Z0-9]{8,})\s*$") -# Session-derived Slack thread targets use ":". -_SLACK_THREAD_TARGET_RE = re.compile(r"^\s*([CGD][A-Z0-9]{8,}):([^\s:]+)\s*$") +_SLACK_TARGET_RE = re.compile(r"^\s*([CGD][A-Z0-9]{8,})\s*$") _WEIXIN_TARGET_RE = re.compile(r"^\s*((?:wxid|gh|v\d+|wm|wb)_[A-Za-z0-9_-]+|[A-Za-z0-9._-]+@chatroom|filehelper)\s*$") _YUANBAO_TARGET_RE = re.compile(r"^\s*((?:group|direct):[^:]+)\s*$") # Discord snowflake IDs are numeric, same regex pattern as Telegram topic targets. @@ -138,7 +136,7 @@ async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs) }, "message": { "type": "string", - "description": "The message text to send. To send an image or file, include MEDIA: (e.g. 'MEDIA:/tmp/report.pdf') in the message — the platform will deliver it as a native media attachment." + "description": "The message text to send. To send an image or file, include MEDIA: (e.g. 'MEDIA:/tmp/hermes/cache/img_xxx.jpg') in the message — the platform will deliver it as a native media attachment." } }, "required": [] @@ -243,14 +241,7 @@ def _handle_send(args): from gateway.platforms.base import BasePlatformAdapter - # Capture [[as_document]] directive before extract_media strips it. - # Image-extension files in this batch will route through send_document - # instead of send_photo so the original bytes survive (e.g. info-graph - # JPGs where Telegram's sendPhoto recompresses to 1280px). - force_document_attachments = "[[as_document]]" in message - media_files, cleaned_message = BasePlatformAdapter.extract_media(message) - media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files) mirror_text = cleaned_message.strip() or _describe_media_for_mirror(media_files) used_home_channel = False @@ -275,28 +266,6 @@ def _handle_send(args): if duplicate_skip: return json.dumps(duplicate_skip) - # Slack: resolve user IDs (U...) to DM channel IDs via conversations.open - if platform_name == "slack" and chat_id and chat_id.startswith("U"): - try: - import aiohttp - async def _open_slack_dm(token, user_id): - url = "https://slack.com/api/conversations.open" - headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} - async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session: - async with session.post(url, headers=headers, json={"users": [user_id]}) as resp: - data = await resp.json() - if data.get("ok"): - return data["channel"]["id"] - return None - from model_tools import _run_async - dm_channel = _run_async(_open_slack_dm(pconfig.token, chat_id)) - if dm_channel: - chat_id = dm_channel - else: - return json.dumps({"error": f"Could not open DM with Slack user {chat_id}. Check bot permissions (im:write)."}) - except Exception as e: - return json.dumps({"error": f"Failed to open Slack DM: {e}"}) - try: from model_tools import _run_async result = _run_async( @@ -307,7 +276,6 @@ async def _open_slack_dm(token, user_id): cleaned_message, thread_id=thread_id, media_files=media_files, - force_document=force_document_attachments, ) ) if used_home_channel and isinstance(result, dict) and result.get("success"): @@ -354,24 +322,9 @@ def _parse_target_ref(platform_name: str, target_ref: str): if match: return match.group(1), match.group(2), True if platform_name == "slack": - match = _SLACK_THREAD_TARGET_RE.fullmatch(target_ref) - if match: - return match.group(1), match.group(2), True match = _SLACK_TARGET_RE.fullmatch(target_ref) if match: - chat_id = match.group(1) - # Slack user IDs (U...) and workspace IDs (W...) are NOT valid - # explicit send targets — chat.postMessage rejects them. A DM - # must be opened first via conversations.open to get a D... - # conversation ID. Caller still gets the chat_id so the U→D - # resolution path in send_message() can run. - is_explicit = chat_id[0] not in {"U", "W"} - return chat_id, None, is_explicit - if platform_name == "matrix": - trimmed = target_ref.strip() - split_idx = trimmed.rfind(":$") - if split_idx > 0: - return trimmed[:split_idx], trimmed[split_idx + 1 :], True + return match.group(1), None, True if platform_name == "weixin": match = _WEIXIN_TARGET_RE.fullmatch(target_ref) if match: @@ -394,9 +347,6 @@ def _parse_target_ref(platform_name: str, target_ref: str): # Matrix room IDs (start with !) and user IDs (start with @) are explicit if platform_name == "matrix" and (target_ref.startswith("!") or target_ref.startswith("@")): return target_ref, None, True - # XMPP JIDs (user@server or room@conference.server) are explicit - if platform_name == "xmpp" and "@" in target_ref: - return target_ref, None, True return None, None, False @@ -465,96 +415,28 @@ def _maybe_skip_cron_duplicate_send(platform_name: str, chat_id: str, thread_id: } -async def _send_via_adapter( - platform, - pconfig, - chat_id, - chunk, - *, - thread_id=None, - media_files=None, - force_document=False, -): - """Send a message via a live gateway adapter, with a standalone fallback - for out-of-process callers (e.g. cron running separately from the gateway). - - Order of attempts: - 1. Live in-process adapter via ``_gateway_runner_ref()`` (the path that - existed before this change). - 2. The plugin's ``standalone_sender_fn`` registered on its - ``PlatformEntry`` (used when the gateway is not in this process, so - the runner weakref is ``None``). - 3. A descriptive error explaining both options. +async def _send_via_adapter(platform, pconfig, chat_id, chunk): + """Send a message via a live gateway adapter (for plugin platforms). + + Falls back to error if no adapter is connected for this platform. """ - runner = None try: from gateway.run import _gateway_runner_ref runner = _gateway_runner_ref() - except Exception: - runner = None - - if runner is not None: - try: + if runner: adapter = runner.adapters.get(platform) - except Exception: - adapter = None - if adapter is not None: - try: - metadata = {"thread_id": thread_id} if thread_id else None - result = await adapter.send(chat_id=chat_id, content=chunk, metadata=metadata) - except asyncio.CancelledError: - raise - except Exception as e: - return {"error": f"Plugin platform send failed: {e}"} - if result.success: - return {"success": True, "message_id": result.message_id} - return {"error": f"Adapter send failed: {result.error}"} - - platform_name = platform.value if hasattr(platform, "value") else str(platform) - entry = None - try: - from gateway.platform_registry import platform_registry - entry = platform_registry.get(platform_name) - except Exception: - entry = None - - if entry is not None and entry.standalone_sender_fn is not None: - try: - result = await entry.standalone_sender_fn( - pconfig, - chat_id, - chunk, - thread_id=thread_id, - media_files=media_files, - force_document=force_document, - ) - except asyncio.CancelledError: - raise - except Exception as e: - logger.debug("Plugin standalone send for %s raised", platform_name, exc_info=True) - return {"error": f"Plugin standalone send failed: {e}"} - - if isinstance(result, dict) and (result.get("success") or result.get("error")): - return result - return { - "error": ( - f"Plugin standalone send for '{platform_name}' returned an " - f"invalid result: expected a dict with 'success' or 'error' " - f"keys, got {type(result).__name__}" - ) - } - - return { - "error": ( - f"No live adapter for platform '{platform_name}'. Is the gateway " - f"running with this platform connected? For out-of-process delivery " - f"(e.g. cron in a separate process), the platform plugin must " - f"register a standalone_sender_fn on its PlatformEntry." - ) - } + if adapter: + from gateway.platforms.base import SendResult + result = await adapter.send(chat_id=chat_id, content=chunk) + if result.success: + return {"success": True, "message_id": result.message_id} + return {"error": f"Adapter send failed: {result.error}"} + except Exception as e: + return {"error": f"Plugin platform send failed: {e}"} + return {"error": f"No live adapter for platform '{platform.value}'. Is the gateway running with this platform connected?"} -async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None, force_document=False): +async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None): """Route a message to the appropriate platform sender. Long messages are automatically chunked to fit within platform limits @@ -563,6 +445,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, """ from gateway.config import Platform from gateway.platforms.base import BasePlatformAdapter, utf16_len + from gateway.platforms.discord import DiscordAdapter from gateway.platforms.slack import SlackAdapter # Telegram adapter import is optional (requires python-telegram-bot) @@ -588,10 +471,10 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, except Exception: logger.debug("Failed to apply Slack mrkdwn formatting in _send_to_platform", exc_info=True) - # Platform message length limits (from adapter class attributes for - # built-in platforms; from PlatformEntry.max_message_length for plugins). + # Platform message length limits (from adapter class attributes) _MAX_LENGTHS = { Platform.TELEGRAM: TelegramAdapter.MAX_MESSAGE_LENGTH if _telegram_available else 4096, + Platform.DISCORD: DiscordAdapter.MAX_MESSAGE_LENGTH, Platform.SLACK: SlackAdapter.MAX_MESSAGE_LENGTH, } if _feishu_available: @@ -630,7 +513,6 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=media_files if is_last else [], thread_id=thread_id, disable_link_previews=disable_link_previews, - force_document=force_document, ) if isinstance(result, dict) and result.get("error"): return result @@ -641,27 +523,17 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, if platform == Platform.WEIXIN: return await _send_weixin(pconfig, chat_id, message, media_files=media_files) - # --- Discord: chunked delivery via the registry's standalone_sender_fn. - # The plugin's ``_standalone_send`` (registered in - # plugins/platforms/discord/adapter.py) handles forum channels, threads, - # and multipart media uploads. ``_send_via_adapter`` tries the live - # in-process adapter first via ``adapter.send()``, but Discord's elif - # historically went straight to the HTTP path; we preserve that by - # explicitly invoking the registry hook here so behavior is unchanged. + # --- Discord: special handling for media attachments --- if platform == Platform.DISCORD: - from gateway.platform_registry import platform_registry - entry = platform_registry.get("discord") - if entry is None or entry.standalone_sender_fn is None: - return {"error": "Discord plugin not registered or missing standalone_sender_fn"} last_result = None for i, chunk in enumerate(chunks): is_last = (i == len(chunks) - 1) - result = await entry.standalone_sender_fn( - pconfig, + result = await _send_discord( + pconfig.token, chat_id, chunk, - thread_id=thread_id, media_files=media_files if is_last else [], + thread_id=thread_id, ) if isinstance(result, dict) and result.get("error"): return result @@ -716,17 +588,14 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, last_result = result return last_result - # --- Feishu: native media attachment support via adapter --- - if platform == Platform.FEISHU and media_files: + # --- QQBot: native media attachment support via running gateway adapter --- + if platform == Platform.QQBOT and media_files: last_result = None for i, chunk in enumerate(chunks): is_last = (i == len(chunks) - 1) - result = await _send_feishu( - pconfig, - chat_id, - chunk, + result = await _send_qqbot( + pconfig, chat_id, chunk, media_files=media_files if is_last else None, - thread_id=thread_id, ) if isinstance(result, dict) and result.get("error"): return result @@ -737,7 +606,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, if media_files and not message.strip(): return { "error": ( - f"send_message MEDIA delivery is currently only supported for telegram, discord, matrix, weixin, signal, yuanbao and feishu; " + f"send_message MEDIA delivery is currently only supported for telegram, discord, matrix, weixin, signal and yuanbao; " f"target {platform.value} had only media attachments" ) } @@ -745,7 +614,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, if media_files: warning = ( f"MEDIA attachments were omitted for {platform.value}; " - "native send_message media delivery is currently only supported for telegram, discord, matrix, weixin, signal, yuanbao and feishu" + "native send_message media delivery is currently only supported for telegram, discord, matrix, weixin, signal and yuanbao" ) last_result = None @@ -760,6 +629,8 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, result = await _send_email(pconfig.extra, chat_id, chunk) elif platform == Platform.SMS: result = await _send_sms(pconfig.api_key, chat_id, chunk) + elif platform == Platform.MATTERMOST: + result = await _send_mattermost(pconfig.token, pconfig.extra, chat_id, chunk) elif platform == Platform.MATRIX: result = await _send_matrix(pconfig.token, pconfig.extra, chat_id, chunk) elif platform == Platform.HOMEASSISTANT: @@ -777,17 +648,9 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, elif platform == Platform.YUANBAO: result = await _send_yuanbao(chat_id, chunk) else: - # Plugin platform: route through the gateway's live adapter if - # available, otherwise the plugin's standalone_sender_fn. - result = await _send_via_adapter( - platform, - pconfig, - chat_id, - chunk, - thread_id=thread_id, - media_files=media_files, - force_document=force_document, - ) + # Plugin platform — route through the gateway's live adapter + # if available, otherwise report the error. + result = await _send_via_adapter(platform, pconfig, chat_id, chunk) if isinstance(result, dict) and result.get("error"): return result @@ -800,16 +663,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, return last_result -def _is_telegram_thread_not_found(error: Exception) -> bool: - """Check if a Telegram error is a thread-not-found failure. - - Matches the gateway adapter's ``_is_thread_not_found_error`` for - the standalone ``_send_telegram`` path (issue #27012). - """ - return "thread not found" in str(error).lower() - - -async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False, force_document=False): +async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False): """Send via Telegram Bot API (one-shot, no polling needed). Applies markdown→MarkdownV2 formatting (same as the gateway adapter) @@ -839,61 +693,14 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No formatted = message send_parse_mode = ParseMode.MARKDOWN_V2 - # Honour a configured proxy (telegram.proxy_url in config.yaml, exported - # as TELEGRAM_PROXY env var by load_gateway_config). Without this, the - # standalone send path bypasses the proxy and times out in regions - # where api.telegram.org is blocked. The in-gateway adapter does the - # same thing in gateway/platforms/telegram.py. - try: - from gateway.platforms.base import resolve_proxy_url - _tg_proxy = resolve_proxy_url("TELEGRAM_PROXY", target_hosts=["api.telegram.org"]) - except Exception: - _tg_proxy = None - if _tg_proxy: - try: - from telegram.request import HTTPXRequest - logger.info("send_message: standalone Telegram send routed through proxy %s", _tg_proxy) - bot = Bot( - token=token, - request=HTTPXRequest(proxy=_tg_proxy), - get_updates_request=HTTPXRequest(proxy=_tg_proxy), - ) - except Exception as _proxy_err: - logger.warning("send_message: failed to attach Telegram proxy (%s), falling back to direct connection", _proxy_err) - bot = Bot(token=token) - else: - bot = Bot(token=token) + bot = Bot(token=token) int_chat_id = int(chat_id) media_files = media_files or [] thread_kwargs = {} if thread_id is not None: - # Reuse the gateway adapter's General-topic mapping: in Telegram - # forum supergroups, the General topic is addressed as - # message_thread_id="1" on incoming updates, but Bot API - # sendMessage rejects message_thread_id=1 with "Message thread - # not found". The adapter's helper maps "1" to None for that - # reason; the send_message tool needs the same mapping or a - # send to a forum group's General topic always errors out - # (see issue #22267). - try: - from gateway.platforms.telegram import TelegramAdapter - effective_thread_id = TelegramAdapter._message_thread_id_for_send( - str(thread_id) - ) - except Exception: - # Fallback: explicit mapping in case the adapter import - # fails (e.g. python-telegram-bot missing in this venv). - effective_thread_id = ( - None if str(thread_id) == "1" else int(thread_id) - ) - if effective_thread_id is not None: - thread_kwargs["message_thread_id"] = effective_thread_id - # disable_web_page_preview is only valid for send_message, not - # send_photo/send_video/etc. Keep it separate so media sends - # don't inherit an invalid parameter (issue #27012). - text_kwargs = dict(thread_kwargs) + thread_kwargs["message_thread_id"] = int(thread_id) if disable_link_previews: - text_kwargs["disable_web_page_preview"] = True + thread_kwargs["disable_web_page_preview"] = True last_msg = None warnings = [] @@ -903,24 +710,11 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No last_msg = await _send_telegram_message_with_retry( bot, chat_id=int_chat_id, text=formatted, - parse_mode=send_parse_mode, **text_kwargs + parse_mode=send_parse_mode, **thread_kwargs ) except Exception as md_error: - # Thread not found — retry without message_thread_id so the - # message still delivers (matching the gateway adapter's - # fallback behaviour, issue #27012). - if _is_telegram_thread_not_found(md_error) and thread_kwargs: - logger.warning( - "Thread %s not found in _send_telegram, retrying without message_thread_id", - thread_kwargs.get("message_thread_id"), - ) - text_kwargs.pop("message_thread_id", None) - last_msg = await _send_telegram_message_with_retry( - bot, - chat_id=int_chat_id, text=formatted, - parse_mode=send_parse_mode, **text_kwargs - ) - elif "parse" in str(md_error).lower() or "markdown" in str(md_error).lower() or "html" in str(md_error).lower(): + # Parse failed, fall back to plain text + if "parse" in str(md_error).lower() or "markdown" in str(md_error).lower() or "html" in str(md_error).lower(): logger.warning( "Parse mode %s failed in _send_telegram, falling back to plain text: %s", send_parse_mode, @@ -937,7 +731,7 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No last_msg = await _send_telegram_message_with_retry( bot, chat_id=int_chat_id, text=plain, - parse_mode=None, **text_kwargs + parse_mode=None, **thread_kwargs ) else: raise @@ -952,61 +746,26 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No ext = os.path.splitext(media_path)[1].lower() try: with open(media_path, "rb") as f: - media_kwargs = dict(thread_kwargs) - try: - if ext in _IMAGE_EXTS and not force_document: - last_msg = await bot.send_photo( - chat_id=int_chat_id, photo=f, **media_kwargs - ) - elif ext in _VIDEO_EXTS: - last_msg = await bot.send_video( - chat_id=int_chat_id, video=f, **media_kwargs - ) - elif ext in _VOICE_EXTS and is_voice: - last_msg = await bot.send_voice( - chat_id=int_chat_id, voice=f, **media_kwargs - ) - elif ext in _TELEGRAM_SEND_AUDIO_EXTS: - last_msg = await bot.send_audio( - chat_id=int_chat_id, audio=f, **media_kwargs - ) - else: - last_msg = await bot.send_document( - chat_id=int_chat_id, document=f, **media_kwargs - ) - except Exception as media_err: - if _is_telegram_thread_not_found(media_err) and media_kwargs.get("message_thread_id"): - # Thread not found for media — retry without - # message_thread_id (issue #27012). - logger.warning( - "Thread %s not found for media send, retrying without message_thread_id", - media_kwargs["message_thread_id"], - ) - # Re-seek the file since the first attempt consumed it - f.seek(0) - media_kwargs.pop("message_thread_id", None) - if ext in _IMAGE_EXTS and not force_document: - last_msg = await bot.send_photo( - chat_id=int_chat_id, photo=f, **media_kwargs - ) - elif ext in _VIDEO_EXTS: - last_msg = await bot.send_video( - chat_id=int_chat_id, video=f, **media_kwargs - ) - elif ext in _VOICE_EXTS and is_voice: - last_msg = await bot.send_voice( - chat_id=int_chat_id, voice=f, **media_kwargs - ) - elif ext in _TELEGRAM_SEND_AUDIO_EXTS: - last_msg = await bot.send_audio( - chat_id=int_chat_id, audio=f, **media_kwargs - ) - else: - last_msg = await bot.send_document( - chat_id=int_chat_id, document=f, **media_kwargs - ) - else: - raise + if ext in _IMAGE_EXTS: + last_msg = await bot.send_photo( + chat_id=int_chat_id, photo=f, **thread_kwargs + ) + elif ext in _VIDEO_EXTS: + last_msg = await bot.send_video( + chat_id=int_chat_id, video=f, **thread_kwargs + ) + elif ext in _VOICE_EXTS and is_voice: + last_msg = await bot.send_voice( + chat_id=int_chat_id, voice=f, **thread_kwargs + ) + elif ext in _TELEGRAM_SEND_AUDIO_EXTS: + last_msg = await bot.send_audio( + chat_id=int_chat_id, audio=f, **thread_kwargs + ) + else: + last_msg = await bot.send_document( + chat_id=int_chat_id, document=f, **thread_kwargs + ) except Exception as e: warning = _sanitize_error_text(f"Failed to send media {media_path}: {e}") logger.error(warning) @@ -1033,6 +792,227 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No return _error(f"Telegram send failed: {e}") +def _derive_forum_thread_name(message: str) -> str: + """Derive a thread name from the first line of the message, capped at 100 chars.""" + first_line = message.strip().split("\n", 1)[0].strip() + # Strip common markdown heading prefixes + first_line = first_line.lstrip("#").strip() + if not first_line: + first_line = "New Post" + return first_line[:100] + + +# Process-local cache for Discord channel-type probes. Avoids re-probing the +# same channel on every send when the directory cache has no entry (e.g. fresh +# install, or channel created after the last directory build). +_DISCORD_CHANNEL_TYPE_PROBE_CACHE: Dict[str, bool] = {} + + +def _remember_channel_is_forum(chat_id: str, is_forum: bool) -> None: + _DISCORD_CHANNEL_TYPE_PROBE_CACHE[str(chat_id)] = bool(is_forum) + + +def _probe_is_forum_cached(chat_id: str) -> Optional[bool]: + return _DISCORD_CHANNEL_TYPE_PROBE_CACHE.get(str(chat_id)) + + +async def _send_discord(token, chat_id, message, thread_id=None, media_files=None): + """Send a single message via Discord REST API (no websocket client needed). + + Chunking is handled by _send_to_platform() before this is called. + + When thread_id is provided, the message is sent directly to that thread + via the /channels/{thread_id}/messages endpoint. + + Media files are uploaded one-by-one via multipart/form-data after the + text message is sent (same pattern as Telegram). + + Forum channels (type 15) reject POST /messages — a thread post is created + automatically via POST /channels/{id}/threads. Media files are uploaded + as multipart attachments on the starter message of the new thread. + + Channel type is resolved from the channel directory first, then a + process-local probe cache, and only as a last resort with a live + GET /channels/{id} probe (whose result is memoized). + """ + try: + import aiohttp + except ImportError: + return {"error": "aiohttp not installed. Run: pip install aiohttp"} + try: + from gateway.platforms.base import resolve_proxy_url, proxy_kwargs_for_aiohttp + _proxy = resolve_proxy_url(platform_env_var="DISCORD_PROXY") + _sess_kw, _req_kw = proxy_kwargs_for_aiohttp(_proxy) + auth_headers = {"Authorization": f"Bot {token}"} + json_headers = {**auth_headers, "Content-Type": "application/json"} + media_files = media_files or [] + last_data = None + warnings = [] + + # Thread endpoint: Discord threads are channels; send directly to the thread ID. + if thread_id: + url = f"https://discord.com/api/v10/channels/{thread_id}/messages" + else: + # Check if the target channel is a forum channel (type 15). + # Forum channels reject POST /messages — create a thread post instead. + # Three-layer detection: directory cache → process-local probe + # cache → GET /channels/{id} probe (with result memoized). + _channel_type = None + try: + from gateway.channel_directory import lookup_channel_type + _channel_type = lookup_channel_type("discord", chat_id) + except Exception: + pass + + if _channel_type == "forum": + is_forum = True + elif _channel_type is not None: + is_forum = False + else: + cached = _probe_is_forum_cached(chat_id) + if cached is not None: + is_forum = cached + else: + is_forum = False + try: + info_url = f"https://discord.com/api/v10/channels/{chat_id}" + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15), **_sess_kw) as info_sess: + async with info_sess.get(info_url, headers=json_headers, **_req_kw) as info_resp: + if info_resp.status == 200: + info = await info_resp.json() + is_forum = info.get("type") == 15 + _remember_channel_is_forum(chat_id, is_forum) + except Exception: + logger.debug("Failed to probe channel type for %s", chat_id, exc_info=True) + + if is_forum: + thread_name = _derive_forum_thread_name(message) + thread_url = f"https://discord.com/api/v10/channels/{chat_id}/threads" + + # Filter to readable media files up front so we can pick the + # right code path (JSON vs multipart) before opening a session. + valid_media = [] + for media_path, _is_voice in media_files: + if not os.path.exists(media_path): + warning = f"Media file not found, skipping: {media_path}" + logger.warning(warning) + warnings.append(warning) + continue + valid_media.append(media_path) + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60), **_sess_kw) as session: + if valid_media: + # Multipart: payload_json + files[N] creates a forum + # thread with the starter message plus attachments in + # a single API call. + attachments_meta = [ + {"id": str(idx), "filename": os.path.basename(path)} + for idx, path in enumerate(valid_media) + ] + starter_message = {"content": message, "attachments": attachments_meta} + payload_json = json.dumps({"name": thread_name, "message": starter_message}) + + form = aiohttp.FormData() + form.add_field("payload_json", payload_json, content_type="application/json") + + # Buffer file bytes up front — aiohttp's FormData can + # read lazily and we don't want handles closing under + # it on retry. + try: + for idx, media_path in enumerate(valid_media): + with open(media_path, "rb") as fh: + form.add_field( + f"files[{idx}]", + fh.read(), + filename=os.path.basename(media_path), + ) + async with session.post(thread_url, headers=auth_headers, data=form, **_req_kw) as resp: + if resp.status not in (200, 201): + body = await resp.text() + return _error(f"Discord forum thread creation error ({resp.status}): {body}") + data = await resp.json() + except Exception as e: + return _error(_sanitize_error_text(f"Discord forum thread upload failed: {e}")) + else: + # No media — simple JSON POST creates the thread with + # just the text starter. + async with session.post( + thread_url, + headers=json_headers, + json={ + "name": thread_name, + "message": {"content": message}, + }, + **_req_kw, + ) as resp: + if resp.status not in (200, 201): + body = await resp.text() + return _error(f"Discord forum thread creation error ({resp.status}): {body}") + data = await resp.json() + + thread_id_created = data.get("id") + starter_msg_id = (data.get("message") or {}).get("id", thread_id_created) + result = { + "success": True, + "platform": "discord", + "chat_id": chat_id, + "thread_id": thread_id_created, + "message_id": starter_msg_id, + } + if warnings: + result["warnings"] = warnings + return result + + url = f"https://discord.com/api/v10/channels/{chat_id}/messages" + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30), **_sess_kw) as session: + # Send text message (skip if empty and media is present) + if message.strip() or not media_files: + async with session.post(url, headers=json_headers, json={"content": message}, **_req_kw) as resp: + if resp.status not in (200, 201): + body = await resp.text() + return _error(f"Discord API error ({resp.status}): {body}") + last_data = await resp.json() + + # Send each media file as a separate multipart upload + for media_path, _is_voice in media_files: + if not os.path.exists(media_path): + warning = f"Media file not found, skipping: {media_path}" + logger.warning(warning) + warnings.append(warning) + continue + try: + form = aiohttp.FormData() + filename = os.path.basename(media_path) + with open(media_path, "rb") as f: + form.add_field("files[0]", f, filename=filename) + async with session.post(url, headers=auth_headers, data=form, **_req_kw) as resp: + if resp.status not in (200, 201): + body = await resp.text() + warning = _sanitize_error_text(f"Failed to send media {media_path}: Discord API error ({resp.status}): {body}") + logger.error(warning) + warnings.append(warning) + continue + last_data = await resp.json() + except Exception as e: + warning = _sanitize_error_text(f"Failed to send media {media_path}: {e}") + logger.error(warning) + warnings.append(warning) + + if last_data is None: + error = "No deliverable text or media remained after processing" + if warnings: + return {"error": error, "warnings": warnings} + return {"error": error} + + result = {"success": True, "platform": "discord", "chat_id": chat_id, "message_id": last_data.get("id")} + if warnings: + result["warnings"] = warnings + return result + except Exception as e: + return _error(f"Discord send failed: {e}") + + async def _send_slack(token, chat_id, message): """Send via Slack Web API.""" try: @@ -1269,6 +1249,7 @@ async def _send_email(extra, chat_id, message): """Send via SMTP (one-shot, no persistent connection needed).""" import smtplib from email.mime.text import MIMEText + from email.utils import formatdate address = extra.get("address") or os.getenv("EMAIL_ADDRESS", "") password = os.getenv("EMAIL_PASSWORD", "") @@ -1354,6 +1335,30 @@ async def _send_sms(auth_token, chat_id, message): return _error(f"SMS send failed: {e}") +async def _send_mattermost(token, extra, chat_id, message): + """Send via Mattermost REST API.""" + try: + import aiohttp + except ImportError: + return {"error": "aiohttp not installed. Run: pip install aiohttp"} + try: + base_url = (extra.get("url") or os.getenv("MATTERMOST_URL", "")).rstrip("/") + token = token or os.getenv("MATTERMOST_TOKEN", "") + if not base_url or not token: + return {"error": "Mattermost not configured (MATTERMOST_URL, MATTERMOST_TOKEN required)"} + url = f"{base_url}/api/v4/posts" + headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session: + async with session.post(url, headers=headers, json={"channel_id": chat_id, "message": message}) as resp: + if resp.status not in (200, 201): + body = await resp.text() + return _error(f"Mattermost API error ({resp.status}): {body}") + data = await resp.json() + return {"success": True, "platform": "mattermost", "chat_id": chat_id, "message_id": data.get("id")} + except Exception as e: + return _error(f"Mattermost send failed: {e}") + + async def _send_matrix(token, extra, chat_id, message): """Send via Matrix Client-Server API. @@ -1389,7 +1394,7 @@ async def _send_matrix(token, extra, chat_id, message): async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session: async with session.put(url, headers=headers, json=payload) as resp: - if resp.status not in {200, 201}: + if resp.status not in (200, 201): body = await resp.text() return _error(f"Matrix API error ({resp.status}): {body}") data = await resp.json() @@ -1473,7 +1478,7 @@ async def _send_homeassistant(token, extra, chat_id, message): headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session: async with session.post(url, headers=headers, json={"message": message, "target": chat_id}) as resp: - if resp.status not in {200, 201}: + if resp.status not in (200, 201): body = await resp.text() return _error(f"Home Assistant API error ({resp.status}): {body}") return {"success": True, "platform": "homeassistant", "chat_id": chat_id} @@ -1645,20 +1650,7 @@ async def _send_feishu(pconfig, chat_id, message, media_files=None, thread_id=No def _check_send_message(): - """Gate send_message on gateway running (always available on messaging platforms). - - Also passes for kanban workers — the dispatcher sets ``HERMES_KANBAN_TASK`` - on every spawned worker, but those workers run with the assignee profile's - ``HERMES_HOME`` which has no ``gateway.pid``, so the gateway-running check - would fail even though the parent gateway is alive. Honoring the env var - lets workers call ``send_message`` to deliver rich content directly to the - originating chat (paired with ``kanban_complete`` for the short notifier - summary), which is the canonical pattern for any worker that needs to - reply with more than the ~200-char first-line truncation the kanban - notifier applies. - """ - if os.environ.get("HERMES_KANBAN_TASK"): - return True + """Gate send_message on gateway running (always available on messaging platforms).""" from gateway.session_context import get_session_env platform = get_session_env("HERMES_SESSION_PLATFORM", "") if platform and platform != "local": @@ -1670,74 +1662,42 @@ def _check_send_message(): return False -async def _send_qqbot(pconfig, chat_id, message): - """Send via QQBot using the REST API directly (no WebSocket needed). +async def _send_qqbot(pconfig, chat_id, message, media_files=None): + """Send via QQBot using the running gateway adapter (supports media). - Uses the QQ Bot Open Platform REST endpoints to get an access token - and post a message. Supports guild channels, C2C (private) chats, - and group chats by trying the appropriate endpoints. + Obtains the live QQAdapter singleton, re-injects MEDIA: tags for any + media_files that were already extracted, and delegates to adapter.send(). """ try: - import httpx + from gateway.platforms.qqbot.adapter import get_active_adapter except ImportError: - return _error("QQBot direct send requires httpx. Run: pip install httpx") + return _error("QQBot adapter module not available.") - extra = pconfig.extra or {} - appid = extra.get("app_id") or os.getenv("QQ_APP_ID", "") - secret = (pconfig.token or extra.get("client_secret") - or os.getenv("QQ_CLIENT_SECRET", "")) - if not appid or not secret: - return _error("QQBot: QQ_APP_ID / QQ_CLIENT_SECRET not configured.") + adapter = get_active_adapter() + if adapter is None: + return _error( + "QQBot adapter is not running. " + "Start the gateway with qqbot platform enabled first." + ) + + # Re-inject MEDIA: tags so adapter.send() → extract_media() picks them up + content = message or "" + if media_files: + tags = [] + for item in media_files: + if isinstance(item, tuple): + path, is_voice = item + tags.append(f"VOICE:{path}" if is_voice else f"MEDIA:{path}") + else: + tags.append(f"MEDIA:{item}") + content = "\n".join(tags) + ("\n" + content if content else "") try: - async with httpx.AsyncClient(timeout=15) as client: - # Step 1: Get access token - token_resp = await client.post( - "https://bots.qq.com/app/getAppAccessToken", - json={"appId": str(appid), "clientSecret": str(secret)}, - ) - if token_resp.status_code != 200: - return _error(f"QQBot token request failed: {token_resp.status_code}") - token_data = token_resp.json() - access_token = token_data.get("access_token") - if not access_token: - return _error(f"QQBot: no access_token in response") - - # Step 2: Send message via REST - # QQ Bot API has separate endpoints for channels, C2C, and groups. - # We try them in order: channel first, then fallback to C2C. - headers = { - "Authorization": f"QQBot {access_token}", - "Content-Type": "application/json", - } - payload = {"content": message[:4000], "msg_type": 0} - - # Try channel endpoint first (works for guild channels) - url = f"https://api.sgroup.qq.com/channels/{chat_id}/messages" - resp = await client.post(url, json=payload, headers=headers) - if resp.status_code in {200, 201}: - data = resp.json() - return {"success": True, "platform": "qqbot", "chat_id": chat_id, - "message_id": data.get("id")} - - # If channel endpoint failed (likely "频道不存在"), try C2C endpoint - url_c2c = f"https://api.sgroup.qq.com/v2/users/{chat_id}/messages" - resp_c2c = await client.post(url_c2c, json=payload, headers=headers) - if resp_c2c.status_code in {200, 201}: - data = resp_c2c.json() - return {"success": True, "platform": "qqbot", "chat_id": chat_id, - "message_id": data.get("id")} - - # If C2C also failed, try group endpoint - url_group = f"https://api.sgroup.qq.com/v2/groups/{chat_id}/messages" - resp_group = await client.post(url_group, json=payload, headers=headers) - if resp_group.status_code in {200, 201}: - data = resp_group.json() - return {"success": True, "platform": "qqbot", "chat_id": chat_id, - "message_id": data.get("id")} - - # All endpoints failed — return the most informative error - return _error(f"QQBot send failed: channel={resp.status_code} c2c={resp_c2c.status_code} group={resp_group.status_code}") + result = await adapter.send(chat_id, content) + if result and result.success: + return {"success": True, "platform": "qqbot", "chat_id": chat_id, + "message_id": getattr(result, "message_id", None)} + return _error(f"QQBot send failed: {getattr(result, 'error', 'unknown')}") except Exception as e: return _error(f"QQBot send failed: {e}")