| 1 | """Personal WeChat (微信) channel using HTTP long-poll API. |
| 2 | |
| 3 | Uses the ilinkai.weixin.qq.com API for personal WeChat messaging. |
| 4 | No WebSocket, no local WeChat client needed — just HTTP requests with a |
| 5 | bot token obtained via QR code login. |
| 6 | |
| 7 | Protocol reverse-engineered from ``@tencent-weixin/openclaw-weixin`` v1.0.3. |
| 8 | """ |
| 9 | |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import asyncio |
| 13 | import base64 |
| 14 | import hashlib |
| 15 | import json |
| 16 | import os |
| 17 | import random |
| 18 | import re |
| 19 | import time |
| 20 | import uuid |
| 21 | from collections import OrderedDict |
| 22 | from pathlib import Path |
| 23 | from typing import Any |
| 24 | from urllib.parse import quote |
| 25 | |
| 26 | import httpx |
| 27 | from loguru import logger |
| 28 | from pydantic import Field |
| 29 | |
| 30 | from nanobot.bus.events import OutboundMessage |
| 31 | from nanobot.bus.queue import MessageBus |
| 32 | from nanobot.channels.base import BaseChannel |
| 33 | from nanobot.config.paths import get_media_dir, get_runtime_subdir |
| 34 | from nanobot.config.schema import Base |
| 35 | from nanobot.utils.helpers import split_message |
| 36 | |
| 37 | # --------------------------------------------------------------------------- |
| 38 | # Protocol constants (from openclaw-weixin types.ts) |
| 39 | # --------------------------------------------------------------------------- |
| 40 | |
| 41 | # MessageItemType |
| 42 | ITEM_TEXT = 1 |
| 43 | ITEM_IMAGE = 2 |
| 44 | ITEM_VOICE = 3 |
| 45 | ITEM_FILE = 4 |
| 46 | ITEM_VIDEO = 5 |
| 47 | |
| 48 | # MessageType (1 = inbound from user, 2 = outbound from bot) |
| 49 | MESSAGE_TYPE_USER = 1 |
| 50 | MESSAGE_TYPE_BOT = 2 |
| 51 | |
| 52 | # MessageState |
| 53 | MESSAGE_STATE_FINISH = 2 |
| 54 | |
| 55 | WEIXIN_MAX_MESSAGE_LEN = 4000 |
| 56 | WEIXIN_CHANNEL_VERSION = "2.1.1" |
| 57 | ILINK_APP_ID = "bot" |
| 58 | |
| 59 | |
| 60 | def _build_client_version(version: str) -> int: |
| 61 | """Encode semantic version as 0x00MMNNPP (major/minor/patch in one uint32).""" |
| 62 | parts = version.split(".") |
| 63 | |
| 64 | def _as_int(idx: int) -> int: |
| 65 | try: |
| 66 | return int(parts[idx]) |
| 67 | except Exception: |
| 68 | return 0 |
| 69 | |
| 70 | major = _as_int(0) |
| 71 | minor = _as_int(1) |
| 72 | patch = _as_int(2) |
| 73 | return ((major & 0xFF) << 16) | ((minor & 0xFF) << 8) | (patch & 0xFF) |
| 74 | |
| 75 | ILINK_APP_CLIENT_VERSION = _build_client_version(WEIXIN_CHANNEL_VERSION) |
| 76 | BASE_INFO: dict[str, str] = {"channel_version": WEIXIN_CHANNEL_VERSION} |
| 77 | |
| 78 | # Session-expired error code |
| 79 | ERRCODE_SESSION_EXPIRED = -14 |
| 80 | SESSION_PAUSE_DURATION_S = 60 * 60 |
| 81 | |
| 82 | # Retry constants (matching the reference plugin's monitor.ts) |
| 83 | MAX_CONSECUTIVE_FAILURES = 3 |
| 84 | BACKOFF_DELAY_S = 30 |
| 85 | RETRY_DELAY_S = 2 |
| 86 | MAX_QR_REFRESH_COUNT = 3 |
| 87 | TYPING_STATUS_TYPING = 1 |
| 88 | TYPING_STATUS_CANCEL = 2 |
| 89 | TYPING_TICKET_TTL_S = 24 * 60 * 60 |
| 90 | TYPING_KEEPALIVE_INTERVAL_S = 5 |
| 91 | CONFIG_CACHE_INITIAL_RETRY_S = 2 |
| 92 | CONFIG_CACHE_MAX_RETRY_S = 60 * 60 |
| 93 | |
| 94 | # Default long-poll timeout; overridden by server via longpolling_timeout_ms. |
| 95 | DEFAULT_LONG_POLL_TIMEOUT_S = 35 |
| 96 | |
| 97 | # Media-type codes for getuploadurl (1=image, 2=video, 3=file, 4=voice) |
| 98 | UPLOAD_MEDIA_IMAGE = 1 |
| 99 | UPLOAD_MEDIA_VIDEO = 2 |
| 100 | UPLOAD_MEDIA_FILE = 3 |
| 101 | UPLOAD_MEDIA_VOICE = 4 |
| 102 | |
| 103 | # File extensions considered as images / videos for outbound media |
| 104 | _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff", ".ico", ".svg"} |
| 105 | _VIDEO_EXTS = {".mp4", ".avi", ".mov", ".mkv", ".webm", ".flv"} |
| 106 | _VOICE_EXTS = {".mp3", ".wav", ".amr", ".silk", ".ogg", ".m4a", ".aac", ".flac"} |
| 107 | |
| 108 | |
| 109 | def _has_downloadable_media_locator(media: dict[str, Any] | None) -> bool: |
| 110 | if not isinstance(media, dict): |
| 111 | return False |
| 112 | return bool(str(media.get("encrypt_query_param", "") or "") or str(media.get("full_url", "") or "").strip()) |
| 113 | |
| 114 | |
| 115 | class WeixinConfig(Base): |
| 116 | """Personal WeChat channel configuration.""" |
| 117 | |
| 118 | enabled: bool = False |
| 119 | allow_from: list[str] = Field(default_factory=list) |
| 120 | base_url: str = "https://ilinkai.weixin.qq.com" |
| 121 | cdn_base_url: str = "https://novac2c.cdn.weixin.qq.com/c2c" |
| 122 | route_tag: str | int | None = None |
| 123 | token: str = "" # Manually set token, or obtained via QR login |
| 124 | state_dir: str = "" # Default: ~/.nanobot/weixin/ |
| 125 | poll_timeout: int = DEFAULT_LONG_POLL_TIMEOUT_S # seconds for long-poll |
| 126 | |
| 127 | |
| 128 | class WeixinChannel(BaseChannel): |
| 129 | """ |
| 130 | Personal WeChat channel using HTTP long-poll. |
| 131 | |
| 132 | Connects to ilinkai.weixin.qq.com API to receive and send personal |
| 133 | WeChat messages. Authentication is via QR code login which produces |
| 134 | a bot token. |
| 135 | """ |
| 136 | |
| 137 | name = "weixin" |
| 138 | display_name = "WeChat" |
| 139 | |
| 140 | @classmethod |
| 141 | def default_config(cls) -> dict[str, Any]: |
| 142 | return WeixinConfig().model_dump(by_alias=True) |
| 143 | |
| 144 | def __init__(self, config: Any, bus: MessageBus): |
| 145 | if isinstance(config, dict): |
| 146 | config = WeixinConfig.model_validate(config) |
| 147 | super().__init__(config, bus) |
| 148 | self.config: WeixinConfig = config |
| 149 | |
| 150 | # State |
| 151 | self._client: httpx.AsyncClient | None = None |
| 152 | self._get_updates_buf: str = "" |
| 153 | self._context_tokens: dict[str, str] = {} # from_user_id -> context_token |
| 154 | self._processed_ids: OrderedDict[str, None] = OrderedDict() |
| 155 | self._state_dir: Path | None = None |
| 156 | self._token: str = "" |
| 157 | self._poll_task: asyncio.Task | None = None |
| 158 | self._next_poll_timeout_s: int = DEFAULT_LONG_POLL_TIMEOUT_S |
| 159 | self._session_pause_until: float = 0.0 |
| 160 | self._typing_tasks: dict[str, asyncio.Task] = {} |
| 161 | self._typing_tickets: dict[str, dict[str, Any]] = {} |
| 162 | |
| 163 | # ------------------------------------------------------------------ |
| 164 | # State persistence |
| 165 | # ------------------------------------------------------------------ |
| 166 | |
| 167 | def _get_state_dir(self) -> Path: |
| 168 | if self._state_dir: |
| 169 | return self._state_dir |
| 170 | if self.config.state_dir: |
| 171 | d = Path(self.config.state_dir).expanduser() |
| 172 | else: |
| 173 | d = get_runtime_subdir("weixin") |
| 174 | d.mkdir(parents=True, exist_ok=True) |
| 175 | self._state_dir = d |
| 176 | return d |
| 177 | |
| 178 | def _load_state(self) -> bool: |
| 179 | """Load saved account state. Returns True if a valid token was found.""" |
| 180 | state_file = self._get_state_dir() / "account.json" |
| 181 | if not state_file.exists(): |
| 182 | return False |
| 183 | try: |
| 184 | data = json.loads(state_file.read_text()) |
| 185 | self._token = data.get("token", "") |
| 186 | self._get_updates_buf = data.get("get_updates_buf", "") |
| 187 | context_tokens = data.get("context_tokens", {}) |
| 188 | if isinstance(context_tokens, dict): |
| 189 | self._context_tokens = { |
| 190 | str(user_id): str(token) |
| 191 | for user_id, token in context_tokens.items() |
| 192 | if str(user_id).strip() and str(token).strip() |
| 193 | } |
| 194 | else: |
| 195 | self._context_tokens = {} |
| 196 | typing_tickets = data.get("typing_tickets", {}) |
| 197 | if isinstance(typing_tickets, dict): |
| 198 | self._typing_tickets = { |
| 199 | str(user_id): ticket |
| 200 | for user_id, ticket in typing_tickets.items() |
| 201 | if str(user_id).strip() and isinstance(ticket, dict) |
| 202 | } |
| 203 | else: |
| 204 | self._typing_tickets = {} |
| 205 | base_url = data.get("base_url", "") |
| 206 | if base_url: |
| 207 | self.config.base_url = base_url |
| 208 | return bool(self._token) |
| 209 | except Exception: |
| 210 | return False |
| 211 | |
| 212 | def _save_state(self) -> None: |
| 213 | state_file = self._get_state_dir() / "account.json" |
| 214 | try: |
| 215 | data = { |
| 216 | "token": self._token, |
| 217 | "get_updates_buf": self._get_updates_buf, |
| 218 | "context_tokens": self._context_tokens, |
| 219 | "typing_tickets": self._typing_tickets, |
| 220 | "base_url": self.config.base_url, |
| 221 | } |
| 222 | state_file.write_text(json.dumps(data, ensure_ascii=False)) |
| 223 | except Exception: |
| 224 | pass |
| 225 | |
| 226 | # ------------------------------------------------------------------ |
| 227 | # HTTP helpers (matches api.ts buildHeaders / apiFetch) |
| 228 | # ------------------------------------------------------------------ |
| 229 | |
| 230 | @staticmethod |
| 231 | def _random_wechat_uin() -> str: |
| 232 | """X-WECHAT-UIN: random uint32 → decimal string → base64. |
| 233 | |
| 234 | Matches the reference plugin's ``randomWechatUin()`` in api.ts. |
| 235 | Generated fresh for **every** request (same as reference). |
| 236 | """ |
| 237 | uint32 = int.from_bytes(os.urandom(4), "big") |
| 238 | return base64.b64encode(str(uint32).encode()).decode() |
| 239 | |
| 240 | def _make_headers(self, *, auth: bool = True) -> dict[str, str]: |
| 241 | """Build per-request headers (new UIN each call, matching reference).""" |
| 242 | headers: dict[str, str] = { |
| 243 | "X-WECHAT-UIN": self._random_wechat_uin(), |
| 244 | "Content-Type": "application/json", |
| 245 | "AuthorizationType": "ilink_bot_token", |
| 246 | "iLink-App-Id": ILINK_APP_ID, |
| 247 | "iLink-App-ClientVersion": str(ILINK_APP_CLIENT_VERSION), |
| 248 | } |
| 249 | if auth and self._token: |
| 250 | headers["Authorization"] = f"Bearer {self._token}" |
| 251 | if self.config.route_tag is not None and str(self.config.route_tag).strip(): |
| 252 | headers["SKRouteTag"] = str(self.config.route_tag).strip() |
| 253 | return headers |
| 254 | |
| 255 | @staticmethod |
| 256 | def _is_retryable_media_download_error(err: Exception) -> bool: |
| 257 | if isinstance(err, httpx.TimeoutException | httpx.TransportError): |
| 258 | return True |
| 259 | if isinstance(err, httpx.HTTPStatusError): |
| 260 | status_code = err.response.status_code if err.response is not None else 0 |
| 261 | return status_code >= 500 |
| 262 | return False |
| 263 | |
| 264 | async def _api_get( |
| 265 | self, |
| 266 | endpoint: str, |
| 267 | params: dict | None = None, |
| 268 | *, |
| 269 | auth: bool = True, |
| 270 | extra_headers: dict[str, str] | None = None, |
| 271 | ) -> dict: |
| 272 | assert self._client is not None |
| 273 | url = f"{self.config.base_url}/{endpoint}" |
| 274 | hdrs = self._make_headers(auth=auth) |
| 275 | if extra_headers: |
| 276 | hdrs.update(extra_headers) |
| 277 | resp = await self._client.get(url, params=params, headers=hdrs) |
| 278 | resp.raise_for_status() |
| 279 | return resp.json() |
| 280 | |
| 281 | async def _api_get_with_base( |
| 282 | self, |
| 283 | *, |
| 284 | base_url: str, |
| 285 | endpoint: str, |
| 286 | params: dict | None = None, |
| 287 | auth: bool = True, |
| 288 | extra_headers: dict[str, str] | None = None, |
| 289 | ) -> dict: |
| 290 | """GET helper that allows overriding base_url for QR redirect polling.""" |
| 291 | assert self._client is not None |
| 292 | url = f"{base_url.rstrip('/')}/{endpoint}" |
| 293 | hdrs = self._make_headers(auth=auth) |
| 294 | if extra_headers: |
| 295 | hdrs.update(extra_headers) |
| 296 | resp = await self._client.get(url, params=params, headers=hdrs) |
| 297 | resp.raise_for_status() |
| 298 | return resp.json() |
| 299 | |
| 300 | async def _api_post( |
| 301 | self, |
| 302 | endpoint: str, |
| 303 | body: dict | None = None, |
| 304 | *, |
| 305 | auth: bool = True, |
| 306 | ) -> dict: |
| 307 | assert self._client is not None |
| 308 | url = f"{self.config.base_url}/{endpoint}" |
| 309 | payload = body or {} |
| 310 | if "base_info" not in payload: |
| 311 | payload["base_info"] = BASE_INFO |
| 312 | resp = await self._client.post(url, json=payload, headers=self._make_headers(auth=auth)) |
| 313 | resp.raise_for_status() |
| 314 | return resp.json() |
| 315 | |
| 316 | # ------------------------------------------------------------------ |
| 317 | # QR Code Login (matches login-qr.ts) |
| 318 | # ------------------------------------------------------------------ |
| 319 | |
| 320 | async def _fetch_qr_code(self) -> tuple[str, str]: |
| 321 | """Fetch a fresh QR code. Returns (qrcode_id, scan_url).""" |
| 322 | data = await self._api_get( |
| 323 | "ilink/bot/get_bot_qrcode", |
| 324 | params={"bot_type": "3"}, |
| 325 | auth=False, |
| 326 | ) |
| 327 | qrcode_img_content = data.get("qrcode_img_content", "") |
| 328 | qrcode_id = data.get("qrcode", "") |
| 329 | if not qrcode_id: |
| 330 | raise RuntimeError(f"Failed to get QR code from WeChat API: {data}") |
| 331 | return qrcode_id, (qrcode_img_content or qrcode_id) |
| 332 | |
| 333 | async def _qr_login(self) -> bool: |
| 334 | """Perform QR code login flow. Returns True on success.""" |
| 335 | try: |
| 336 | refresh_count = 0 |
| 337 | qrcode_id, scan_url = await self._fetch_qr_code() |
| 338 | self._print_qr_code(scan_url) |
| 339 | current_poll_base_url = self.config.base_url |
| 340 | |
| 341 | while self._running: |
| 342 | try: |
| 343 | status_data = await self._api_get_with_base( |
| 344 | base_url=current_poll_base_url, |
| 345 | endpoint="ilink/bot/get_qrcode_status", |
| 346 | params={"qrcode": qrcode_id}, |
| 347 | auth=False, |
| 348 | ) |
| 349 | except Exception as e: |
| 350 | if self._is_retryable_qr_poll_error(e): |
| 351 | await asyncio.sleep(1) |
| 352 | continue |
| 353 | raise |
| 354 | |
| 355 | if not isinstance(status_data, dict): |
| 356 | await asyncio.sleep(1) |
| 357 | continue |
| 358 | |
| 359 | status = status_data.get("status", "") |
| 360 | if status == "confirmed": |
| 361 | token = status_data.get("bot_token", "") |
| 362 | bot_id = status_data.get("ilink_bot_id", "") |
| 363 | base_url = status_data.get("baseurl", "") |
| 364 | user_id = status_data.get("ilink_user_id", "") |
| 365 | if token: |
| 366 | self._token = token |
| 367 | if base_url: |
| 368 | self.config.base_url = base_url |
| 369 | self._save_state() |
| 370 | logger.info( |
| 371 | "WeChat login successful! bot_id={} user_id={}", |
| 372 | bot_id, |
| 373 | user_id, |
| 374 | ) |
| 375 | return True |
| 376 | else: |
| 377 | logger.error("Login confirmed but no bot_token in response") |
| 378 | return False |
| 379 | elif status == "scaned_but_redirect": |
| 380 | redirect_host = str(status_data.get("redirect_host", "") or "").strip() |
| 381 | if redirect_host: |
| 382 | if redirect_host.startswith("http://") or redirect_host.startswith("https://"): |
| 383 | redirected_base = redirect_host |
| 384 | else: |
| 385 | redirected_base = f"https://{redirect_host}" |
| 386 | if redirected_base != current_poll_base_url: |
| 387 | current_poll_base_url = redirected_base |
| 388 | elif status == "expired": |
| 389 | refresh_count += 1 |
| 390 | if refresh_count > MAX_QR_REFRESH_COUNT: |
| 391 | logger.warning( |
| 392 | "QR code expired too many times ({}/{}), giving up.", |
| 393 | refresh_count - 1, |
| 394 | MAX_QR_REFRESH_COUNT, |
| 395 | ) |
| 396 | return False |
| 397 | qrcode_id, scan_url = await self._fetch_qr_code() |
| 398 | current_poll_base_url = self.config.base_url |
| 399 | self._print_qr_code(scan_url) |
| 400 | continue |
| 401 | # status == "wait" — keep polling |
| 402 | |
| 403 | await asyncio.sleep(1) |
| 404 | |
| 405 | except Exception as e: |
| 406 | logger.error("WeChat QR login failed: {}", e) |
| 407 | |
| 408 | return False |
| 409 | |
| 410 | @staticmethod |
| 411 | def _is_retryable_qr_poll_error(err: Exception) -> bool: |
| 412 | if isinstance(err, httpx.TimeoutException | httpx.TransportError): |
| 413 | return True |
| 414 | if isinstance(err, httpx.HTTPStatusError): |
| 415 | status_code = err.response.status_code if err.response is not None else 0 |
| 416 | if status_code >= 500: |
| 417 | return True |
| 418 | return False |
| 419 | |
| 420 | @staticmethod |
| 421 | def _print_qr_code(url: str) -> None: |
| 422 | try: |
| 423 | import qrcode as qr_lib |
| 424 | |
| 425 | qr = qr_lib.QRCode(border=1) |
| 426 | qr.add_data(url) |
| 427 | qr.make(fit=True) |
| 428 | qr.print_ascii(invert=True) |
| 429 | except ImportError: |
| 430 | print(f"\nLogin URL: {url}\n") |
| 431 | |
| 432 | # ------------------------------------------------------------------ |
| 433 | # Channel lifecycle |
| 434 | # ------------------------------------------------------------------ |
| 435 | |
| 436 | async def login(self, force: bool = False) -> bool: |
| 437 | """Perform QR code login and save token. Returns True on success.""" |
| 438 | if force: |
| 439 | self._token = "" |
| 440 | self._get_updates_buf = "" |
| 441 | state_file = self._get_state_dir() / "account.json" |
| 442 | if state_file.exists(): |
| 443 | state_file.unlink() |
| 444 | if self._token or self._load_state(): |
| 445 | return True |
| 446 | |
| 447 | # Initialize HTTP client for the login flow |
| 448 | self._client = httpx.AsyncClient( |
| 449 | timeout=httpx.Timeout(60, connect=30), |
| 450 | follow_redirects=True, |
| 451 | ) |
| 452 | self._running = True # Enable polling loop in _qr_login() |
| 453 | try: |
| 454 | return await self._qr_login() |
| 455 | finally: |
| 456 | self._running = False |
| 457 | if self._client: |
| 458 | await self._client.aclose() |
| 459 | self._client = None |
| 460 | |
| 461 | async def start(self) -> None: |
| 462 | self._running = True |
| 463 | self._next_poll_timeout_s = self.config.poll_timeout |
| 464 | self._client = httpx.AsyncClient( |
| 465 | timeout=httpx.Timeout(self._next_poll_timeout_s + 10, connect=30), |
| 466 | follow_redirects=True, |
| 467 | ) |
| 468 | |
| 469 | if self.config.token: |
| 470 | self._token = self.config.token |
| 471 | elif not self._load_state(): |
| 472 | if not await self._qr_login(): |
| 473 | logger.error("WeChat login failed. Run 'nanobot channels login weixin' to authenticate.") |
| 474 | self._running = False |
| 475 | return |
| 476 | |
| 477 | logger.info("WeChat channel starting with long-poll...") |
| 478 | |
| 479 | consecutive_failures = 0 |
| 480 | while self._running: |
| 481 | try: |
| 482 | await self._poll_once() |
| 483 | consecutive_failures = 0 |
| 484 | except httpx.TimeoutException: |
| 485 | # Normal for long-poll, just retry |
| 486 | continue |
| 487 | except Exception: |
| 488 | if not self._running: |
| 489 | break |
| 490 | consecutive_failures += 1 |
| 491 | if consecutive_failures >= MAX_CONSECUTIVE_FAILURES: |
| 492 | consecutive_failures = 0 |
| 493 | await asyncio.sleep(BACKOFF_DELAY_S) |
| 494 | else: |
| 495 | await asyncio.sleep(RETRY_DELAY_S) |
| 496 | |
| 497 | async def stop(self) -> None: |
| 498 | self._running = False |
| 499 | if self._poll_task and not self._poll_task.done(): |
| 500 | self._poll_task.cancel() |
| 501 | for chat_id in list(self._typing_tasks): |
| 502 | await self._stop_typing(chat_id, clear_remote=False) |
| 503 | if self._client: |
| 504 | await self._client.aclose() |
| 505 | self._client = None |
| 506 | self._save_state() |
| 507 | # ------------------------------------------------------------------ |
| 508 | # Polling (matches monitor.ts monitorWeixinProvider) |
| 509 | # ------------------------------------------------------------------ |
| 510 | |
| 511 | def _pause_session(self, duration_s: int = SESSION_PAUSE_DURATION_S) -> None: |
| 512 | self._session_pause_until = time.time() + duration_s |
| 513 | |
| 514 | def _session_pause_remaining_s(self) -> int: |
| 515 | remaining = int(self._session_pause_until - time.time()) |
| 516 | if remaining <= 0: |
| 517 | self._session_pause_until = 0.0 |
| 518 | return 0 |
| 519 | return remaining |
| 520 | |
| 521 | def _assert_session_active(self) -> None: |
| 522 | remaining = self._session_pause_remaining_s() |
| 523 | if remaining > 0: |
| 524 | remaining_min = max((remaining + 59) // 60, 1) |
| 525 | raise RuntimeError( |
| 526 | f"WeChat session paused, {remaining_min} min remaining (errcode {ERRCODE_SESSION_EXPIRED})" |
| 527 | ) |
| 528 | |
| 529 | async def _poll_once(self) -> None: |
| 530 | remaining = self._session_pause_remaining_s() |
| 531 | if remaining > 0: |
| 532 | await asyncio.sleep(remaining) |
| 533 | return |
| 534 | |
| 535 | body: dict[str, Any] = { |
| 536 | "get_updates_buf": self._get_updates_buf, |
| 537 | "base_info": BASE_INFO, |
| 538 | } |
| 539 | |
| 540 | # Adjust httpx timeout to match the current poll timeout |
| 541 | assert self._client is not None |
| 542 | self._client.timeout = httpx.Timeout(self._next_poll_timeout_s + 10, connect=30) |
| 543 | |
| 544 | data = await self._api_post("ilink/bot/getupdates", body) |
| 545 | |
| 546 | # Check for API-level errors (monitor.ts checks both ret and errcode) |
| 547 | ret = data.get("ret", 0) |
| 548 | errcode = data.get("errcode", 0) |
| 549 | is_error = (ret is not None and ret != 0) or (errcode is not None and errcode != 0) |
| 550 | |
| 551 | if is_error: |
| 552 | if errcode == ERRCODE_SESSION_EXPIRED or ret == ERRCODE_SESSION_EXPIRED: |
| 553 | self._pause_session() |
| 554 | remaining = self._session_pause_remaining_s() |
| 555 | logger.warning( |
| 556 | "WeChat session expired (errcode {}). Pausing {} min.", |
| 557 | errcode, |
| 558 | max((remaining + 59) // 60, 1), |
| 559 | ) |
| 560 | return |
| 561 | raise RuntimeError( |
| 562 | f"getUpdates failed: ret={ret} errcode={errcode} errmsg={data.get('errmsg', '')}" |
| 563 | ) |
| 564 | |
| 565 | # Honour server-suggested poll timeout (monitor.ts:102-105) |
| 566 | server_timeout_ms = data.get("longpolling_timeout_ms") |
| 567 | if server_timeout_ms and server_timeout_ms > 0: |
| 568 | self._next_poll_timeout_s = max(server_timeout_ms // 1000, 5) |
| 569 | |
| 570 | # Update cursor |
| 571 | new_buf = data.get("get_updates_buf", "") |
| 572 | if new_buf: |
| 573 | self._get_updates_buf = new_buf |
| 574 | self._save_state() |
| 575 | |
| 576 | # Process messages (WeixinMessage[] from types.ts) |
| 577 | msgs: list[dict] = data.get("msgs", []) or [] |
| 578 | for msg in msgs: |
| 579 | try: |
| 580 | await self._process_message(msg) |
| 581 | except Exception: |
| 582 | pass |
| 583 | |
| 584 | # ------------------------------------------------------------------ |
| 585 | # Inbound message processing (matches inbound.ts + process-message.ts) |
| 586 | # ------------------------------------------------------------------ |
| 587 | |
| 588 | async def _process_message(self, msg: dict) -> None: |
| 589 | """Process a single WeixinMessage from getUpdates.""" |
| 590 | # Skip bot's own messages (message_type 2 = BOT) |
| 591 | if msg.get("message_type") == MESSAGE_TYPE_BOT: |
| 592 | return |
| 593 | |
| 594 | # Deduplication by message_id |
| 595 | msg_id = str(msg.get("message_id", "") or msg.get("seq", "")) |
| 596 | if not msg_id: |
| 597 | msg_id = f"{msg.get('from_user_id', '')}_{msg.get('create_time_ms', '')}" |
| 598 | if msg_id in self._processed_ids: |
| 599 | return |
| 600 | self._processed_ids[msg_id] = None |
| 601 | while len(self._processed_ids) > 1000: |
| 602 | self._processed_ids.popitem(last=False) |
| 603 | |
| 604 | from_user_id = msg.get("from_user_id", "") or "" |
| 605 | if not from_user_id: |
| 606 | return |
| 607 | |
| 608 | # Cache context_token (required for all replies — inbound.ts:23-27) |
| 609 | ctx_token = msg.get("context_token", "") |
| 610 | if ctx_token: |
| 611 | self._context_tokens[from_user_id] = ctx_token |
| 612 | self._save_state() |
| 613 | |
| 614 | # Parse item_list (WeixinMessage.item_list — types.ts:161) |
| 615 | item_list: list[dict] = msg.get("item_list") or [] |
| 616 | content_parts: list[str] = [] |
| 617 | media_paths: list[str] = [] |
| 618 | has_top_level_downloadable_media = False |
| 619 | |
| 620 | for item in item_list: |
| 621 | item_type = item.get("type", 0) |
| 622 | |
| 623 | if item_type == ITEM_TEXT: |
| 624 | text = (item.get("text_item") or {}).get("text", "") |
| 625 | if text: |
| 626 | # Handle quoted/ref messages (inbound.ts:86-98) |
| 627 | ref = item.get("ref_msg") |
| 628 | if ref: |
| 629 | ref_item = ref.get("message_item") |
| 630 | # If quoted message is media, just pass the text |
| 631 | if ref_item and ref_item.get("type", 0) in ( |
| 632 | ITEM_IMAGE, |
| 633 | ITEM_VOICE, |
| 634 | ITEM_FILE, |
| 635 | ITEM_VIDEO, |
| 636 | ): |
| 637 | content_parts.append(text) |
| 638 | else: |
| 639 | parts: list[str] = [] |
| 640 | if ref.get("title"): |
| 641 | parts.append(ref["title"]) |
| 642 | if ref_item: |
| 643 | ref_text = (ref_item.get("text_item") or {}).get("text", "") |
| 644 | if ref_text: |
| 645 | parts.append(ref_text) |
| 646 | if parts: |
| 647 | content_parts.append(f"[引用: {' | '.join(parts)}]\n{text}") |
| 648 | else: |
| 649 | content_parts.append(text) |
| 650 | else: |
| 651 | content_parts.append(text) |
| 652 | |
| 653 | elif item_type == ITEM_IMAGE: |
| 654 | image_item = item.get("image_item") or {} |
| 655 | if _has_downloadable_media_locator(image_item.get("media")): |
| 656 | has_top_level_downloadable_media = True |
| 657 | file_path = await self._download_media_item(image_item, "image") |
| 658 | if file_path: |
| 659 | content_parts.append(f"[image]\n[Image: source: {file_path}]") |
| 660 | media_paths.append(file_path) |
| 661 | else: |
| 662 | content_parts.append("[image]") |
| 663 | |
| 664 | elif item_type == ITEM_VOICE: |
| 665 | voice_item = item.get("voice_item") or {} |
| 666 | # Voice-to-text provided by WeChat (inbound.ts:101-103) |
| 667 | voice_text = voice_item.get("text", "") |
| 668 | if voice_text: |
| 669 | content_parts.append(f"[voice] {voice_text}") |
| 670 | else: |
| 671 | if _has_downloadable_media_locator(voice_item.get("media")): |
| 672 | has_top_level_downloadable_media = True |
| 673 | file_path = await self._download_media_item(voice_item, "voice") |
| 674 | if file_path: |
| 675 | transcription = await self.transcribe_audio(file_path) |
| 676 | if transcription: |
| 677 | content_parts.append(f"[voice] {transcription}") |
| 678 | else: |
| 679 | content_parts.append(f"[voice]\n[Audio: source: {file_path}]") |
| 680 | media_paths.append(file_path) |
| 681 | else: |
| 682 | content_parts.append("[voice]") |
| 683 | |
| 684 | elif item_type == ITEM_FILE: |
| 685 | file_item = item.get("file_item") or {} |
| 686 | if _has_downloadable_media_locator(file_item.get("media")): |
| 687 | has_top_level_downloadable_media = True |
| 688 | file_name = file_item.get("file_name", "unknown") |
| 689 | file_path = await self._download_media_item( |
| 690 | file_item, |
| 691 | "file", |
| 692 | file_name, |
| 693 | ) |
| 694 | if file_path: |
| 695 | content_parts.append(f"[file: {file_name}]\n[File: source: {file_path}]") |
| 696 | media_paths.append(file_path) |
| 697 | else: |
| 698 | content_parts.append(f"[file: {file_name}]") |
| 699 | |
| 700 | elif item_type == ITEM_VIDEO: |
| 701 | video_item = item.get("video_item") or {} |
| 702 | if _has_downloadable_media_locator(video_item.get("media")): |
| 703 | has_top_level_downloadable_media = True |
| 704 | file_path = await self._download_media_item(video_item, "video") |
| 705 | if file_path: |
| 706 | content_parts.append(f"[video]\n[Video: source: {file_path}]") |
| 707 | media_paths.append(file_path) |
| 708 | else: |
| 709 | content_parts.append("[video]") |
| 710 | |
| 711 | # Fallback: when no top-level media was downloaded, try quoted/referenced media. |
| 712 | # This aligns with the reference plugin behavior that checks ref_msg.message_item |
| 713 | # when main item_list has no downloadable media. |
| 714 | if not media_paths and not has_top_level_downloadable_media: |
| 715 | ref_media_item: dict[str, Any] | None = None |
| 716 | for item in item_list: |
| 717 | if item.get("type", 0) != ITEM_TEXT: |
| 718 | continue |
| 719 | ref = item.get("ref_msg") or {} |
| 720 | candidate = ref.get("message_item") or {} |
| 721 | if candidate.get("type", 0) in (ITEM_IMAGE, ITEM_VOICE, ITEM_FILE, ITEM_VIDEO): |
| 722 | ref_media_item = candidate |
| 723 | break |
| 724 | |
| 725 | if ref_media_item: |
| 726 | ref_type = ref_media_item.get("type", 0) |
| 727 | if ref_type == ITEM_IMAGE: |
| 728 | image_item = ref_media_item.get("image_item") or {} |
| 729 | file_path = await self._download_media_item(image_item, "image") |
| 730 | if file_path: |
| 731 | content_parts.append(f"[image]\n[Image: source: {file_path}]") |
| 732 | media_paths.append(file_path) |
| 733 | elif ref_type == ITEM_VOICE: |
| 734 | voice_item = ref_media_item.get("voice_item") or {} |
| 735 | file_path = await self._download_media_item(voice_item, "voice") |
| 736 | if file_path: |
| 737 | transcription = await self.transcribe_audio(file_path) |
| 738 | if transcription: |
| 739 | content_parts.append(f"[voice] {transcription}") |
| 740 | else: |
| 741 | content_parts.append(f"[voice]\n[Audio: source: {file_path}]") |
| 742 | media_paths.append(file_path) |
| 743 | elif ref_type == ITEM_FILE: |
| 744 | file_item = ref_media_item.get("file_item") or {} |
| 745 | file_name = file_item.get("file_name", "unknown") |
| 746 | file_path = await self._download_media_item(file_item, "file", file_name) |
| 747 | if file_path: |
| 748 | content_parts.append(f"[file: {file_name}]\n[File: source: {file_path}]") |
| 749 | media_paths.append(file_path) |
| 750 | elif ref_type == ITEM_VIDEO: |
| 751 | video_item = ref_media_item.get("video_item") or {} |
| 752 | file_path = await self._download_media_item(video_item, "video") |
| 753 | if file_path: |
| 754 | content_parts.append(f"[video]\n[Video: source: {file_path}]") |
| 755 | media_paths.append(file_path) |
| 756 | |
| 757 | content = "\n".join(content_parts) |
| 758 | if not content: |
| 759 | return |
| 760 | |
| 761 | logger.info( |
| 762 | "WeChat inbound: from={} items={} bodyLen={}", |
| 763 | from_user_id, |
| 764 | ",".join(str(i.get("type", 0)) for i in item_list), |
| 765 | len(content), |
| 766 | ) |
| 767 | |
| 768 | await self._start_typing(from_user_id, ctx_token) |
| 769 | |
| 770 | await self._handle_message( |
| 771 | sender_id=from_user_id, |
| 772 | chat_id=from_user_id, |
| 773 | content=content, |
| 774 | media=media_paths or None, |
| 775 | metadata={"message_id": msg_id}, |
| 776 | ) |
| 777 | |
| 778 | # ------------------------------------------------------------------ |
| 779 | # Media download (matches media-download.ts + pic-decrypt.ts) |
| 780 | # ------------------------------------------------------------------ |
| 781 | |
| 782 | async def _download_media_item( |
| 783 | self, |
| 784 | typed_item: dict, |
| 785 | media_type: str, |
| 786 | filename: str | None = None, |
| 787 | ) -> str | None: |
| 788 | """Download + AES-decrypt a media item. Returns local path or None.""" |
| 789 | try: |
| 790 | media = typed_item.get("media") or {} |
| 791 | encrypt_query_param = str(media.get("encrypt_query_param", "") or "") |
| 792 | full_url = str(media.get("full_url", "") or "").strip() |
| 793 | |
| 794 | if not encrypt_query_param and not full_url: |
| 795 | return None |
| 796 | |
| 797 | # Resolve AES key (media-download.ts:43-45, pic-decrypt.ts:40-52) |
| 798 | # image_item.aeskey is a raw hex string (16 bytes as 32 hex chars). |
| 799 | # media.aes_key is always base64-encoded. |
| 800 | # For images, prefer image_item.aeskey; for others use media.aes_key. |
| 801 | raw_aeskey_hex = typed_item.get("aeskey", "") |
| 802 | media_aes_key_b64 = media.get("aes_key", "") |
| 803 | |
| 804 | aes_key_b64: str = "" |
| 805 | if raw_aeskey_hex: |
| 806 | # Convert hex → raw bytes → base64 (matches media-download.ts:43-44) |
| 807 | aes_key_b64 = base64.b64encode(bytes.fromhex(raw_aeskey_hex)).decode() |
| 808 | elif media_aes_key_b64: |
| 809 | aes_key_b64 = media_aes_key_b64 |
| 810 | |
| 811 | # Reference protocol behavior: VOICE/FILE/VIDEO require aes_key; |
| 812 | # only IMAGE may be downloaded as plain bytes when key is missing. |
| 813 | if media_type != "image" and not aes_key_b64: |
| 814 | return None |
| 815 | |
| 816 | assert self._client is not None |
| 817 | fallback_url = "" |
| 818 | if encrypt_query_param: |
| 819 | fallback_url = ( |
| 820 | f"{self.config.cdn_base_url}/download" |
| 821 | f"?encrypted_query_param={quote(encrypt_query_param)}" |
| 822 | ) |
| 823 | |
| 824 | download_candidates: list[tuple[str, str]] = [] |
| 825 | if full_url: |
| 826 | download_candidates.append(("full_url", full_url)) |
| 827 | if fallback_url and (not full_url or fallback_url != full_url): |
| 828 | download_candidates.append(("encrypt_query_param", fallback_url)) |
| 829 | |
| 830 | data = b"" |
| 831 | for idx, (download_source, cdn_url) in enumerate(download_candidates): |
| 832 | try: |
| 833 | resp = await self._client.get(cdn_url) |
| 834 | resp.raise_for_status() |
| 835 | data = resp.content |
| 836 | break |
| 837 | except Exception as e: |
| 838 | has_more_candidates = idx + 1 < len(download_candidates) |
| 839 | should_fallback = ( |
| 840 | download_source == "full_url" |
| 841 | and has_more_candidates |
| 842 | and self._is_retryable_media_download_error(e) |
| 843 | ) |
| 844 | if should_fallback: |
| 845 | logger.warning( |
| 846 | "WeChat media download failed via full_url, falling back to encrypt_query_param: type={} err={}", |
| 847 | media_type, |
| 848 | e, |
| 849 | ) |
| 850 | continue |
| 851 | raise |
| 852 | |
| 853 | if aes_key_b64 and data: |
| 854 | data = _decrypt_aes_ecb(data, aes_key_b64) |
| 855 | |
| 856 | if not data: |
| 857 | return None |
| 858 | |
| 859 | media_dir = get_media_dir("weixin") |
| 860 | ext = _ext_for_type(media_type) |
| 861 | if not filename: |
| 862 | ts = int(time.time()) |
| 863 | hash_seed = encrypt_query_param or full_url |
| 864 | h = abs(hash(hash_seed)) % 100000 |
| 865 | filename = f"{media_type}_{ts}_{h}{ext}" |
| 866 | safe_name = os.path.basename(filename) |
| 867 | file_path = media_dir / safe_name |
| 868 | file_path.write_bytes(data) |
| 869 | return str(file_path) |
| 870 | |
| 871 | except Exception as e: |
| 872 | logger.error("Error downloading WeChat media: {}", e) |
| 873 | return None |
| 874 | |
| 875 | # ------------------------------------------------------------------ |
| 876 | # Outbound (matches send.ts buildTextMessageReq + sendMessageWeixin) |
| 877 | # ------------------------------------------------------------------ |
| 878 | |
| 879 | async def _get_typing_ticket(self, user_id: str, context_token: str = "") -> str: |
| 880 | """Get typing ticket with per-user refresh + failure backoff cache.""" |
| 881 | now = time.time() |
| 882 | entry = self._typing_tickets.get(user_id) |
| 883 | if entry and now < float(entry.get("next_fetch_at", 0)): |
| 884 | return str(entry.get("ticket", "") or "") |
| 885 | |
| 886 | body: dict[str, Any] = { |
| 887 | "ilink_user_id": user_id, |
| 888 | "context_token": context_token or None, |
| 889 | "base_info": BASE_INFO, |
| 890 | } |
| 891 | data = await self._api_post("ilink/bot/getconfig", body) |
| 892 | if data.get("ret", 0) == 0: |
| 893 | ticket = str(data.get("typing_ticket", "") or "") |
| 894 | self._typing_tickets[user_id] = { |
| 895 | "ticket": ticket, |
| 896 | "ever_succeeded": True, |
| 897 | "next_fetch_at": now + (random.random() * TYPING_TICKET_TTL_S), |
| 898 | "retry_delay_s": CONFIG_CACHE_INITIAL_RETRY_S, |
| 899 | } |
| 900 | return ticket |
| 901 | |
| 902 | prev_delay = float(entry.get("retry_delay_s", CONFIG_CACHE_INITIAL_RETRY_S)) if entry else CONFIG_CACHE_INITIAL_RETRY_S |
| 903 | next_delay = min(prev_delay * 2, CONFIG_CACHE_MAX_RETRY_S) |
| 904 | if entry: |
| 905 | entry["next_fetch_at"] = now + next_delay |
| 906 | entry["retry_delay_s"] = next_delay |
| 907 | return str(entry.get("ticket", "") or "") |
| 908 | |
| 909 | self._typing_tickets[user_id] = { |
| 910 | "ticket": "", |
| 911 | "ever_succeeded": False, |
| 912 | "next_fetch_at": now + CONFIG_CACHE_INITIAL_RETRY_S, |
| 913 | "retry_delay_s": CONFIG_CACHE_INITIAL_RETRY_S, |
| 914 | } |
| 915 | return "" |
| 916 | |
| 917 | async def _send_typing(self, user_id: str, typing_ticket: str, status: int) -> None: |
| 918 | """Best-effort sendtyping wrapper.""" |
| 919 | if not typing_ticket: |
| 920 | return |
| 921 | body: dict[str, Any] = { |
| 922 | "ilink_user_id": user_id, |
| 923 | "typing_ticket": typing_ticket, |
| 924 | "status": status, |
| 925 | "base_info": BASE_INFO, |
| 926 | } |
| 927 | await self._api_post("ilink/bot/sendtyping", body) |
| 928 | |
| 929 | async def _typing_keepalive_loop(self, user_id: str, typing_ticket: str, stop_event: asyncio.Event) -> None: |
| 930 | try: |
| 931 | while not stop_event.is_set(): |
| 932 | await asyncio.sleep(TYPING_KEEPALIVE_INTERVAL_S) |
| 933 | if stop_event.is_set(): |
| 934 | break |
| 935 | try: |
| 936 | await self._send_typing(user_id, typing_ticket, TYPING_STATUS_TYPING) |
| 937 | except Exception: |
| 938 | pass |
| 939 | finally: |
| 940 | pass |
| 941 | |
| 942 | async def send(self, msg: OutboundMessage) -> None: |
| 943 | if not self._client or not self._token: |
| 944 | logger.warning("WeChat client not initialized or not authenticated") |
| 945 | return |
| 946 | try: |
| 947 | self._assert_session_active() |
| 948 | except RuntimeError: |
| 949 | return |
| 950 | |
| 951 | is_progress = bool((msg.metadata or {}).get("_progress", False)) |
| 952 | if not is_progress: |
| 953 | await self._stop_typing(msg.chat_id, clear_remote=True) |
| 954 | |
| 955 | content = msg.content.strip() |
| 956 | ctx_token = self._context_tokens.get(msg.chat_id, "") |
| 957 | if not ctx_token: |
| 958 | logger.warning( |
| 959 | "WeChat: no context_token for chat_id={}, cannot send", |
| 960 | msg.chat_id, |
| 961 | ) |
| 962 | return |
| 963 | |
| 964 | typing_ticket = "" |
| 965 | try: |
| 966 | typing_ticket = await self._get_typing_ticket(msg.chat_id, ctx_token) |
| 967 | except Exception: |
| 968 | typing_ticket = "" |
| 969 | |
| 970 | if typing_ticket: |
| 971 | try: |
| 972 | await self._send_typing(msg.chat_id, typing_ticket, TYPING_STATUS_TYPING) |
| 973 | except Exception: |
| 974 | pass |
| 975 | |
| 976 | typing_keepalive_stop = asyncio.Event() |
| 977 | typing_keepalive_task: asyncio.Task | None = None |
| 978 | if typing_ticket: |
| 979 | typing_keepalive_task = asyncio.create_task( |
| 980 | self._typing_keepalive_loop(msg.chat_id, typing_ticket, typing_keepalive_stop) |
| 981 | ) |
| 982 | |
| 983 | try: |
| 984 | # --- Send media files first (following Telegram channel pattern) --- |
| 985 | for media_path in (msg.media or []): |
| 986 | try: |
| 987 | await self._send_media_file(msg.chat_id, media_path, ctx_token) |
| 988 | except (httpx.TimeoutException, httpx.TransportError) as net_err: |
| 989 | # Network/transport errors: do NOT fall back to text — |
| 990 | # the text send would also likely fail, and the outer |
| 991 | # except will re-raise so ChannelManager retries properly. |
| 992 | logger.error( |
| 993 | "Network error sending WeChat media {}: {}", |
| 994 | media_path, |
| 995 | net_err, |
| 996 | ) |
| 997 | raise |
| 998 | except httpx.HTTPStatusError as http_err: |
| 999 | status_code = ( |
| 1000 | http_err.response.status_code |
| 1001 | if http_err.response is not None |
| 1002 | else 0 |
| 1003 | ) |
| 1004 | if status_code >= 500: |
| 1005 | # Server-side / retryable HTTP error — same as network. |
| 1006 | logger.error( |
| 1007 | "Server error ({} {}) sending WeChat media {}: {}", |
| 1008 | status_code, |
| 1009 | http_err.response.reason_phrase |
| 1010 | if http_err.response is not None |
| 1011 | else "", |
| 1012 | media_path, |
| 1013 | http_err, |
| 1014 | ) |
| 1015 | raise |
| 1016 | # 4xx client errors are NOT retryable — fall back to text. |
| 1017 | filename = Path(media_path).name |
| 1018 | logger.error("Failed to send WeChat media {}: {}", media_path, http_err) |
| 1019 | await self._send_text( |
| 1020 | msg.chat_id, f"[Failed to send: {filename}]", ctx_token, |
| 1021 | ) |
| 1022 | except Exception as e: |
| 1023 | # Non-network errors (format, file-not-found, etc.): |
| 1024 | # notify the user via text fallback. |
| 1025 | filename = Path(media_path).name |
| 1026 | logger.error("Failed to send WeChat media {}: {}", media_path, e) |
| 1027 | # Notify user about failure via text |
| 1028 | await self._send_text( |
| 1029 | msg.chat_id, f"[Failed to send: {filename}]", ctx_token, |
| 1030 | ) |
| 1031 | |
| 1032 | # --- Send text content --- |
| 1033 | if not content: |
| 1034 | return |
| 1035 | |
| 1036 | chunks = split_message(content, WEIXIN_MAX_MESSAGE_LEN) |
| 1037 | for chunk in chunks: |
| 1038 | await self._send_text(msg.chat_id, chunk, ctx_token) |
| 1039 | except Exception as e: |
| 1040 | logger.error("Error sending WeChat message: {}", e) |
| 1041 | raise |
| 1042 | finally: |
| 1043 | if typing_keepalive_task: |
| 1044 | typing_keepalive_stop.set() |
| 1045 | typing_keepalive_task.cancel() |
| 1046 | try: |
| 1047 | await typing_keepalive_task |
| 1048 | except asyncio.CancelledError: |
| 1049 | pass |
| 1050 | |
| 1051 | if typing_ticket and not is_progress: |
| 1052 | try: |
| 1053 | await self._send_typing(msg.chat_id, typing_ticket, TYPING_STATUS_CANCEL) |
| 1054 | except Exception: |
| 1055 | pass |
| 1056 | |
| 1057 | async def _start_typing(self, chat_id: str, context_token: str = "") -> None: |
| 1058 | """Start typing indicator immediately when a message is received.""" |
| 1059 | if not self._client or not self._token or not chat_id: |
| 1060 | return |
| 1061 | await self._stop_typing(chat_id, clear_remote=False) |
| 1062 | try: |
| 1063 | ticket = await self._get_typing_ticket(chat_id, context_token) |
| 1064 | if not ticket: |
| 1065 | return |
| 1066 | await self._send_typing(chat_id, ticket, TYPING_STATUS_TYPING) |
| 1067 | except Exception as e: |
| 1068 | logger.debug("WeChat typing indicator start failed for {}: {}", chat_id, e) |
| 1069 | return |
| 1070 | |
| 1071 | stop_event = asyncio.Event() |
| 1072 | |
| 1073 | async def keepalive() -> None: |
| 1074 | try: |
| 1075 | while not stop_event.is_set(): |
| 1076 | await asyncio.sleep(TYPING_KEEPALIVE_INTERVAL_S) |
| 1077 | if stop_event.is_set(): |
| 1078 | break |
| 1079 | try: |
| 1080 | await self._send_typing(chat_id, ticket, TYPING_STATUS_TYPING) |
| 1081 | except Exception: |
| 1082 | pass |
| 1083 | finally: |
| 1084 | pass |
| 1085 | |
| 1086 | task = asyncio.create_task(keepalive()) |
| 1087 | task._typing_stop_event = stop_event # type: ignore[attr-defined] |
| 1088 | self._typing_tasks[chat_id] = task |
| 1089 | |
| 1090 | async def _stop_typing(self, chat_id: str, *, clear_remote: bool) -> None: |
| 1091 | """Stop typing indicator for a chat.""" |
| 1092 | task = self._typing_tasks.pop(chat_id, None) |
| 1093 | if task and not task.done(): |
| 1094 | stop_event = getattr(task, "_typing_stop_event", None) |
| 1095 | if stop_event: |
| 1096 | stop_event.set() |
| 1097 | task.cancel() |
| 1098 | try: |
| 1099 | await task |
| 1100 | except asyncio.CancelledError: |
| 1101 | pass |
| 1102 | if not clear_remote: |
| 1103 | return |
| 1104 | entry = self._typing_tickets.get(chat_id) |
| 1105 | ticket = str(entry.get("ticket", "") or "") if isinstance(entry, dict) else "" |
| 1106 | if not ticket: |
| 1107 | return |
| 1108 | try: |
| 1109 | await self._send_typing(chat_id, ticket, TYPING_STATUS_CANCEL) |
| 1110 | except Exception as e: |
| 1111 | logger.debug("WeChat typing clear failed for {}: {}", chat_id, e) |
| 1112 | |
| 1113 | async def _send_text( |
| 1114 | self, |
| 1115 | to_user_id: str, |
| 1116 | text: str, |
| 1117 | context_token: str, |
| 1118 | ) -> None: |
| 1119 | """Send a text message matching the exact protocol from send.ts.""" |
| 1120 | client_id = f"nanobot-{uuid.uuid4().hex[:12]}" |
| 1121 | |
| 1122 | item_list: list[dict] = [] |
| 1123 | if text: |
| 1124 | item_list.append({"type": ITEM_TEXT, "text_item": {"text": text}}) |
| 1125 | |
| 1126 | weixin_msg: dict[str, Any] = { |
| 1127 | "from_user_id": "", |
| 1128 | "to_user_id": to_user_id, |
| 1129 | "client_id": client_id, |
| 1130 | "message_type": MESSAGE_TYPE_BOT, |
| 1131 | "message_state": MESSAGE_STATE_FINISH, |
| 1132 | } |
| 1133 | if item_list: |
| 1134 | weixin_msg["item_list"] = item_list |
| 1135 | if context_token: |
| 1136 | weixin_msg["context_token"] = context_token |
| 1137 | |
| 1138 | body: dict[str, Any] = { |
| 1139 | "msg": weixin_msg, |
| 1140 | "base_info": BASE_INFO, |
| 1141 | } |
| 1142 | |
| 1143 | data = await self._api_post("ilink/bot/sendmessage", body) |
| 1144 | errcode = data.get("errcode", 0) |
| 1145 | if errcode and errcode != 0: |
| 1146 | logger.warning( |
| 1147 | "WeChat send error (code {}): {}", |
| 1148 | errcode, |
| 1149 | data.get("errmsg", ""), |
| 1150 | ) |
| 1151 | |
| 1152 | async def _send_media_file( |
| 1153 | self, |
| 1154 | to_user_id: str, |
| 1155 | media_path: str, |
| 1156 | context_token: str, |
| 1157 | ) -> None: |
| 1158 | """Upload a local file to WeChat CDN and send it as a media message. |
| 1159 | |
| 1160 | Follows the exact protocol from ``@tencent-weixin/openclaw-weixin`` v1.0.3: |
| 1161 | 1. Generate a random 16-byte AES key (client-side). |
| 1162 | 2. Call ``getuploadurl`` with file metadata + hex-encoded AES key. |
| 1163 | 3. AES-128-ECB encrypt the file and POST to CDN (``{cdnBaseUrl}/upload``). |
| 1164 | 4. Read ``x-encrypted-param`` header from CDN response as the download param. |
| 1165 | 5. Send a ``sendmessage`` with the appropriate media item referencing the upload. |
| 1166 | """ |
| 1167 | p = Path(media_path) |
| 1168 | if not p.is_file(): |
| 1169 | raise FileNotFoundError(f"Media file not found: {media_path}") |
| 1170 | |
| 1171 | raw_data = p.read_bytes() |
| 1172 | raw_size = len(raw_data) |
| 1173 | raw_md5 = hashlib.md5(raw_data).hexdigest() |
| 1174 | |
| 1175 | # Determine upload media type from extension |
| 1176 | ext = p.suffix.lower() |
| 1177 | if ext in _IMAGE_EXTS: |
| 1178 | upload_type = UPLOAD_MEDIA_IMAGE |
| 1179 | item_type = ITEM_IMAGE |
| 1180 | item_key = "image_item" |
| 1181 | elif ext in _VIDEO_EXTS: |
| 1182 | upload_type = UPLOAD_MEDIA_VIDEO |
| 1183 | item_type = ITEM_VIDEO |
| 1184 | item_key = "video_item" |
| 1185 | elif ext in _VOICE_EXTS: |
| 1186 | upload_type = UPLOAD_MEDIA_VOICE |
| 1187 | item_type = ITEM_VOICE |
| 1188 | item_key = "voice_item" |
| 1189 | else: |
| 1190 | upload_type = UPLOAD_MEDIA_FILE |
| 1191 | item_type = ITEM_FILE |
| 1192 | item_key = "file_item" |
| 1193 | |
| 1194 | # Generate client-side AES-128 key (16 random bytes) |
| 1195 | aes_key_raw = os.urandom(16) |
| 1196 | aes_key_hex = aes_key_raw.hex() |
| 1197 | |
| 1198 | # Compute encrypted size: PKCS7 padding to 16-byte boundary |
| 1199 | # Matches aesEcbPaddedSize: Math.ceil((size + 1) / 16) * 16 |
| 1200 | padded_size = ((raw_size + 1 + 15) // 16) * 16 |
| 1201 | |
| 1202 | # Step 1: Get upload URL from server (prefer upload_full_url, fallback to upload_param) |
| 1203 | file_key = os.urandom(16).hex() |
| 1204 | upload_body: dict[str, Any] = { |
| 1205 | "filekey": file_key, |
| 1206 | "media_type": upload_type, |
| 1207 | "to_user_id": to_user_id, |
| 1208 | "rawsize": raw_size, |
| 1209 | "rawfilemd5": raw_md5, |
| 1210 | "filesize": padded_size, |
| 1211 | "no_need_thumb": True, |
| 1212 | "aeskey": aes_key_hex, |
| 1213 | } |
| 1214 | |
| 1215 | assert self._client is not None |
| 1216 | upload_resp = await self._api_post("ilink/bot/getuploadurl", upload_body) |
| 1217 | |
| 1218 | upload_full_url = str(upload_resp.get("upload_full_url", "") or "").strip() |
| 1219 | upload_param = str(upload_resp.get("upload_param", "") or "") |
| 1220 | if not upload_full_url and not upload_param: |
| 1221 | raise RuntimeError( |
| 1222 | "getuploadurl returned no upload URL " |
| 1223 | f"(need upload_full_url or upload_param): {upload_resp}" |
| 1224 | ) |
| 1225 | |
| 1226 | # Step 2: AES-128-ECB encrypt and POST to CDN |
| 1227 | aes_key_b64 = base64.b64encode(aes_key_raw).decode() |
| 1228 | encrypted_data = _encrypt_aes_ecb(raw_data, aes_key_b64) |
| 1229 | |
| 1230 | if upload_full_url: |
| 1231 | cdn_upload_url = upload_full_url |
| 1232 | else: |
| 1233 | cdn_upload_url = ( |
| 1234 | f"{self.config.cdn_base_url}/upload" |
| 1235 | f"?encrypted_query_param={quote(upload_param)}" |
| 1236 | f"&filekey={quote(file_key)}" |
| 1237 | ) |
| 1238 | |
| 1239 | cdn_resp = await self._client.post( |
| 1240 | cdn_upload_url, |
| 1241 | content=encrypted_data, |
| 1242 | headers={"Content-Type": "application/octet-stream"}, |
| 1243 | ) |
| 1244 | cdn_resp.raise_for_status() |
| 1245 | |
| 1246 | # The download encrypted_query_param comes from CDN response header |
| 1247 | download_param = cdn_resp.headers.get("x-encrypted-param", "") |
| 1248 | if not download_param: |
| 1249 | raise RuntimeError( |
| 1250 | "CDN upload response missing x-encrypted-param header; " |
| 1251 | f"status={cdn_resp.status_code} headers={dict(cdn_resp.headers)}" |
| 1252 | ) |
| 1253 | |
| 1254 | # Step 3: Send message with the media item |
| 1255 | # aes_key for CDNMedia is the hex key encoded as base64 |
| 1256 | # (matches: Buffer.from(uploaded.aeskey).toString("base64")) |
| 1257 | cdn_aes_key_b64 = base64.b64encode(aes_key_hex.encode()).decode() |
| 1258 | |
| 1259 | media_item: dict[str, Any] = { |
| 1260 | "media": { |
| 1261 | "encrypt_query_param": download_param, |
| 1262 | "aes_key": cdn_aes_key_b64, |
| 1263 | "encrypt_type": 1, |
| 1264 | }, |
| 1265 | } |
| 1266 | |
| 1267 | if item_type == ITEM_IMAGE: |
| 1268 | media_item["mid_size"] = padded_size |
| 1269 | elif item_type == ITEM_VIDEO: |
| 1270 | media_item["video_size"] = padded_size |
| 1271 | elif item_type == ITEM_FILE: |
| 1272 | media_item["file_name"] = p.name |
| 1273 | media_item["len"] = str(raw_size) |
| 1274 | |
| 1275 | # Send each media item as its own message (matching reference plugin) |
| 1276 | client_id = f"nanobot-{uuid.uuid4().hex[:12]}" |
| 1277 | item_list: list[dict] = [{"type": item_type, item_key: media_item}] |
| 1278 | |
| 1279 | weixin_msg: dict[str, Any] = { |
| 1280 | "from_user_id": "", |
| 1281 | "to_user_id": to_user_id, |
| 1282 | "client_id": client_id, |
| 1283 | "message_type": MESSAGE_TYPE_BOT, |
| 1284 | "message_state": MESSAGE_STATE_FINISH, |
| 1285 | "item_list": item_list, |
| 1286 | } |
| 1287 | if context_token: |
| 1288 | weixin_msg["context_token"] = context_token |
| 1289 | |
| 1290 | body: dict[str, Any] = { |
| 1291 | "msg": weixin_msg, |
| 1292 | "base_info": BASE_INFO, |
| 1293 | } |
| 1294 | |
| 1295 | data = await self._api_post("ilink/bot/sendmessage", body) |
| 1296 | errcode = data.get("errcode", 0) |
| 1297 | if errcode and errcode != 0: |
| 1298 | raise RuntimeError( |
| 1299 | f"WeChat send media error (code {errcode}): {data.get('errmsg', '')}" |
| 1300 | ) |
| 1301 | |
| 1302 | |
| 1303 | # --------------------------------------------------------------------------- |
| 1304 | # AES-128-ECB encryption / decryption (matches pic-decrypt.ts / aes-ecb.ts) |
| 1305 | # --------------------------------------------------------------------------- |
| 1306 | |
| 1307 | |
| 1308 | def _parse_aes_key(aes_key_b64: str) -> bytes: |
| 1309 | """Parse a base64-encoded AES key, handling both encodings seen in the wild. |
| 1310 | |
| 1311 | From ``pic-decrypt.ts parseAesKey``: |
| 1312 | |
| 1313 | * ``base64(raw 16 bytes)`` → images (media.aes_key) |
| 1314 | * ``base64(hex string of 16 bytes)`` → file / voice / video |
| 1315 | |
| 1316 | In the second case base64-decoding yields 32 ASCII hex chars which must |
| 1317 | then be parsed as hex to recover the actual 16-byte key. |
| 1318 | """ |
| 1319 | decoded = base64.b64decode(aes_key_b64) |
| 1320 | if len(decoded) == 16: |
| 1321 | return decoded |
| 1322 | if len(decoded) == 32 and re.fullmatch(rb"[0-9a-fA-F]{32}", decoded): |
| 1323 | # hex-encoded key: base64 → hex string → raw bytes |
| 1324 | return bytes.fromhex(decoded.decode("ascii")) |
| 1325 | raise ValueError( |
| 1326 | f"aes_key must decode to 16 raw bytes or 32-char hex string, got {len(decoded)} bytes" |
| 1327 | ) |
| 1328 | |
| 1329 | |
| 1330 | def _encrypt_aes_ecb(data: bytes, aes_key_b64: str) -> bytes: |
| 1331 | """Encrypt data with AES-128-ECB and PKCS7 padding for CDN upload.""" |
| 1332 | try: |
| 1333 | key = _parse_aes_key(aes_key_b64) |
| 1334 | except Exception as e: |
| 1335 | logger.warning("Failed to parse AES key for encryption, sending raw: {}", e) |
| 1336 | return data |
| 1337 | |
| 1338 | # PKCS7 padding |
| 1339 | pad_len = 16 - len(data) % 16 |
| 1340 | padded = data + bytes([pad_len] * pad_len) |
| 1341 | |
| 1342 | try: |
| 1343 | from Crypto.Cipher import AES |
| 1344 | |
| 1345 | cipher = AES.new(key, AES.MODE_ECB) |
| 1346 | return cipher.encrypt(padded) |
| 1347 | except ImportError: |
| 1348 | pass |
| 1349 | |
| 1350 | try: |
| 1351 | from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes |
| 1352 | |
| 1353 | cipher_obj = Cipher(algorithms.AES(key), modes.ECB()) |
| 1354 | encryptor = cipher_obj.encryptor() |
| 1355 | return encryptor.update(padded) + encryptor.finalize() |
| 1356 | except ImportError: |
| 1357 | logger.warning("Cannot encrypt media: install 'pycryptodome' or 'cryptography'") |
| 1358 | return data |
| 1359 | |
| 1360 | |
| 1361 | def _decrypt_aes_ecb(data: bytes, aes_key_b64: str) -> bytes: |
| 1362 | """Decrypt AES-128-ECB media data. |
| 1363 | |
| 1364 | ``aes_key_b64`` is always base64-encoded (caller converts hex keys first). |
| 1365 | """ |
| 1366 | try: |
| 1367 | key = _parse_aes_key(aes_key_b64) |
| 1368 | except Exception as e: |
| 1369 | logger.warning("Failed to parse AES key, returning raw data: {}", e) |
| 1370 | return data |
| 1371 | |
| 1372 | decrypted: bytes | None = None |
| 1373 | |
| 1374 | try: |
| 1375 | from Crypto.Cipher import AES |
| 1376 | |
| 1377 | cipher = AES.new(key, AES.MODE_ECB) |
| 1378 | decrypted = cipher.decrypt(data) |
| 1379 | except ImportError: |
| 1380 | pass |
| 1381 | |
| 1382 | if decrypted is None: |
| 1383 | try: |
| 1384 | from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes |
| 1385 | |
| 1386 | cipher_obj = Cipher(algorithms.AES(key), modes.ECB()) |
| 1387 | decryptor = cipher_obj.decryptor() |
| 1388 | decrypted = decryptor.update(data) + decryptor.finalize() |
| 1389 | except ImportError: |
| 1390 | logger.warning("Cannot decrypt media: install 'pycryptodome' or 'cryptography'") |
| 1391 | return data |
| 1392 | |
| 1393 | return _pkcs7_unpad_safe(decrypted) |
| 1394 | |
| 1395 | |
| 1396 | def _pkcs7_unpad_safe(data: bytes, block_size: int = 16) -> bytes: |
| 1397 | """Safely remove PKCS7 padding when valid; otherwise return original bytes.""" |
| 1398 | if not data: |
| 1399 | return data |
| 1400 | if len(data) % block_size != 0: |
| 1401 | return data |
| 1402 | pad_len = data[-1] |
| 1403 | if pad_len < 1 or pad_len > block_size: |
| 1404 | return data |
| 1405 | if data[-pad_len:] != bytes([pad_len]) * pad_len: |
| 1406 | return data |
| 1407 | return data[:-pad_len] |
| 1408 | |
| 1409 | |
| 1410 | def _ext_for_type(media_type: str) -> str: |
| 1411 | return { |
| 1412 | "image": ".jpg", |
| 1413 | "voice": ".silk", |
| 1414 | "video": ".mp4", |
| 1415 | "file": "", |
| 1416 | }.get(media_type, "") |
| 1417 |