"""飞书入站事件路由：Trace 会话准备、Executor 提交、绑定 Agent ``trace_id``。""" from __future__ import annotations import logging from collections.abc import Mapping from typing import Any from gateway.core.channels.feishu.connector import FeishuConnector from gateway.core.channels.feishu.protocols import ( FeishuExecutorBackend, FeishuUserIdentityResolver, ) from gateway.core.channels.feishu.types import ( FeishuReplyContext, IncomingFeishuEvent, feishu_event_to_mapping, ) from gateway.core.channels.protocols import TraceBackend from gateway.core.channels.router import ChannelTraceRouter from gateway.core.channels.types import CHANNEL_FEISHU, RouteResult logger = logging.getLogger(__name__) def routing_from_card_action_raw(raw: dict[str, Any]) -> tuple[str | None, str | None, str | None]: """ 当规范化 JSON 未带 chat_id 时，从飞书 card.action.trigger 原始体兜底解析。 常见路径：event.context.open_chat_id / open_message_id（或顶层 open_chat_id）。 """ if not raw: return None, None, None ev: Any = raw.get("event") if isinstance(raw.get("event"), dict) else raw if not isinstance(ev, dict): return None, None, None ctx = ev.get("context") if isinstance(ev.get("context"), dict) else {} def pick(*keys: str) -> str | None: for d in (ev, ctx): for k in keys: v = d.get(k) if isinstance(v, str) and v.strip(): return v.strip() return None chat_id = pick("open_chat_id", "chat_id") message_id = pick("open_message_id", "message_id") chat_type = pick("chat_type") return chat_id, message_id, chat_type def as_opt_str(v: Any) -> str | None: if v is None: return None s = str(v) return s if s else None class FeishuMessageRouter(ChannelTraceRouter): """ 飞书消息路由：prepare_session → Executor → bind_agent_trace_id。 ``reaction`` / ``card_action`` 是否触发续跑由 ``dispatch_*`` 控制； ``card_action`` 多用于 OAuth 等卡片交互后让 Agent 继续执行。 """ def __init__( self, *, connector: FeishuConnector, trace_backend: TraceBackend, executor_backend: FeishuExecutorBackend, identity_resolver: FeishuUserIdentityResolver, workspace_prefix: str = CHANNEL_FEISHU, default_agent_type: str = "personal_assistant", auto_create_trace: bool = True, dispatch_reactions: bool = False, dispatch_card_actions: bool = False, ) -> None: super().__init__( trace_backend=trace_backend, workspace_prefix=workspace_prefix, default_agent_type=default_agent_type, ) self._connector = connector self._executor = executor_backend self._identity = identity_resolver self._auto_create = auto_create_trace self._dispatch_reactions = dispatch_reactions self._dispatch_card_actions = dispatch_card_actions def _reply_context_from_event(self, event: IncomingFeishuEvent) -> FeishuReplyContext | None: chat_id = event.chat_id message_id = event.message_id if not chat_id and event.event_type == "card_action": c, m, _ = routing_from_card_action_raw(event.raw) chat_id = chat_id or c message_id = message_id or m if not chat_id: logger.warning("missing chat_id, cannot reply: %s", feishu_event_to_mapping(event)) return None return FeishuReplyContext( account_id=event.account_id, app_id=event.app_id, chat_id=chat_id, message_id=message_id, open_id=event.open_id, ) def _synthetic_text_for_event(self, event: IncomingFeishuEvent) -> str | None: if event.event_type == "reaction" and event.emoji: return f"[系统-表情] {event.emoji} message_id={event.message_id or ''}" if event.event_type == "card_action": return ( "[系统-卡片交互] 用户已在飞书内完成卡片操作（如授权确认），" "请结合当前上下文继续执行未完成任务，必要时重试刚才失败的工具。" f" action={event.action or ''} operation_id={event.operation_id or ''}" ) return None async def route_feishu_inbound_event(self, event: IncomingFeishuEvent) -> RouteResult: """处理 Feishu HTTP 适配服务转发的规范化入站事件。""" user_id = self._identity.resolve_user_id(event) workspace_id = self._workspace_id_for_user(user_id) dispatch = False text: str | None = None if event.event_type == "message": dispatch = True text = event.content or "" elif event.event_type == "reaction" and self._dispatch_reactions: dispatch = True text = self._synthetic_text_for_event(event) or "" elif event.event_type == "card_action" and self._dispatch_card_actions: dispatch = True text = self._synthetic_text_for_event(event) or "" if not dispatch: return RouteResult( ok=True, skipped=True, reason=f"event_type_not_dispatched:{event.event_type}", user_id=user_id, ) if not self._auto_create: return RouteResult(ok=False, error="auto_create_trace_disabled", user_id=user_id) meta = feishu_event_to_mapping(event) await self._trace.prepare_session( channel=CHANNEL_FEISHU, user_id=user_id, workspace_id=workspace_id, agent_type=self._agent_type, metadata=meta, ) existing_agent_trace_id = await self._trace.get_existing_trace_id(CHANNEL_FEISHU, user_id) ctx = self._reply_context_from_event(event) if ctx is None: return RouteResult( ok=False, error="missing_chat_id_for_reply", trace_id=existing_agent_trace_id or "", user_id=user_id, workspace_id=workspace_id, ) task_id, agent_trace_id = await self._executor.handle_inbound_message( existing_agent_trace_id or "", text or "", ctx, self._connector, event=event, ) if agent_trace_id: await self._trace.bind_agent_trace_id( channel=CHANNEL_FEISHU, user_id=user_id, workspace_id=workspace_id, agent_trace_id=agent_trace_id, agent_type=self._agent_type, metadata=meta, ) return RouteResult( ok=True, trace_id=agent_trace_id or existing_agent_trace_id or "", task_id=task_id, user_id=user_id, workspace_id=workspace_id, ) async def route_message(self, channel: str, user_id: str, message: Mapping[str, Any]) -> str: """ 通用入口：message 建议包含 text、可选飞书上下文字段 （account_id, app_id, chat_id, message_id, open_id）。 """ text = str(message.get("text") or message.get("content") or "") workspace_id = self._workspace_id_for_user(user_id) meta = dict(message) if isinstance(message, dict) else {} await self._trace.prepare_session( channel=channel, user_id=user_id, workspace_id=workspace_id, agent_type=self._agent_type, metadata=meta, ) existing = await self._trace.get_existing_trace_id(channel, user_id) ctx = FeishuReplyContext( account_id=as_opt_str(message.get("account_id")), app_id=str(message.get("app_id") or ""), chat_id=str(message.get("chat_id") or ""), message_id=as_opt_str(message.get("message_id")), open_id=as_opt_str(message.get("open_id")), ) if not ctx.app_id or not ctx.chat_id: raise ValueError("route_message requires app_id and chat_id in message for Feishu reply") synthetic = IncomingFeishuEvent( event_type="message", app_id=ctx.app_id, account_id=ctx.account_id, open_id=ctx.open_id, chat_type=as_opt_str(message.get("chat_type")), chat_id=ctx.chat_id, message_id=ctx.message_id, content=text, raw=dict(message) if isinstance(message, dict) else {}, ) task_id, agent_trace_id = await self._executor.handle_inbound_message( existing or "", text, ctx, self._connector, event=synthetic ) if agent_trace_id: await self._trace.bind_agent_trace_id( channel=channel, user_id=user_id, workspace_id=workspace_id, agent_trace_id=agent_trace_id, agent_type=self._agent_type, metadata=meta, ) return task_id async def send_agent_reply( self, trace_id: str, content: str, metadata: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """ Executor 完成后由业务调用：根据 metadata 中的飞书上下文发消息。 metadata 键：account_id?, app_id, chat_id, message_id?, open_id? """ meta = dict(metadata or {}) ctx = FeishuReplyContext( account_id=as_opt_str(meta.get("account_id")), app_id=str(meta.get("app_id") or ""), chat_id=str(meta.get("chat_id") or ""), message_id=as_opt_str(meta.get("message_id")), open_id=as_opt_str(meta.get("open_id")), ) if not ctx.chat_id: return {"ok": False, "error": "metadata missing chat_id", "trace_id": trace_id} _ = trace_id return await self._connector.send_text(ctx, content) __all__ = ["FeishuMessageRouter", "routing_from_card_action_raw", "as_opt_str"]