ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-14' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:57:17 -07:00
parent 9834afc084 deab4b731a
commit a76d7e7f86
5 changed files with 800 additions and 1633 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/platforms/whatsapp_cloud.py
+435 -825
View File
File diff suppressed because it is too large Load Diff
gateway/platforms/whatsapp_common.py
+89 -153
View File
@@ -3,20 +3,10 @@ Transport-agnostic WhatsApp behavior shared by the Baileys bridge adapter and th
Cloud API adapter: allow-list / DM / group gating, mention detection, quoted-reply-
to-bot detection, broadcast filtering, WhatsApp markdown conversion, chunk budgeting.
Mixin contract — the adapter must set these on ``self`` before any of the
mixin's methods are called (typically in ``__init__``):
self.config # gateway.config.PlatformConfig
self.name # str — adapter name (used in log lines)
self._dm_policy # str: "open" | "allowlist" | "disabled"
self._allow_from # set[str]
self._group_policy # str: "open" | "allowlist" | "disabled"
self._group_allow_from # set[str]
self._mention_patterns # list[re.Pattern]
self._reply_prefix # Optional[str]
Class attributes ``MAX_MESSAGE_LENGTH`` and ``DEFAULT_REPLY_PREFIX`` are
defined on the mixin and may be overridden per-adapter if needed.
Mixin contract — the host adapter sets these on ``self`` before calling any mixin
method: ``config`` (PlatformConfig), ``name``, ``_dm_policy`` / ``_group_policy``
("open" | "allowlist" | "disabled"), ``_allow_from`` / ``_group_allow_from`` (set[str]),
``_mention_patterns`` (list[re.Pattern]), ``_reply_prefix`` (Optional[str]).
"""
from __future__ import annotations
@@ -33,6 +23,27 @@ from gateway.platforms._shared import get_scoped_secret as _get_wsecret
logger = logging.getLogger(__name__)
_TRUTHY = {"true", "1", "yes", "on"}
_OPTIN_TRUTHY = {"true", "1", "yes"}
def _stash(pattern: str, text: str, tag: str) -> tuple[str, list[str]]:
"""Replace every ``pattern`` match with a ``\\x00<tag><n>\\x00`` placeholder."""
saved: list[str] = []
def keep(m: re.Match) -> str:
saved.append(m.group(0))
return f"\x00{tag}{len(saved) - 1}\x00"
return re.sub(pattern, keep, text), saved
def _header_to_bold(m: re.Match) -> str:
"""``# Header`` → ``*Header*``, stripping already-bolded ``*...*`` so ``# **Title**``
doesn't render with literal asterisks."""
inner = m.group(1).strip()
while len(inner) > 1 and inner.startswith("*") and inner.endswith("*"):
inner = inner[1:-1].strip()
return f"*{inner}*"
class WhatsAppBehaviorMixin:
@@ -54,8 +65,7 @@ class WhatsAppBehaviorMixin:
spaces — WhatsApp renders them as mojibake prefixes. Emoji joiners are kept."""
if not content:
return content
content = cls._OUTBOUND_INVISIBLE_CHARS_RE.sub("", content)
return cls._OUTBOUND_ODD_SPACE_RE.sub(" ", content)
return cls._OUTBOUND_ODD_SPACE_RE.sub(" ", cls._OUTBOUND_INVISIBLE_CHARS_RE.sub("", content))
@property
def enforces_own_access_policy(self) -> bool:
@@ -64,8 +74,7 @@ class WhatsAppBehaviorMixin:
def _effective_reply_prefix(self) -> str:
"""Prefix for outgoing replies in self-chat mode (Cloud API overrides to ``""``)."""
whatsapp_mode = _get_wsecret("WHATSAPP_MODE", default="self-chat") or "self-chat"
if whatsapp_mode != "self-chat":
if (_get_wsecret("WHATSAPP_MODE", default="self-chat") or "self-chat") != "self-chat":
return ""
if self._reply_prefix is not None:
return self._reply_prefix.replace("\\n", "\n")
@@ -80,11 +89,11 @@ class WhatsAppBehaviorMixin:
def _whatsapp_require_mention(self) -> bool:
configured = self.config.extra.get("require_mention")
if configured is not None:
if isinstance(configured, str):
return configured.lower() in _TRUTHY
return bool(configured)
return (_get_wsecret("WHATSAPP_REQUIRE_MENTION", default="false") or "false").lower() in _TRUTHY
if configured is None:
configured = _get_wsecret("WHATSAPP_REQUIRE_MENTION", default="false") or "false"
if isinstance(configured, str):
return configured.lower() in _TRUTHY
return bool(configured)
def _whatsapp_free_response_chats(self) -> set[str]:
raw = self.config.extra.get("free_response_chats")
@@ -97,15 +106,13 @@ class WhatsAppBehaviorMixin:
"""Parse allow_from / group_allow_from from config (list) or env var (CSV)."""
if raw is None:
return set()
if isinstance(raw, list):
return {str(part).strip() for part in raw if str(part).strip()}
return {part.strip() for part in str(raw).split(",") if part.strip()}
parts = raw if isinstance(raw, list) else str(raw).split(",")
return {str(part).strip() for part in parts if str(part).strip()}
def _select_dm_allowlist(self, extra: Dict[str, Any], env_keys, read_env) -> Any:
"""Pick the raw DM allowlist by key *presence*: ``allow_from``/``allowFrom`` in
config (an explicit empty list stays authoritative), then the first truthy
env carrier. Records the winning source in ``_dm_allowlist_source`` so live
DM checks keep the same precedence."""
"""Pick the raw DM allowlist by key *presence*: ``allow_from``/``allowFrom`` in config (an
explicit empty list stays authoritative), then the first truthy env carrier. Records the
winning source in ``_dm_allowlist_source`` so live DM checks keep the same precedence."""
for key in ("allow_from", "allowFrom"):
if key in extra:
self._dm_allowlist_source = "config"
@@ -118,18 +125,13 @@ class WhatsAppBehaviorMixin:
return None
def _live_dm_allow_from(self) -> set[str]:
"""Allowlist currently enforced for DM intake / strict DM auth.
Env-seeded adapters re-read the same key so pairing approve/revoke takes
effect without restart; a removed key (sole-entry revoke) means empty, not
the construction snapshot. Config-seeded adapters keep the in-memory set
(pairing revoke purges it in place) — a stale env value must not broaden access.
"""
"""Allowlist currently enforced for DM intake / strict DM auth. Env-seeded adapters re-read
the same key so pairing approve/revoke takes effect without restart; a removed key (sole-entry
revoke) means empty, not the construction snapshot. Config-seeded adapters keep the in-memory
set (pairing revoke purges it in place) — a stale env value must not broaden access."""
source = getattr(self, "_dm_allowlist_source", None)
if isinstance(source, str) and source != "config":
if source in os.environ:
return self._coerce_allow_list(os.environ.get(source, ""))
return set()
return self._coerce_allow_list(os.environ[source]) if source in os.environ else set()
return set(self._allow_from or ())
# ------------------------------------------------------------------ JID helpers
@@ -146,52 +148,40 @@ class WhatsAppBehaviorMixin:
def _is_broadcast_chat(chat_id: str) -> bool:
"""Status updates (Stories) and Channel/Newsletter broadcasts — never reply
(answering a Story spams the status feed; Channel posts aren't addressable)."""
if not chat_id:
return False
cid = chat_id.strip().lower()
return cid == "status@broadcast" or cid.endswith("@broadcast") or cid.endswith("@newsletter")
cid = (chat_id or "").strip().lower()
return cid == "status@broadcast" or cid.endswith(("@broadcast", "@newsletter"))
# ------------------------------------------------------------------ gating
def _open_dm_opted_in(self) -> bool:
if os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in {"true", "1", "yes"}:
if os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in _OPTIN_TRUTHY:
return True
return (_get_wsecret("WHATSAPP_ALLOW_ALL_USERS", default="") or "").lower() in {"true", "1", "yes"}
return (_get_wsecret("WHATSAPP_ALLOW_ALL_USERS", default="") or "").lower() in _OPTIN_TRUTHY
@staticmethod
def _matches_whatsapp_allowlist(candidate: str, allow_from) -> bool:
"""Match a WhatsApp identifier against an allowlist across phone/LID forms.
Inbound senders arrive as ``<id>@lid`` while allowlists hold phone numbers
(or vice versa), so resolve both sides through the bridge's lid-mapping
files via ``gateway.whatsapp_identity``.
"""
"""Match a WhatsApp identifier against an allowlist across phone/LID forms. Inbound senders
arrive as ``<id>@lid`` while allowlists hold phone numbers (or vice versa), so resolve both
sides through the bridge's lid-mapping files via ``gateway.whatsapp_identity``."""
if not allow_from:
return False
if candidate in allow_from:
return True
from gateway.whatsapp_identity import (
expand_whatsapp_aliases,
normalize_whatsapp_identifier,
)
from gateway.whatsapp_identity import expand_whatsapp_aliases, normalize_whatsapp_identifier
candidate_aliases = expand_whatsapp_aliases(candidate)
if not candidate_aliases:
return False
for entry in allow_from:
if entry == "*":
return True
if normalize_whatsapp_identifier(entry) in candidate_aliases:
return True
if expand_whatsapp_aliases(entry) & candidate_aliases:
return True
return False
return any(
entry == "*"
or normalize_whatsapp_identifier(entry) in candidate_aliases
or expand_whatsapp_aliases(entry) & candidate_aliases
for entry in allow_from
)
def _is_dm_allowed(self, sender_id: str) -> bool:
"""Strict DM authorization — pairing does not imply access."""
if self._dm_policy == "allowlist":
return self._matches_whatsapp_allowlist(sender_id, self._live_dm_allow_from())
if self._dm_policy == "open":
return self._open_dm_opted_in()
return False
return self._dm_policy == "open" and self._open_dm_opted_in()
def _is_dm_intake_allowed(self, sender_id: str) -> bool:
"""Whether a DM may reach the gateway intake (pairing handshake path)."""
@@ -202,9 +192,7 @@ class WhatsAppBehaviorMixin:
return self._matches_whatsapp_allowlist(principal, self._live_dm_allow_from())
if self._dm_policy == "pairing":
return True
if self._dm_policy == "open":
return self._open_dm_opted_in()
return False
return self._dm_policy == "open" and self._open_dm_opted_in()
def _is_group_allowed(self, chat_id: str) -> bool:
"""Check whether a group chat should be processed."""
@@ -220,18 +208,15 @@ class WhatsAppBehaviorMixin:
try:
patterns = json.loads(raw)
except Exception:
patterns = [part.strip() for part in raw.splitlines() if part.strip()]
if not patterns:
patterns = [part.strip() for part in raw.split(",") if part.strip()]
# Plain text: one pattern per line, else comma-separated.
patterns = [p.strip() for p in raw.splitlines() if p.strip()]
patterns = patterns or [p.strip() for p in raw.split(",") if p.strip()]
if patterns is None:
return []
if isinstance(patterns, str):
patterns = [patterns]
if not isinstance(patterns, list):
logger.warning(
"[%s] whatsapp mention_patterns must be a list or string; got %s",
self.name, type(patterns).__name__,
)
logger.warning("[%s] whatsapp mention_patterns must be a list or string; got %s", self.name, type(patterns).__name__)
return []
compiled = []
for pattern in patterns:
@@ -246,42 +231,28 @@ class WhatsAppBehaviorMixin:
return compiled
def _bot_ids_from_message(self, data: Dict[str, Any]) -> set[str]:
bot_ids = set()
for candidate in data.get("botIds") or []:
normalized = self._normalize_whatsapp_id(candidate)
if normalized:
bot_ids.add(normalized)
return bot_ids
return {nid for c in (data.get("botIds") or []) if (nid := self._normalize_whatsapp_id(c))}
def _message_is_reply_to_bot(self, data: Dict[str, Any]) -> bool:
quoted_participant = self._normalize_whatsapp_id(data.get("quotedParticipant"))
if not quoted_participant:
return False
return quoted_participant in self._bot_ids_from_message(data)
return bool(quoted_participant) and quoted_participant in self._bot_ids_from_message(data)
def _message_mentions_bot(self, data: Dict[str, Any]) -> bool:
bot_ids = self._bot_ids_from_message(data)
if not bot_ids:
return False
mentioned_ids = {
nid
for candidate in (data.get("mentionedIds") or [])
if (nid := self._normalize_whatsapp_id(candidate))
}
if mentioned_ids & bot_ids:
mentioned = {nid for c in (data.get("mentionedIds") or []) if (nid := self._normalize_whatsapp_id(c))}
if mentioned & bot_ids:
return True
lower_body = str(data.get("body") or "").lower()
for bot_id in bot_ids:
bare_id = bot_id.split("@", 1)[0].lower()
if bare_id and (f"@{bare_id}" in lower_body or bare_id in lower_body):
return True
return False
return any(
bare and (f"@{bare}" in lower_body or bare in lower_body)
for bare in (bot_id.split("@", 1)[0].lower() for bot_id in bot_ids)
)
def _message_matches_mention_patterns(self, data: Dict[str, Any]) -> bool:
if not self._mention_patterns:
return False
body = str(data.get("body") or "")
return any(pattern.search(body) for pattern in self._mention_patterns)
return any(pattern.search(body) for pattern in self._mention_patterns or ())
def _clean_bot_mention_text(self, text: str, data: Dict[str, Any]) -> str:
if not text:
@@ -300,23 +271,18 @@ class WhatsAppBehaviorMixin:
return False
if not data.get("isGroup", False):
# DMs that pass the policy gate are always processed
sender_id = str(data.get("senderId") or data.get("from") or "")
return self._is_dm_intake_allowed(sender_id)
return self._is_dm_intake_allowed(str(data.get("senderId") or data.get("from") or ""))
if not self._is_group_allowed(chat_id):
return False
# Group messages: check mention / free-response settings
if chat_id in self._whatsapp_free_response_chats():
if chat_id in self._whatsapp_free_response_chats() or not self._whatsapp_require_mention():
return True
if not self._whatsapp_require_mention():
return True
body = str(data.get("body") or "").strip()
if body.startswith("/"):
return True
if self._message_is_reply_to_bot(data):
return True
if self._message_mentions_bot(data):
return True
return self._message_matches_mention_patterns(data)
return (
str(data.get("body") or "").strip().startswith("/")
or self._message_is_reply_to_bot(data)
or self._message_mentions_bot(data)
or self._message_matches_mention_patterns(data)
)
# ------------------------------------------------------------------ formatting
def format_message(self, content: str) -> str:
@@ -324,46 +290,19 @@ class WhatsAppBehaviorMixin:
inline code are protected via placeholder substitution."""
if not content:
return content
content = self._sanitize_outbound_text(content)
_FENCE_PH = "\x00FENCE"
fences: list[str] = []
def _save_fence(m: re.Match) -> str:
fences.append(m.group(0))
return f"{_FENCE_PH}{len(fences) - 1}\x00"
result = re.sub(r"```[\s\S]*?```", _save_fence, content)
_CODE_PH = "\x00CODE"
codes: list[str] = []
def _save_code(m: re.Match) -> str:
codes.append(m.group(0))
return f"{_CODE_PH}{len(codes) - 1}\x00"
result = re.sub(r"`[^`\n]+`", _save_code, result)
result, fences = _stash(r"```[\s\S]*?```", self._sanitize_outbound_text(content), "FENCE")
result, codes = _stash(r"`[^`\n]+`", result, "CODE")
# Italic *text* → _text_ BEFORE bold so **bold** doesn't become italic;
# lookarounds skip list bullets and bold delimiters.
result = re.sub(r"(?<!\*)\*(?!\s|\*)([^*\n]*?\S[^*\n]*?)\*(?!\*)", r"_\1_", result)
result = re.sub(r"\*\*(.+?)\*\*", r"*\1*", result)
result = re.sub(r"__(.+?)__", r"*\1*", result)
result = re.sub(r"~~(.+?)~~", r"~\1~", result)
# "# Header" → *Header*, stripping *...* already produced above so
# "# **Title**" doesn't render with literal asterisks.
def _header_to_bold(m: re.Match) -> str:
inner = m.group(1).strip()
while len(inner) > 1 and inner.startswith("*") and inner.endswith("*"):
inner = inner[1:-1].strip()
return f"*{inner}*"
result = re.sub(r"^#{1,6}\s+(.+)$", _header_to_bold, result, flags=re.MULTILINE)
# [text](url) → text (url)
result = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)", result)
for i, fence in enumerate(fences):
result = result.replace(f"{_FENCE_PH}{i}\x00", fence)
for i, code in enumerate(codes):
result = result.replace(f"{_CODE_PH}{i}\x00", code)
result = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)", result) # [text](url) → text (url)
for tag, saved in (("FENCE", fences), ("CODE", codes)):
for i, original in enumerate(saved):
result = result.replace(f"\x00{tag}{i}\x00", original)
return result
@@ -376,14 +315,11 @@ def resolve_whatsapp_bridge_dir() -> Path:
install_bridge = _Path(__file__).resolve().parents[2] / "scripts" / "whatsapp-bridge"
hermes_home_bridge = get_hermes_home() / "scripts" / "whatsapp-bridge"
try:
test_file = install_bridge / ".write_test"
test_file.touch()
test_file.unlink()
install_writable = True
except (OSError, PermissionError):
install_writable = False
if install_writable:
(install_bridge / ".write_test").touch()
(install_bridge / ".write_test").unlink()
return install_bridge
except OSError:
pass
if hermes_home_bridge.exists():
return hermes_home_bridge
try:
gateway/platforms/yuanbao_media.py
+72 -174
View File
@@ -1,14 +1,8 @@
"""
yuanbao_media.py — 元宝平台媒体处理模块
提供 COS 上传、文件下载、TIM 媒体消息构建等功能。移植自 TypeScript 版 media.ts
（yuanbao-openclaw-plugin），使用 httpx 替代 cos-nodejs-sdk-v5。
COS 上传流程：
1. genUploadInfo 获取临时凭证（tmpSecretId/tmpSecretKey/sessionToken）
2. 临时凭证 HMAC-SHA1 签名构建 Authorization 头
3. HTTP PUT 上传到 COS
yuanbao_media.py — 元宝平台媒体处理：COS 上传、URL 下载（SSRF 防护）、TIM 媒体消息体。
移植自 TypeScript 版 media.ts（yuanbao-openclaw-plugin），用 httpx 替代 cos-nodejs-sdk-v5。
COS 上传流程：genUploadInfo 获取临时凭证 → 临时凭证 HMAC-SHA1 签名 Authorization 头 → HTTP PUT。
TIM 消息体：build_image_msg_body() → TIMImageElem，build_file_msg_body() → TIMFileElem
"""
@@ -32,9 +26,6 @@ UPLOAD_INFO_PATH = "/api/resource/genUploadInfo"
DEFAULT_API_DOMAIN = "yuanbao.tencent.com"
DEFAULT_MAX_SIZE_MB = 50
# COS 加速域名（优先使用全球加速）
COS_USE_ACCELERATE = True
# MIME → image_format 数字（TIM 协议字段）
_MIME_TO_IMAGE_FORMAT: dict[str, int] = {
"image/jpeg": 1, "image/jpg": 1, "image/gif": 2, "image/png": 3, "image/bmp": 4,
@@ -42,15 +33,8 @@ _MIME_TO_IMAGE_FORMAT: dict[str, int] = {
}
_EXT_TO_MIME: dict[str, str] = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".gif": "image/gif",
".webp": "image/webp",
".bmp": "image/bmp",
".heic": "image/heic",
".tiff": "image/tiff",
".ico": "image/x-icon",
".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp",
".bmp": "image/bmp", ".heic": "image/heic", ".tiff": "image/tiff", ".ico": "image/x-icon",
".pdf": "application/pdf",
".doc": "application/msword",
".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
@@ -58,31 +42,27 @@ _EXT_TO_MIME: dict[str, str] = {
".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
".ppt": "application/vnd.ms-powerpoint",
".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
".txt": "text/plain",
".zip": "application/zip",
".tar": "application/x-tar",
".gz": "application/gzip",
".mp3": "audio/mpeg",
".mp4": "video/mp4",
".wav": "audio/wav",
".ogg": "audio/ogg",
".webm": "video/webm",
".txt": "text/plain", ".zip": "application/zip", ".tar": "application/x-tar", ".gz": "application/gzip",
".mp3": "audio/mpeg", ".mp4": "video/mp4", ".wav": "audio/wav", ".ogg": "audio/ogg", ".webm": "video/webm",
}
_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".heic", ".tiff", ".ico"}
def _ext(filename: str) -> str:
return os.path.splitext(filename)[-1].lower()
def guess_mime_type(filename: str) -> str:
"""根据文件扩展名猜测 MIME 类型。"""
return _EXT_TO_MIME.get(os.path.splitext(filename)[-1].lower(), "application/octet-stream")
return _EXT_TO_MIME.get(_ext(filename), "application/octet-stream")
def is_image(filename: str, mime_type: str = "") -> bool:
"""判断是否为图片类型（MIME 前缀或扩展名）。"""
return mime_type.startswith("image/") or os.path.splitext(filename)[-1].lower() in _IMAGE_EXTS
"""MIME 前缀或扩展名判断是否为图片。"""
return mime_type.startswith("image/") or _ext(filename) in _IMAGE_EXTS
def get_image_format(mime_type: str) -> int:
"""获取 TIM 图片格式编号。"""
"""TIM 图片格式编号（未知 → 255）。"""
return _MIME_TO_IMAGE_FORMAT.get(mime_type.lower(), 255)
@@ -97,11 +77,6 @@ def generate_file_id() -> str:
# ============ 图片尺寸解析（纯 Python，无需 Pillow） ============
def parse_image_size(data: bytes) -> Optional[dict[str, int]]:
"""解析图片宽高（JPEG/PNG/GIF/WebP），返回 {"width", "height"} 或 None。"""
return _parse_png_size(data) or _parse_jpeg_size(data) or _parse_gif_size(data) or _parse_webp_size(data)
def _wh(w: int, h: int) -> dict[str, int]:
return {"width": w, "height": h}
@@ -113,17 +88,16 @@ def _parse_png_size(buf: bytes) -> Optional[dict[str, int]]:
def _parse_jpeg_size(buf: bytes) -> Optional[dict[str, int]]:
if len(buf) < 4 or buf[0] != 0xFF or buf[1] != 0xD8:
if len(buf) < 4 or buf[:2] != b"\xff\xd8":
return None
i = 2
while i < len(buf) - 9:
if buf[i] != 0xFF:
i += 1
continue
if buf[i + 1] in {0xC0, 0xC2}: # SOF0 / SOF2: height then width
elif buf[i + 1] in {0xC0, 0xC2}: # SOF0 / SOF2: height then width
h, w = struct.unpack(">HH", buf[i + 5: i + 9])
return _wh(w, h)
if i + 3 < len(buf):
elif i + 3 < len(buf):
i += 2 + struct.unpack(">H", buf[i + 2: i + 4])[0]
else:
break
@@ -131,7 +105,7 @@ def _parse_jpeg_size(buf: bytes) -> Optional[dict[str, int]]:
def _parse_gif_size(buf: bytes) -> Optional[dict[str, int]]:
if len(buf) < 10 or buf[:6].decode("ascii", errors="replace") not in {"GIF87a", "GIF89a"}:
if len(buf) < 10 or buf[:6] not in (b"GIF87a", b"GIF89a"):
return None
return _wh(*struct.unpack("<HH", buf[6:10]))
@@ -139,20 +113,23 @@ def _parse_gif_size(buf: bytes) -> Optional[dict[str, int]]:
def _parse_webp_size(buf: bytes) -> Optional[dict[str, int]]:
if len(buf) < 16 or buf[:4] != b"RIFF" or buf[8:12] != b"WEBP":
return None
chunk = buf[12:16].decode("ascii", errors="replace")
if chunk == "VP8 ":
if len(buf) >= 30 and buf[23:26] == b"\x9d\x01\x2a":
w, h = struct.unpack("<HH", buf[26:30])
return _wh(w & 0x3FFF, h & 0x3FFF)
elif chunk == "VP8L":
if len(buf) >= 25 and buf[20] == 0x2F:
bits = struct.unpack("<I", buf[21:25])[0]
return _wh((bits & 0x3FFF) + 1, ((bits >> 14) & 0x3FFF) + 1)
elif chunk == "VP8X" and len(buf) >= 30:
chunk = buf[12:16]
if chunk == b"VP8 " and len(buf) >= 30 and buf[23:26] == b"\x9d\x01\x2a":
w, h = struct.unpack("<HH", buf[26:30])
return _wh(w & 0x3FFF, h & 0x3FFF)
if chunk == b"VP8L" and len(buf) >= 25 and buf[20] == 0x2F:
bits = struct.unpack("<I", buf[21:25])[0]
return _wh((bits & 0x3FFF) + 1, ((bits >> 14) & 0x3FFF) + 1)
if chunk == b"VP8X" and len(buf) >= 30:
return _wh(int.from_bytes(buf[24:27], "little") + 1, int.from_bytes(buf[27:30], "little") + 1)
return None
def parse_image_size(data: bytes) -> Optional[dict[str, int]]:
"""解析图片宽高（JPEG/PNG/GIF/WebP），返回 {"width", "height"} 或 None。"""
return _parse_png_size(data) or _parse_jpeg_size(data) or _parse_gif_size(data) or _parse_webp_size(data)
# ============ URL 下载 ============
async def download_url(url: str, max_size_mb: int = DEFAULT_MAX_SIZE_MB) -> tuple[bytes, str]:
@@ -180,17 +157,14 @@ async def download_url(url: str, max_size_mb: int = DEFAULT_MAX_SIZE_MB) -> tupl
async with create_ssrf_safe_async_client(
timeout=30.0, follow_redirects=True, event_hooks={"response": [_redirect_guard]},
) as client:
# HEAD 预检大小；部分服务器不支持 HEAD，忽略
try:
try: # HEAD 预检大小；部分服务器不支持 HEAD，忽略
head = await client.head(url)
content_length = int(head.headers.get("content-length", 0) or 0)
if content_length > 0 and content_length > max_bytes:
raise ValueError(f"文件过大: {content_length / 1024 / 1024:.1f} MB > {max_size_mb} MB")
except httpx.HTTPStatusError:
pass
# 流式 GET，防止超限
async with client.stream("GET", url) as resp:
async with client.stream("GET", url) as resp: # 流式 GET，防止超限
resp.raise_for_status()
content_type = resp.headers.get("content-type", "").split(";")[0].strip()
chunks: list[bytes] = []
@@ -215,89 +189,51 @@ def _sorted_kv(d: dict[str, str]) -> list[tuple[str, str]]:
def _cos_sign(
method: str,
path: str,
params: dict[str, str],
headers: dict[str, str],
secret_id: str,
secret_key: str,
start_time: Optional[int] = None,
expire_seconds: int = 3600,
method: str, path: str, params: dict[str, str], headers: dict[str, str], secret_id: str, secret_key: str,
start_time: Optional[int] = None, expire_seconds: int = 3600,
) -> str:
"""构建 COS 请求签名（q-sign-algorithm=sha1；https://cloud.tencent.com/document/product/436/7778）。
"""COS Authorization 头（q-sign-algorithm=sha1；https://cloud.tencent.com/document/product/436/7778）。
Args:
method / path: HTTP 方法（小写）与 URL-encoded 路径
params / headers: 参与签名的查询参数与请求头（key 会被小写）
secret_id / secret_key: 临时 tmpSecretId / tmpSecretKey
start_time: 签名起始 Unix 时间戳（默认 now）；expire_seconds: 有效期（秒）
Returns:
Authorization header 值
method 小写、path 已 URL-encode；params / headers 的 key 会被小写；secret_* 为临时 tmpSecretId/Key；
start_time 默认 now。
"""
now = int(time.time())
q_sign_time = f"{start_time or now};{(start_time or now) + expire_seconds}"
# SignKey = HMAC-SHA1(SecretKey, q-sign-time)
sign_key = _hmac_sha1_hex(secret_key, q_sign_time)
start = start_time or int(time.time())
q_sign_time = f"{start};{start + expire_seconds}"
sign_key = _hmac_sha1_hex(secret_key, q_sign_time) # SignKey = HMAC-SHA1(SecretKey, q-sign-time)
sorted_params = _sorted_kv(params)
sorted_headers = _sorted_kv(headers)
http_string = "\n".join([
method.lower(),
path,
"&".join(f"{k}={v}" for k, v in sorted_params),
"&".join(f"{k}={v}" for k, v in sorted_headers),
"",
])
kv = lambda pairs: "&".join(f"{k}={v}" for k, v in pairs) # noqa: E731
http_string = "\n".join([method.lower(), path, kv(sorted_params), kv(sorted_headers), ""])
string_to_sign = "\n".join(["sha1", q_sign_time, hashlib.sha1(http_string.encode("utf-8")).hexdigest(), ""])
signature = _hmac_sha1_hex(sign_key, string_to_sign)
return (
f"q-sign-algorithm=sha1"
f"&q-ak={secret_id}"
f"&q-sign-time={q_sign_time}"
f"&q-key-time={q_sign_time}"
f"q-sign-algorithm=sha1&q-ak={secret_id}&q-sign-time={q_sign_time}&q-key-time={q_sign_time}"
f"&q-header-list={';'.join(k for k, _ in sorted_headers)}"
f"&q-url-param-list={';'.join(k for k, _ in sorted_params)}"
f"&q-signature={signature}"
f"&q-signature={_hmac_sha1_hex(sign_key, string_to_sign)}"
)
# ============ 主要公开 API ============
async def get_cos_credentials(
app_key: str,
api_domain: str,
token: str,
filename: str = "file",
file_id: Optional[str] = None,
bot_id: str = "",
route_env: str = "",
app_key: str, api_domain: str, token: str, filename: str = "file", file_id: Optional[str] = None,
bot_id: str = "", route_env: str = "",
) -> dict:
"""调用 genUploadInfo 获取 COS 临时密钥及上传配置。
Args:
app_key / bot_id: X-ID 头（bot_id 优先）; api_domain: 如 https://bot.yuanbao.tencent.com
token: 当前有效签票 token（X-Token 头）; file_id: 不传则自动生成
Returns:
接口 data dict：bucketName, region, location (COS key), encryptTmpSecretId,
encryptTmpSecretKey, encryptToken, startTime, expiredTime, resourceUrl, resourceID(可选)
Raises:
RuntimeError: 接口返回非 0 code 或 bucketName/location 缺失
X-ID 头取 bot_id（优先）或 app_key；api_domain 如 https://bot.yuanbao.tencent.com；file_id 不传则自动生成。
返回接口 data dict：bucketName, region, location (COS key), encryptTmpSecretId, encryptTmpSecretKey,
encryptToken, startTime, expiredTime, resourceUrl, resourceID(可选)。
Raises RuntimeError: 接口返回非 0 code 或 bucketName/location 缺失
"""
if file_id is None:
file_id = generate_file_id()
upload_url = f"{api_domain.rstrip('/')}{UPLOAD_INFO_PATH}"
headers = {"Content-Type": "application/json", "X-Token": token, "X-ID": bot_id or app_key, "X-Source": "web"}
if route_env:
headers["X-Route-Env"] = route_env
body = {"fileName": filename, "fileId": file_id, "docFrom": "localDoc", "docOpenId": ""}
body = {"fileName": filename, "fileId": file_id if file_id is not None else generate_file_id(), "docFrom": "localDoc", "docOpenId": ""}
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.post(upload_url, json=body, headers=headers)
resp = await client.post(f"{api_domain.rstrip('/')}{UPLOAD_INFO_PATH}", json=body, headers=headers)
resp.raise_for_status()
result: dict[str, Any] = resp.json()
code = result.get("code")
if code != 0 and code is not None:
raise RuntimeError(f"genUploadInfo 失败: code={code}, msg={result.get('msg', '')}")
@@ -309,75 +245,44 @@ async def get_cos_credentials(
async def upload_to_cos(
file_bytes: bytes,
filename: str,
content_type: str,
credentials: dict,
bucket: str,
region: str,
file_bytes: bytes, filename: str, content_type: str, credentials: dict, bucket: str, region: str,
) -> dict:
"""用临时凭证 HMAC-SHA1 签名，httpx PUT 上传到 COS。
"""用临时凭证（get_cos_credentials() 返回的 dict）HMAC-SHA1 签名，httpx PUT 上传到 COS（走全球加速域名）。
Args:
credentials: get_cos_credentials() 返回的 dict（encryptTmpSecretId/Key, encryptToken,
location, resourceUrl, startTime, expiredTime）
bucket / region: 如 chatbot-1234567890 / ap-guangzhou
Returns:
{url, uuid (内容 MD5), size, width?, height? (仅图片)}
Raises:
httpx.HTTPStatusError: COS 返回非 2xx
RuntimeError: credentials 字段缺失
Returns {url, uuid (内容 MD5), size, width?, height? (仅图片)}
Raises httpx.HTTPStatusError（COS 非 2xx）/ RuntimeError（credentials 字段缺失）
"""
secret_id: str = credentials.get("encryptTmpSecretId", "")
secret_key: str = credentials.get("encryptTmpSecretKey", "")
session_token: str = credentials.get("encryptToken", "")
cos_key: str = credentials.get("location", "")
resource_url: str = credentials.get("resourceUrl", "")
secret_id, secret_key, session_token, cos_key = (
credentials.get(k, "") for k in ("encryptTmpSecretId", "encryptTmpSecretKey", "encryptToken", "location")
)
start_time: Optional[int] = credentials.get("startTime")
expired_time: Optional[int] = credentials.get("expiredTime")
if not secret_id or not secret_key or not cos_key:
raise RuntimeError(
f"COS credentials 不完整: secretId={bool(secret_id)}, "
f"secretKey={bool(secret_key)}, location={bool(cos_key)}"
f"COS credentials 不完整: secretId={bool(secret_id)}, secretKey={bool(secret_key)}, location={bool(cos_key)}"
)
cos_host = f"{bucket}.cos.accelerate.myqcloud.com" if COS_USE_ACCELERATE else f"{bucket}.cos.{region}.myqcloud.com"
cos_host = f"{bucket}.cos.accelerate.myqcloud.com"
encoded_key = urllib.parse.quote(cos_key, safe="/").lstrip("/")
cos_url = f"https://{cos_host}/{encoded_key}"
if not content_type or content_type == "application/octet-stream":
content_type = guess_mime_type(filename) if is_image(filename) else "application/octet-stream"
file_uuid = md5_hex(file_bytes)
file_size = len(file_bytes)
now = int(time.time())
authorization = _cos_sign(
method="put",
path=f"/{encoded_key}",
params={},
method="put", path=f"/{encoded_key}", params={},
headers={"host": cos_host, "content-type": content_type, "x-cos-security-token": session_token},
secret_id=secret_id,
secret_key=secret_key,
secret_id=secret_id, secret_key=secret_key,
start_time=start_time if start_time else now,
expire_seconds=(expired_time - now) if expired_time and expired_time > now else 3600,
)
put_headers = {"Authorization": authorization, "Content-Type": content_type, "x-cos-security-token": session_token}
logger.info("COS PUT: bucket=%s region=%s key=%s size=%d mime=%s", bucket, region, cos_key, file_size, content_type)
async with httpx.AsyncClient(timeout=120.0) as client:
resp = await client.put(cos_url, content=file_bytes, headers=put_headers)
resp.raise_for_status()
result: dict[str, Any] = {"url": resource_url or cos_url, "uuid": file_uuid, "size": file_size}
result: dict[str, Any] = {"url": credentials.get("resourceUrl", "") or cos_url, "uuid": md5_hex(file_bytes), "size": file_size}
if content_type.startswith("image/"):
size_info = parse_image_size(file_bytes)
if size_info:
result["width"] = size_info["width"]
result["height"] = size_info["height"]
result.update(parse_image_size(file_bytes) or {})
logger.info("COS 上传成功: url=%s size=%d", result["url"], file_size)
return result
@@ -385,29 +290,22 @@ async def upload_to_cos(
# ============ TIM 媒体消息构建（https://cloud.tencent.com/document/product/269/2720） ============
def build_image_msg_body(
url: str,
uuid: Optional[str] = None,
filename: Optional[str] = None,
size: int = 0,
width: int = 0,
height: int = 0,
mime_type: str = "",
url: str, uuid: Optional[str] = None, filename: Optional[str] = None, size: int = 0, width: int = 0,
height: int = 0, mime_type: str = "",
) -> list[dict]:
"""构建 TIMImageElem 消息体（可直接放入 msg_body）。uuid 缺省依次退到 filename / URL basename / "image"。"""
"""TIMImageElem 消息体（可直接放入 msg_body）。uuid 缺省依次退到 filename / URL basename / "image"。"""
return [{
"msg_type": "TIMImageElem",
"msg_content": {
"uuid": uuid or filename or _basename_from_url(url) or "image",
"image_format": get_image_format(mime_type) if mime_type else 255,
"image_info_array": [
{"type": 1, "size": size, "width": width, "height": height, "url": url}, # type 1 = 原图
],
"image_info_array": [{"type": 1, "size": size, "width": width, "height": height, "url": url}], # type 1 = 原图
},
}]
def build_file_msg_body(url: str, filename: str, uuid: Optional[str] = None, size: int = 0) -> list[dict]:
"""构建 TIMFileElem 消息体（可直接放入 msg_body）。uuid 缺省使用 filename。"""
"""TIMFileElem 消息体（可直接放入 msg_body）。uuid 缺省使用 filename。"""
return [{
"msg_type": "TIMFileElem",
"msg_content": {"uuid": uuid or filename, "file_name": filename, "file_size": size, "url": url},
gateway/platforms/yuanbao_proto.py
+172 -399
View File
@@ -1,52 +1,23 @@
"""
yuanbao_proto.py - Yuanbao WebSocket 协议编解码（纯 Python 实现）
"""yuanbao_proto.py - Yuanbao WebSocket 协议编解码（手写 protobuf wire-format，不依赖 google.protobuf）
协议层级：
WebSocket frame
└── ConnMsg (protobuf: trpc.yuanbao.conn_common.ConnMsg)
├── head: Head (cmd_type, cmd, seq_no, msg_id, module, ...)
└── data: bytes (业务 payload，标准 protobuf)
└── InboundMessagePush / SendC2CMessageReq / SendGroupMessageReq / ...
(trpc.yuanbao.yuanbao_conn.yuanbao_openclaw_proxy.*)
注意：conn 层（ConnMsg）本身是标准 protobuf，不是自定义二进制格式；conn.proto 注释里的
magic+head_len+body_len 格式仅用于 quic/tcp。WebSocket 每个 frame = 一条 ConnMsg protobuf。
实现方式：手写 varint / protobuf wire-format 编解码，不依赖第三方 protobuf 库。
每个 WebSocket frame = 一条 ConnMsg protobuf（标准 protobuf；conn.proto 注释里的 magic+len 二进制格式只用于 quic/tcp）：
ConnMsg { Head head=1 (cmd_type, cmd, seq_no, msg_id, module, ...); bytes data=2 }
data = 业务 payload（InboundMessagePush / SendC2CMessageReq / ...，包 trpc.yuanbao.yuanbao_conn.yuanbao_openclaw_proxy.*）
"""
from __future__ import annotations
import logging
import threading
import time
from typing import Optional
logger = logging.getLogger(__name__)
DEBUG_MODE = False
def _dbg(label: str, data: bytes) -> None:
if DEBUG_MODE:
hex_str = " ".join(f"{b:02x}" for b in data[:64])
ellipsis = "..." if len(data) > 64 else ""
logger.debug("[yuanbao_proto] %s (%dB): %s", label, len(data), hex_str + ellipsis)
# ============================================================
# 常量
# ============================================================
# conn 层消息类型（ConnMsg.Head.cmd_type）
PB_MSG_TYPES = {
n: f"trpc.yuanbao.conn_common.{n}"
for n in ("ConnMsg", "AuthBindReq", "AuthBindRsp", "PingReq", "PingRsp", "KickoutMsg", "DirectedPush", "PushMsg")
}
# cmd_type: 上行请求 / 请求回包 / 下行推送 / 推送 ACK
CMD_TYPE = {"Request": 0, "Response": 1, "Push": 2, "PushAck": 3}
# 内置命令字 / 模块名
CMD = {"AuthBind": "auth-bind", "Ping": "ping", "Kickout": "kickout", "UpdateMeta": "update-meta"}
MODULE = {"ConnAccess": "conn_access"}
@@ -54,28 +25,16 @@ MODULE = {"ConnAccess": "conn_access"}
_BIZ_PKG = "yuanbao_openclaw_proxy"
BIZ_SERVICES = {
n: f"{_BIZ_PKG}.{n}"
for n in (
"InboundMessagePush",
"SendC2CMessageReq", "SendC2CMessageRsp",
"SendGroupMessageReq", "SendGroupMessageRsp",
"QueryGroupInfoReq", "QueryGroupInfoRsp",
"GetGroupMemberListReq", "GetGroupMemberListRsp",
"SendPrivateHeartbeatReq", "SendPrivateHeartbeatRsp",
"SendGroupHeartbeatReq", "SendGroupHeartbeatRsp",
for n in ("InboundMessagePush",) + tuple(
f"{m}{k}" for m in ("SendC2CMessage", "SendGroupMessage", "QueryGroupInfo", "GetGroupMemberList",
"SendPrivateHeartbeat", "SendGroupHeartbeat") for k in ("Req", "Rsp")
)
}
# openclaw instance_id（固定值 17）
HERMES_INSTANCE_ID = 17
# Reply Heartbeat 状态常量
HERMES_INSTANCE_ID = 17 # openclaw instance_id（固定值）
WS_HEARTBEAT_RUNNING = 1
WS_HEARTBEAT_FINISH = 2
# ============================================================
# 序列号生成
# ============================================================
_seq_lock = threading.Lock()
_seq_counter = 0
_SEQ_MAX = 2 ** 32 - 1 # uint32 上限
@@ -90,9 +49,7 @@ def next_seq_no() -> int:
return val
# ============================================================
# Protobuf wire-format 基础工具（手写，不依赖 google.protobuf）
# ============================================================
# ---- Protobuf wire-format 基础工具
WT_VARINT = 0
WT_64BIT = 1
@@ -102,19 +59,17 @@ _FIXED_SIZE = {WT_64BIT: 8, WT_32BIT: 4}
def _encode_varint(value: int) -> bytes:
"""将整数编码为 protobuf varint（负数按 64-bit two's complement）"""
"""protobuf varint（负数按 64-bit two's complement）"""
if value < 0:
value = value & 0xFFFFFFFFFFFFFFFF
value &= 0xFFFFFFFFFFFFFFFF
out = []
while True:
bits = value & 0x7F
value >>= 7
if value:
out.append(bits | 0x80)
else:
if not value:
out.append(bits)
break
return bytes(out)
return bytes(out)
out.append(bits | 0x80)
def _decode_varint(data: bytes, pos: int) -> tuple[int, int]:
@@ -134,21 +89,18 @@ def _decode_varint(data: bytes, pos: int) -> tuple[int, int]:
def _encode_field(field_number: int, wire_type: int, value: bytes) -> bytes:
"""编码一个 protobuf field（tag + value）"""
return _encode_varint((field_number << 3) | wire_type) + value
def _encode_string(s: str) -> bytes:
"""length-prefixed UTF-8 string value"""
encoded = s.encode("utf-8")
return _encode_varint(len(encoded)) + encoded
def _encode_message(b: bytes) -> bytes:
"""length-prefixed bytes / 嵌套 message value"""
return _encode_varint(len(b)) + b
def _encode_string(s: str) -> bytes:
return _encode_message(s.encode("utf-8"))
# 完整 field 编码快捷方式：string / varint / 嵌套 message
def _s(fn: int, s: str) -> bytes:
return _encode_field(fn, WT_LEN, _encode_string(s))
@@ -163,33 +115,29 @@ def _m(fn: int, b: bytes) -> bytes:
def _parse_fields(data: bytes) -> list[tuple[int, int, bytes | int]]:
"""解析 message 的所有字段 → [(field_number, wire_type, raw_value)]；
raw_value 为 int（VARINT）或 bytes（LEN / 64BIT / 32BIT）。"""
"""→ [(field_number, wire_type, raw_value)]；raw_value 为 int（VARINT）或 bytes（LEN / 64BIT / 32BIT）"""
fields = []
pos = 0
n = len(data)
while pos < n:
while pos < len(data):
tag, pos = _decode_varint(data, pos)
field_number = tag >> 3
wire_type = tag & 0x07
if wire_type == WT_VARINT:
val, pos = _decode_varint(data, pos)
elif wire_type == WT_LEN:
length, pos = _decode_varint(data, pos)
else:
if wire_type == WT_LEN:
length, pos = _decode_varint(data, pos)
elif wire_type in _FIXED_SIZE:
length = _FIXED_SIZE[wire_type]
else:
raise ValueError(f"unknown wire type {wire_type} at pos {pos - 1}")
val = data[pos: pos + length]
pos += length
elif wire_type in _FIXED_SIZE:
size = _FIXED_SIZE[wire_type]
val = data[pos: pos + size]
pos += size
else:
raise ValueError(f"unknown wire type {wire_type} at pos {pos - 1}")
fields.append((field_number, wire_type, val))
fields.append((tag >> 3, wire_type, val))
return fields
def _fields_to_dict(fields: list) -> dict[int, list]:
"""fields 列表 → {field_number: [(wire_type, value), ...]}（repeated 字段有多个）"""
"""→ {field_number: [(wire_type, value), ...]}（repeated 字段有多个）"""
d: dict[int, list] = {}
for fn, wt, val in fields:
d.setdefault(fn, []).append((wt, val))
@@ -201,18 +149,14 @@ def _parse_dict(data: bytes) -> dict[int, list]:
def _first(fdict: dict, fn: int, wt: int):
"""第一个 wire type 匹配的字段值，无则 None"""
"""第一个字段值（仅当其 wire type 匹配），无则 None"""
entries = fdict.get(fn)
if entries and entries[0][0] == wt:
return entries[0][1]
return None
return entries[0][1] if entries and entries[0][0] == wt else None
def _get_string(fdict: dict, fn: int, default: str = "") -> str:
val = _first(fdict, fn, WT_LEN)
if isinstance(val, (bytes, bytearray)):
return val.decode("utf-8", errors="replace")
return default
return val.decode("utf-8", errors="replace") if isinstance(val, (bytes, bytearray)) else default
def _get_varint(fdict: dict, fn: int, default: int = 0) -> int:
@@ -229,6 +173,28 @@ def _get_repeated_bytes(fdict: dict, fn: int) -> list[bytes]:
return [bytes(val) for wt, val in fdict.get(fn, []) if wt == WT_LEN]
def _parse_repeated(fdict: dict, fn: int) -> list[dict]:
return [_parse_dict(b) for b in _get_repeated_bytes(fdict, fn)]
# 字段表编码：parts = [(field_number, kind, value)]；kind:
# "S" string 总是编码 "s" string 非空才编码 "v" varint 非零才编码 "n" varint 非 None 才编码 "m" 嵌套 bytes 非空才编码
# "b" repeated MsgBodyElement "t" LogInfoExt{1 trace_id} 非空才编码
_PART_ENCODERS = {
"S": _s, "s": _s, "v": _v, "n": _v, "m": _m,
"b": lambda fn, body: b"".join(_m(fn, _encode_msg_body_element(el)) for el in body),
"t": lambda fn, trace_id: _m(fn, _s(1, trace_id)),
}
def _encode_parts(parts: list) -> bytes:
buf = b""
for fn, kind, val in parts:
if kind == "S" or (val is not None if kind == "n" else val):
buf += _PART_ENCODERS[kind](fn, val)
return buf
# 字段表驱动编解码：spec = [(field_number, key, kind)]，kind:
# "s" string（编码时 str(v)） "r" string（原值） "i" varint（编码时 int(v)）
# 编码跳过 falsy 值；解码只保留 truthy 值。spec 顺序即 wire 顺序和 dict 插入顺序。
@@ -253,9 +219,7 @@ def _decode_spec(fdict: dict, spec: list) -> dict:
return out
# ============================================================
# ConnMsg 层编解码
# ============================================================
# ---- ConnMsg 层编解码
# message Head { uint32 cmd_type=1; string cmd=2; uint32 seq_no=3; string msg_id=4;
# string module=5; bool need_ack=6; ... int32 status=10; }
# message ConnMsg { Head head=1; bytes data=2; }
@@ -264,76 +228,51 @@ def _decode_spec(fdict: dict, spec: list) -> dict:
def _encode_head(
cmd_type: int, cmd: str, seq_no: int, msg_id: str, module: str, need_ack: bool = False, status: int = 0,
) -> bytes:
buf = b""
if cmd_type != 0:
buf += _v(1, cmd_type)
if cmd:
buf += _s(2, cmd)
if seq_no != 0:
buf += _v(3, seq_no)
if msg_id:
buf += _s(4, msg_id)
if module:
buf += _s(5, module)
if need_ack:
buf += _v(6, 1)
if status != 0:
buf += _v(10, status & 0xFFFFFFFFFFFFFFFF)
return buf
return _encode_parts([
(1, "v", cmd_type), (2, "s", cmd), (3, "v", seq_no), (4, "s", msg_id), (5, "s", module),
(6, "v", 1 if need_ack else 0), (10, "v", status & 0xFFFFFFFFFFFFFFFF),
])
def _decode_head(data: bytes) -> dict:
fdict = _parse_dict(data)
fd = _parse_dict(data)
return {
"cmd_type": _get_varint(fdict, 1, 0),
"cmd": _get_string(fdict, 2, ""),
"seq_no": _get_varint(fdict, 3, 0),
"msg_id": _get_string(fdict, 4, ""),
"module": _get_string(fdict, 5, ""),
"need_ack": bool(_get_varint(fdict, 6, 0)),
"status": _get_varint(fdict, 10, 0),
"cmd_type": _get_varint(fd, 1), "cmd": _get_string(fd, 2), "seq_no": _get_varint(fd, 3), "msg_id": _get_string(fd, 4),
"module": _get_string(fd, 5), "need_ack": bool(_get_varint(fd, 6)), "status": _get_varint(fd, 10),
}
def _conn_msg(label: str, cmd_type: int, cmd: str, seq_no: int, msg_id: str, module: str, data: bytes, need_ack: bool) -> bytes:
buf = _m(1, _encode_head(cmd_type, cmd, seq_no, msg_id, module, need_ack))
if data:
buf += _m(2, data)
_dbg(label, buf)
return buf
def encode_conn_msg(msg_type: int, seq_no: int, data: bytes) -> bytes:
"""编码 ConnMsg（简化接口：仅 cmd_type + seq_no + payload）"""
return _conn_msg("encode_conn_msg", msg_type, "", seq_no, "", "", data, False)
def encode_conn_msg_full(
cmd_type: int, cmd: str, seq_no: int, msg_id: str, module: str, data: bytes, need_ack: bool = False,
) -> bytes:
"""编码完整的 ConnMsg（含 cmd/msg_id/module 等 head 字段）"""
return _conn_msg("encode_conn_msg_full", cmd_type, cmd, seq_no, msg_id, module, data, need_ack)
buf = _m(1, _encode_head(cmd_type, cmd, seq_no, msg_id, module, need_ack))
return buf + _m(2, data) if data else buf
def encode_conn_msg(msg_type: int, seq_no: int, data: bytes) -> bytes:
"""编码 ConnMsg（简化接口：仅 cmd_type + seq_no + payload）"""
return encode_conn_msg_full(msg_type, "", seq_no, "", "", data)
def decode_conn_msg(data: bytes) -> dict:
"""解码 ConnMsg → {msg_type, seq_no, data, head}（head 为完整 Head dict）"""
_dbg("decode_conn_msg", data)
fdict = _parse_dict(data)
head = _decode_head(_get_bytes(fdict, 1))
return {"msg_type": head["cmd_type"], "seq_no": head["seq_no"], "data": _get_bytes(fdict, 2), "head": head}
# ============================================================
# BizMsg 层：业务 body 包装成 ConnMsg（head.cmd = method, head.module = service）
def _conn_request(cmd_type: int, cmd: str, msg_id: str, module: str, data: bytes = b"") -> bytes:
return encode_conn_msg_full(cmd_type, cmd, next_seq_no(), msg_id, module, data)
# ---- BizMsg 层：业务 body 包装成 ConnMsg（head.cmd = method, head.module = service）
# 与 conn-codec.ts buildBusinessConnMsg(cmd, module, bizData, msgId) 行为一致。
# ============================================================
def encode_biz_msg(service: str, method: str, req_id: str, body: bytes) -> bytes:
"""将已编码的业务 protobuf 包装为可直接发送的 ConnMsg bytes"""
return encode_conn_msg_full(
cmd_type=CMD_TYPE["Request"], cmd=method, seq_no=next_seq_no(), msg_id=req_id, module=service, data=body,
)
return _conn_request(CMD_TYPE["Request"], method, req_id, service, body)
def decode_biz_msg(data: bytes) -> dict:
@@ -341,24 +280,17 @@ def decode_biz_msg(data: bytes) -> dict:
result = decode_conn_msg(data)
head = result["head"]
return {
"service": head["module"],
"method": head["cmd"],
"req_id": head["msg_id"],
"body": result["data"],
"is_response": head["cmd_type"] == CMD_TYPE["Response"],
"head": head,
"service": head["module"], "method": head["cmd"], "req_id": head["msg_id"], "body": result["data"],
"is_response": head["cmd_type"] == CMD_TYPE["Response"], "head": head,
}
def _biz_request(method: str, prefix: str, body: bytes, msg_id: str = "") -> bytes:
"""biz 请求 ConnMsg；req_id 为 msg_id，空则 '<prefix>_<seq>'（seq 在 conn seq_no 之前分配）"""
req_id = msg_id or f"{prefix}_{next_seq_no()}"
return encode_biz_msg(service=_BIZ_PKG, method=method, req_id=req_id, body=body)
return encode_biz_msg(_BIZ_PKG, method, msg_id or f"{prefix}_{next_seq_no()}", body)
# ============================================================
# 业务 protobuf 消息编解码（biz payload）
# ============================================================
# ---- 业务 protobuf 消息编解码（biz payload）
# MsgContent：1 text, 2 uuid, 3 image_format, 4 data, 5 desc, 6 ext, 7 sound,
# 8 image_info_array (repeated), 9 index, 10 url, 11 file_size, 12 file_name,
@@ -388,104 +320,53 @@ def _encode_msg_content(content: dict) -> bytes:
def _decode_msg_content(data: bytes) -> dict:
fdict = _parse_dict(data)
content = _decode_spec(fdict, _MSG_CONTENT_SPEC)
imgs = [img for img in (_decode_spec(_parse_dict(b), _IMAGE_INFO_SPEC) for b in _get_repeated_bytes(fdict, 8)) if img]
if imgs:
content["image_info_array"] = imgs
ext_map: dict[str, str] = {}
for entry_bytes in _get_repeated_bytes(fdict, 999):
efdict = _parse_dict(entry_bytes)
k = _get_string(efdict, 1)
if k:
ext_map[k] = _get_string(efdict, 2)
if ext_map:
content["ext_map"] = ext_map
imgs = [img for img in (_decode_spec(d, _IMAGE_INFO_SPEC) for d in _parse_repeated(fdict, 8)) if img]
ext_map = {_get_string(e, 1): _get_string(e, 2) for e in _parse_repeated(fdict, 999) if _get_string(e, 1)}
content.update({k: v for k, v in (("image_info_array", imgs), ("ext_map", ext_map)) if v})
return content
# MsgBodyElement：1 msg_type (string, e.g. "TIMTextElem"), 2 msg_content (MsgContent)
def _encode_msg_body_element(element: dict) -> bytes:
buf = b""
msg_type = element.get("msg_type", "")
if msg_type:
buf += _s(1, msg_type)
content = element.get("msg_content", {})
if content:
buf += _m(2, _encode_msg_content(content))
return buf
return _encode_parts([(1, "s", element.get("msg_type", "")), (2, "m", _encode_msg_content(content) if content else b"")])
def _decode_msg_body_element(data: bytes) -> dict:
fdict = _parse_dict(data)
content_bytes = _get_bytes(fdict, 2)
return {"msg_type": _get_string(fdict, 1, ""), "msg_content": _decode_msg_content(content_bytes) if content_bytes else {}}
return {"msg_type": _get_string(fdict, 1), "msg_content": _decode_msg_content(content_bytes) if content_bytes else {}}
def _encode_msg_body(fn: int, msg_body: list) -> bytes:
return b"".join(_m(fn, _encode_msg_body_element(el)) for el in msg_body)
# ---- 入站消息解析
def _encode_log_ext(trace_id: str) -> bytes:
"""LogInfoExt：1 trace_id"""
return _s(1, trace_id) if trace_id else b""
# ============================================================
# 入站消息解析
# ============================================================
# InboundMessagePush: 1 callback_command, 2 from_account, 3 to_account, 4 sender_nickname,
# 5 group_id, 6 group_code, 7 group_name, 8 msg_seq, 9 msg_random, 10 msg_time, 11 msg_key,
# 12 msg_id, 13 msg_body (repeated MsgBodyElement), 14 cloud_custom_data, 15 event_time,
# 16 bot_owner_id, 17 recall_msg_seq_list (repeated ImMsgSeq{1 msg_seq, 2 msg_id}),
# 18 claw_msg_type, 19 private_from_group_code, 20 log_ext (LogInfoExt{1 trace_id})
# InboundMessagePush 字段表 [(field_number, key, getter)]；getter 为 _get_string / _get_varint 或自定义 (fdict, fn) -> value
_INBOUND_PUSH_SPEC = [
(1, "callback_command", _get_string), (2, "from_account", _get_string), (3, "to_account", _get_string),
(4, "sender_nickname", _get_string), (5, "group_id", _get_string), (6, "group_code", _get_string),
(7, "group_name", _get_string), (8, "msg_seq", _get_varint), (9, "msg_random", _get_varint),
(10, "msg_time", _get_varint), (11, "msg_key", _get_string), (12, "msg_id", _get_string),
(13, "msg_body", lambda fd, fn: [_decode_msg_body_element(b) for b in _get_repeated_bytes(fd, fn)]),
(14, "cloud_custom_data", _get_string), (15, "event_time", _get_varint), (16, "bot_owner_id", _get_string),
(17, "recall_msg_seq_list", lambda fd, fn: [ # repeated ImMsgSeq{1 msg_seq, 2 msg_id}
{"msg_seq": _get_varint(d, 1), "msg_id": _get_string(d, 2)} for d in _parse_repeated(fd, fn)] or None),
(18, "claw_msg_type", _get_varint), (19, "private_from_group_code", _get_string),
(20, "trace_id", lambda fd, fn: _get_string(_parse_dict(_get_bytes(fd, fn)), 1) if _get_bytes(fd, fn) else ""), # LogInfoExt
]
def decode_inbound_push(data: bytes) -> Optional[dict]:
"""解析 InboundMessagePush biz payload。
Returns: 上述字段的 dict（空值已过滤，msg_body / msg_seq 始终保留；
recall_msg_seq_list 为 [{msg_seq, msg_id}] 或 None），解析失败返回 None。
"""
"""解析 InboundMessagePush biz payload；空值已过滤（msg_body / msg_seq 始终保留），解析失败返回 None。"""
try:
_dbg("decode_inbound_push input", data)
fdict = _parse_dict(data)
log_ext_bytes = _get_bytes(fdict, 20)
result: dict = {
"callback_command": _get_string(fdict, 1),
"from_account": _get_string(fdict, 2),
"to_account": _get_string(fdict, 3),
"sender_nickname": _get_string(fdict, 4),
"group_id": _get_string(fdict, 5),
"group_code": _get_string(fdict, 6),
"group_name": _get_string(fdict, 7),
"msg_seq": _get_varint(fdict, 8),
"msg_random": _get_varint(fdict, 9),
"msg_time": _get_varint(fdict, 10),
"msg_key": _get_string(fdict, 11),
"msg_id": _get_string(fdict, 12),
"msg_body": [_decode_msg_body_element(b) for b in _get_repeated_bytes(fdict, 13)],
"cloud_custom_data": _get_string(fdict, 14),
"event_time": _get_varint(fdict, 15),
"bot_owner_id": _get_string(fdict, 16),
"recall_msg_seq_list": [
{"msg_seq": _get_varint(d, 1), "msg_id": _get_string(d, 2)}
for d in map(_parse_dict, _get_repeated_bytes(fdict, 17))
] or None,
"claw_msg_type": _get_varint(fdict, 18),
"private_from_group_code": _get_string(fdict, 19),
"trace_id": _get_string(_parse_dict(log_ext_bytes), 1) if log_ext_bytes else "",
}
result = {key: get(fdict, fn) for fn, key, get in _INBOUND_PUSH_SPEC}
return {k: v for k, v in result.items() if v or k in {"msg_body", "msg_seq"}}
except Exception as e:
if DEBUG_MODE:
logger.debug("[yuanbao_proto] decode_inbound_push failed: %s", e)
except Exception:
return None
# ============================================================
# WeChat forwarded chat-history parsing (ForwardMsgData)
# ============================================================
# ---- WeChat forwarded chat-history parsing (ForwardMsgData)
# ext_map["wexin_forward_msg_<id>_<userid>"] = base64(ForwardMsgData) — protobuf, NOT JSON.
# Verified against live captures:
# ForwardMsgData { uint32 sub_type=1 (1 = WeChat chat-history forward); uint32 begin_time=2;
@@ -503,189 +384,98 @@ def _decode_forward_msg_content(data: bytes) -> dict:
"""MsgContent → {type, text?, multimedia?}（shape 与 _format_multimedia 对齐）"""
fdict = _parse_dict(data)
content: dict = {"type": _get_varint(fdict, 1)}
text = _get_string(fdict, 2)
if text:
content["text"] = text
multimedia = [_decode_spec(_parse_dict(b), _FORWARD_MULTIMEDIA_SPEC) for b in _get_repeated_bytes(fdict, 3)]
if multimedia:
content["multimedia"] = multimedia
if _get_string(fdict, 2):
content["text"] = _get_string(fdict, 2)
if _get_repeated_bytes(fdict, 3):
content["multimedia"] = [_decode_spec(d, _FORWARD_MULTIMEDIA_SPEC) for d in _parse_repeated(fdict, 3)]
return content
def _decode_forward_msg(data: bytes) -> dict:
fdict = _parse_dict(data)
return {
"sender": _get_string(fdict, 1),
"time": _get_varint(fdict, 2),
"plainText": _get_string(fdict, 3),
"msgContent": [_decode_forward_msg_content(b) for b in _get_repeated_bytes(fdict, 4)],
}
def _decode_forward_msg(fd: dict) -> dict:
return {"sender": _get_string(fd, 1), "time": _get_varint(fd, 2), "plainText": _get_string(fd, 3),
"msgContent": [_decode_forward_msg_content(b) for b in _get_repeated_bytes(fd, 4)]}
def decode_forward_msg_data(data: bytes) -> Optional[dict]:
"""Parse ForwardMsgData protobuf bytes (the base64-decoded ext_map value).
Returns the ``sub_type`` / ``nick_name`` / ``msg`` structure consumed by
``ForwardedRecordsParseMiddleware.build_forward_text``; ``None`` on parse failure.
"""
"""Parse ForwardMsgData bytes (base64-decoded ext_map value) into the {sub_type, nick_name, msg, ...}
structure consumed by ForwardedRecordsParseMiddleware.build_forward_text; None on parse failure."""
try:
fdict = _parse_dict(data)
fd = _parse_dict(data)
return {
"sub_type": _get_varint(fdict, 1),
"begin_time": _get_varint(fdict, 2),
"end_time": _get_varint(fdict, 3),
"nick_name": _get_string(fdict, 4),
"msg": [_decode_forward_msg(b) for b in _get_repeated_bytes(fdict, 5)],
"sub_type": _get_varint(fd, 1), "begin_time": _get_varint(fd, 2), "end_time": _get_varint(fd, 3),
"nick_name": _get_string(fd, 4), "msg": [_decode_forward_msg(d) for d in _parse_repeated(fd, 5)],
}
except Exception as e:
if DEBUG_MODE:
logger.debug("[yuanbao_proto] decode_forward_msg_data failed: %s", e)
except Exception:
return None
# ============================================================
# Outbound message encoding
# ============================================================
# ---- Outbound message encoding
def encode_send_c2c_message(
to_account: str,
msg_body: list,
from_account: str,
msg_id: str = "",
msg_random: int = 0,
msg_seq: Optional[int] = None,
group_code: str = "",
trace_id: str = "",
to_account: str, msg_body: list, from_account: str, msg_id: str = "", msg_random: int = 0,
msg_seq: Optional[int] = None, group_code: str = "", trace_id: str = "",
) -> bytes:
"""Encode a SendC2CMessageReq and return the full ConnMsg bytes (ready to send over WebSocket).
"""SendC2CMessageReq → 完整 ConnMsg bytes（可直接发送）。
SendC2CMessageReq: 1 msg_id, 2 to_account, 3 from_account, 4 msg_random,
5 msg_body (repeated MsgBodyElement), 6 group_code, 7 msg_seq, 8 log_ext.
msg_body items are {"msg_type": str, "msg_content": dict}; msg_id doubles as req_id when set;
group_code is filled for the "private chat originating from a group" case.
"""
biz_bytes = b""
if msg_id:
biz_bytes += _s(1, msg_id)
biz_bytes += _s(2, to_account)
if from_account:
biz_bytes += _s(3, from_account)
if msg_random:
biz_bytes += _v(4, msg_random)
biz_bytes += _encode_msg_body(5, msg_body)
if group_code:
biz_bytes += _s(6, group_code)
if msg_seq is not None:
biz_bytes += _v(7, msg_seq)
if trace_id:
biz_bytes += _m(8, _encode_log_ext(trace_id))
_dbg("encode_send_c2c biz payload", biz_bytes)
return _biz_request("send_c2c_message", "c2c", biz_bytes, msg_id)
return _biz_request("send_c2c_message", "c2c", _encode_parts([
(1, "s", msg_id), (2, "S", to_account), (3, "s", from_account), (4, "v", msg_random),
(5, "b", msg_body), (6, "s", group_code), (7, "n", msg_seq), (8, "t", trace_id),
]), msg_id)
def encode_send_group_message(
group_code: str,
msg_body: list,
from_account: str,
msg_id: str = "",
to_account: str = "",
random: str = "",
msg_seq: Optional[int] = None,
ref_msg_id: str = "",
trace_id: str = "",
group_code: str, msg_body: list, from_account: str, msg_id: str = "", to_account: str = "", random: str = "",
msg_seq: Optional[int] = None, ref_msg_id: str = "", trace_id: str = "",
) -> bytes:
"""Encode a SendGroupMessageReq and return the full ConnMsg bytes (ready to send over WebSocket).
SendGroupMessageReq: 1 msg_id, 2 group_code, 3 from_account, 4 to_account (usually empty),
5 random (string), 6 msg_body, 7 ref_msg_id (quoted message), 8 msg_seq, 9 log_ext.
"""
biz_bytes = b""
if msg_id:
biz_bytes += _s(1, msg_id)
biz_bytes += _s(2, group_code)
if from_account:
biz_bytes += _s(3, from_account)
if to_account:
biz_bytes += _s(4, to_account)
if random:
biz_bytes += _s(5, random)
biz_bytes += _encode_msg_body(6, msg_body)
if ref_msg_id:
biz_bytes += _s(7, ref_msg_id)
if msg_seq is not None:
biz_bytes += _v(8, msg_seq)
if trace_id:
biz_bytes += _m(9, _encode_log_ext(trace_id))
_dbg("encode_send_group biz payload", biz_bytes)
return _biz_request("send_group_message", "grp", biz_bytes, msg_id)
"""SendGroupMessageReq → 完整 ConnMsg bytes。to_account usually empty; ref_msg_id = quoted message."""
return _biz_request("send_group_message", "grp", _encode_parts([
(1, "s", msg_id), (2, "S", group_code), (3, "s", from_account), (4, "s", to_account), (5, "s", random),
(6, "b", msg_body), (7, "s", ref_msg_id), (8, "n", msg_seq), (9, "t", trace_id),
]), msg_id)
# ============================================================
# AuthBind / Ping / PushAck
# ============================================================
# ---- AuthBind / Ping / PushAck
def encode_auth_bind(
biz_id: str,
uid: str,
source: str,
token: str,
msg_id: str,
app_version: str = "",
operation_system: str = "",
bot_version: str = "",
route_env: str = "",
biz_id: str, uid: str, source: str, token: str, msg_id: str, app_version: str = "", operation_system: str = "",
bot_version: str = "", route_env: str = "",
) -> bytes:
"""构造 auth-bind 请求 ConnMsg bytes。
"""auth-bind 请求 ConnMsg bytes。
AuthBindReq: 1 biz_id, 2 auth_info (AuthInfo{1 uid, 2 source, 3 token}),
3 device_info (DeviceInfo{1 app_version, 2 app_operation_system, 10 instance_id, 24 bot_version}),
5 env_name
"""
auth_buf = _s(1, uid) + _s(2, source) + _s(3, token)
dev_buf = b""
if app_version:
dev_buf += _s(1, app_version)
if operation_system:
dev_buf += _s(2, operation_system)
dev_buf += _s(10, str(HERMES_INSTANCE_ID))
if bot_version:
dev_buf += _s(24, bot_version)
req_buf = _s(1, biz_id) + _m(2, auth_buf) + _m(3, dev_buf)
if route_env:
req_buf += _s(5, route_env)
return encode_conn_msg_full(
cmd_type=CMD_TYPE["Request"], cmd=CMD["AuthBind"], seq_no=next_seq_no(), msg_id=msg_id,
module=MODULE["ConnAccess"], data=req_buf,
)
dev_buf = _encode_parts([
(1, "s", app_version), (2, "s", operation_system), (10, "S", str(HERMES_INSTANCE_ID)), (24, "s", bot_version),
])
req_buf = _encode_parts([
(1, "S", biz_id), (2, "m", _s(1, uid) + _s(2, source) + _s(3, token)), (3, "m", dev_buf), (5, "s", route_env),
])
return _conn_request(CMD_TYPE["Request"], CMD["AuthBind"], msg_id, MODULE["ConnAccess"], req_buf)
def encode_ping(msg_id: str) -> bytes:
"""构造 ping 请求 ConnMsg bytes（PingReq 为空消息）"""
return encode_conn_msg_full(
cmd_type=CMD_TYPE["Request"], cmd=CMD["Ping"], seq_no=next_seq_no(), msg_id=msg_id,
module=MODULE["ConnAccess"], data=b"",
)
"""ping 请求 ConnMsg bytes（PingReq 为空消息）"""
return _conn_request(CMD_TYPE["Request"], CMD["Ping"], msg_id, MODULE["ConnAccess"])
def encode_push_ack(original_head: dict) -> bytes:
"""构造 push ACK 回包（回显原 head 的 cmd / msg_id / module）"""
return encode_conn_msg_full(
cmd_type=CMD_TYPE["PushAck"], cmd=original_head.get("cmd", ""), seq_no=next_seq_no(),
msg_id=original_head.get("msg_id", ""), module=original_head.get("module", ""), data=b"",
"""push ACK 回包（回显原 head 的 cmd / msg_id / module）"""
return _conn_request(
CMD_TYPE["PushAck"], original_head.get("cmd", ""), original_head.get("msg_id", ""), original_head.get("module", ""),
)
# ============================================================
# Heartbeat / 群信息 / 群成员列表
# ============================================================
# ---- Heartbeat / 群信息 / 群成员列表
def encode_send_private_heartbeat(from_account: str, to_account: str, heartbeat: int = WS_HEARTBEAT_RUNNING) -> bytes:
"""SendPrivateHeartbeatReq{1 from_account, 2 to_account, 3 heartbeat (RUNNING=1, FINISH=2)} → ConnMsg bytes"""
buf = _s(1, from_account) + _s(2, to_account) + _v(3, heartbeat)
return _biz_request("send_private_heartbeat", "hb_priv", buf)
return _biz_request("send_private_heartbeat", "hb_priv", _s(1, from_account) + _s(2, to_account) + _v(3, heartbeat))
def encode_send_group_heartbeat(
@@ -693,8 +483,7 @@ def encode_send_group_heartbeat(
) -> bytes:
"""SendGroupHeartbeatReq{1 from_account, 2 to_account (群场景留空), 3 group_code,
4 send_time (ms; 0 → now), 5 heartbeat} → ConnMsg bytes"""
import time as _time
ts = send_time or int(_time.time() * 1000)
ts = send_time or int(time.time() * 1000)
buf = _s(1, from_account) + _s(2, "") + _s(3, group_code) + _v(4, ts) + _v(5, heartbeat)
return _biz_request("send_group_heartbeat", "hb_grp", buf)
@@ -705,26 +494,22 @@ def encode_query_group_info(group_code: str) -> bytes:
def decode_query_group_info_rsp(data: bytes) -> Optional[dict]:
"""解码 QueryGroupInfoRsp（对齐 TS biz-codec / member.ts queryGroupInfo）。
QueryGroupInfoRsp{1 code, 2 message, 3 GroupInfo{1 group_name, 2 group_owner_user_id,
3 group_owner_nickname, 4 group_size}} → {code, message?, group_name, owner_id,
owner_nickname, member_count}，解析失败返回 None。
"""
"""QueryGroupInfoRsp{1 code, 2 message, 3 GroupInfo{1 group_name, 2 group_owner_user_id,
3 group_owner_nickname, 4 group_size}} → {code, message?, group_name, owner_id, owner_nickname,
member_count}（对齐 TS member.ts queryGroupInfo）；解析失败返回 None。"""
try:
fdict = _parse_dict(data)
result: dict = {"code": _get_varint(fdict, 1, 0)}
msg = _get_string(fdict, 2)
if msg:
result["message"] = msg
result: dict = {"code": _get_varint(fdict, 1)}
if _get_string(fdict, 2):
result["message"] = _get_string(fdict, 2)
# field 3 taken regardless of wire type; non-bytes payloads fall back to defaults
gi_entries = fdict.get(3, [])
gi_bytes = gi_entries[0][1] if gi_entries else b""
gi = _parse_dict(gi_bytes) if gi_bytes and isinstance(gi_bytes, (bytes, bytearray)) else {}
result["group_name"] = _get_string(gi, 1)
result["owner_id"] = _get_string(gi, 2)
result["owner_nickname"] = _get_string(gi, 3)
result["member_count"] = _get_varint(gi, 4, 0)
result.update(
group_name=_get_string(gi, 1), owner_id=_get_string(gi, 2), owner_nickname=_get_string(gi, 3),
member_count=_get_varint(gi, 4),
)
return result
except Exception:
return None
@@ -732,36 +517,24 @@ def decode_query_group_info_rsp(data: bytes) -> Optional[dict]:
def encode_get_group_member_list(group_code: str, offset: int = 0, limit: int = 200) -> bytes:
"""GetGroupMemberListReq{1 group_code, 2 offset, 3 limit} → ConnMsg bytes"""
buf = _s(1, group_code)
if offset:
buf += _v(2, offset)
buf += _v(3, limit)
return _biz_request("get_group_member_list", "gml", buf)
return _biz_request("get_group_member_list", "gml", _s(1, group_code) + (_v(2, offset) if offset else b"") + _v(3, limit))
def decode_get_group_member_list_rsp(data: bytes) -> Optional[dict]:
"""解码 GetGroupMemberListRsp{1 code, 2 message, 3 members (repeated MemberInfo), 4 next_offset,
5 is_complete}；MemberInfo{1 user_id, 2 nickname, 3 role (0=member,1=admin,2=owner),
4 join_time, 5 name_card (群昵称)}。member dict 过滤空值但保留 role；解析失败返回 None。
"""
"""GetGroupMemberListRsp{1 code, 2 message, 3 members (repeated MemberInfo), 4 next_offset, 5 is_complete}；
MemberInfo{1 user_id, 2 nickname, 3 role (0=member,1=admin,2=owner), 4 join_time, 5 name_card (群昵称)}。
member dict 过滤空值但保留 role；解析失败返回 None。"""
try:
fdict = _parse_dict(data)
members = []
for mdict in map(_parse_dict, _get_repeated_bytes(fdict, 3)):
member = {
"user_id": _get_string(mdict, 1),
"nickname": _get_string(mdict, 2),
"role": _get_varint(mdict, 3),
"join_time": _get_varint(mdict, 4),
"name_card": _get_string(mdict, 5),
}
members.append({k: v for k, v in member.items() if v or k == "role"})
members = [
{"user_id": _get_string(m, 1), "nickname": _get_string(m, 2), "role": _get_varint(m, 3),
"join_time": _get_varint(m, 4), "name_card": _get_string(m, 5)}
for m in _parse_repeated(fdict, 3)
]
return {
"code": _get_varint(fdict, 1, 0),
"message": _get_string(fdict, 2),
"members": members,
"next_offset": _get_varint(fdict, 4),
"is_complete": bool(_get_varint(fdict, 5)),
"code": _get_varint(fdict, 1), "message": _get_string(fdict, 2),
"members": [{k: v for k, v in mem.items() if v or k == "role"} for mem in members],
"next_offset": _get_varint(fdict, 4), "is_complete": bool(_get_varint(fdict, 5)),
}
except Exception:
return None
gateway/platforms/yuanbao_sticker.py
+32 -82
View File
@@ -1,11 +1,9 @@
"""
Yuanbao sticker (TIMFaceElem) support. Ported from yuanbao-openclaw-plugin/src/sticker/.
TIMFaceElem wire format:
{"msg_type": "TIMFaceElem",
"msg_content": {"index": 0, # always 0 per Yuanbao convention
"data": "<json>"}} # serialised sticker metadata so the receiver
# can look up the asset in the emoji pack
TIMFaceElem wire format: {"msg_type": "TIMFaceElem", "msg_content": {"index": 0, "data": "<json>"}}
index is always 0 per Yuanbao convention; data is serialised sticker metadata so the receiver
can look up the asset in the emoji pack.
"""
from __future__ import annotations
@@ -14,6 +12,7 @@ import json
import random
import re
import unicodedata
from collections import Counter
from typing import Optional
# Sticker catalogue – ported from builtin-stickers.json. Every builtin sticker is in
@@ -90,11 +89,7 @@ STICKER_MAP: dict[str, dict] = {
def get_sticker_by_name(name: str) -> Optional[dict]:
"""按名称查找贴纸，支持模糊匹配。
匹配优先级：完全相等 → name 与查询词互为子串 → description 包含查询词 →
search_stickers 模糊评分最高的一条。找不到返回 None。
"""
"""完全相等 → name 与查询词互为子串 → description 包含查询词 → search_stickers 最高分；找不到返回 None。"""
if not name:
return None
query = name.strip()
@@ -104,7 +99,7 @@ def get_sticker_by_name(name: str) -> Optional[dict]:
if query in key or key in query:
return sticker
for sticker in STICKER_MAP.values():
if query in sticker.get("description", ""):
if query in sticker["description"]:
return sticker
matches = search_stickers(query, limit=1)
return matches[0] if matches else None
@@ -112,27 +107,18 @@ def get_sticker_by_name(name: str) -> Optional[dict]:
def get_random_sticker(category: str = None) -> dict:
"""随机贴纸；指定 category 时优先在 description/name 含该词的贴纸中选取。"""
if category:
candidates = [
s for s in STICKER_MAP.values()
if category in s.get("description", "") or category in s.get("name", "")
]
if candidates:
return random.choice(candidates)
return random.choice(list(STICKER_MAP.values()))
candidates = [s for s in STICKER_MAP.values() if category in s["description"] or category in s["name"]] if category else []
return random.choice(candidates or list(STICKER_MAP.values()))
def get_sticker_by_id(sticker_id: str) -> Optional[dict]:
"""按 sticker_id 精确查找贴纸。"""
if not sticker_id:
return None
sid = str(sticker_id).strip()
return next((s for s in STICKER_MAP.values() if s.get("sticker_id") == sid), None)
return next((s for s in STICKER_MAP.values() if s["sticker_id"] == sid), None)
# ---------------------------------------------------------------------------
# 模糊搜索（对齐 chatbot-web yuanbao-openclaw-plugin/sticker-cache.ts.searchStickers）
# ---------------------------------------------------------------------------
_PUNCT_RE = re.compile(r"[\s\u3000\-_·.,，。!！?？\"“”'‘’、/\\]+")
@@ -146,28 +132,14 @@ def _compact_text(raw: str) -> str:
def _multiset_char_hit_ratio(needle: str, haystack: str) -> float:
if not needle:
return 0.0
bag: dict[str, int] = {}
for ch in haystack:
bag[ch] = bag.get(ch, 0) + 1
hits = 0
for ch in needle:
n = bag.get(ch, 0)
if n > 0:
hits += 1
bag[ch] = n - 1
return hits / len(needle)
return sum((Counter(needle) & Counter(haystack)).values()) / len(needle) if needle else 0.0
def _bigram_jaccard(a: str, b: str) -> float:
if len(a) < 2 or len(b) < 2:
return 0.0
A = {a[i:i + 2] for i in range(len(a) - 1)}
B = {b[i:i + 2] for i in range(len(b) - 1)}
inter = len(A & B)
union = len(A) + len(B) - inter
return inter / union if union else 0.0
A, B = ({x[i:i + 2] for i in range(len(x) - 1)} for x in (a, b))
return len(A & B) / len(A | B)
def _longest_subsequence_ratio(needle: str, haystack: str) -> float:
@@ -177,8 +149,7 @@ def _longest_subsequence_ratio(needle: str, haystack: str) -> float:
for ch in haystack:
if j >= len(needle):
break
if ch == needle[j]:
j += 1
j += ch == needle[j]
return j / len(needle)
@@ -189,21 +160,16 @@ def _score_field(haystack: str, query: str) -> float:
return 0.0
hay_c = _compact_text(haystack)
q_c = _compact_text(query)
best = 0.0
if hay == q:
best = max(best, 100.0)
if q in hay:
best = max(best, 92 + min(6, len(q)))
if len(q) >= 2 and hay.startswith(q):
best = max(best, 88.0)
if q_c and q_c in hay_c:
best = max(best, 86.0)
best = max(best, _multiset_char_hit_ratio(q_c, hay_c) * 62)
best = max(best, _bigram_jaccard(q_c, hay_c) * 58)
best = max(best, _longest_subsequence_ratio(q_c, hay_c) * 52)
if len(q) == 1 and q in hay:
best = max(best, 68.0)
return best
return max(
100.0 if hay == q else 0.0,
92 + min(6, len(q)) if q in hay else 0.0,
88.0 if len(q) >= 2 and hay.startswith(q) else 0.0,
86.0 if q_c and q_c in hay_c else 0.0,
_multiset_char_hit_ratio(q_c, hay_c) * 62,
_bigram_jaccard(q_c, hay_c) * 58,
_longest_subsequence_ratio(q_c, hay_c) * 52,
68.0 if len(q) == 1 and q in hay else 0.0,
)
def search_stickers(query: str, limit: int = 10) -> list[dict]:
@@ -216,35 +182,23 @@ def search_stickers(query: str, limit: int = 10) -> list[dict]:
q_norm = _normalize_text(query)
if not query or not q_norm:
return list(STICKER_MAP.values())[:safe_limit]
scored: list[tuple[float, dict]] = []
for sticker in STICKER_MAP.values():
name_s = _score_field(sticker.get("name", ""), query)
desc_s = _score_field(sticker.get("description", ""), query) * 0.88
sid_norm = _normalize_text(str(sticker.get("sticker_id", "")).strip())
sid_norm = _normalize_text(sticker["sticker_id"].strip())
id_s = 100.0 if sid_norm == q_norm else 84.0 if sid_norm and q_norm in sid_norm else 0.0
scored.append((max(name_s, desc_s, id_s), sticker))
scored.append((max(_score_field(sticker["name"], query), _score_field(sticker["description"], query) * 0.88, id_s), sticker))
scored.sort(key=lambda x: x[0], reverse=True)
top = scored[0][0] if scored else 0
if top <= 0:
return [s for _, s in scored[:safe_limit]]
if top >= 22:
floor = 18.0
elif top >= 12:
floor = max(10.0, top * 0.5)
else:
floor = max(6.0, top * 0.35)
floor = 18.0 if top >= 22 else max(10.0, top * 0.5) if top >= 12 else max(6.0, top * 0.35)
filtered = [pair for pair in scored if pair[0] >= floor]
return [s for _, s in (filtered or scored)[:safe_limit]]
def build_face_msg_body(face_index: int, face_type: int = 1, data: Optional[str] = None) -> list:
"""构造 TIMFaceElem 消息体。
Yuanbao 约定 index 固定 0（服务端通过 data 字段识别表情）；face_index > 0 视为旧版 QQ 表情 ID。
face_type 为兼容旧接口保留，不影响 wire format。data 为 None 时仅传 index。
"""
"""TIMFaceElem 消息体。Yuanbao 约定 index 固定 0（服务端通过 data 字段识别表情）；face_index > 0 视为旧版 QQ
表情 ID。face_type 为兼容旧接口保留，不影响 wire format。data 为 None 时仅传 index。"""
msg_content: dict = {"index": face_index}
if data is not None:
msg_content["data"] = data
@@ -255,14 +209,10 @@ def build_sticker_msg_body(sticker: dict) -> list:
"""从 STICKER_MAP 的 sticker dict 构造 TIMFaceElem 消息体（data 字段与原始 JS 插件一致）。"""
data_payload = json.dumps(
{
"sticker_id": sticker["sticker_id"],
"package_id": sticker["package_id"],
"width": sticker.get("width", 128),
"height": sticker.get("height", 128),
"formats": sticker.get("formats", "png"),
"name": sticker["name"],
"sticker_id": sticker["sticker_id"], "package_id": sticker["package_id"],
"width": sticker.get("width", 128), "height": sticker.get("height", 128),
"formats": sticker.get("formats", "png"), "name": sticker["name"],
},
ensure_ascii=False,
separators=(",", ":"),
ensure_ascii=False, separators=(",", ":"),
)
return build_face_msg_body(face_index=0, data=data_payload)