From 51d8df6e6aea40e3eb8cbfbf8babe77c2c90911d Mon Sep 17 00:00:00 2001 From: qihu-bot Date: Sat, 15 Aug 2026 18:40:01 +0800 Subject: [PATCH] =?UTF-8?q?#971=20=E5=BE=AE=E4=BF=A1Bot=20P2=20-=20?= =?UTF-8?q?=E6=8E=A5=E6=94=B6=E4=BE=A7=E5=AA=92=E4=BD=93=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=E8=A7=A3=E5=AF=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现CDN媒体下载解密(parse_aes_key支持base64(raw16B)和base64(hex32)两种编码)+extract_inbound_media(图片/语音/文件/视频)。接收循环检测媒体消息下载解密保存到/tmp。图片已保存供后续vision模型接入。 --- backend/python/app/api/im_bot_user.py | 41 +++++++- backend/python/app/api/im_ilink_media.py | 116 +++++++++++++++++++++++ 2 files changed, 156 insertions(+), 1 deletion(-) diff --git a/backend/python/app/api/im_bot_user.py b/backend/python/app/api/im_bot_user.py index d319d4e..bdd2281 100644 --- a/backend/python/app/api/im_bot_user.py +++ b/backend/python/app/api/im_bot_user.py @@ -425,6 +425,42 @@ def _wechat_extract_text(msg: dict) -> str: return "\n".join(parts) +def _extract_and_save_inbound_media(msg: dict) -> str: + """检测入站媒体消息(图片/语音/文件/视频)，下载解密保存，返回给 LLM 的提示文本。 + + 无媒体或解密失败返回空字符串。图片已保存到本地，供后续 vision 模型接入。 + """ + try: + from app.api.im_ilink_media import extract_inbound_media + import os as _os + for it in (msg.get("item_list") or []): + if it.get("type") in (2, 3, 4, 5): + kind, data = extract_inbound_media(it) + if kind and data: + _dir = "/tmp/wechat_inbound_media" + _os.makedirs(_dir, exist_ok=True) + _ext = {"image": "jpg", "voice": "silk", "file": "bin", "video": "mp4"}.get(kind, "bin") + _path = _os.path.join(_dir, "%d_%s.%s" % (int(time.time() * 1000), kind, _ext)) + with open(_path, "wb") as _f: + _f.write(data) + _label = {"image": "图片", "voice": "语音", "file": "文件", "video": "视频"}.get(kind, kind) + try: + with open("/tmp/wechat_inbound_media.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " %s saved=%s size=%d\n" + % (kind, _path, len(data))) + except Exception: + pass + return "[用户发送了%s，已保存到 %s（%d 字节）]" % (_label, _path, len(data)) + return "" + except Exception as _e: + try: + with open("/tmp/wechat_inbound_media.log", "a", encoding="utf-8") as _f: + _f.write(_beijing_now().isoformat() + " MEDIA_EXTRACT_ERR %s\n" % (_e,)) + except Exception: + pass + return "" + + def _wechat_sender_id(m: dict) -> str: """从入站消息中提取发送者 id（兼容多套字段名）。""" for k in ("from_user_id", "from_user_name", "from_username", "user_id", @@ -576,7 +612,10 @@ def _wechat_receiver_loop(account_id: int): continue # 仅跳过 bot 自己发出的回声，其余（含后续追问）均尝试处理 content = _wechat_extract_text(m) if not content: - continue + # 入站媒体（图片/语音/文件/视频）：下载解密保存，供 LLM 感知 + content = _extract_and_save_inbound_media(m) + if not content: + continue sender = _wechat_sender_id(m) ctx = _wechat_context_token(m) # 诊断：记录原始入站消息，定位字段名 / 第二轮不回复 diff --git a/backend/python/app/api/im_ilink_media.py b/backend/python/app/api/im_ilink_media.py index 1039697..af5277d 100644 --- a/backend/python/app/api/im_ilink_media.py +++ b/backend/python/app/api/im_ilink_media.py @@ -19,6 +19,7 @@ import base64 import hashlib import math import os +import re from urllib.parse import quote import requests @@ -212,3 +213,118 @@ def build_video_item(uploaded: dict) -> dict: "video_size": uploaded["filesize"], }, } + + +# ============================================================ +# 接收侧媒体下载解密（P2） +# 参考 openclaw cdn/pic-decrypt.ts + media/media-download.ts +# ============================================================ + +def parse_aes_key(aes_key_base64: str) -> bytes: + """解析 CDNMedia.aes_key 为原始 16 字节 AES key。 + + 两种真实编码（openclaw parseAesKey）： + - base64(raw 16 bytes) → 图片 (media.aes_key) + - base64(hex 32 chars 字符串) → 文件/语音/视频 + 第二种需 base64 解码后得到 32 个 ASCII hex 字符，再 hex 解码成 16 字节。 + """ + decoded = base64.b64decode(aes_key_base64) + if len(decoded) == 16: + return decoded + if len(decoded) == 32 and re.match(r"^[0-9a-fA-F]{32}$", decoded.decode("ascii", "ignore")): + return bytes.fromhex(decoded.decode("ascii")) + raise ValueError( + f"aes_key 解码后应为 16 字节或 32 字符 hex，实际 {len(decoded)} 字节" + ) + + +def download_and_decrypt_media( + encrypted_query_param: str, + aes_key_base64: str, + cdn_base_url: str = WECHAT_CDN_BASE, + full_url: str = None, +) -> bytes: + """下载并 AES-128-ECB 解密 CDN 媒体文件，返回明文字节。""" + key = parse_aes_key(aes_key_base64) + if full_url: + url = full_url + else: + url = ( + f"{cdn_base_url}/download" + f"?encrypted_query_param={quote(encrypted_query_param, safe='')}" + ) + resp = requests.get(url, timeout=60) + if not resp.ok: + raise RuntimeError(f"CDN download failed HTTP {resp.status_code}: {resp.text[:150]}") + encrypted = resp.content + return decrypt_aes_ecb(encrypted, key) + + +def extract_inbound_media(item: dict) -> tuple: + """从入站 MessageItem 提取媒体并下载解密。 + + 返回 (media_kind, media_bytes)，不支持的类型或失败返回 (None, None)。 + media_kind ∈ {"image", "voice", "file", "video"} + """ + try: + t = item.get("type") + if t == 2: # 图片 + img = item.get("image_item") or {} + media = img.get("media") or {} + eqp = media.get("encrypt_query_param") or "" + full = media.get("full_url") + if not eqp and not full: + return None, None + # 图片 key 优先用 image_item.aeskey（hex string）转 base64 + if img.get("aeskey"): + aes_key_b64 = base64.b64encode(bytes.fromhex(img["aeskey"])).decode() + else: + aes_key_b64 = media.get("aes_key") or "" + if aes_key_b64: + return "image", download_and_decrypt_media(eqp, aes_key_b64, full_url=full) + return "image", _download_plain(eqp, full) + elif t == 3: # 语音 + voice = item.get("voice_item") or {} + media = voice.get("media") or {} + eqp = media.get("encrypt_query_param") or "" + full = media.get("full_url") + aes_key_b64 = media.get("aes_key") or "" + if not aes_key_b64: + return None, None + return "voice", download_and_decrypt_media(eqp, aes_key_b64, full_url=full) + elif t == 4: # 文件 + fitem = item.get("file_item") or {} + media = fitem.get("media") or {} + eqp = media.get("encrypt_query_param") or "" + full = media.get("full_url") + aes_key_b64 = media.get("aes_key") or "" + if not aes_key_b64: + return None, None + return "file", download_and_decrypt_media(eqp, aes_key_b64, full_url=full) + elif t == 5: # 视频 + vitem = item.get("video_item") or {} + media = vitem.get("media") or {} + eqp = media.get("encrypt_query_param") or "" + full = media.get("full_url") + aes_key_b64 = media.get("aes_key") or "" + if not aes_key_b64: + return None, None + return "video", download_and_decrypt_media(eqp, aes_key_b64, full_url=full) + except Exception: + return None, None + return None, None + + +def _download_plain(encrypted_query_param: str, full_url: str = None) -> bytes: + """下载未加密的 CDN 媒体（无 aes_key 时）。""" + if full_url: + url = full_url + else: + url = ( + f"{WECHAT_CDN_BASE}/download" + f"?encrypted_query_param={quote(encrypted_query_param, safe='')}" + ) + resp = requests.get(url, timeout=60) + if not resp.ok: + raise RuntimeError(f"CDN plain download failed HTTP {resp.status_code}") + return resp.content