diff --git a/runtime/application/gateway/inbound_service.py b/runtime/application/gateway/inbound_service.py index da17a545..e8311506 100644 --- a/runtime/application/gateway/inbound_service.py +++ b/runtime/application/gateway/inbound_service.py @@ -300,6 +300,49 @@ def _parse_generic_inbound(channel_name: str, payload: dict[str, Any]) -> Inboun ) +def _parse_message_attachments(raw: Any) -> list[dict[str, Any]]: + if raw is None: + return [] + obj = raw + if isinstance(raw, str): + s = str(raw or "").strip() + if not s: + return [] + try: + import json + + obj = json.loads(s) + except Exception: + return [] + if not isinstance(obj, list): + return [] + return [x for x in obj if isinstance(x, dict)] + + +def _collect_reply_attachments_from_history(*, store: Any, session_id: str, reply_text: str) -> list[dict[str, Any]]: + sid = str(session_id or "").strip() + if not sid: + return [] + try: + rows = store.get_messages(session_id=sid, limit=120) + except Exception: + rows = [] + target = str(reply_text or "").strip() + fallback: list[dict[str, Any]] = [] + matched: list[dict[str, Any]] = [] + for row in rows or []: + role = str(getattr(row, "role", "") or "").strip().lower() + if role != "assistant": + continue + atts = _parse_message_attachments(getattr(row, "attachments", None)) + if atts: + fallback = atts + content = str(getattr(row, "content", "") or "").strip() + if target and content == target and atts: + matched = atts + return matched or fallback + + def _should_suppress_channel_reply(*, channel: str, text: str) -> bool: ch = str(channel or "").strip().lower() if ch not in {"wechat", "weixin"}: @@ -364,6 +407,7 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: reply = ("绑定成功。\n\n" + _menu_text()) if info else "绑定失败:无效或已使用的绑定码。" else: reply = "" + reply_attachments: list[dict[str, Any]] = [] ident = store.resolve_user_by_channel_identity_v2( channel=inbound.channel, account_id=account_id, @@ -506,10 +550,17 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: ).reply_text or "" ).strip() + reply_attachments = _collect_reply_attachments_from_history( + store=store, + session_id=str(session_id), + reply_text=reply, + ) except Exception as e: reply = f"抱歉,处理消息时出错:{type(e).__name__}: {e}" + reply_attachments = [] else: reply = "收到消息,但内容为空。请直接发送文本。" + reply_attachments = [] if preface: if reply: reply = f"{preface}\n\n{reply}" @@ -527,7 +578,7 @@ def process_inbound_payload(payload: dict[str, Any]) -> dict[str, Any]: "channel": inbound.channel, "chat_id": inbound.external_chat_id, "text": reply, - "attachments": [], + "attachments": list(reply_attachments or []), "metadata": {}, } ] diff --git a/runtime/operations/weixin_bridge/official_runner.ts b/runtime/operations/weixin_bridge/official_runner.ts index 569d8e1b..aa956d57 100644 --- a/runtime/operations/weixin_bridge/official_runner.ts +++ b/runtime/operations/weixin_bridge/official_runner.ts @@ -281,6 +281,32 @@ function buildAttachmentsFromMedia(mediaOpts: Json): Json[] { return out; } +function _decodeReplyBase64Attachment(att: Json): { filePath: string; mime: string } | null { + const b64Raw = String( + att.data_base64 || att.media_base64 || att.image_base64 || att.video_base64 || att.audio_base64 || att.data || "", + ).trim(); + if (!b64Raw) return null; + let mime = String(att.media_type || att.mime_type || att.mime || "application/octet-stream").trim(); + let payload = b64Raw; + const m = b64Raw.match(/^data:([^;,]+);base64,(.*)$/i); + if (m) { + mime = String(m[1] || mime).trim() || mime; + payload = String(m[2] || "").trim(); + } + try { + const buf = Buffer.from(payload.replace(/\s+/g, ""), "base64"); + if (!buf.length) return null; + const dir = path.join(STATE_DIR, "media", "outbound"); + ensureDir(dir); + const ext = inferExtension(mime || "application/octet-stream", String(att.name || att.filename || "").trim()); + const filePath = path.join(dir, `${Date.now()}-${Math.random().toString(16).slice(2, 10)}${ext}`); + fs.writeFileSync(filePath, buf); + return { filePath, mime }; + } catch { + return null; + } +} + async function handleInboundMessage( modules: OfficialModules, params: { @@ -349,8 +375,18 @@ async function handleInboundMessage( const replyContextToken = String( reply.context_token || modules.getContextToken(params.accountId, deliverTo) || contextToken || "", ).trim(); - if (mediaPath || mediaUrl) { - const filePath = mediaPath || mediaUrl; + const replyAtts = Array.isArray(reply.attachments) ? (reply.attachments as Json[]) : []; + let inlineMediaPath = ""; + for (const att of replyAtts) { + if (!att || typeof att !== "object") continue; + const decoded = _decodeReplyBase64Attachment(att); + if (decoded?.filePath) { + inlineMediaPath = decoded.filePath; + break; + } + } + if (mediaPath || mediaUrl || inlineMediaPath) { + const filePath = mediaPath || mediaUrl || inlineMediaPath; await modules.sendWeixinMediaFile({ filePath, to: deliverTo, diff --git a/runtime/operations/whatsapp_bridge/baileys_runner.ts b/runtime/operations/whatsapp_bridge/baileys_runner.ts index 27110e8c..34880a6c 100644 --- a/runtime/operations/whatsapp_bridge/baileys_runner.ts +++ b/runtime/operations/whatsapp_bridge/baileys_runner.ts @@ -76,6 +76,72 @@ function buildInboundPayload(params: { }; } +function decodeBase64Payload(raw: string): { mime: string; data: Buffer } | null { + const s = String(raw || "").trim(); + if (!s) return null; + let mime = "application/octet-stream"; + let payload = s; + const m = s.match(/^data:([^;,]+);base64,(.*)$/i); + if (m) { + mime = String(m[1] || mime).trim() || mime; + payload = String(m[2] || "").trim(); + } + try { + const data = Buffer.from(payload.replace(/\s+/g, ""), "base64"); + if (!data.length) return null; + return { mime, data }; + } catch { + return null; + } +} + +async function sendReplyWithAttachments(params: { + sock: ReturnType | null; + deliverTo: string; + text: string; + reply: Json; +}): Promise { + const s = params.sock; + if (!s) return; + const reply = params.reply || {}; + const outText = String(params.text || "").trim(); + const mediaPath = String((reply as any).media_path || (reply as any).mediaPath || "").trim(); + const mediaUrl = String((reply as any).media_url || (reply as any).mediaUrl || "").trim(); + const attachments = Array.isArray((reply as any).attachments) ? ((reply as any).attachments as Json[]) : []; + + const sendMediaRef = async (source: string): Promise => { + if (!source) return false; + const msg: Json = { document: source as any }; + if (outText) (msg as any).caption = outText; + await s.sendMessage(params.deliverTo, msg as any); + return true; + }; + + if (await sendMediaRef(mediaPath || mediaUrl)) return; + + for (const att of attachments) { + if (!att || typeof att !== "object") continue; + const raw = String( + (att as any).data_base64 || (att as any).media_base64 || (att as any).image_base64 || (att as any).data || "", + ).trim(); + if (!raw) continue; + const decoded = decodeBase64Payload(raw); + if (!decoded) continue; + const msg: Json = { + document: decoded.data as any, + mimetype: decoded.mime, + fileName: String((att as any).name || (att as any).filename || "attachment.bin"), + }; + if (outText) (msg as any).caption = outText; + await s.sendMessage(params.deliverTo, msg as any); + return; + } + + if (outText) { + await s.sendMessage(params.deliverTo, { text: outText }); + } +} + async function postInbound(payload: Json): Promise { const url = `${LOCAL_BASE_URL.replace(/\/+$/, "")}/inbound/whatsapp`; const res = await fetch(url, { @@ -250,9 +316,8 @@ async function main(): Promise { if (VERBOSE) log(`inbound ok replies=${replies.length}`); for (const r of replies) { const outText = String((r as any).text || "").trim(); - if (!outText) continue; const deliverTo = String((r as any).chat_id || chatId).trim() || chatId; - await sock?.sendMessage(deliverTo, { text: outText }); + await sendReplyWithAttachments({ sock, deliverTo, text: outText, reply: r }); } } catch (err) { log(`handle message error: ${String(err)}`); diff --git a/tests/test_inbound_service_reply_suppress.py b/tests/test_inbound_service_reply_suppress.py index 67813950..60c4ef11 100644 --- a/tests/test_inbound_service_reply_suppress.py +++ b/tests/test_inbound_service_reply_suppress.py @@ -1,6 +1,12 @@ from __future__ import annotations -from oclaw.runtime.application.gateway.inbound_service import _should_suppress_channel_reply +from dataclasses import dataclass + +from oclaw.runtime.application.gateway.inbound_service import ( + _collect_reply_attachments_from_history, + _parse_message_attachments, + _should_suppress_channel_reply, +) def test_should_suppress_weixin_openai_missing_api_key_message() -> None: @@ -13,3 +19,35 @@ def test_should_not_suppress_non_weixin_channel() -> None: text = '⚠️ Missing API key for provider "openai". Configure the gateway auth for that provider, then try again.' assert _should_suppress_channel_reply(channel="admin_chat", text=text) is False + +def test_parse_message_attachments_accepts_json_string() -> None: + out = _parse_message_attachments('[{"type":"image_ref","attachment_id":"a1"}]') + assert len(out) == 1 + assert out[0].get("attachment_id") == "a1" + + +@dataclass +class _Row: + role: str + content: str + attachments: object + + +class _FakeStore: + def __init__(self, rows: list[_Row]) -> None: + self._rows = rows + + def get_messages(self, *, session_id: str, limit: int = 120) -> list[_Row]: + _ = (session_id, limit) + return list(self._rows) + + +def test_collect_reply_attachments_prefers_matching_assistant_text() -> None: + rows = [ + _Row(role="assistant", content="old", attachments='[{"attachment_id":"old"}]'), + _Row(role="assistant", content="target", attachments='[{"attachment_id":"new"}]'), + ] + out = _collect_reply_attachments_from_history(store=_FakeStore(rows), session_id="s1", reply_text="target") + assert len(out) == 1 + assert out[0].get("attachment_id") == "new" + diff --git a/tests/test_weixin_ilink_api.py b/tests/test_weixin_ilink_api.py index 60b3a1a9..683af49e 100644 --- a/tests/test_weixin_ilink_api.py +++ b/tests/test_weixin_ilink_api.py @@ -92,6 +92,14 @@ class WeixinIlinkApiTests(unittest.TestCase): { "chat_id": str(payload.get("chat_id") or ""), "text": f"native:{text}", + "attachments": [ + { + "type": "image_ref", + "name": "demo.png", + "mime_type": "image/png", + "data_base64": "aGVsbG8=", + } + ], } ], } @@ -118,6 +126,9 @@ class WeixinIlinkApiTests(unittest.TestCase): replies = data.get("replies") if isinstance(data.get("replies"), list) else [] self.assertEqual(len(replies), 1, data) self.assertEqual(str((replies[0] or {}).get("text") or ""), "native:hello native") + atts = (replies[0] or {}).get("attachments") if isinstance((replies[0] or {}).get("attachments"), list) else [] + self.assertEqual(len(atts), 1, data) + self.assertEqual(str((atts[0] or {}).get("name") or ""), "demo.png") finally: weixin_ilink_api._process_inbound_payload_usecase = old_usecase # type: ignore[assignment] @@ -134,6 +145,14 @@ class WeixinIlinkApiTests(unittest.TestCase): { "chat_id": str(payload.get("chat_id") or ""), "text": f"wa:{text}", + "attachments": [ + { + "type": "binary_ref", + "name": "demo.txt", + "mime_type": "text/plain", + "data_base64": "d2E=", + } + ], } ], } @@ -158,6 +177,9 @@ class WeixinIlinkApiTests(unittest.TestCase): replies = data.get("replies") if isinstance(data.get("replies"), list) else [] self.assertEqual(len(replies), 1, data) self.assertEqual(str((replies[0] or {}).get("text") or ""), "wa:hello whatsapp") + atts = (replies[0] or {}).get("attachments") if isinstance((replies[0] or {}).get("attachments"), list) else [] + self.assertEqual(len(atts), 1, data) + self.assertEqual(str((atts[0] or {}).get("name") or ""), "demo.txt") finally: weixin_ilink_api._process_inbound_payload_usecase = old_usecase # type: ignore[assignment] diff --git a/tests/test_weixin_official_scripts.py b/tests/test_weixin_official_scripts.py index f97be181..b98b165d 100644 --- a/tests/test_weixin_official_scripts.py +++ b/tests/test_weixin_official_scripts.py @@ -49,3 +49,11 @@ def test_official_runner_logs_active_bridge_path() -> None: text = _read("runtime/operations/weixin_bridge/official_runner.ts") assert "official runner started account=" in text assert "native reply failed; no fallback enabled" in text + + +def test_official_runner_supports_reply_attachments_base64() -> None: + text = _read("runtime/operations/weixin_bridge/official_runner.ts") + assert "_decodeReplyBase64Attachment" in text + assert "data_base64" in text + assert "media_base64" in text + assert "reply.attachments" in text diff --git a/tests/test_whatsapp_inbound_payload_shape.py b/tests/test_whatsapp_inbound_payload_shape.py index 1e0bbb06..141e41e4 100644 --- a/tests/test_whatsapp_inbound_payload_shape.py +++ b/tests/test_whatsapp_inbound_payload_shape.py @@ -15,7 +15,23 @@ def test_inbound_whatsapp_accepts_basic_payload_and_returns_replies() -> None: assert str(payload.get("account_id") or "") == "wa-default" assert str(payload.get("user_id") or "") == "111@s.whatsapp.net" assert str(payload.get("chat_id") or "") == "111@s.whatsapp.net" - return {"ok": True, "replies": [{"chat_id": payload.get("chat_id"), "text": "ok"}]} + return { + "ok": True, + "replies": [ + { + "chat_id": payload.get("chat_id"), + "text": "ok", + "attachments": [ + { + "type": "binary_ref", + "name": "hello.txt", + "mime_type": "text/plain", + "data_base64": "aGVsbG8=", + } + ], + } + ], + } try: fastapi_app.process_inbound_payload_usecase = _fake_usecase # type: ignore[assignment] @@ -35,6 +51,9 @@ def test_inbound_whatsapp_accepts_basic_payload_and_returns_replies() -> None: replies = data.get("replies") if isinstance(data.get("replies"), list) else [] assert replies and isinstance(replies[0], dict) assert str(replies[0].get("text") or "") == "ok" + atts = replies[0].get("attachments") if isinstance(replies[0].get("attachments"), list) else [] + assert len(atts) == 1 + assert str(atts[0].get("name") or "") == "hello.txt" finally: fastapi_app.process_inbound_payload_usecase = old_usecase # type: ignore[assignment] diff --git a/tests/test_whatsapp_ops_scripts.py b/tests/test_whatsapp_ops_scripts.py index 290643d9..7fd0e221 100644 --- a/tests/test_whatsapp_ops_scripts.py +++ b/tests/test_whatsapp_ops_scripts.py @@ -37,3 +37,12 @@ def test_start_all_gracefully_skips_missing_channel_sidecars() -> None: text = _read("runtime/operations/scripts/start_all.ps1") assert 'Warn "weixin sidecar skipped:' in text assert 'Warn "whatsapp sidecar skipped:' in text + + +def test_whatsapp_runner_supports_reply_attachments_base64() -> None: + text = _read("runtime/operations/whatsapp_bridge/baileys_runner.ts") + assert "sendReplyWithAttachments" in text + assert "data_base64" in text + assert "media_base64" in text + assert "media_path" in text + assert "media_url" in text