""" 外部消息接入、会话管理、交互路由与智能体调用编排。 """ import asyncio import base64 import json import mimetypes import re import threading import uuid from collections.abc import Mapping from concurrent.futures import CancelledError as FutureCancelledError from concurrent.futures import Future from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Protocol, Tuple, Union, cast from urllib.parse import unquote, urlparse from app.application.agent import ( get_running_agent_manager, is_audio_input_available, supports_image_input, transcribe_audio, ) from app.application.messaging import router as interaction_router from app.application.messaging.interaction import InteractionContext, InteractionDispatch from app.application.messaging.interaction.agent import ( agent_interaction_manager, is_agent_learning_command, parse_agent_choice_callback, ) from app.application.messaging.media import media_interaction_manager from app.application.messaging.plugin import PluginInputInteractionHandler from app.application.messaging.session import MessageSessionService from app.application.messaging.site import site_interaction_manager from app.application.messaging.skill import SkillInteractionHandler, skill_interaction_manager from app.application.messaging.subscribe import subscribe_interaction_manager from app.chain.base import ChainBase from app.chain.interaction import MediaInteractionChain as _MediaInteractionChain from app.chain.site import SiteChain from app.chain.subscribe.facade import SubscribeChain from app.chain.system import SystemChain from app.chain.transfer.facade import TransferChain from app.runtime.log import logger from app.runtime.loop import main_loop_registry from app.runtime.tasks import get_task_registry from app.schemas.message import IncomingMessage, Message from app.schemas.notification import ChannelCapabilityManager from app.schemas.types import EventType, NotificationChannel class MessageResponsePort(Protocol): """ 消息链读取附件所需的最小同步 HTTP 响应契约。 :ivar content: 远程附件的原始字节内容 :ivar headers: 附件响应头，用于判断媒体类型及编码 """ content: bytes headers: Mapping[str, str] def close(self) -> None: """ 释放响应与连接资源。 """ ... class MessageHttpPort(Protocol): """ 消息链读取远程附件所需的同步 GET 端口。 """ def get(self, url: str, *, timeout: int) -> Optional[MessageResponsePort]: """ 读取附件响应，并保留无响应与有响应两态。 :param url: 远程图片或音频附件的完整地址 :param timeout: 请求超时秒数 :return: 原始附件响应，未取得响应时为 None；调用方读取后负责关闭 """ ... _message_http_lock = threading.RLock() _message_http_port: Optional[MessageHttpPort] = None def configure_message_http_port(http: MessageHttpPort) -> Optional[MessageHttpPort]: """ 由启动组合根装配消息附件 HTTP 端口，并返回旧实现。 """ global _message_http_port with _message_http_lock: previous = _message_http_port _message_http_port = http return previous def reset_message_http_port(http: Optional[MessageHttpPort] = None) -> None: """ 恢复指定消息 HTTP 端口；省略参数时回到未装配状态。 """ global _message_http_port with _message_http_lock: _message_http_port = http def _message_http_snapshot() -> MessageHttpPort: """ 读取消息 HTTP 端口快照，未装配时稳定失败。 """ with _message_http_lock: http = _message_http_port if http is None: raise RuntimeError("消息附件 HTTP 端口尚未由启动组合根装配") return http def _read_message_http( url: str, *, timeout: int = 30, ) -> tuple[Optional[bytes], Mapping[str, str]]: """ 读取远程消息附件并在复制所需字段后立即释放响应。 """ response = _message_http_snapshot().get(url, timeout=timeout) if response is None: return None, {} try: return response.content or None, dict(response.headers) finally: try: response.close() except Exception as err: logger.debug(f"释放消息附件响应失败：{str(err)}") class MessageChain(ChainBase): """ 外来消息处理链 """ _ai_prefix = "/ai" _no_ai_prefix = "/noai" # 用户会话信息按渠道、接入源与用户组合键隔离，旧直接调用保留原始键。 _user_sessions: Dict[Union[str, int], tuple[str, datetime]] = {} # 会话超时时间（分钟） _session_timeout_minutes: int = 24 * 60 @staticmethod def _schedule_agent_session_clear(session_id: str, userid: Union[str, int]) -> None: """ 异步调度 Agent 会话清理，避免同步消息链阻塞在模型资源释放上。 """ if not session_id: return manager = get_running_agent_manager() if manager is None: return clear_task = None try: clear_task = manager.clear_session( session_id=session_id, user_id=str(userid) ) get_task_registry().submit_threadsafe( clear_task, loop=main_loop_registry.require(), owner="chain.message.agent_session_clear", ) except Exception as e: if clear_task: clear_task.close() logger.warning(f"调度清理智能体会话失败: {e}") def _cleanup_expired_user_sessions(self, current_time: datetime) -> None: """ 清理超过复用窗口的用户会话映射，并同步释放旧 Agent 实例。 """ self._message_session_service().cleanup(current_time) def _message_session_service(self) -> MessageSessionService: """ 用类级兼容映射构建可测试的用户会话服务。 """ def expired(session_id: str, user_key: Union[str, int]) -> None: """ 映射键只隔离入口，会话回收仍使用真正的渠道用户身份。 """ user_id = user_key if isinstance(user_key, str) and user_key.startswith("message:"): try: identity = json.loads(user_key.removeprefix("message:")) if isinstance(identity, list) and len(identity) == 3 and isinstance(identity[2], str): user_id = identity[2] except ValueError: pass self._schedule_agent_session_clear(session_id, user_id) return MessageSessionService( sessions=self._user_sessions, timeout_minutes=self._session_timeout_minutes, expired_handler=expired, ) def _plugin_input_interaction_handler(self) -> PluginInputInteractionHandler: """ 构造使用当前 Chain 消息与事件端口的插件输入处理器。 """ return PluginInputInteractionHandler( messenger=self, event_publisher=self.eventmanager, ) @dataclass class _ProcessingStatus: """ 保存回复处理中状态所需的渠道、消息和会话定位信息。 :param channel: 需要显示处理状态的通知渠道 :param source: 渠道接入来源标识 :param userid: 接收状态反馈的用户标识，None 表示未提供 :param message_id: 用于关联或更新的原消息标识，None 表示未提供 :param chat_id: 会话或群聊标识，None 表示使用渠道默认定位 :param metadata: 渠道扩展上下文，None 在序列化时转换为空字典 """ channel: NotificationChannel source: str userid: Optional[Union[str, int]] = None message_id: Optional[Union[str, int]] = None chat_id: Optional[Union[str, int]] = None metadata: Optional[Dict[str, Any]] = None def to_dict(self) -> Dict[str, Any]: """ 转换为模块接口可安全传递的普通字典。 """ return { "channel": self.channel.value, "source": self.source, "userid": self.userid, "message_id": self.message_id, "chat_id": self.chat_id, "metadata": self.metadata or {}, } def process(self, body: Any, form: Any, args: Any) -> None: """ 调用模块识别消息内容 """ source = args.get("source") info = self.message_parser(source=source, body=body, form=form, args=args) if not info: logger.info("消息链路未识别到有效消息: source=%s", source) return source = str(info.source or source or "") channel = info.channel if channel is None: logger.warning("消息缺少有效通知渠道：source=%s", source) return userid = info.userid # 用户名（当渠道未提供公开用户名时，回退为 userid 的字符串，避免后续类型校验异常） username = ( str(info.username) if info.username not in (None, "") else str(userid) ) if userid is None or userid == "": logger.debug(f"未识别到用户ID：{body}{form}{args}") return text = str(info.text).strip() if info.text else "" images = info.images audio_refs = info.audio_refs files = info.files # 结构化按钮回调数据，优先于 CALLBACK: 文本前缀 callback_data = ( str(info.callback_data).strip() if info.callback_data else None ) if not text and not callback_data and not images and not audio_refs and not files: logger.debug(f"未识别到消息内容：：{body}{form}{args}") return original_message_id = info.message_id original_chat_id = info.chat_id reply_to_message_id = info.reply_to_message_id self.handle_message( channel=channel, source=source, userid=userid, username=username, is_channel_admin=info.is_channel_admin, text=text, original_message_id=original_message_id, original_chat_id=original_chat_id, reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, callback_data=callback_data, ) def handle_message( self, channel: NotificationChannel, source: str, userid: Union[str, int], username: str, text: Optional[str], original_message_id: Optional[Union[str, int]] = None, original_chat_id: Optional[str] = None, images: Optional[List[IncomingMessage.MessageImage]] = None, audio_refs: Optional[List[str]] = None, files: Optional[List[IncomingMessage.MessageAttachment]] = None, reply_to_message_id: Optional[Union[str, int]] = None, is_channel_admin: Optional[bool] = None, callback_data: Optional[str] = None, ) -> None: """ 识别消息内容，执行操作 """ text = text or "" images = IncomingMessage.MessageImage.normalize_list(images) # 兼容归一化：结构化回调优先，CALLBACK: 文本前缀作为旧渠道和插件直接调用的兼容入口 normalized_callback = str(callback_data or "").strip() or None if normalized_callback is None and str(text or "").startswith("CALLBACK:"): normalized_callback = str(text)[9:].strip() or None processing_status = None processing_finish_deferred = False try: # 语音输入只用于转写为文本，不默认改变回复形式。 has_audio_input = bool(audio_refs) if audio_refs: transcript = self._transcribe_audio_refs(audio_refs, channel, source) merged_parts = [] seen_parts = set() for item in [text.strip() if text else "", transcript or ""]: normalized = item.strip() if not normalized or normalized in seen_parts: continue seen_parts.add(normalized) merged_parts.append(normalized) text = "\n".join(merged_parts).strip() if not text: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="语音识别失败，请稍后重试", save_history=False, ) ) return if self._handle_secret_confirmation_control( channel=channel, source=source, userid=userid, username=username, is_channel_admin=is_channel_admin, text=text, original_message_id=original_message_id, original_chat_id=original_chat_id, images=images, audio_refs=audio_refs, files=files, has_audio_input=has_audio_input, ): return interaction_context = InteractionContext( channel=channel, source=source, user_id=userid, username=username, original_message_id=original_message_id, original_chat_id=original_chat_id, is_channel_admin=is_channel_admin, ) if self._plugin_input_interaction_handler().handle_text( context=interaction_context, text=text, reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, has_audio_input=has_audio_input, ): return is_agent_message = self._is_agent_message( userid=userid, text=text, callback_data=normalized_callback, images=images, files=files, has_audio_input=has_audio_input, ) # 回调消息不写入普通用户消息历史 if normalized_callback is None and not is_agent_message: self._record_user_message( channel=channel, source=source, userid=userid, username=username, text=text, ) if not is_agent_message: processing_status = self._mark_message_processing_started( channel=channel, source=source, userid=userid, original_message_id=original_message_id, original_chat_id=original_chat_id, text=text, ) processing_finish_deferred = self._handle_message_core( channel=channel, source=source, userid=userid, username=username, is_channel_admin=is_channel_admin, text=text, original_message_id=original_message_id, original_chat_id=original_chat_id, reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, has_audio_input=has_audio_input, processing_status=processing_status, callback_data=normalized_callback, ) is True finally: if not processing_finish_deferred: self._mark_message_processing_finished( channel=channel, source=source, userid=userid, status=processing_status, original_message_id=original_message_id, original_chat_id=original_chat_id, ) def _handle_secret_confirmation_control( self, channel: NotificationChannel, source: str, userid: Union[str, int], username: str, text: Optional[str], original_message_id: Optional[Union[str, int]] = None, original_chat_id: Optional[str] = None, images: Optional[List[IncomingMessage.MessageImage]] = None, audio_refs: Optional[List[str]] = None, files: Optional[List[IncomingMessage.MessageAttachment]] = None, has_audio_input: bool = False, is_channel_admin: Optional[bool] = None, ) -> bool: """ 将 TG/飞书中的确认控制文本交回所属 Agent 会话。 """ if channel not in {NotificationChannel.Telegram, NotificationChannel.Feishu}: return False if str(text or "").strip() not in {"确认", "取消"}: return False if images or audio_refs or files or has_audio_input: return False session_info = self._user_sessions.get(self._session_user_key(userid, channel, source)) if not session_info: return False session_id, _ = session_info manager = get_running_agent_manager() if manager is None or not manager.matches_secret_confirmation( session_id, str(userid), channel=channel.value, source=source, ): return False return self._handle_ai_message( text=str(text).strip(), channel=channel, source=source, userid=userid, username=username, is_channel_admin=is_channel_admin, original_message_id=original_message_id, original_chat_id=original_chat_id, images=images, files=files, session_id=session_id, has_audio_input=has_audio_input, ) def _handle_message_core( self, channel: NotificationChannel, source: str, userid: Union[str, int], username: str, text: Optional[str], original_message_id: Optional[Union[str, int]] = None, original_chat_id: Optional[str] = None, images: Optional[List[IncomingMessage.MessageImage]] = None, audio_refs: Optional[List[str]] = None, files: Optional[List[IncomingMessage.MessageAttachment]] = None, has_audio_input: bool = False, processing_status: Optional[_ProcessingStatus] = None, reply_to_message_id: Optional[Union[str, int]] = None, is_channel_admin: Optional[bool] = None, callback_data: Optional[str] = None, ) -> bool: """ 执行实际消息路由，便于统一包裹处理中状态。 """ text = text or "" context = InteractionContext( channel=channel, source=source, user_id=userid, username=username, original_message_id=original_message_id, original_chat_id=original_chat_id, is_channel_admin=is_channel_admin, ) if callback_data: if ChannelCapabilityManager.supports_callbacks(channel): return self._handle_callback( callback_data=callback_data, context=context, ) else: logger.warning( "渠道 %s 不支持回调，但收到了回调消息：%s", channel.value, callback_data, ) return False if self._plugin_input_interaction_handler().handle_text( context=context, text=text, reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, has_audio_input=has_audio_input, ): return False no_ai_requested, no_ai_text = self._strip_no_ai_prefix(text) if no_ai_requested: text = no_ai_text if not text: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="请输入要使用传统交互处理的内容", save_history=False, ) ) return False if text.startswith("/") and not self._has_ai_prefix(text) and not is_agent_learning_command(text): self.eventmanager.send_event( EventType.CommandExcute, { "cmd": text, "user": userid, "channel": channel, "source": source, "processing_status": processing_status.to_dict() if processing_status else None, }, ) return bool(processing_status) if not no_ai_requested and (self._has_ai_prefix(text) or is_agent_learning_command(text)): return self._handle_ai_message( text=text, channel=channel, source=source, userid=userid, username=username, is_channel_admin=is_channel_admin, original_message_id=original_message_id, original_chat_id=original_chat_id, images=images, files=files, has_audio_input=has_audio_input, ) # 最近活动的传统交互会话（按创建时间选择，避免旧会话抢占新输入） if self._interaction_router().dispatch_active_text(context, text): return False if ( not no_ai_requested and self.runtime_config.ai_agent_enable and ( self.runtime_config.ai_agent_global or images or files or has_audio_input ) ): return self._handle_ai_message( text=text, channel=channel, source=source, userid=userid, username=username, is_channel_admin=is_channel_admin, original_message_id=original_message_id, original_chat_id=original_chat_id, images=images, files=files, has_audio_input=has_audio_input, ) if _MediaInteractionChain().handle_text_interaction( channel=channel, source=source, userid=userid, username=username, text=text, ): return False self.eventmanager.send_event( EventType.UserMessage, { "text": text, "userid": userid, "channel": channel, "source": source, "chat_id": original_chat_id, "reply_to_message_id": reply_to_message_id, }, ) return False @classmethod def _strip_no_ai_prefix(cls, text: str) -> Tuple[bool, str]: """ 解析 /noai 前缀，显式要求本条消息绕过全局智能体。 """ normalized = (text or "").strip() pattern = rf"^{re.escape(cls._no_ai_prefix)}(?:\s+|[:：]\s*|$)(.*)$" match = re.match(pattern, normalized, re.IGNORECASE | re.DOTALL) if not match: return False, text return True, match.group(1).strip() @classmethod def _has_ai_prefix(cls, text: str) -> bool: """ 判断消息是否使用显式 AI 前缀。 """ return (text or "").lower().startswith(cls._ai_prefix) def _is_agent_message( self, userid: Union[str, int], text: str, callback_data: Optional[str] = None, images: Optional[List[IncomingMessage.MessageImage]] = None, files: Optional[List[IncomingMessage.MessageAttachment]] = None, has_audio_input: bool = False, ) -> bool: """ 判断本条消息是否会进入 Agent worker，由 Agent worker 管理 typing 生命周期。 """ if callback_data: return parse_agent_choice_callback(callback_data) is not None if self._has_ai_prefix(text) or is_agent_learning_command(text): return True if text.startswith("/"): return False if not ( self.runtime_config.ai_agent_enable and ( self.runtime_config.ai_agent_global or images or files or has_audio_input ) ): return False if self._interaction_router().has_pending(userid): return False return True def _mark_message_processing_started( self, channel: NotificationChannel, source: str, userid: Union[str, int], original_message_id: Optional[Union[str, int]], original_chat_id: Optional[Union[str, int]], text: str, ) -> Optional[_ProcessingStatus]: """ 为支持的渠道标记“消息正在处理”。 """ status = self.start_message_processing_status( channel=channel, source=source, userid=userid, message_id=original_message_id, chat_id=original_chat_id, text=text, ) if not status: return None metadata = status.get("metadata") return self._ProcessingStatus( channel=channel, source=source, userid=status.get("userid", userid), message_id=status.get("message_id", original_message_id), chat_id=status.get("chat_id", original_chat_id), metadata=metadata if isinstance(metadata, dict) else {}, ) def _mark_message_processing_finished( self, channel: NotificationChannel, source: str, userid: Union[str, int], status: Optional[_ProcessingStatus] = None, original_message_id: Optional[Union[str, int]] = None, original_chat_id: Optional[Union[str, int]] = None, ) -> None: """ 结束渠道侧“消息正在处理”状态。 不同渠道的表现可能是 reaction、typing 等，消息链只负责调用通用模块接口。 """ if not status: return self.finish_message_processing_status( status=status.to_dict(), channel=channel, source=source, userid=userid, message_id=status.message_id or original_message_id, chat_id=status.chat_id or original_chat_id, ) def _interaction_router(self) -> interaction_router.InteractionRouter: """ 构造交互路由器，文本会话按创建时间选择，回调路由注册顺序即优先级。 """ from app.application.messaging.update import update_interaction_manager session_routes = [ interaction_router.SessionRoute( name="sites", get_pending=site_interaction_manager.get_by_user, handle_text=interaction_router.adapt_session_text_handler( lambda **kw: SiteChain().handle_text_interaction(**kw) ), ), interaction_router.SessionRoute( name="subscribes", get_pending=subscribe_interaction_manager.get_by_user, handle_text=interaction_router.adapt_session_text_handler( lambda **kw: SubscribeChain().handle_text_interaction(**kw) ), ), interaction_router.SessionRoute( name="skills", get_pending=skill_interaction_manager.get_by_user, handle_text=interaction_router.adapt_session_text_handler( lambda **kw: SkillInteractionHandler(messenger=self).handle_text_interaction(**kw) ), ), interaction_router.SessionRoute( name="media", get_pending=media_interaction_manager.get_by_user, handle_text=interaction_router.adapt_session_text_handler( lambda **kw: _MediaInteractionChain().handle_text_interaction(**kw) ), ), interaction_router.SessionRoute( name="update", get_pending=update_interaction_manager.get_by_user, handle_text=interaction_router.adapt_session_text_handler(lambda **kw: SystemChain().handle_update_text_interaction(**kw)), ), ] def _dispatch_agent_choice(callback_data: str, context: InteractionContext) -> InteractionDispatch: """ 将已解析的选择回调交给当前 Agent，并返回是否处理的结果。 """ handled = self._handle_agent_choice_callback( callback_data=callback_data, context=context, ) # Agent 选择回调会接续会话，需要延迟结束处理中状态 return InteractionDispatch(handled=handled, defer_processing_finish=handled) def _dispatch_plugin_callback(callback_data: str, context: InteractionContext) -> InteractionDispatch: """ 校验插件回调格式后派发交互，不接收无法识别的载荷。 """ parsed = PluginInputInteractionHandler.parse_callback(callback_data) if not parsed: return InteractionDispatch(handled=False) plugin_id, content = parsed self.eventmanager.send_event( EventType.MessageAction, { "plugin_id": plugin_id, "text": content, "userid": context.user_id, "channel": context.channel, "source": context.source, "original_message_id": context.original_message_id, "original_chat_id": context.original_chat_id, }, ) return InteractionDispatch(handled=True) callback_routes = [ interaction_router.CallbackRoute( name="transfer", matches=lambda data: TransferChain.parse_failed_transfer_callback(data) is not None, dispatch=lambda data, context: InteractionDispatch( handled=TransferChain().handle_failed_transfer_callback( callback_data=data, channel=context.channel, source=context.source, userid=context.user_id, username=context.username, ) ), ), interaction_router.CallbackRoute( name="skill", matches=lambda data: data.startswith("skills:"), dispatch=interaction_router.adapt_callback_handler( lambda **kw: SkillInteractionHandler(messenger=self).handle_callback_interaction(**kw) ), ), interaction_router.CallbackRoute( name="site", matches=lambda data: data.startswith("sites:"), dispatch=interaction_router.adapt_callback_handler( lambda **kw: SiteChain().handle_callback_interaction(**kw) ), ), interaction_router.CallbackRoute( name="subscribe", matches=lambda data: data.startswith("subscribes:"), dispatch=interaction_router.adapt_callback_handler( lambda **kw: SubscribeChain().handle_callback_interaction(**kw) ), ), interaction_router.CallbackRoute( name="media", matches=lambda data: _MediaInteractionChain.parse_callback(data) is not None, dispatch=interaction_router.adapt_callback_handler( lambda **kw: _MediaInteractionChain().handle_callback_interaction(**kw) ), ), interaction_router.CallbackRoute( name="update", matches=lambda data: data.startswith("update:"), dispatch=interaction_router.adapt_callback_handler(lambda **kw: SystemChain().handle_update_callback_interaction(**kw)), ), interaction_router.CallbackRoute( name="agent_choice", matches=lambda data: parse_agent_choice_callback(data) is not None, dispatch=_dispatch_agent_choice, ), interaction_router.CallbackRoute( name="plugin", matches=lambda data: data.startswith("[PLUGIN]"), dispatch=_dispatch_plugin_callback, ), ] return interaction_router.InteractionRouter( session_routes=session_routes, callback_routes=callback_routes, ) def _handle_callback( self, callback_data: str, context: InteractionContext, ) -> bool: """ 处理按钮回调。 :return: 是否延迟结束处理中状态（Agent 选择回调会等待会话接续） """ logger.info(f"处理按钮回调：{callback_data}") result = self._interaction_router().dispatch_callback(context, callback_data) if result.handled: return result.defer_processing_finish logger.error(f"回调数据格式错误：{callback_data}") self.post_message( Message( channel=context.channel, source=context.source, userid=context.user_id, username=context.username, title="回调数据格式错误，请检查！", save_history=False, ) ) return False def _handle_agent_choice_callback( self, *, callback_data: str, context: InteractionContext, ) -> bool: """ 将 Agent 按钮选择回传为同一会话中的下一条用户消息。 """ callback = parse_agent_choice_callback(callback_data) if not callback: return False request_id, option_index = callback resolved = agent_interaction_manager.resolve( request_id=request_id, option_index=option_index, user_id=str(context.user_id), ) if not resolved: self.post_message( Message( channel=context.channel, source=context.source, userid=context.user_id, username=context.username, title="该选择已失效，请重新发起选择", save_history=False, ) ) return False request, option = resolved selected_text = option.value self._update_interaction_message_feedback( channel=context.channel, source=context.source or "", original_message_id=context.original_message_id, original_chat_id=context.original_chat_id, title=request.title, prompt=request.prompt, selected_label=option.label, ) self._bind_session_id(self._session_user_key(context.user_id, context.channel, context.source), request.session_id) return self._handle_ai_message( text=selected_text, channel=context.channel, source=context.source or "", userid=context.user_id, username=context.username or str(context.user_id), is_channel_admin=context.is_channel_admin, session_id=request.session_id, ) def _update_interaction_message_feedback( self, channel: NotificationChannel, source: str, original_message_id: Optional[Union[str, int]], original_chat_id: Optional[str], prompt: str, selected_label: str, title: Optional[str] = None, ) -> None: """ 在用户点击交互按钮后，立即更新原消息，明确显示已选择的内容。 """ if not original_message_id or not original_chat_id: return lines = [prompt.strip()] if selected_label: lines.append(f"已选择：{selected_label}") feedback_text = "\n\n".join(line for line in lines if line) self.edit_message( channel=channel, source=source, message_id=original_message_id, chat_id=original_chat_id, title=title, text=feedback_text, ) @staticmethod def _session_user_key(userid: Union[str, int], channel: NotificationChannel, source: Optional[str]) -> str: """ 以无歧义序列化生成渠道和接入源隔离的会话键，不回退读取旧裸用户绑定。 :param userid: 渠道内原始用户标识 :param channel: 接收消息的渠道 :param source: 该渠道的配置实例，未提供时使用空标识 :return: 内部会话映射键，不作为发送消息的用户ID """ return "message:" + json.dumps([channel.value, source or "", str(userid)], ensure_ascii=False, separators=(",", ":")) def _get_or_create_session_id(self, userid: Union[str, int]) -> str: """ 获取或创建会话ID 如果用户上次会话在15分钟内，则复用相同的会话ID；否则创建新的会话ID """ resolution = self._message_session_service().resolve(userid) if resolution.reused: logger.info( f"复用会话ID: {resolution.session_id}, 用户: {userid}, " f"距离上次会话: {resolution.inactive_minutes:.1f}分钟" ) else: logger.info(f"创建新会话ID: {resolution.session_id}, 用户: {userid}") return resolution.session_id def _bind_session_id(self, userid: Union[str, int], session_id: str) -> None: """ 将用户会话绑定到指定的 session_id，并刷新最后活动时间。 """ self._message_session_service().bind(userid, session_id) def bind_user_session(self, userid: Union[str, int], session_id: str) -> None: """ 绑定用户与指定智能体会话，供非传统入口复用远程命令状态查询。 :param userid: 用户 ID :param session_id: 智能体会话 ID """ self._bind_session_id(self._session_user_key(userid, NotificationChannel.WebAgent, "web-agent"), session_id) def _record_user_message( self, channel: NotificationChannel, source: str, userid: Union[str, int], username: str, text: str, ) -> None: """ 保存一条用户消息到消息历史与数据库。 """ self.messagehelper.put( IncomingMessage( userid=userid, username=username, channel=channel, source=source, text=text, ), role="user", ) self.messageoper.add( channel=channel, source=source, userid=username or userid, text=text, action=0, ) def clear_user_session(self, userid: Union[str, int]) -> bool: """ 清除指定用户的会话信息 返回是否成功清除 """ session_id = self._message_session_service().clear(userid) if session_id: logger.info(f"已清除用户 {userid} 的会话: {session_id}") return True return False def remote_clear_session( self, channel: NotificationChannel, userid: Union[str, int], source: Optional[str] = None, ) -> None: """ 清除用户会话（远程命令接口） """ session_id = self._message_session_service().clear(self._session_user_key(userid, channel, source)) if session_id: logger.info(f"已清除用户 {userid} 的会话: {session_id}") # 如果有会话ID，同时清除智能体的会话记忆 if session_id: self._schedule_agent_session_clear(session_id, userid) self.post_message( Message( channel=channel, source=source, title="智能体会话已清除，下次将创建新的会话", userid=userid, save_history=False, ) ) else: self.post_message( Message( channel=channel, source=source, title="您当前没有活跃的智能体会话", userid=userid, save_history=False, ) ) def remote_stop_agent( self, channel: NotificationChannel, userid: Union[str, int], source: Optional[str] = None, ) -> None: """ 应急停止当前正在执行的Agent推理（远程命令接口）。 与 /clear_session 不同，此命令不会清除会话和记忆， 停止后用户仍可继续对话。 """ # 查找用户的会话ID（不弹出，保留会话） session_info = self._message_session_service().get(self._session_user_key(userid, channel, source)) if session_info: session_id, _ = session_info manager = get_running_agent_manager() try: if manager is None: stopped = False else: future = asyncio.run_coroutine_threadsafe( manager.stop_current_task(session_id=session_id), main_loop_registry.require(), ) stopped = future.result(timeout=10) except Exception as e: logger.warning(f"停止Agent推理失败: {e}") stopped = False if stopped: self.post_message( Message( channel=channel, source=source, title="智能体推理已应急停止，会话记忆已保留，您可以继续对话", userid=userid, save_history=False, ) ) else: self.post_message( Message( channel=channel, source=source, title="当前没有正在执行的智能体任务", userid=userid, save_history=False, ) ) else: self.post_message( Message( channel=channel, source=source, title="您当前没有活跃的智能体会话", userid=userid, save_history=False, ) ) @staticmethod def _format_token_count(value: Optional[int]) -> str: """ 格式化 token 数量，未报告的用量显示为未知。 """ return f"{value:,}" if value is not None else "未知" @classmethod def _format_session_status_text(cls, status: Dict[str, Any]) -> str: """ 把会话上下文、模型调用和用量统计转换为用户可读状态。 """ context_window_tokens = status.get("context_window_tokens") last_input_tokens = status.get("last_input_tokens") if context_window_tokens and status.get("model_call_count"): context_ratio = status.get("last_context_usage_ratio") if ( context_ratio is None and status.get("last_input_usage_available") is True and last_input_tokens is not None ): context_ratio = last_input_tokens / context_window_tokens context_usage_text = ( f"{cls._format_token_count(last_input_tokens)} / " f"{cls._format_token_count(context_window_tokens)} " f"({context_ratio * 100:.2f}%)" if context_ratio is not None else f"{cls._format_token_count(last_input_tokens)} / " f"{cls._format_token_count(context_window_tokens)}" ) else: context_usage_text = "暂无模型调用数据" lines = [ f"会话ID: {status.get('session_id') or '未知'}", f"执行状态: {'运行中' if status.get('is_processing') else '空闲'}", f"当前模型: {status.get('model') or '未知'}", f"上下文窗口: {cls._format_token_count(context_window_tokens)} tokens", f"最近一次上下文占用: {context_usage_text}", ] if status.get("last_request_estimate_available"): estimated_tokens = status.get("last_estimated_input_tokens") estimated_ratio = status.get("last_estimated_input_ratio") estimate_text = ( f"{cls._format_token_count(estimated_tokens)} / " f"{cls._format_token_count(context_window_tokens)}" ) if estimated_ratio is not None: estimate_text += f" ({estimated_ratio * 100:.2f}%)" if status.get("last_estimated_over_input_limit"): estimate_text += "，估算已超输入上限" lines.extend( [ f"最终请求估算: {estimate_text}", "估算组成: " f"消息 {cls._format_token_count(status.get('last_estimated_message_tokens'))} / " f"系统 {cls._format_token_count(status.get('last_estimated_system_tokens'))} / " f"工具 {cls._format_token_count(status.get('last_estimated_tool_tokens'))} / " f"其中图片固定成本 {cls._format_token_count(status.get('last_estimated_multimodal_tokens'))}", ] ) actual_input_tokens = status.get("last_actual_input_tokens") estimate_error_tokens = status.get("last_estimate_error_tokens") if actual_input_tokens is not None and estimate_error_tokens is not None: estimate_error_ratio = status.get("last_estimate_error_ratio") error_text = ( f"实际 {cls._format_token_count(actual_input_tokens)} / " f"误差 {estimate_error_tokens:+,}" ) if estimate_error_ratio is not None: error_text += f" ({estimate_error_ratio:+.2%})" lines.append(f"估算校准: {error_text}") lines.append( f"最近一次 tokens: 输入 {cls._format_token_count(status.get('last_input_tokens'))} / 输出 {cls._format_token_count(status.get('last_output_tokens'))} / 总计 {cls._format_token_count(status.get('last_total_tokens'))}" ) if status.get("last_cache_usage_available"): last_cache_ratio = status.get("last_cache_hit_ratio") lines.append( "最近一次缓存: " f"命中 {cls._format_token_count(status.get('last_cache_read_input_tokens'))} / " f"写入 {cls._format_token_count(status.get('last_cache_write_input_tokens'))} / " f"未命中 {cls._format_token_count(status.get('last_uncached_input_tokens'))}" + ( f" ({last_cache_ratio * 100:.2f}%)" if last_cache_ratio is not None else "" ), ) if status.get("cache_usage_available"): total_cache_ratio = status.get("total_cache_hit_ratio") lines.append( "当前会话累计缓存: " f"命中 {cls._format_token_count(status.get('total_cache_read_input_tokens'))} / " f"写入 {cls._format_token_count(status.get('total_cache_write_input_tokens'))} / " f"未命中 {cls._format_token_count(status.get('total_uncached_input_tokens'))}" + ( f" ({total_cache_ratio * 100:.2f}%)" if total_cache_ratio is not None else "" ), ) pending_messages = status.get("pending_messages", 0) queue_capacity = status.get("queue_capacity") pending_text = ( f"{pending_messages} / {queue_capacity}" if queue_capacity else str(pending_messages) ) lines.extend( [ f"当前会话累计 tokens: 输入 {cls._format_token_count(status.get('total_input_tokens'))} / 输出 {cls._format_token_count(status.get('total_output_tokens'))} / 总计 {cls._format_token_count(status.get('total_tokens'))}", f"模型调用次数: {status.get('model_call_count', 0)}", f"排队消息数: {pending_text}", f"最后更新: {status.get('last_updated_at') or '暂无'}", ] ) if status.get("queue_rejections"): lines.append(f"排队拒绝次数: {status['queue_rejections']}") if status.get("shutdown_pending"): lines.append("会话状态: 正在停止") return "\n".join(lines) def remote_session_status( self, channel: NotificationChannel, userid: Union[str, int], source: Optional[str] = None, ) -> None: """ 查询当前用户的智能体会话状态。 """ session_info = self._message_session_service().get(self._session_user_key(userid, channel, source)) if not session_info: self.post_message( Message( channel=channel, source=source, title="您当前没有活跃的智能体会话", userid=userid, save_history=False, ) ) return session_id, _ = session_info manager = get_running_agent_manager() if manager is None: self.post_message( Message( channel=channel, source=source, title="您当前没有活跃的智能体会话", userid=userid, save_history=False, ) ) return status = manager.get_session_status(session_id=session_id) self.post_message( Message( channel=channel, source=source, title="当前智能体会话状态", text=self._format_session_status_text(status), userid=userid, save_history=False, ) ) def _handle_ai_message( self, text: str, channel: NotificationChannel, source: str, userid: Union[str, int], username: str, original_message_id: Optional[Union[str, int]] = None, original_chat_id: Optional[str] = None, images: Optional[List[IncomingMessage.MessageImage]] = None, files: Optional[List[IncomingMessage.MessageAttachment]] = None, session_id: Optional[str] = None, has_audio_input: bool = False, is_channel_admin: Optional[bool] = None, ) -> bool: """ 处理AI智能体消息 """ try: if not self.runtime_config.ai_agent_enable: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="MoviePilot智能助手未启用，请在系统设置中启用", save_history=False, ) ) return False manager = get_running_agent_manager() if manager is None: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="MoviePilot智能助手服务尚未就绪，请稍后重试", save_history=False, ) ) return False images = IncomingMessage.MessageImage.normalize_list(images) if self._has_ai_prefix(text): # 前缀匹配不区分大小写，但保留原始正文避免改变用户输入内容。 user_message = text[len(self._ai_prefix):].strip() else: user_message = text.strip() if not user_message and not images and not files: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="请输入您的问题或需求", save_history=False, ) ) return False session_key = self._session_user_key(userid, channel, source) session_id = session_id or self._get_or_create_session_id(session_key) self._bind_session_id(session_key, session_id) # 将可直接输入给 LLM 的附件统一转换为 data URL original_images = images image_data_urls: Optional[list[str]] = None all_files = list(files or []) if images and supports_image_input( provider=self.runtime_config.llm_provider, model=self.runtime_config.llm_model, ): image_data_urls = self._download_attachments_to_data_urls( images, channel, source ) if original_images and not image_data_urls and not user_message and not files: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="附件读取失败，请稍后重试", save_history=False, ) ) return False elif images: image_attachments = self._build_image_attachments(images) if ( original_images and not image_attachments and not user_message and not files ): self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="附件读取失败，请稍后重试", save_history=False, ) ) return False all_files.extend(image_attachments) images = None prepared_files = self._prepare_agent_files( session_id=session_id, files=all_files, channel=channel, source=source, ) if all_files and not prepared_files and not user_message and not image_data_urls: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title="文件读取失败，请稍后重试", save_history=False, ) ) return False process_kwargs = { "session_id": session_id, "user_id": str(userid), "message": user_message, "images": image_data_urls, "files": prepared_files, "channel": channel.value if channel else None, "source": source, "username": username, "is_channel_admin": is_channel_admin, "original_message_id": str(original_message_id) if original_message_id else None, "original_chat_id": original_chat_id, } if has_audio_input: process_kwargs["has_audio_input"] = True # 在事件循环中处理，并消费跨线程 Future 的失败，避免队列满时静默丢消息。 submission_future = asyncio.run_coroutine_threadsafe( manager.process_message(**process_kwargs), main_loop_registry.require(), ) def _report_agent_submission_failure(completed: Future[Any]) -> None: """ 消费跨线程消息提交结果并反馈排队或服务错误。 """ try: completed.result() except BaseException as error: if isinstance( error, (asyncio.CancelledError, FutureCancelledError), ): return error_code = getattr(error, "code", None) if error_code == "agent_manager_queue_full": title = "智能助手当前排队已满，请稍后重试" elif error_code == "agent_manager_unavailable": title = "智能助手服务暂不可用，请稍后重试" else: title = "智能助手处理失败，请查看日志" logger.warning(f"Agent 消息提交失败: {error}") try: self.post_message( Message( channel=channel, source=source, userid=userid, username=username, title=title, original_message_id=original_message_id, original_chat_id=original_chat_id, save_history=False, ) ) except Exception as report_error: logger.error(f"发送 Agent 提交失败提示失败: {report_error}") if submission_future is not None: submission_future.add_done_callback(_report_agent_submission_failure) return True except Exception as e: logger.error(f"处理AI智能体消息失败: {e}", exc_info=True) self.messagehelper.put("智能助手执行失败，请稍后重试", role="system", title="MoviePilot助手", level="error") return False def _transcribe_audio_refs( self, audio_refs: List[str], channel: NotificationChannel, source: str ) -> Optional[str]: """ 下载并识别语音消息，仅处理当前已接入的渠道。 """ if not audio_refs: return None if not is_audio_input_available(): logger.warning("音频输入能力未配置或未启用，跳过语音识别") return None transcripts = [] for audio_ref in audio_refs: try: if audio_ref.startswith("tg://voice_file_id/"): file_id = audio_ref.replace("tg://voice_file_id/", "", 1) content = self.run_module( "download_telegram_file_bytes", file_id=file_id, source=source ) filename = "input.ogg" elif audio_ref.startswith("tg://audio_file_id/"): file_id = audio_ref.replace("tg://audio_file_id/", "", 1) content = self.run_module( "download_telegram_file_bytes", file_id=file_id, source=source ) filename = "input.mp3" elif audio_ref.startswith("wxwork://voice_media_id/"): content = self.run_module( "download_wechat_media_bytes", media_ref=audio_ref, source=source, ) filename = "input.amr" elif audio_ref.startswith("wxclaw://voice/"): content = self.run_module( "download_wechat_media_bytes", media_ref=audio_ref, source=source, ) filename = self._guess_audio_filename( audio_ref, default="input.amr" ) elif audio_ref.startswith("slack://file/"): content = self.run_module( "download_slack_file_bytes", file_ref=audio_ref, source=source ) filename = self._guess_audio_filename( audio_ref, default="input.ogg" ) elif audio_ref.startswith("discord://file/"): content = self.run_module( "download_discord_file_bytes", file_ref=audio_ref, source=source ) filename = self._guess_audio_filename( audio_ref, default="input.ogg" ) elif audio_ref.startswith("qq://file/"): content = self.run_module( "download_qq_file_bytes", file_ref=audio_ref, source=source ) filename = self._guess_audio_filename( audio_ref, default="input.ogg" ) elif audio_ref.startswith("vocechat://file/"): content = self.run_module( "download_vocechat_file_bytes", file_ref=audio_ref, source=source, ) filename = self._guess_audio_filename( audio_ref, default="input.ogg" ) elif audio_ref.startswith("synology://file/"): content = self.run_module( "download_synologychat_file_bytes", file_ref=audio_ref, source=source, ) filename = self._guess_audio_filename( audio_ref, default="input.ogg" ) elif audio_ref.startswith("wxbot://voice"): continue elif audio_ref.startswith("feishu://file/"): content = self.run_module( "download_feishu_file_bytes", file_ref=audio_ref, source=source ) filename = self._guess_audio_filename( audio_ref, default="input.opus" ) elif audio_ref.startswith("http"): content, _ = _read_message_http(audio_ref) filename = self._guess_audio_filename( audio_ref, default="input.ogg" ) else: logger.debug( "暂不支持的语音引用: channel=%s, source=%s, ref=%s", channel.value if channel else None, source, audio_ref, ) continue if not content: logger.warning( "语音下载失败，跳过识别: channel=%s, source=%s, ref=%s", channel.value if channel else None, source, audio_ref, ) continue transcript = transcribe_audio( content=content, filename=filename ) if transcript: transcripts.append(transcript) logger.info( "语音识别成功: channel=%s, source=%s, ref=%s, text_len=%s", channel.value if channel else None, source, audio_ref, len(transcript), ) except Exception as err: logger.error(f"语音识别失败: {err}") return "\n".join(transcripts).strip() if transcripts else None @staticmethod def _guess_audio_filename(audio_ref: str, default: str = "input.ogg") -> str: """ 根据引用中的扩展名推测音频文件名，便于 STT 服务识别格式。 """ if not audio_ref: return default raw_ref = unquote(audio_ref).split("?", 1)[0].split("#", 1)[0] match = re.search( r"([^/]+\.(mp3|m4a|wav|ogg|oga|opus|aac|amr|flac|mpga|mpeg|webm))$", raw_ref, flags=re.IGNORECASE, ) if match: return match.group(1) return default def _download_attachments_to_data_urls( self, attachments: List[IncomingMessage.MessageImage], channel: NotificationChannel, source: str, ) -> Optional[List[str]]: """ 下载可直接提供给 LLM 的附件内容，并统一转换为 data URL。 """ normalized_attachments = IncomingMessage.MessageImage.normalize_list(attachments) or [] if not normalized_attachments: return None data_urls: list[str] = [] for attachment in normalized_attachments: attachment_ref = attachment.ref try: before_count = len(data_urls) if attachment_ref.startswith("data:"): data_urls.append(attachment_ref) elif attachment_ref.startswith("tg://file_id/"): file_id = attachment_ref.replace("tg://file_id/", "") base64_data = self.run_module( "download_telegram_file_to_base64", file_id=file_id, source=source, ) if base64_data: data_urls.append(f"data:image/jpeg;base64,{base64_data}") elif attachment_ref.startswith( "wxwork://media_id/" ) or attachment_ref.startswith( "wxbot://image/" ) or attachment_ref.startswith( "wxclaw://image/" ): data_url = self.run_module( "download_wechat_image_to_data_url", image_ref=attachment_ref, source=source, ) if data_url: data_urls.append(data_url) elif attachment_ref.startswith("feishu://image/"): data_url = self.run_module( "download_feishu_image_to_data_url", image_ref=attachment_ref, source=source, ) if data_url: data_urls.append(data_url) elif channel == NotificationChannel.Slack: data_url = self.run_module( "download_slack_file_to_data_url", file_url=attachment_ref, source=source, ) if data_url: data_urls.append(data_url) elif attachment_ref.startswith("vocechat://file/"): data_url = self.run_module( "download_vocechat_image_to_data_url", image_ref=attachment_ref, source=source, ) if data_url: data_urls.append(data_url) elif attachment_ref.startswith("http"): content, headers = _read_message_http(attachment_ref) if content: base64_data = base64.b64encode(content).decode() mime_type = headers.get("Content-Type", "image/jpeg") data_urls.append(f"data:{mime_type};base64,{base64_data}") else: logger.debug( "暂不支持直接转换为 data URL 的附件引用: channel=%s, source=%s, ref=%s", channel.value if channel else None, source, attachment_ref, ) continue if len(data_urls) > before_count: logger.info( "附件读取成功并已转换为 data URL: channel=%s, source=%s, ref=%s, mime_type=%s", channel.value if channel else None, source, attachment_ref, attachment.mime_type, ) except Exception as err: logger.error( "附件读取失败，无法转换为 data URL: channel=%s, source=%s, ref=%s, error=%s", channel.value if channel else None, source, attachment_ref, err, ) return data_urls if data_urls else None def _build_image_attachments( self, images: List[IncomingMessage.MessageImage] ) -> List[IncomingMessage.MessageAttachment]: """ 将图片引用转换为附件描述，以便按文件方式交给 Agent 处理。 """ images = IncomingMessage.MessageImage.normalize_list(images) or [] if not images: return [] attachments = [] for index, image in enumerate(images, start=1): image_ref = image.ref if not image_ref: continue name = image.name or self._guess_image_attachment_name(image_ref, index) mime_type = image.mime_type or self._guess_image_mime_type(image_ref, name) attachments.append( IncomingMessage.MessageAttachment( ref=image_ref, name=name, mime_type=mime_type, size=image.size, ) ) return attachments def _prepare_agent_files( self, session_id: str, files: Optional[List[IncomingMessage.MessageAttachment]], channel: NotificationChannel, source: str, ) -> Optional[List[dict[str, Any]]]: """ 下载用户上传的附件，落盘到临时目录，并生成 Agent 可消费的文件描述。 """ if not files: return None prepared_files = [] for attachment in files: payload = { "name": attachment.name, "mime_type": attachment.mime_type, "size": attachment.size, "ref": attachment.ref, "status": "download_failed", } try: content = self._download_message_file_bytes( file_ref=attachment.ref, channel=channel, source=source, ) if not content: prepared_files.append(payload) continue local_path = self._save_agent_attachment( session_id=session_id, filename=attachment.name, content=content, mime_type=attachment.mime_type, ) payload.update( { "local_path": str(local_path), "status": "ready", } ) except Exception as err: logger.error(f"准备附件上下文失败: {attachment.ref}, error: {err}", exc_info=True) payload["error"] = "附件读取失败，请稍后重试" prepared_files.append(payload) return prepared_files or None def _download_message_file_bytes( self, file_ref: str, channel: NotificationChannel, source: str ) -> Optional[bytes]: """ 下载消息附件的原始字节内容。 """ if not file_ref: return None if file_ref.startswith("data:"): return self._decode_data_url_bytes(file_ref) if file_ref.startswith("tg://file_id/"): file_id = file_ref.replace("tg://file_id/", "", 1) return cast(Optional[bytes], self.run_module( "download_telegram_file_bytes", file_id=file_id, source=source )) if file_ref.startswith("tg://document_file_id/"): file_id = file_ref.replace("tg://document_file_id/", "", 1) return cast(Optional[bytes], self.run_module( "download_telegram_file_bytes", file_id=file_id, source=source )) if file_ref.startswith("wxwork://media_id/"): return cast(Optional[bytes], self.run_module( "download_wechat_media_bytes", media_ref=file_ref, source=source )) if file_ref.startswith("wxwork://file_media_id/"): return cast(Optional[bytes], self.run_module( "download_wechat_media_bytes", media_ref=file_ref, source=source )) if file_ref.startswith("wxbot://image/"): data_url = self.run_module( "download_wechat_image_to_data_url", image_ref=file_ref, source=source ) return self._decode_data_url_bytes(data_url) if data_url else None if file_ref.startswith("wxclaw://image/"): data_url = self.run_module( "download_wechat_image_to_data_url", image_ref=file_ref, source=source ) return self._decode_data_url_bytes(data_url) if data_url else None if file_ref.startswith("wxbot://file/"): file_url = unquote(file_ref.replace("wxbot://file/", "", 1)) content, _ = _read_message_http(file_url) return content if file_ref.startswith("wxclaw://file/") or file_ref.startswith("wxclaw://voice/"): return cast(Optional[bytes], self.run_module( "download_wechat_media_bytes", media_ref=file_ref, source=source )) if file_ref.startswith("feishu://file/"): return cast(Optional[bytes], self.run_module( "download_feishu_file_bytes", file_ref=file_ref, source=source )) if file_ref.startswith("slack://file/"): return cast(Optional[bytes], self.run_module( "download_slack_file_bytes", file_ref=file_ref, source=source )) if file_ref.startswith("discord://file/"): return cast(Optional[bytes], self.run_module( "download_discord_file_bytes", file_ref=file_ref, source=source )) if file_ref.startswith("qq://file/"): return cast(Optional[bytes], self.run_module( "download_qq_file_bytes", file_ref=file_ref, source=source )) if file_ref.startswith("vocechat://file/"): return cast(Optional[bytes], self.run_module( "download_vocechat_file_bytes", file_ref=file_ref, source=source )) if file_ref.startswith("synology://file/"): return cast(Optional[bytes], self.run_module( "download_synologychat_file_bytes", file_ref=file_ref, source=source )) if file_ref.startswith("http"): if channel == NotificationChannel.Slack: data_url = self.run_module( "download_slack_file_to_data_url", file_url=file_ref, source=source ) return self._decode_data_url_bytes(data_url) if data_url else None content, _ = _read_message_http(file_ref) return content logger.debug( "暂不支持的附件引用: channel=%s, source=%s, ref=%s", channel.value if channel else None, source, file_ref, ) return None def _save_agent_attachment( self, session_id: str, filename: Optional[str], content: bytes, mime_type: Optional[str] = None, ) -> Path: """ 将用户上传文件写入临时目录，并返回本地路径。 """ safe_name = self._sanitize_attachment_name(filename, mime_type) base_dir = self.runtime_config.temporary_path / "agent_uploads" / session_id base_dir.mkdir(parents=True, exist_ok=True) file_id = uuid.uuid4().hex[:8] local_path = base_dir / f"{file_id}_{safe_name}" local_path.write_bytes(content or b"") return local_path @staticmethod def _sanitize_attachment_name( filename: Optional[str], mime_type: Optional[str] = None ) -> str: """ 规范化附件文件名，避免路径穿越和非法字符。 """ name = Path(filename or "attachment").name name = re.sub(r"[^\w.\-]+", "_", name, flags=re.ASCII).strip("._") if not name: name = "attachment" if "." not in name: mime = (mime_type or "").split(";", 1)[0].strip().lower() default_ext = { "image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif", "image/webp": ".webp", "image/bmp": ".bmp", "application/json": ".json", "text/plain": ".txt", "text/markdown": ".md", "text/csv": ".csv", }.get(mime) if default_ext: name = f"{name}{default_ext}" return name @staticmethod def _guess_image_attachment_name(image_ref: str, index: int) -> str: """ 根据图片引用推测附件名。 """ if not image_ref: return f"image_{index}.jpg" if image_ref.startswith("data:"): mime_part = image_ref[5:].split(";", 1)[0].strip().lower() ext = mimetypes.guess_extension(mime_part) or ".jpg" return f"image_{index}{ext}" parsed = urlparse(unquote(image_ref)) name = Path(parsed.path).name if parsed.path else "" if name and "." in name: return name return f"image_{index}.jpg" @staticmethod def _guess_image_mime_type(image_ref: str, filename: Optional[str]) -> str: """ 根据图片引用或文件名推测 MIME 类型。 """ if image_ref and image_ref.startswith("data:"): mime = image_ref[5:].split(";", 1)[0].strip().lower() return mime or "image/jpeg" guessed, _ = mimetypes.guess_type(filename or "") if guessed and guessed.startswith("image/"): return guessed return "image/jpeg" @staticmethod def _decode_data_url_bytes(data_url: Optional[str]) -> Optional[bytes]: """ 将 data URL 解码为原始字节。 """ if not data_url or not data_url.startswith("data:"): return None try: _, payload = data_url.split(",", 1) except ValueError: return None try: return base64.b64decode(payload) except Exception as e: logger.error(e) return None