diff --git a/backend/app/api/channels.py b/backend/app/api/channels.py index ea8a25ed..89f54b46 100644 --- a/backend/app/api/channels.py +++ b/backend/app/api/channels.py @@ -2,12 +2,15 @@ import json import logging +import re import secrets import threading import time from datetime import timedelta +from urllib.parse import quote from fastapi import APIRouter, Depends, HTTPException, Query, Response +from fastapi.responses import Response as FastAPIResponse from sqlalchemy import case, text, update from sqlalchemy.exc import IntegrityError from sqlmodel import Session, select @@ -94,6 +97,23 @@ router = APIRouter(prefix="/api/enterprise/channels", tags=["enterprise:channels"]) +def _channel_attachment_metadata(metadata: object) -> list[ChannelConversationAttachmentRead]: + if not isinstance(metadata, dict): + return [] + raw_attachments = metadata.get("attachments") + if not isinstance(raw_attachments, list): + return [] + attachments: list[ChannelConversationAttachmentRead] = [] + for raw in raw_attachments: + if not isinstance(raw, dict): + continue + try: + attachments.append(ChannelConversationAttachmentRead.model_validate(raw)) + except ValueError: + continue + return attachments + + def _patch_binding_config_key( db: Session, tenant_id: str, @@ -1332,17 +1352,65 @@ def list_channel_conversation_messages( role=row.role, content=row.content, created_at=row.created_at.isoformat(), - attachments=[ - ChannelConversationAttachmentRead( - id=item.get("id"), - filename=item.get("filename"), - content_type=item.get("content_type"), - size=item.get("size"), - kind=item.get("kind"), - ) - for item in ((row.metadata_json or {}).get("attachments") or []) - ] - or None, + attachments=_channel_attachment_metadata(row.metadata_json) or None, ) for row in rows ] + + +@router.get("/{binding_id}/conversations/{session_id}/messages/{message_id}/attachments/{attachment_id}") +def get_channel_conversation_attachment( + binding_id: str, + session_id: str, + message_id: str, + attachment_id: str, + tenant_id: str = Query(...), + current_user: User = Depends(get_current_user), + db: Session = Depends(get_session), +) -> FastAPIResponse: + from app.session.attachment_store import read_staged_chat_attachment + from app.session.session_schema import ChatAttachmentRead + + ensure_current_user_tenant(tenant_id, current_user) + binding = _get_binding(db, tenant_id, binding_id) + _ensure_binding_manager(db, tenant_id, binding, current_user) + session_ids = {row.id for row in _binding_channel_sessions(db, binding)} + if session_id not in session_ids: + raise HTTPException(status_code=404, detail="Channel conversation not found") + message = db.get(Message, message_id) + if not message or message.tenant_id != tenant_id or message.session_id != session_id: + raise HTTPException(status_code=404, detail="Channel message not found") + raw_attachments = (message.metadata_json or {}).get("attachments") + if not isinstance(raw_attachments, list): + raise HTTPException(status_code=404, detail="Attachment not found") + raw = next( + (item for item in raw_attachments if isinstance(item, dict) and item.get("id") == attachment_id), + None, + ) + session = db.get(ChatSession, session_id) + if raw is None or not session or not session.user_id: + raise HTTPException(status_code=404, detail="Attachment not found") + try: + attachment = ChatAttachmentRead.model_validate(raw) + except ValueError as exc: + raise HTTPException(status_code=404, detail="Attachment not found") from exc + data = read_staged_chat_attachment( + attachment, + tenant_id=tenant_id, + user_id=session.user_id, + ) + if data is None: + raise HTTPException(status_code=404, detail="Attachment content not found") + filename = attachment.filename or "attachment" + ascii_filename = re.sub(r"[^\x20-\x7e]", "_", filename).replace('"', "'") + encoded_filename = quote(filename, safe="") + return FastAPIResponse( + content=data, + media_type=attachment.content_type or "application/octet-stream", + headers={ + "Content-Disposition": ( + f'attachment; filename="{ascii_filename}"; ' + f"filename*=UTF-8''{encoded_filename}" + ) + }, + ) diff --git a/backend/app/channels/adapters/wechat.py b/backend/app/channels/adapters/wechat.py index 8edd4c80..7bec563e 100644 --- a/backend/app/channels/adapters/wechat.py +++ b/backend/app/channels/adapters/wechat.py @@ -1,25 +1,40 @@ from __future__ import annotations +import asyncio import base64 import json import logging import os +import shutil +import ssl +import subprocess import threading import time from datetime import timedelta from typing import Any +from urllib.parse import urlparse, urlunparse from uuid import uuid4 +import aiohttp +import certifi import httpx +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from sqlalchemy import text from sqlmodel import Session, select from app.channels.adapters.base import ( ChannelInbound, + ChannelInboundAttachment, register_channel_adapter, split_channel_text, ) from app.channels.crypto import decrypt_channel_secret +from app.channels.media import ( + MAX_CHANNEL_MEDIA_BYTES, + MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES, + ensure_channel_media_size, +) from app.config import get_settings from app.db import engine from app.db.models import ChannelBinding, utc_now @@ -41,11 +56,151 @@ RECOVERY_COOLDOWN_SECONDS = 3600.0 # 连续恢复失败达上限才判真过期(expired + 清游标 + 线程退出) RECOVERY_MAX_FAILURES = 5 +WECHAT_MEDIA_DOWNLOAD_ATTEMPTS = 4 +WECHAT_CURL_DOWNLOAD_ATTEMPTS = 6 # 腾讯官方接入域名:业务请求携带 bot_token,redirect/baseurl 必须限制在官方域内 WECHAT_ALLOWED_HOSTS = ("ilinkai.weixin.qq.com",) +async def _download_wechat_cdn_limited(url: str) -> tuple[bytes, str]: + ssl_context = ssl.create_default_context(cafile=certifi.where()) + for attempt in range(WECHAT_MEDIA_DOWNLOAD_ATTEMPTS): + try: + timeout = aiohttp.ClientTimeout(total=15.0) + connector = aiohttp.TCPConnector(ssl=ssl_context) + async with ( + aiohttp.ClientSession(timeout=timeout, connector=connector) as client, + client.get(url) as response, + ): + response.raise_for_status() + content_length = int(response.headers.get("content-length") or 0) + ensure_channel_media_size(content_length, encrypted=True) + chunks: list[bytes] = [] + total = 0 + async for chunk in response.content.iter_chunked(64 * 1024): + total += len(chunk) + if total > MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES: + raise ValueError( + "微信媒体密文超过大小上限: " + f"size>{MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES}" + ) + chunks.append(chunk) + return b"".join(chunks), response.headers.get("content-type", "") + except (TimeoutError, aiohttp.ClientConnectionError, aiohttp.ClientPayloadError): + if attempt == WECHAT_MEDIA_DOWNLOAD_ATTEMPTS - 1: + raise + # 微信 CDN occasionally rejects a TLS handshake; retry with a new + # session/connection before dropping the channel attachment. + await asyncio.sleep(0.5 * (attempt + 1)) + raise RuntimeError("微信媒体下载失败") + + +async def _download_wechat_cdn_httpx(url: str) -> tuple[bytes, str]: + """Fallback for CDN nodes that reject aiohttp's TLS handshake.""" + async with httpx.AsyncClient( + verify=certifi.where(), + http2=False, + timeout=15.0, + ) as client, client.stream("GET", url) as response: + response.raise_for_status() + content_type = response.headers.get("content-type", "") + chunks: list[bytes] = [] + total = 0 + async for chunk in response.aiter_bytes(64 * 1024): + total += len(chunk) + if total > MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES: + raise ValueError("微信媒体密文超过大小上限") + chunks.append(chunk) + return b"".join(chunks), content_type + + +async def _download_wechat_cdn_curl(url: str) -> tuple[bytes, str]: + """Use the system TLS stack for CDN nodes incompatible with Python TLS.""" + curl = shutil.which("curl") + if not curl: + raise RuntimeError("微信媒体下载失败: curl 不可用") + + def run() -> bytes: + command = [ + curl, + "--silent", + "--show-error", + "--fail", + "--location", + "--http1.1", + "--user-agent", + "Mozilla/5.0", + "--max-time", + "30", + "--connect-timeout", + "10", + "--ignore-content-length", + url, + ] + last_error = "" + for attempt in range(WECHAT_CURL_DOWNLOAD_ATTEMPTS): + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + chunks: list[bytes] = [] + total = 0 + assert process.stdout is not None + while True: + chunk = process.stdout.read(64 * 1024) + if not chunk: + break + total += len(chunk) + if total > MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES: + process.kill() + process.wait() + raise ValueError("微信媒体密文超过大小上限") + chunks.append(chunk) + stderr = process.stderr.read() if process.stderr else b"" + return_code = process.wait() + # 微信 CDN sometimes advertises a stale Content-Length. curl + # returns 18 or 28 after receiving the complete encrypted payload; + # let AES/expected_size validation decide whether it is usable. + data = b"".join(chunks) + if return_code == 0 or (return_code in {18, 28} and data): + return data + last_error = stderr.decode("utf-8", errors="replace")[:200] + if attempt < WECHAT_CURL_DOWNLOAD_ATTEMPTS - 1: + time.sleep(0.5 * (attempt + 1)) + raise RuntimeError( + f"微信媒体下载失败: curl exit={return_code} {last_error}" + ) + + data = await asyncio.to_thread(run) + ensure_channel_media_size(len(data), encrypted=True) + return data, "" + + +def decrypt_wechat_media(data: bytes, aes_key: str, *, expected_size: int = 0) -> bytes: + """Decrypt iLink CDN media using the observed AES-ECB/PKCS#7 format.""" + if not aes_key: + return data + try: + decoded = base64.b64decode(aes_key, validate=True) + key = bytes.fromhex(decoded.decode("ascii")) + if len(key) not in {16, 24, 32} or len(data) % 16: + raise ValueError + decryptor = Cipher(algorithms.AES(key), modes.ECB()).decryptor() + padded = decryptor.update(data) + decryptor.finalize() + unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder() + decrypted = unpadder.update(padded) + unpadder.finalize() + except (ValueError, TypeError) as exc: + raise WeChatApiError(-1, "微信媒体解密失败") from exc + if expected_size > 0 and len(decrypted) != expected_size: + raise WeChatApiError( + -1, + f"微信媒体解密后大小不匹配 expected={expected_size} actual={len(decrypted)}", + ) + return decrypted + + def _patch_runtime_config( db_engine, binding_id: str, @@ -165,7 +320,7 @@ def __init__( self._client = httpx.Client(transport=transport) @classmethod - def for_binding(cls, binding: ChannelBinding) -> "WeChatClient": + def for_binding(cls, binding: ChannelBinding) -> WeChatClient: config = dict(binding.config_json or {}) # 防御纵深:存量 config 里的非法 baseurl 一律钳制回默认官方地址 base_url = sanitize_wechat_baseurl( @@ -290,6 +445,91 @@ def send_typing(self, ilink_user_id: str, typing_ticket: str, status: int = 1) - ) resp.raise_for_status() + def download_media(self, context_token: str, media_id: str) -> bytes: + request = self._client.build_request( + "POST", + f"{self.base_url}/ilink/bot/downloadmedia", + headers=self._business_headers(), + json={ + "context_token": context_token, + "media_id": media_id, + "base_info": self._base_info(), + }, + ) + response = self._client.send(request, stream=True) + try: + response.raise_for_status() + content_type = response.headers.get("content-type", "") + if "application/json" not in content_type.lower(): + chunks: list[bytes] = [] + total = 0 + for chunk in response.iter_bytes(64 * 1024): + total += len(chunk) + if total > MAX_CHANNEL_MEDIA_BYTES: + raise ValueError("微信媒体超过大小上限") + chunks.append(chunk) + return b"".join(chunks) + chunks = [] + total = 0 + for chunk in response.iter_bytes(64 * 1024): + total += len(chunk) + if total > MAX_CHANNEL_MEDIA_BYTES: + raise ValueError("微信媒体响应超过大小上限") + chunks.append(chunk) + raw = b"".join(chunks) + finally: + response.close() + try: + data = json.loads(raw) + except ValueError as exc: + raise WeChatApiError(-1, "下载响应格式无效") from exc + errcode = int(data.get("errcode") or data.get("ret") or 0) + if errcode: + raise WeChatApiError(errcode, str(data.get("errmsg") or "")) + raise WeChatApiError(-1, "下载响应缺少二进制内容") + + def download_media_url( + self, + full_url: str, + *, + aes_key: str = "", + expected_size: int = 0, + ) -> bytes: + """Download the CDN URL supplied by an iLink image item.""" + parsed = urlparse(full_url) + if parsed.scheme != "https" or not validate_wechat_host(parsed.hostname or ""): + raise WeChatApiError(-1, "微信媒体 URL 域名不受信任") + download_url = urlunparse(parsed) + try: + raw, content_type = asyncio.run(_download_wechat_cdn_limited(download_url)) + except (TimeoutError, aiohttp.ClientError, OSError) as exc: + logger.warning( + "微信 CDN aiohttp 下载失败，切换 httpx: host=%s error=%s", + parsed.hostname, + type(exc).__name__, + ) + try: + raw, content_type = asyncio.run(_download_wechat_cdn_httpx(download_url)) + except (TimeoutError, httpx.HTTPError, OSError) as httpx_exc: + logger.warning( + "微信 CDN httpx 下载失败，切换系统 curl: host=%s error=%s", + parsed.hostname, + type(httpx_exc).__name__, + ) + raw, content_type = asyncio.run(_download_wechat_cdn_curl(download_url)) + if "application/json" not in content_type.lower(): + return decrypt_wechat_media( + raw, + aes_key, + expected_size=expected_size, + ) + try: + data = json.loads(raw) + except ValueError as exc: + raise WeChatApiError(-1, "媒体下载响应格式无效") from exc + errcode = int(data.get("errcode") or data.get("ret") or 0) + raise WeChatApiError(errcode or -1, str(data.get("errmsg") or "媒体下载返回 JSON")) + class WeChatAdapter: """微信适配器:出站 sendmessage + 归一化 + typing + ingress(poll manager)。""" @@ -300,6 +540,29 @@ def __init__(self, client_factory=None): def normalize(self, raw: dict[str, Any]) -> ChannelInbound | None: return normalize_wechat_message(raw) + def download_media( + self, + binding: ChannelBinding, + attachment: ChannelInboundAttachment, + *, + max_bytes: int = 0, + ) -> bytes: + context_token = str(attachment.download_params.get("context_token") or "").strip() + if not context_token: + raise ValueError("微信附件下载缺少 context_token") + client = self._client_factory(binding) + full_url = str(attachment.download_params.get("full_url") or "").strip() + if full_url: + declared_size = int(attachment.download_params.get("declared_size") or 0) + expected_size = int(attachment.download_params.get("expected_size") or 0) + ensure_channel_media_size(declared_size or expected_size) + return client.download_media_url( + full_url, + aes_key=str(attachment.download_params.get("aes_key") or ""), + expected_size=expected_size, + ) + return client.download_media(context_token, attachment.media_id) + def send( self, binding: ChannelBinding, @@ -411,6 +674,89 @@ def extract_message_text(msg: dict[str, Any]) -> str: return str(msg.get("text") or msg.get("content") or "").strip() +def extract_message_attachments(msg: dict[str, Any]) -> list[ChannelInboundAttachment]: + """Extract assumed iLink image/file item descriptors.""" + items = msg.get("item_list") + if not isinstance(items, list): + return [] + context_token = str(msg.get("context_token") or "").strip() + attachments: list[ChannelInboundAttachment] = [] + for item in items: + if not isinstance(item, dict): + continue + item_type = item.get("type") + if item_type == 2: + info = item.get("image_item") or {} + media = info.get("media") or {} + full_url = str(media.get("full_url") or "").strip() if isinstance(media, dict) else "" + media_id = str( + info.get("media_id") + or info.get("file_id") + or (media.get("media_id") if isinstance(media, dict) else "") + or full_url + ).strip() + if media_id: + message_id = str(msg.get("message_id") or msg.get("msg_id") or media_id).strip() + download_params = {"context_token": context_token} + if full_url: + download_params.update( + { + "full_url": full_url, + "encrypt_query_param": str( + media.get("encrypt_query_param") or "" + ).strip(), + "aes_key": str(media.get("aes_key") or info.get("aeskey") or "").strip(), + # full_url may return a higher-resolution variant with channel + # trailer bytes, so these sizes are only a pre-download limit hint. + "declared_size": max( + int(info.get("mid_size") or 0), + int(info.get("hd_size") or 0), + ), + } + ) + attachments.append( + ChannelInboundAttachment( + media_id=media_id, + kind="image", + filename=f"{message_id}.jpg", + content_type="image/jpeg", + download_params=download_params, + ) + ) + elif item_type == 4: + info = item.get("file_item") or {} + media = info.get("media") or {} + full_url = str(media.get("full_url") or "").strip() if isinstance(media, dict) else "" + media_id = str( + info.get("media_id") + or info.get("file_id") + or (media.get("media_id") if isinstance(media, dict) else "") + or full_url + ).strip() + if media_id: + download_params = {"context_token": context_token} + if full_url: + download_params.update( + { + "full_url": full_url, + "encrypt_query_param": str( + media.get("encrypt_query_param") or "" + ).strip(), + "aes_key": str(media.get("aes_key") or "").strip(), + "expected_size": int(info.get("len") or 0), + } + ) + attachments.append( + ChannelInboundAttachment( + media_id=media_id, + kind="file", + filename=str(info.get("file_name") or info.get("name") or media_id).strip(), + download_params=download_params, + ) + ) + return attachments + + def normalize_wechat_message(msg: dict[str, Any], *, ilink_bot_id: str = "") -> WeChatInbound | None: """归一化 getupdates 消息；自身消息/无文本/无 context_token 返回 None（丢弃）。""" if not isinstance(msg, dict) or is_self_message(msg, ilink_bot_id): @@ -420,7 +766,41 @@ def normalize_wechat_message(msg: dict[str, Any], *, ilink_bot_id: str = "") -> return None context_token = str(msg.get("context_token") or "").strip() text = extract_message_text(msg) - if not context_token or not text: + attachments = extract_message_attachments(msg) + items = msg.get("item_list") + if isinstance(items, list): + nested_keys = [] + for item in items: + if not isinstance(item, dict): + continue + for field_name in ("image_item", "file_item", "voice_item"): + nested = item.get(field_name) + if isinstance(nested, dict): + entry = { + "field": field_name, + "keys": sorted(nested.keys()), + "value_types": { + key: type(value).__name__ for key, value in nested.items() + }, + } + media = nested.get("media") + if isinstance(media, dict): + entry["media_keys"] = sorted(media.keys()) + entry["media_value_types"] = { + key: type(value).__name__ for key, value in media.items() + } + nested_keys.append(entry) + logger.warning( + "微信入站消息附件诊断 message_id=%s item_types=%s item_keys=%s " + "nested_keys=%s recognized_attachments=%s has_text=%s", + str(msg.get("message_id") or msg.get("msg_id") or "").strip(), + [item.get("type") for item in items if isinstance(item, dict)], + [sorted(item.keys()) for item in items if isinstance(item, dict)], + nested_keys, + len(attachments), + bool(text), + ) + if not context_token or (not text and not attachments): return None event_id = str(msg.get("message_id") or msg.get("msg_id") or msg.get("client_id") or "").strip() if not event_id: @@ -442,6 +822,7 @@ def normalize_wechat_message(msg: dict[str, Any], *, ilink_bot_id: str = "") -> text=text, is_group=is_group, raw=msg, + attachments=attachments, ) diff --git a/backend/app/channels/adapters/wecom.py b/backend/app/channels/adapters/wecom.py index 6ee01764..540a00c5 100644 --- a/backend/app/channels/adapters/wecom.py +++ b/backend/app/channels/adapters/wecom.py @@ -1,23 +1,34 @@ from __future__ import annotations import asyncio +import json import logging import queue +import re import threading import time +from collections.abc import Callable from datetime import timedelta from typing import Any +from urllib.parse import unquote, urlparse +import httpx from sqlalchemy import text, update from sqlalchemy.pool import NullPool from sqlmodel import Session, create_engine, select from app.channels.adapters.base import ( ChannelInbound, + ChannelInboundAttachment, register_channel_adapter, split_channel_text, ) from app.channels.crypto import decrypt_channel_secret +from app.channels.media import ( + MAX_CHANNEL_MEDIA_BYTES, + MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES, + ensure_channel_media_size, +) from app.db import engine from app.db.models import ChannelBinding, utc_now @@ -27,6 +38,44 @@ SEND_TIMEOUT_SECONDS = 15.0 # 企微长连接持续未 connected 超过该阈值时,给绑定创建者发一次性断开告警 WECOM_DISCONNECT_ALERT_MINUTES = 15 +WECOM_API_BASE = "https://qyapi.weixin.qq.com/cgi-bin" +WECOM_TOKEN_REFRESH_SKEW_SECONDS = 300 +WECOM_MEDIA_HOSTS = {"ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com"} + + +def validate_wecom_media_url(url: str) -> bool: + try: + parsed = urlparse(url) + except ValueError: + return False + return parsed.scheme == "https" and (parsed.hostname or "").lower() in WECOM_MEDIA_HOSTS + + +async def _download_wecom_media_limited(url: str, aes_key: str) -> tuple[bytes, str | None]: + from aibot import decrypt_file + + async with httpx.AsyncClient(timeout=15.0) as client, client.stream("GET", url) as response: + response.raise_for_status() + content_length = int(response.headers.get("content-length") or 0) + ensure_channel_media_size(content_length, encrypted=True) + chunks: list[bytes] = [] + total = 0 + async for chunk in response.aiter_bytes(): + total += len(chunk) + if total > MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES: + raise ValueError( + f"企微附件超过大小上限: size>{MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES}" + ) + chunks.append(chunk) + encrypted = b"".join(chunks) + data = decrypt_file(encrypted, aes_key) if aes_key else encrypted + if len(data) > MAX_CHANNEL_MEDIA_BYTES: + raise ValueError(f"企微附件解密后超过大小上限: size={len(data)}") + disposition = response.headers.get("content-disposition", "") + match = re.search(r"filename\*=UTF-8''([^;\s]+)", disposition, re.IGNORECASE) + if not match: + match = re.search(r'filename="?([^";\s]+)', disposition, re.IGNORECASE) + return data, unquote(match.group(1)) if match else None def is_self_frame(frame: dict[str, Any]) -> bool: @@ -44,12 +93,93 @@ def normalize_wecom_frame(frame: dict[str, Any], *, account_scope: str = "") -> body = frame.get("body") or {} msgtype = str(body.get("msgtype") or "") text = "" + attachments: list[ChannelInboundAttachment] = [] if msgtype == "text": text = str((body.get("text") or {}).get("content") or "").strip() elif msgtype == "voice": # 语音帧 body.voice.content 为微信侧转写文本 text = str((body.get("voice") or {}).get("content") or "").strip() - if not text: + elif msgtype == "image": + image = body.get("image") or {} + media_url = str(image.get("url") or "").strip() + media_id = str(image.get("media_id") or image.get("file_id") or "").strip() + if media_url and validate_wecom_media_url(media_url): + media_id = media_id or media_url + if media_id: + attachments.append( + ChannelInboundAttachment( + media_id=media_id, + kind="image", + filename=f"{body.get('msgid') or 'image'}.jpg", + content_type="image/jpeg", + download_params={ + "url": media_url, + "aes_key": str(image.get("aeskey") or "").strip(), + }, + ) + ) + elif msgtype == "file": + file_info = body.get("file") or {} + media_url = str(file_info.get("url") or "").strip() + media_id = str(file_info.get("media_id") or file_info.get("file_id") or "").strip() + if media_url and validate_wecom_media_url(media_url): + media_id = media_id or media_url + if media_id: + attachments.append( + ChannelInboundAttachment( + media_id=media_id, + kind="file", + filename=str( + file_info.get("file_name") + or file_info.get("filename") + or body.get("msgid") + or "attachment.bin" + ).strip(), + download_params={ + "url": media_url, + "aes_key": str(file_info.get("aeskey") or "").strip(), + }, + ) + ) + elif msgtype == "mixed": + mixed = body.get("mixed") or {} + items = mixed.get("msg_item") or [] + if isinstance(items, list): + text_parts: list[str] = [] + for index, item in enumerate(items, start=1): + if not isinstance(item, dict): + continue + item_type = str(item.get("msgtype") or "") + if item_type == "text": + content = str((item.get("text") or {}).get("content") or "").strip() + if content: + text_parts.append(content) + elif item_type in {"image", "file"}: + info = item.get(item_type) or {} + media_url = str(info.get("url") or "").strip() + if not media_url or not validate_wecom_media_url(media_url): + continue + filename = str( + info.get("file_name") + or info.get("filename") + or f"{body.get('msgid') or 'attachment'}-{index}" + ).strip() + if item_type == "image" and "." not in filename: + filename = f"{filename}.jpg" + attachments.append( + ChannelInboundAttachment( + media_id=media_url, + kind=item_type, + filename=filename, + content_type="image/jpeg" if item_type == "image" else "", + download_params={ + "url": media_url, + "aes_key": str(info.get("aeskey") or "").strip(), + }, + ) + ) + text = "\n".join(text_parts) + if not text and not attachments: return None from_user_id = str((body.get("from") or {}).get("userid") or "").strip() if not from_user_id: @@ -81,6 +211,7 @@ def normalize_wecom_frame(frame: dict[str, Any], *, account_scope: str = "") -> raw=frame, sender_name=sender_name, account_scope=account_scope, + attachments=attachments, ) @@ -607,6 +738,84 @@ class WeComAdapter: def normalize(self, raw: dict[str, Any]) -> ChannelInbound | None: return normalize_wecom_frame(raw) + _token_provider: WeComTokenProvider | None = None + + def _get_token_provider(self) -> WeComTokenProvider: + if self._token_provider is None: + self._token_provider = WeComTokenProvider() + return self._token_provider + + def download_media( + self, + binding: ChannelBinding, + attachment: ChannelInboundAttachment, + *, + max_bytes: int = 0, + ) -> bytes: + media_url = str(attachment.download_params.get("url") or "").strip() + if media_url: + if not validate_wecom_media_url(media_url): + raise ValueError("企微媒体 URL 域名不受信任") + from app.channels import get_wecom_stream_manager + + stream = get_wecom_stream_manager().get_stream(binding.id) + if not stream: + raise RuntimeError(f"企微连接未就绪 binding={binding.id}") + client, loop = stream + future = asyncio.run_coroutine_threadsafe( + _download_wecom_media_limited( + media_url, + str(attachment.download_params.get("aes_key") or "").strip(), + ), + loop, + ) + data, downloaded_filename = future.result(timeout=15.0) + if downloaded_filename: + attachment.filename = downloaded_filename + return data + provider = self._get_token_provider() + token = provider.get(binding) + for attempt in range(2): + try: + with httpx.Client(timeout=15.0) as client, client.stream( + "GET", + f"{WECOM_API_BASE}/media/get", + params={"access_token": token, "media_id": attachment.media_id}, + ) as response: + response.raise_for_status() + content_type = response.headers.get("content-type", "") + if "application/json" not in content_type.lower(): + chunks: list[bytes] = [] + total = 0 + for chunk in response.iter_bytes(64 * 1024): + total += len(chunk) + if total > MAX_CHANNEL_MEDIA_BYTES: + raise ValueError("企微附件超过大小上限") + chunks.append(chunk) + return b"".join(chunks) + chunks = [] + total = 0 + for chunk in response.iter_bytes(64 * 1024): + total += len(chunk) + if total > MAX_CHANNEL_MEDIA_BYTES: + raise ValueError("企微媒体响应超过大小上限") + chunks.append(chunk) + raw = b"".join(chunks) + except httpx.HTTPError as exc: + raise WeComTokenError("企微媒体下载请求失败") from exc + try: + data = json.loads(raw) + except ValueError as exc: + raise WeComTokenError("企微 media/get 响应格式无效") from exc + errcode = int(data.get("errcode") or 0) + if errcode in {40014, 42001} and attempt == 0: + token = provider.get(binding, force_refresh=True) + continue + raise WeComTokenError( + f"企微 media/get 错误: errcode={errcode} msg={data.get('errmsg')}" + ) + raise WeComTokenError("企微 media/get 下载失败") + def send( self, binding: ChannelBinding, @@ -641,5 +850,58 @@ def stop_ingress(self, binding_id: str) -> None: get_wecom_stream_manager().stop_binding(binding_id) +class WeComTokenError(RuntimeError): + pass + + +class WeComTokenProvider: + """Cache enterprise access tokens for the configured corp and revision.""" + + def __init__(self, *, client_factory: Callable[[], httpx.Client] | None = None): + self._client_factory = client_factory or (lambda: httpx.Client(timeout=10.0)) + self._cache: dict[str, tuple[str, float]] = {} + self._lock = threading.Lock() + + @staticmethod + def _key(binding: ChannelBinding) -> str: + config = dict(binding.config_json or {}) + return f"{config.get('corp_id', '')}:{binding.config_revision}" + + def get(self, binding: ChannelBinding, *, force_refresh: bool = False) -> str: + key = self._key(binding) + with self._lock: + cached = self._cache.get(key) + if cached and not force_refresh and cached[1] > time.monotonic(): + return cached[0] + config = dict(binding.config_json or {}) + corp_id = str(config.get("corp_id") or "").strip() + if not corp_id or not binding.credentials_enc: + raise WeComTokenError("企微绑定缺少 corp_id 或 secret") + secret = decrypt_channel_secret(binding.credentials_enc) + try: + with self._client_factory() as client: + response = client.get( + f"{WECOM_API_BASE}/gettoken", + params={"corpid": corp_id, "corpsecret": secret}, + ) + data = response.json() + except (httpx.HTTPError, ValueError) as exc: + raise WeComTokenError("企微 token 请求失败") from exc + if response.status_code >= 400 or int(data.get("errcode") or -1) != 0: + raise WeComTokenError( + f"企微 token 请求失败: errcode={data.get('errcode')} msg={data.get('errmsg')}" + ) + token = str(data.get("access_token") or "").strip() + expires_in = int(data.get("expires_in") or 0) + if not token or expires_in <= 0: + raise WeComTokenError("企微 token 响应缺少必要字段") + with self._lock: + self._cache[key] = ( + token, + time.monotonic() + max(1, expires_in - WECOM_TOKEN_REFRESH_SKEW_SECONDS), + ) + return token + + # 模块导入即注册企微适配器(渠道内核按注册表发现渠道) register_channel_adapter("wecom", WeComAdapter()) diff --git a/backend/app/channels/attachment_bridge.py b/backend/app/channels/attachment_bridge.py index 89277aa7..a2af4d44 100644 --- a/backend/app/channels/attachment_bridge.py +++ b/backend/app/channels/attachment_bridge.py @@ -4,113 +4,66 @@ from typing import Any from app.channels.adapters.base import ChannelInbound +from app.channels.media import ( + MAX_CHANNEL_MEDIA_BYTES, + filename_with_extension, + normalize_image_media, +) from app.db.models import ChannelBinding from app.session.session_schema import ChatAttachmentRead logger = logging.getLogger(__name__) -MAX_CHANNEL_MEDIA_BYTES = 25 * 1024 * 1024 # 25MB - -# 图片 magic bytes 签名 → (content_type, extension) -_IMAGE_SIGNATURES: list[tuple[bytes, str, str]] = [ - (b"\x89PNG\r\n\x1a\n", "image/png", ".png"), - (b"\xff\xd8\xff", "image/jpeg", ".jpg"), - (b"GIF87a", "image/gif", ".gif"), - (b"GIF89a", "image/gif", ".gif"), - (b"RIFF", "image/webp", ".webp"), # 需后续确认 WebP 标记 - (b"BM", "image/bmp", ".bmp"), -] - - -def _detect_image_type(data: bytes) -> tuple[str, str] | None: - """从字节签名推断图片 content_type 和扩展名。返回 (content_type, ext) 或 None。""" - if len(data) < 12: - return None - for sig, content_type, ext in _IMAGE_SIGNATURES: - if data.startswith(sig): - if sig == b"RIFF" and data[8:12] != b"WEBP": - continue - return content_type, ext - return None - - -def _resolve_content_type( - att_content_type: str, - att_filename: str, - data: bytes, -) -> tuple[str, str]: - """根据下载字节修正 content_type 和 filename。 - - 渠道 normalize 阶段可能无法确定真实 MIME(飞书 image_key 不含扩展名, - 钉钉 picture 消息也不提供类型),因此下载后用 magic bytes 覆盖。 - """ - detected = _detect_image_type(data) - if detected: - content_type, ext = detected - filename = att_filename - if not filename.lower().endswith(ext): - filename = f"{att_filename}{ext}" - return content_type, filename - # 非图片或无法识别:保留渠道侧提供的值,空则传 None 让 parse 自动推断 - ct = (att_content_type or "").strip() - return (ct or None, att_filename) # type: ignore[return-value] - def inbound_attachments_to_chat( binding: ChannelBinding, inbound: ChannelInbound, *, - db_engine: Any, + db_engine: Any = None, tenant_id: str, user_id: str, ) -> list[ChatAttachmentRead]: - """下载渠道附件,暂存原始字节,转为 ChatAttachmentRead 列表。 - - 调用各适配器的 download_media 方法获取原始字节,然后复用 web chat 的 - parse_chat_attachment + stage_chat_attachment 完成解析和暂存。 - - 单个附件失败不影响其他附件和主链路(intake 侧再降级为纯文本轮)。 - """ - # 延迟 import 避免 app.core -> app.session.attachment_store 的循环依赖 + """Download channel media and stage it through the web attachment pipeline.""" from app.channels.adapters.base import get_channel_adapter - - adapter = get_channel_adapter(inbound.channel) - download_media = getattr(adapter, "download_media", None) - if download_media is None: - logger.warning("渠道 %s 未实现 download_media,跳过附件", inbound.channel) - return [] - - # 真正需要下载/暂存时才 import,避免循环依赖与无谓加载 from app.session.attachment_store import stage_chat_attachment from app.session.attachments import parse_chat_attachment + download_media = getattr(get_channel_adapter(inbound.channel), "download_media", None) + if not callable(download_media): + logger.warning("渠道 %s 未实现 download_media，跳过附件", inbound.channel) + return [] results: list[ChatAttachmentRead] = [] - for att in inbound.attachments: + for descriptor in inbound.attachments: try: - data = download_media(binding, att, max_bytes=MAX_CHANNEL_MEDIA_BYTES) - if not data: + data = download_media(binding, descriptor, max_bytes=MAX_CHANNEL_MEDIA_BYTES) + if not data or len(data) > MAX_CHANNEL_MEDIA_BYTES: + logger.warning( + "渠道附件为空或超过大小上限 binding=%s media_id=%s size=%s", + binding.id, + descriptor.media_id, + len(data) if data else 0, + ) continue - content_type, filename = _resolve_content_type( - att.content_type, - att.filename or att.media_id, - data, - ) - attachment = parse_chat_attachment( - filename, - content_type, + normalized_image = normalize_image_media(data) + if descriptor.kind == "image" and normalized_image is None: + raise ValueError("渠道图片内容不是受支持的图片格式") + if normalized_image is not None: + data, descriptor.content_type, extension = normalized_image + descriptor.kind = "image" + descriptor.filename = filename_with_extension( + descriptor.filename or descriptor.media_id, + extension, + ) + parsed = parse_chat_attachment( + descriptor.filename or descriptor.media_id, + descriptor.content_type or None, data, ) - staged = stage_chat_attachment( - attachment, - data, - tenant_id=tenant_id, - user_id=user_id, + results.append( + stage_chat_attachment(parsed, data, tenant_id=tenant_id, user_id=user_id) ) - results.append(staged) except Exception: logger.exception( - "渠道附件处理失败 binding=%s media_id=%s", - binding.id, - att.media_id, + "渠道附件处理失败 binding=%s media_id=%s", binding.id, descriptor.media_id ) return results diff --git a/backend/app/channels/media.py b/backend/app/channels/media.py new file mode 100644 index 00000000..a679fe7a --- /dev/null +++ b/backend/app/channels/media.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +from pathlib import PurePath + +MAX_CHANNEL_MEDIA_BYTES = 25 * 1024 * 1024 +MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES = MAX_CHANNEL_MEDIA_BYTES + 32 + + +class ChannelMediaTooLargeError(ValueError): + pass + + +def ensure_channel_media_size(size: int, *, encrypted: bool = False) -> None: + limit = MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES if encrypted else MAX_CHANNEL_MEDIA_BYTES + if size > limit: + raise ChannelMediaTooLargeError(f"渠道附件超过大小上限: size={size} limit={limit}") + + +def collect_limited_media(chunks, *, encrypted: bool = False) -> bytes: + limit = MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES if encrypted else MAX_CHANNEL_MEDIA_BYTES + parts: list[bytes] = [] + total = 0 + for chunk in chunks: + if not chunk: + continue + total += len(chunk) + if total > limit: + raise ChannelMediaTooLargeError( + f"渠道附件超过大小上限: size>{limit} limit={limit}" + ) + parts.append(chunk) + return b"".join(parts) + + +def normalize_image_media(data: bytes) -> tuple[bytes, str, str] | None: + if data.startswith(b"\xff\xd8\xff"): + eoi = data.rfind(b"\xff\xd9") + if eoi >= 3: + return data[: eoi + 2], "image/jpeg", ".jpg" + detected = detect_image_media_type(data) + if detected is None: + return None + content_type, extension = detected + return data, content_type, extension + + +def detect_image_media_type(data: bytes) -> tuple[str, str] | None: + if data.startswith(b"\xff\xd8\xff") and b"\xff\xd9" in data[3:]: + return "image/jpeg", ".jpg" + if data.startswith(b"\x89PNG\r\n\x1a\n"): + return "image/png", ".png" + if data.startswith((b"GIF87a", b"GIF89a")): + return "image/gif", ".gif" + if len(data) >= 12 and data.startswith(b"RIFF") and data[8:12] == b"WEBP": + return "image/webp", ".webp" + if data.startswith(b"BM"): + return "image/bmp", ".bmp" + sample = data[:1024].lstrip().lower() + if sample.startswith(b" str: + name = PurePath(filename or "image").name + stem = PurePath(name).stem or "image" + return f"{stem}{extension}" + + +__all__ = [ + "MAX_CHANNEL_MEDIA_BYTES", + "MAX_ENCRYPTED_CHANNEL_MEDIA_BYTES", + "ChannelMediaTooLargeError", + "collect_limited_media", + "detect_image_media_type", + "ensure_channel_media_size", + "filename_with_extension", + "normalize_image_media", +] diff --git a/backend/app/channels/schema.py b/backend/app/channels/schema.py index 6a7cddaa..532dc56e 100644 --- a/backend/app/channels/schema.py +++ b/backend/app/channels/schema.py @@ -154,11 +154,11 @@ class ChannelConversationAttachmentRead(BaseModel): 内部字段或大体量内容塞进会话列表响应。 """ - id: Optional[str] = None - filename: Optional[str] = None - content_type: Optional[str] = None - size: Optional[int] = None - kind: Optional[str] = None + id: str + filename: str + content_type: str + size: int + kind: str class ChannelConversationMessageRead(BaseModel): diff --git a/backend/app/channels/service_intake.py b/backend/app/channels/service_intake.py index f18a5d46..c6f7f91f 100644 --- a/backend/app/channels/service_intake.py +++ b/backend/app/channels/service_intake.py @@ -494,15 +494,23 @@ def _stage_received_reaction( def _message_text(binding: ChannelBinding, inbound: ChannelInbound) -> str: + text = inbound.text.strip() + if not text and inbound.attachments: + kinds = {attachment.kind for attachment in inbound.attachments} + text = ( + "请识别并描述这张图片的内容。" + if kinds == {"image"} + else "请读取并概述这个文件。" + ) if not inbound.is_group: - return inbound.text + return text sender_label = inbound.sender_name or external_identity_for_message( binding.channel, is_group=False, conv_key="", from_user_id=inbound.from_user_id, )[1] - return f"[发送者: {sender_label}]\n{inbound.text}" + return f"[发送者: {sender_label}]\n{text}" def _run_bind_command( @@ -946,7 +954,6 @@ def process_inbound( binding.id, inbound.event_id, ) - request = ChatTurnRequest( tenant_id=binding.tenant_id, session_id=session_id, diff --git a/backend/app/llm/client.py b/backend/app/llm/client.py index 59f8b3cd..e6d21a99 100644 --- a/backend/app/llm/client.py +++ b/backend/app/llm/client.py @@ -205,6 +205,8 @@ def generate_text( ) context_messages, serialized = _prepare_user_input(user_payload) request_messages = _request_messages(system_prompt, context_messages, serialized) + if response_format and response_format.get("type") == "json_object": + request_messages = _with_json_mode_instruction(request_messages) request_messages = _fit_request_messages(request_messages) if isinstance(user_payload, dict) and isinstance( user_payload.get(STAGE_PROTOCOL_KEY), dict @@ -756,6 +758,24 @@ def _request_messages( return messages +def _with_json_mode_instruction(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + projected = copy.deepcopy(messages) + instruction = ( + "Return exactly one valid json object. Do not output Markdown, code fences, " + "explanations, or extra text." + ) + for message in reversed(projected): + if message.get("role") != "user": + continue + content = message.get("content") + if isinstance(content, list): + content.append({"type": "text", "text": instruction}) + else: + message["content"] = f"{str(content or '').rstrip()}\n\n{instruction}".strip() + break + return projected + + def _messages_have_images(messages: list[dict[str, Any]]) -> bool: return any( isinstance(part, dict) diff --git a/backend/app/session/attachments.py b/backend/app/session/attachments.py index e54430c1..6f23761f 100644 --- a/backend/app/session/attachments.py +++ b/backend/app/session/attachments.py @@ -2,9 +2,9 @@ import base64 import csv +import hashlib import io import json -import hashlib import mimetypes import re from collections.abc import Iterable @@ -106,6 +106,8 @@ def image_payloads_from_attachments(attachments: Iterable[ChatAttachmentRead | d for attachment in normalized: if not attachment or not _attachment_is_supported_image(attachment) or not attachment.data_url: continue + if not _data_url_has_valid_image_signature(attachment.data_url, attachment.content_type): + continue payloads.append( { "type": "image_url", @@ -229,6 +231,8 @@ def _validated_image_data_url( raise ValueError(f"{filename} 的图片 data URL 无效") from exc if len(decoded) > IMAGE_DATA_URL_LIMIT_BYTES or len(decoded) != size: raise ValueError(f"{filename} 的图片 data URL 大小不一致或超限") + if not _image_bytes_match_content_type(decoded, content_type): + raise ValueError(f"{filename} 的图片内容与 MIME 类型不一致") if attachment.sha256 and hashlib.sha256(decoded).hexdigest() != attachment.sha256.lower(): raise ValueError(f"{filename} 的图片 data URL 与上传文件不一致") return raw @@ -434,6 +438,35 @@ def _attachment_is_supported_image(attachment: ChatAttachmentRead) -> bool: return attachment.kind == "image" and _is_supported_image_file(attachment.filename.lower(), attachment.content_type) +def _data_url_has_valid_image_signature(data_url: str, content_type: str) -> bool: + prefix = f"data:{content_type};base64," + if not data_url.startswith(prefix): + return False + try: + data = base64.b64decode(data_url.removeprefix(prefix), validate=True) + except (ValueError, TypeError): + return False + return _image_bytes_match_content_type(data, content_type) + + +def _image_bytes_match_content_type(data: bytes, content_type: str) -> bool: + normalized = content_type.lower() + if normalized == "image/jpeg": + return data.startswith(b"\xff\xd8\xff") and b"\xff\xd9" in data[3:] + if normalized == "image/png": + return data.startswith(b"\x89PNG\r\n\x1a\n") + if normalized == "image/gif": + return data.startswith((b"GIF87a", b"GIF89a")) + if normalized == "image/webp": + return len(data) >= 12 and data.startswith(b"RIFF") and data[8:12] == b"WEBP" + if normalized == "image/bmp": + return data.startswith(b"BM") + if normalized == "image/svg+xml": + sample = data[:1024].lstrip().lower() + return sample.startswith(b" str: normalized = content_type.lower() if normalized in SUPPORTED_IMAGE_CONTENT_TYPES: diff --git a/backend/tests/test_channel_api.py b/backend/tests/test_channel_api.py index b477c160..8fa0a0f7 100644 --- a/backend/tests/test_channel_api.py +++ b/backend/tests/test_channel_api.py @@ -13,6 +13,7 @@ ChannelBinding, ChannelDelivery, ChannelInboundEvent, + Message, Tenant, User, utc_now, @@ -985,6 +986,26 @@ def test_list_channel_conversation_messages_order_and_404() -> None: users = _seed_users(engine) binding_id = _seed_binding(engine) _seed_conversations(engine, binding_id) + with Session(engine) as db: + message = db.get(Message, "m1") + message.metadata_json = { + "attachments": [ + { + "id": "file-1", + "filename": "image.png", + "content_type": "image/png", + "size": 123, + "kind": "image", + "data_url": "data:image/png;base64,SECRET", + "sandbox_path": "/workspace/attachments/internal.png", + "sha256": "a" * 64, + "text": "internal text", + "python_summary": "internal summary", + } + ] + } + db.add(message) + db.commit() client = _make_client(engine) response = client.get( @@ -998,6 +1019,16 @@ def test_list_channel_conversation_messages_order_and_404() -> None: assert rows[0]["role"] == "user" assert rows[0]["content"] == "你好" assert rows[0]["created_at"] + assert rows[0]["attachments"] == [ + { + "id": "file-1", + "filename": "image.png", + "content_type": "image/png", + "size": 123, + "kind": "image", + } + ] + assert rows[1]["attachments"] is None # 其他绑定的会话 → 404 other = client.get( diff --git a/backend/tests/test_channel_attachment_bridge.py b/backend/tests/test_channel_attachment_bridge.py index fe24ac78..c052dfc1 100644 --- a/backend/tests/test_channel_attachment_bridge.py +++ b/backend/tests/test_channel_attachment_bridge.py @@ -21,11 +21,10 @@ ) from app.channels.attachment_bridge import ( MAX_CHANNEL_MEDIA_BYTES, - _detect_image_type, - _resolve_content_type, inbound_attachments_to_chat, ) from app.channels.crypto import encrypt_channel_secret +from app.channels.media import detect_image_media_type, filename_with_extension from app.db.models import ChannelBinding from app.session.session_schema import ChatAttachmentRead @@ -157,7 +156,8 @@ def test_bridge_returns_empty_when_adapter_has_no_download_media() -> None: def test_bridge_downloads_and_stages_attachments() -> None: """完整链路:download_media -> parse_chat_attachment -> stage_chat_attachment。""" - fake_adapter = _FakeAdapter(b"PNG-data") + image = b"\x89PNG\r\n\x1a\nimage-data" + fake_adapter = _FakeAdapter(image) previous = get_channel_adapter("feishu") register_channel_adapter("feishu", fake_adapter) try: @@ -198,7 +198,7 @@ def test_bridge_downloads_and_stages_attachments() -> None: assert len(fake_adapter.download_calls) == 1 assert fake_adapter.download_calls[0][1].media_id == "img_v3_001" # parse_chat_attachment 收到原始字节和文件名 - mock_parse.assert_called_once_with("img_v3_001.jpg", "image/jpeg", b"PNG-data") + mock_parse.assert_called_once_with("img_v3_001.png", "image/png", image) # stage_chat_attachment 收到 attachment + 字节 + tenant/user mock_stage.assert_called_once() stage_kwargs = mock_stage.call_args.kwargs @@ -247,7 +247,7 @@ def download_media(self, binding, attachment, *, max_bytes=0): def test_bridge_continues_on_single_attachment_failure() -> None: """单个附件下载异常不影响其他附件。""" - good = b"PNG-data" + good = b"\x89PNG\r\n\x1a\nimage-data" call_count = {"n": 0} class MixedAdapter: @@ -319,47 +319,33 @@ def test_bridge_passes_empty_content_type_as_none() -> None: register_channel_adapter("feishu", previous) -def test_detect_image_type_recognizes_common_formats() -> None: +def test_detect_image_media_type_recognizes_common_formats() -> None: """magic bytes 签名能识别 PNG/JPEG/GIF/WebP/BMP。""" - assert _detect_image_type(b"\x89PNG\r\n\x1a\n" + b"\x00" * 20) == ("image/png", ".png") - assert _detect_image_type(b"\xff\xd8\xff\xe0" + b"\x00" * 20) == ("image/jpeg", ".jpg") - assert _detect_image_type(b"GIF89a" + b"\x00" * 20) == ("image/gif", ".gif") - assert _detect_image_type(b"RIFF" + b"\x00" * 4 + b"WEBP" + b"\x00" * 20) == ("image/webp", ".webp") - assert _detect_image_type(b"BM" + b"\x00" * 20) == ("image/bmp", ".bmp") + assert detect_image_media_type(b"\x89PNG\r\n\x1a\n" + b"\x00" * 20) == ( + "image/png", + ".png", + ) + assert detect_image_media_type(b"\xff\xd8\xff\xe0data\xff\xd9") == ( + "image/jpeg", + ".jpg", + ) + assert detect_image_media_type(b"GIF89a" + b"\x00" * 20) == ("image/gif", ".gif") + assert detect_image_media_type( + b"RIFF" + b"\x00" * 4 + b"WEBP" + b"\x00" * 20 + ) == ("image/webp", ".webp") + assert detect_image_media_type(b"BM" + b"\x00" * 20) == ("image/bmp", ".bmp") -def test_detect_image_type_rejects_non_image_data() -> None: +def test_detect_image_media_type_rejects_non_image_data() -> None: """非图片字节或太短的数据返回 None。""" - assert _detect_image_type(b"") is None - assert _detect_image_type(b"short") is None - assert _detect_image_type(b"RIFF" + b"\x00" * 4 + b"XXXX" + b"\x00" * 20) is None - - -def test_resolve_content_type_overrides_with_magic_bytes() -> None: - """下载后用 magic bytes 覆盖渠道侧空的 content_type 和 filename。""" - png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 20 - ct, fn = _resolve_content_type("", "img_v3_001", png) - assert ct == "image/png" - assert fn == "img_v3_001.png" - - -def test_resolve_content_type_preserves_existing_filename_extension() -> None: - """filename 已含正确扩展名时不重复追加。""" - png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 20 - ct, fn = _resolve_content_type("", "photo.png", png) - assert ct == "image/png" - assert fn == "photo.png" - + assert detect_image_media_type(b"") is None + assert detect_image_media_type(b"short") is None + assert detect_image_media_type(b"RIFF" + b"\x00" * 4 + b"XXXX" + b"\x00" * 20) is None -def test_resolve_content_type_passes_through_non_image() -> None: - """非图片文件保留渠道侧 content_type,空则返回 None。""" - ct, fn = _resolve_content_type("", "report.pdf", b"%PDF-1.4") - assert ct is None - assert fn == "report.pdf" - ct, fn = _resolve_content_type("application/pdf", "report.pdf", b"%PDF-1.4") - assert ct == "application/pdf" - assert fn == "report.pdf" +def test_filename_with_extension_replaces_or_adds_extension() -> None: + assert filename_with_extension("img_v3_001", ".png") == "img_v3_001.png" + assert filename_with_extension("photo.jpg", ".png") == "photo.png" def test_stream_download_with_limit_enforces_content_length() -> None: diff --git a/backend/tests/test_channel_intake.py b/backend/tests/test_channel_intake.py index 498437f4..ade7777b 100644 --- a/backend/tests/test_channel_intake.py +++ b/backend/tests/test_channel_intake.py @@ -1,6 +1,6 @@ +import os import threading import time -import os import pytest from sqlalchemy.pool import StaticPool @@ -8,14 +8,16 @@ import app.channels.service_intake as intake_module import app.core.agent_loop as agent_loop_module +from app.channels.adapters.base import ChannelInbound, ChannelInboundAttachment from app.channels.service_identity import channel_username from app.channels.service_intake import ( - _send_wechat_typing as _real_send_wechat_typing, -) -from app.channels.service_intake import ( + _message_text, _session_lock, process_inbound, ) +from app.channels.service_intake import ( + _send_wechat_typing as _real_send_wechat_typing, +) from app.db.models import ( ChannelBinding, ChannelDelivery, @@ -84,6 +86,33 @@ def _load_binding(engine, binding_id: str) -> ChannelBinding: return binding +def test_channel_only_attachment_uses_default_message_intent() -> None: + binding = ChannelBinding(tenant_id="tenant_demo", agent_id="agent_1", channel="wecom") + image = ChannelInbound( + channel="wecom", + event_id="evt-image", + from_user_id="user-1", + to_user_id="bot-1", + session_id="user-1", + group_id="", + context_token="user-1", + text="", + is_group=False, + raw={}, + attachments=[ChannelInboundAttachment(media_id="image", kind="image")], + ) + file = ChannelInbound( + **{ + **image.__dict__, + "event_id": "evt-file", + "attachments": [ChannelInboundAttachment(media_id="file", kind="file")], + } + ) + + assert _message_text(binding, image) == "请识别并描述这张图片的内容。" + assert _message_text(binding, file) == "请读取并概述这个文件。" + + class RecordingAgentLoop: """替代真实 AgentLoop：记录请求并模拟用户/助手消息落库。""" diff --git a/backend/tests/test_channel_wechat.py b/backend/tests/test_channel_wechat.py index 32716efd..263d00f5 100644 --- a/backend/tests/test_channel_wechat.py +++ b/backend/tests/test_channel_wechat.py @@ -4,6 +4,8 @@ import time import httpx +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from sqlalchemy.pool import StaticPool from sqlmodel import Session, SQLModel, create_engine @@ -11,6 +13,7 @@ WeChatAdapter, WeChatClient, WeChatPollManager, + decrypt_wechat_media, is_self_message, normalize_wechat_message, random_wechat_uin, @@ -36,6 +39,18 @@ def _client(handler) -> WeChatClient: return WeChatClient(BASE_URL, "bot_token_x", transport=httpx.MockTransport(handler)) +def test_decrypt_wechat_media_aes_ecb_pkcs7() -> None: + key = b"0123456789abcdef" + aes_key = base64.b64encode(key.hex().encode("ascii")).decode("ascii") + plaintext = b"\xff\xd8\xffjpeg-data\xff\xd9" + padder = padding.PKCS7(algorithms.AES.block_size).padder() + padded = padder.update(plaintext) + padder.finalize() + encryptor = Cipher(algorithms.AES(key), modes.ECB()).encryptor() + encrypted = encryptor.update(padded) + encryptor.finalize() + + assert decrypt_wechat_media(encrypted, aes_key, expected_size=len(plaintext)) == plaintext + + def _text_message(**overrides) -> dict: msg = { "seq": 429, @@ -133,6 +148,83 @@ def handler(request: httpx.Request) -> httpx.Response: assert inbound.external_conv_id == "wechat_p2p_user_ab12cd34@im.wechat" +def test_normalize_image_and_file_items() -> None: + image = normalize_wechat_message( + _text_message( + item_list=[{"type": 2, "image_item": {"media_id": "image-1"}}], + ) + ) + assert image is not None + assert image.attachments[0].media_id == "image-1" + assert image.attachments[0].download_params["context_token"] == "ctx_token_1" + + file = normalize_wechat_message( + _text_message( + item_list=[ + { + "type": 4, + "file_item": { + "file_name": "a.txt", + "len": "12", + "md5": "md5", + "media": { + "aes_key": "aes", + "encrypt_query_param": "encrypted", + "full_url": f"{BASE_URL}/c2c/download?encrypted_query_param=encrypted&taskid=task", + }, + }, + } + ], + ) + ) + assert file is not None + assert file.attachments[0].media_id.endswith("/c2c/download?encrypted_query_param=encrypted&taskid=task") + assert file.attachments[0].filename == "a.txt" + + +def test_normalize_actual_image_media_shape() -> None: + inbound = normalize_wechat_message( + _text_message( + item_list=[ + { + "type": 2, + "image_item": { + "aeskey": "aes", + "media": { + "aes_key": "aes", + "encrypt_query_param": "encrypted", + "full_url": f"{BASE_URL}/c2c/download?encrypted_query_param=encrypted&taskid=task", + }, + "mid_size": 10, + }, + } + ], + ) + ) + assert inbound is not None + attachment = inbound.attachments[0] + assert attachment.kind == "image" + assert attachment.download_params["full_url"].endswith("taskid=task") + assert attachment.download_params["aes_key"] == "aes" + assert attachment.download_params["declared_size"] == 10 + assert "expected_size" not in attachment.download_params + + +def test_download_media_request() -> None: + captured = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["url"] = str(request.url) + captured["body"] = json.loads(request.content) + return httpx.Response(200, content=b"image-bytes", headers={"content-type": "image/jpeg"}) + + data = _client(handler).download_media("ctx-1", "media-1") + assert data == b"image-bytes" + assert captured["url"] == f"{BASE_URL}/ilink/bot/downloadmedia" + assert captured["body"]["context_token"] == "ctx-1" + assert captured["body"]["media_id"] == "media-1" + + def test_send_message_payload() -> None: captured = {} diff --git a/backend/tests/test_channel_wecom.py b/backend/tests/test_channel_wecom.py index cec2d19f..aea7ec48 100644 --- a/backend/tests/test_channel_wecom.py +++ b/backend/tests/test_channel_wecom.py @@ -13,6 +13,7 @@ import app.core.agent_loop as agent_loop_module from app.channels.adapters.wecom import ( WeComAdapter, + WeComTokenProvider, WeComStreamManager, is_self_frame, normalize_wecom_frame, @@ -115,6 +116,174 @@ def test_normalize_voice_frame_uses_transcript() -> None: assert inbound.text == "我下午三点到" +def test_normalize_image_and_file_frames() -> None: + image = normalize_wecom_frame( + _text_frame( + msgtype="image", + text=None, + image={ + "url": "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/image", + "aeskey": "image-key", + }, + ) + ) + assert image is not None + assert image.text == "" + assert image.attachments[0].media_id.endswith("/image") + assert image.attachments[0].kind == "image" + assert image.attachments[0].download_params["aes_key"] == "image-key" + + file = normalize_wecom_frame( + _text_frame( + msgtype="file", + text=None, + file={ + "url": "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/file", + "aeskey": "file-key", + }, + ) + ) + assert file is not None + assert file.attachments[0].media_id.endswith("/file") + assert file.attachments[0].filename == "msg_1" + assert file.attachments[0].download_params["aes_key"] == "file-key" + + +def test_normalize_mixed_frame_extracts_text_and_image() -> None: + inbound = normalize_wecom_frame( + _text_frame( + msgtype="mixed", + text=None, + mixed={ + "msg_item": [ + { + "msgtype": "image", + "image": { + "url": "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/image", + "aeskey": "image-key", + }, + }, + {"msgtype": "text", "text": {"content": "图片里是什么"}}, + ] + }, + ) + ) + + assert inbound is not None + assert inbound.text == "图片里是什么" + assert inbound.attachments[0].kind == "image" + + +def test_url_download_uses_content_disposition_filename(monkeypatch) -> None: + import app.channels + import app.channels.adapters.wecom as wecom_module + from app.channels.adapters.base import ChannelInboundAttachment + + attachment = ChannelInboundAttachment( + media_id="https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/file", + kind="file", + filename="message-id", + download_params={ + "url": "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/file", + "aes_key": "key", + }, + ) + binding = ChannelBinding(tenant_id="tenant", agent_id="agent", channel="wecom") + manager = SimpleNamespace(get_stream=lambda _binding_id: (object(), object())) + + def run_coroutine_threadsafe(coroutine, _loop): + coroutine.close() + return SimpleNamespace(result=lambda timeout: (b"# document", "项目文档.md")) + + monkeypatch.setattr(app.channels, "get_wecom_stream_manager", lambda: manager) + monkeypatch.setattr( + wecom_module.asyncio, + "run_coroutine_threadsafe", + run_coroutine_threadsafe, + ) + + data = WeComAdapter().download_media(binding, attachment) + + assert data == b"# document" + assert attachment.filename == "项目文档.md" + + +def test_file_message_with_image_bytes_is_promoted_to_image(monkeypatch) -> None: + import app.channels.adapters.base as base_module + import app.channels.attachment_bridge as bridge_module + import app.session.attachment_store as attachment_store_module + import app.session.attachments as attachments_module + from app.channels.adapters.base import ChannelInbound, ChannelInboundAttachment + from app.session.session_schema import ChatAttachmentRead + + descriptor = ChannelInboundAttachment( + media_id="media", + kind="file", + filename="message-id", + ) + inbound = ChannelInbound( + channel="wecom", + event_id="event", + from_user_id="user", + to_user_id="bot", + session_id="user", + group_id="", + context_token="user", + text="", + is_group=False, + raw={}, + attachments=[descriptor], + ) + binding = ChannelBinding(tenant_id="tenant", agent_id="agent", channel="wecom") + adapter = SimpleNamespace( + download_media=lambda _binding, _descriptor, **_kwargs: b"\x89PNG\r\n\x1a\ndata" + ) + captured = {} + + def parse(filename, content_type, data): + captured.update(filename=filename, content_type=content_type, data=data) + return ChatAttachmentRead( + id="file-1", + filename=filename, + content_type=content_type, + size=len(data), + kind="image", + ) + + monkeypatch.setattr(base_module, "get_channel_adapter", lambda _channel: adapter) + monkeypatch.setattr(attachments_module, "parse_chat_attachment", parse) + monkeypatch.setattr( + attachment_store_module, + "stage_chat_attachment", + lambda attachment, *_args, **_kwargs: attachment, + ) + + result = bridge_module.inbound_attachments_to_chat( + binding, + inbound, + tenant_id="tenant", + user_id="user", + ) + + assert descriptor.kind == "image" + assert captured["filename"] == "message-id.png" + assert captured["content_type"] == "image/png" + assert result[0].kind == "image" + + +def test_wecom_replay_restores_attachment_dataclass() -> None: + from app.channels.service_wecom_inbox import decode_wecom_replay_envelope, encode_wecom_replay_envelope + + inbound = normalize_wecom_frame( + _text_frame(msgtype="image", text=None, image={"media_id": "media-image"}) + ) + assert inbound is not None + restored = decode_wecom_replay_envelope( + encode_wecom_replay_envelope(inbound, account_scope="corp") + ) + assert restored.attachments[0].media_id == "media-image" + + def test_normalize_group_frame() -> None: frame = _text_frame( chatid="wrQoP7CwAAA", @@ -134,7 +303,7 @@ def test_normalize_drops_self_and_invalid_frames() -> None: self_frame = _text_frame(**{"from": {"userid": "aib_bot1"}}) assert is_self_frame(self_frame) is True assert normalize_wecom_frame(self_frame) is None - # 图片消息(本期不支持) + # 非受信媒体 URL 不进入附件处理。 image_frame = _text_frame(msgtype="image", text=None, image={"url": "x"}) assert normalize_wecom_frame(image_frame) is None # 缺 msgid/req_id diff --git a/backend/tests/test_chat_attachments.py b/backend/tests/test_chat_attachments.py index ccb94d30..08008194 100644 --- a/backend/tests/test_chat_attachments.py +++ b/backend/tests/test_chat_attachments.py @@ -3,6 +3,7 @@ import pytest from app.api.chat import _user_message_metadata +from app.channels.media import normalize_image_media from app.session.attachments import ( image_payloads_from_attachments, message_content_with_attachment_context, @@ -45,7 +46,8 @@ def test_user_message_metadata_keeps_attachments() -> None: def test_image_attachment_uses_supported_extension_and_builds_image_payload() -> None: - attachment = parse_chat_attachment("screen.PNG", "application/octet-stream", b"image-bytes") + image = b"\x89PNG\r\n\x1a\nimage-bytes" + attachment = parse_chat_attachment("screen.PNG", "application/octet-stream", image) assert attachment.kind == "image" assert attachment.content_type == "image/png" @@ -61,6 +63,21 @@ def test_image_attachment_uses_supported_extension_and_builds_image_payload() -> ] +def test_historical_image_payload_rejects_invalid_image_signature() -> None: + attachment = parse_chat_attachment("screen.jpg", "image/jpeg", b"encrypted-bytes") + + assert attachment.kind == "image" + assert image_payloads_from_attachments([attachment]) == [] + + +def test_jpeg_with_trailing_channel_bytes_is_normalized() -> None: + jpeg = b"\xff\xd8\xffjpeg-data\xff\xd9" + + normalized = normalize_image_media(jpeg + b"\x00channel-trailer") + + assert normalized == (jpeg, "image/jpeg", ".jpg") + + def test_message_context_uses_sandbox_path_without_inlining_text() -> None: attachment = parse_chat_attachment("readme.md", "text/markdown", b"# Title\ncontent") attachment = attachment.model_copy(update={"sandbox_path": "/workspace/attachments/readme.md"}) diff --git a/backend/tests/test_llm_client.py b/backend/tests/test_llm_client.py index 80ffa20d..7987b8eb 100644 --- a/backend/tests/test_llm_client.py +++ b/backend/tests/test_llm_client.py @@ -966,6 +966,7 @@ def test_generate_json_requests_json_object_mode(): assert client.generate_json("prompt", {}) == {"ok": True} assert client.client.chat.completions.calls[0]["response_format"] == {"type": "json_object"} + assert "json" in str(client.client.chat.completions.calls[0]["messages"][-1]["content"]) def test_internal_json_operation_caps_output_without_mutating_system_prompt(): diff --git a/frontend-enterprise/src/i18n/en.json b/frontend-enterprise/src/i18n/en.json index 64bd46a2..c2ff5b0b 100644 --- a/frontend-enterprise/src/i18n/en.json +++ b/frontend-enterprise/src/i18n/en.json @@ -1203,6 +1203,8 @@ "头像会显示在我的数字员工、数字员工档案页和对话端的员工选择中。": "The avatar appears in My Digital Employees, Employee Profile, and the employee picker in Chat.", "头像图片不能超过 5MB": "Avatar images cannot exceed 5 MB", "图片": "Image", + "图片加载中…": "Loading image...", + "图片暂不可用": "Image unavailable", "推荐场景": "Recommended Scenarios", "推荐单端口启动：企业端、对话端和 API 文档都由同一个 FastAPI 进程挂载，适合本地演示和外部隧道测试。": "Recommended single-port startup: the console, chat, and API docs are mounted by one FastAPI process for local demos and external tunnel testing.", "推进执行": "Advance Execute", diff --git a/frontend-enterprise/src/pages/ChannelsPage.tsx b/frontend-enterprise/src/pages/ChannelsPage.tsx index 13249d1c..5ce2bd02 100644 --- a/frontend-enterprise/src/pages/ChannelsPage.tsx +++ b/frontend-enterprise/src/pages/ChannelsPage.tsx @@ -33,6 +33,7 @@ import type { ChannelBindingRead, ChannelBindCodeRead, ChannelConversationMessageRead, + ChannelConversationAttachment, ChannelConversationRead, ChannelDeliveryDay, ChannelDeliveryDayPage, @@ -100,6 +101,67 @@ function messageDisplay( return { label: msg.role, content: msg.content }; } +function ChannelAttachmentView({ + attachment, + bindingId, + sessionId, + messageId, +}: { + attachment: ChannelConversationAttachment; + bindingId: string; + sessionId: string; + messageId: string; +}) { + const [url, setUrl] = useState(null); + const [loading, setLoading] = useState(false); + const path = `/api/enterprise/channels/${bindingId}/conversations/${sessionId}/messages/${messageId}/attachments/${attachment.id}?tenant_id=${TENANT_ID}`; + + useEffect(() => { + if (attachment.kind !== 'image') return; + let disposed = false; + let objectUrl: string | null = null; + setLoading(true); + void api.blob(path).then((blob) => { + objectUrl = URL.createObjectURL(blob); + if (!disposed) setUrl(objectUrl); + else URL.revokeObjectURL(objectUrl); + }).catch(() => { + if (!disposed) setUrl(null); + }).finally(() => { + if (!disposed) setLoading(false); + }); + return () => { + disposed = true; + if (objectUrl) URL.revokeObjectURL(objectUrl); + }; + }, [attachment.id, attachment.kind, path]); + + if (attachment.kind === 'image') { + return url ? ( + {attachment.filename} + ) : {loading ? '图片加载中…' : '图片暂不可用'}; + } + return ( + + ); +} + function isSessionRecovering(binding: ChannelBindingRead): boolean { return ( !binding.connected && @@ -966,9 +1028,22 @@ export default function ChannelsPage({ {shown.label} · {formatTime(msg.created_at)} - - {shown.content} - +
+ {shown.content} + {msg.attachments?.length ? ( + + {msg.attachments.map((attachment) => ( + + ))} + + ) : null} +
); })} diff --git a/frontend-enterprise/src/types/index.ts b/frontend-enterprise/src/types/index.ts index 793a4171..584d7c02 100644 --- a/frontend-enterprise/src/types/index.ts +++ b/frontend-enterprise/src/types/index.ts @@ -905,12 +905,12 @@ export type ChannelConversationRead = { updated_at: string; }; -export type ChannelConversationAttachmentRead = { - id?: string; - filename?: string; - content_type?: string; - size?: number; - kind?: string; +export type ChannelConversationAttachment = { + id: string; + filename: string; + content_type: string; + size: number; + kind: 'text' | 'pdf' | 'image' | 'binary'; }; export type ChannelConversationMessageRead = { @@ -918,7 +918,7 @@ export type ChannelConversationMessageRead = { role: string; content: string; created_at: string; - attachments?: ChannelConversationAttachmentRead[]; + attachments?: ChannelConversationAttachment[] | null; }; export type ChannelBindCodeRead = {