from __future__ import annotations import logging from collections.abc import Mapping, Sequence from typing import Any, Literal, assert_never import httpx from gateway.core.channels.feishu.types import ( FeishuReplyContext, IncomingFeishuEvent, mapping_to_feishu_event, ) logger = logging.getLogger(__name__) # 飞书开放平台 reaction emoji_type，与 openclaw-lark FeishuEmoji.TYPING 一致 TYPING_REACTION_EMOJI = "Typing" class WebhookParseError(ValueError): pass class FeishuHttpAdapterClient: """调用本仓库 Feishu 独立 HTTP 适配服务（默认端口 4380）。""" def __init__(self, base_url: str, *, timeout: float = 120.0) -> None: self._base = base_url.rstrip("/") self._timeout = timeout async def post_json(self, path: str, body: Mapping[str, Any]) -> dict[str, Any]: url = f"{self._base}{path if path.startswith('/') else '/' + path}" async with httpx.AsyncClient(timeout=self._timeout) as client: resp = await client.post(url, json=dict(body)) try: data = resp.json() except Exception: data = {"ok": False, "error": "invalid_json", "status_code": resp.status_code} if not isinstance(data, dict): return {"ok": False, "error": "unexpected_response_shape"} if resp.status_code >= 400 and "ok" not in data: data = {**data, "ok": False, "status_code": resp.status_code} return data async def get_json(self, path: str) -> dict[str, Any]: url = f"{self._base}{path if path.startswith('/') else '/' + path}" async with httpx.AsyncClient(timeout=self._timeout) as client: resp = await client.get(url) try: data = resp.json() except Exception: return {"ok": False, "error": "invalid_json", "status_code": resp.status_code} return data if isinstance(data, dict) else {"ok": False, "error": "unexpected_response_shape"} FeishuHttpMessageKind = Literal["text", "card", "media", "raw"] class FeishuConnector: """ 飞书连接器：解析 Feishu HTTP 适配层规范化入站事件，经适配服务发送消息。 入站 JSON 校验/解析见 `parse_feishu_inbound_event`（不依赖 HTTP 客户端，可按类调用）。 内部使用 ``FeishuHttpAdapterClient`` 访问 Node 适配器；不持有 app_secret，鉴权与 token 均在 Node 侧完成。 发送能力与适配器 POST ``/feishu/send-message`` 一致： ``text``（富文本 post）、``card``（交互卡片）、``media``（URL 上传后按类型发送）、 ``raw``（显式 ``msg_type`` + JSON ``content``，覆盖 text/post/image/分享名片/贴纸等开放平台类型）。 """ @staticmethod def parse_feishu_inbound_event(body: Mapping[str, Any]) -> IncomingFeishuEvent: """ 校验并解析 Feishu HTTP 适配层 ``forwardEventToGateway`` 写入的 JSON。 支持 event_type: message | reaction | card_action（与适配服务 server.ts 一致）。 """ if not body: raise WebhookParseError("empty body") event_type = body.get("event_type") if not isinstance(event_type, str) or not event_type: raise WebhookParseError("missing or invalid event_type") app_id = body.get("app_id") if not isinstance(app_id, str) or not app_id: raise WebhookParseError("missing or invalid app_id") event = mapping_to_feishu_event(body) if event_type == "message": if not event.chat_id: raise WebhookParseError("message event requires chat_id") if event.message_id is None or event.message_id == "": raise WebhookParseError("message event requires message_id") elif event_type == "reaction": if not event.chat_id: raise WebhookParseError("reaction event requires chat_id") if not event.message_id: raise WebhookParseError("reaction event requires message_id") elif event_type == "card_action": pass else: pass return event def __init__( self, *, feishu_http_base_url: str = "http://127.0.0.1:4380", timeout: float = 120.0, ) -> None: self._http = FeishuHttpAdapterClient(feishu_http_base_url, timeout=timeout) def handle_webhook(self, event: Mapping[str, Any]) -> dict[str, Any]: """校验 payload（对应 channels.md 的 handle_webhook）。""" try: ev: IncomingFeishuEvent = self.parse_feishu_inbound_event(event) return {"ok": True, "parsed": True, "event_type": ev.event_type} except WebhookParseError as e: return {"ok": False, "error": str(e)} def parse_normalized_event(self, payload: Mapping[str, Any]) -> IncomingFeishuEvent: return self.parse_feishu_inbound_event(payload) async def send_message( self, user_id: str, text: str, *, reply_context: FeishuReplyContext | None = None, ) -> dict[str, Any]: """ channels.md 签名兼容：user_id 未使用时需传入 reply_context。 实际发送依赖 chat_id（与 server.ts /feishu/send-message 一致）。 """ if reply_context is None: return {"ok": False, "error": "reply_context_required_for_feishu_adapter"} return await self.send_text(reply_context, text) def _send_message_base( self, ctx: FeishuReplyContext, *, reply_in_thread: bool | None = None, ) -> dict[str, Any]: body: dict[str, Any] = {"chat_id": ctx.chat_id} if ctx.account_id: body["account_id"] = ctx.account_id if ctx.message_id: body["reply_to_message_id"] = ctx.message_id if reply_in_thread is not None: body["reply_in_thread"] = reply_in_thread return body async def post_send_message(self, body: Mapping[str, Any]) -> dict[str, Any]: """POST ``/feishu/send-message``；``body`` 需含 ``kind`` 及对应字段（与适配器一致）。""" payload = dict(body) result = await self._http.post_json("/feishu/send-message", payload) if not result.get("ok"): logger.warning("feishu send-message failed: %s", result) return result async def send_text( self, ctx: FeishuReplyContext, text: str, *, reply_in_thread: bool | None = None, ) -> dict[str, Any]: body = self._send_message_base(ctx, reply_in_thread=reply_in_thread) body["kind"] = "text" body["text"] = text return await self.post_send_message(body) async def send_card( self, ctx: FeishuReplyContext, card: Mapping[str, Any], *, reply_in_thread: bool | None = None, ) -> dict[str, Any]: """发送交互卡片（``msg_type: interactive``）。""" body = self._send_message_base(ctx, reply_in_thread=reply_in_thread) body["kind"] = "card" body["card"] = dict(card) return await self.post_send_message(body) async def send_media( self, ctx: FeishuReplyContext, media_url: str, *, reply_in_thread: bool | None = None, media_local_roots: Sequence[str] | None = None, file_name: str | None = None, ) -> dict[str, Any]: """ 发送图片 / 视频 / 普通文件 / 音频（opus/mp4 等由适配器根据扩展名与内容判定）。 ``media_url`` 可为 http(s) 或适配器允许的本地路径；本地路径需配合服务端 ``media_local_roots`` 白名单。 """ body = self._send_message_base(ctx, reply_in_thread=reply_in_thread) body["kind"] = "media" body["media_url"] = media_url if media_local_roots is not None: body["media_local_roots"] = list(media_local_roots) if file_name: body["file_name"] = file_name return await self.post_send_message(body) async def send_raw_im( self, ctx: FeishuReplyContext, msg_type: str, content: str | Mapping[str, Any], *, reply_in_thread: bool | None = None, uuid: str | None = None, ) -> dict[str, Any]: """ 使用机器人身份发送任意已支持的 ``msg_type``，``content`` 为合法 JSON 字符串或对象。 常用示例：``text`` → ``{"text": "你好"}``；``share_chat`` → ``{"chat_id": "oc_xxx"}``； ``image`` → ``{"image_key": "img_xxx"}``（需先上传）。具体格式以飞书 IM 开放文档为准。 """ body = self._send_message_base(ctx, reply_in_thread=reply_in_thread) body["kind"] = "raw" body["msg_type"] = msg_type body["content"] = content if isinstance(content, str) else dict(content) if uuid: body["uuid"] = uuid return await self.post_send_message(body) async def send( self, ctx: FeishuReplyContext, *, kind: FeishuHttpMessageKind = "text", reply_in_thread: bool | None = None, text: str | None = None, card: Mapping[str, Any] | None = None, media_url: str | None = None, media_local_roots: Sequence[str] | None = None, file_name: str | None = None, msg_type: str | None = None, content: str | Mapping[str, Any] | None = None, uuid: str | None = None, ) -> dict[str, Any]: """与适配器 ``kind`` 对齐的统一发送入口。""" if kind == "text": if not text: return {"ok": False, "error": "missing_text"} return await self.send_text(ctx, text, reply_in_thread=reply_in_thread) if kind == "card": if card is None: return {"ok": False, "error": "missing_card"} return await self.send_card(ctx, card, reply_in_thread=reply_in_thread) if kind == "media": if not media_url: return {"ok": False, "error": "missing_media_url"} return await self.send_media( ctx, media_url, reply_in_thread=reply_in_thread, media_local_roots=media_local_roots, file_name=file_name, ) if kind == "raw": if not msg_type or content is None: return {"ok": False, "error": "raw_requires_msg_type_and_content"} return await self.send_raw_im( ctx, msg_type, content, reply_in_thread=reply_in_thread, uuid=uuid, ) assert_never(kind) async def list_feishu_app_accounts(self) -> dict[str, Any]: """ 拉取适配器 GET `/accounts`：已启用、已配置的飞书应用（机器人）列表。 与终端用户 open_id 无关；若要用户姓名/头像等需走开放平台 contact 等接口。 """ try: data = await self._http.get_json("/accounts") if data.get("ok") and "accounts" in data: return {"ok": True, "accounts": data.get("accounts")} except Exception as e: logger.debug("list_feishu_app_accounts: %s", e) return {"ok": True, "accounts": []} async def get_user_info(self, user_id: str) -> dict[str, Any]: """ 飞书「用户」个人资料：当前 HTTP 适配器未暴露联系人查询，仅回传占位。 已配置应用列表请用 `list_feishu_app_accounts()`（勿与 `/accounts` 语义混淆）。 """ return {"ok": True, "user_id": user_id, "profile": None} async def add_message_reaction( self, message_id: str, emoji: str, *, account_id: str | None = None, ) -> dict[str, Any]: """POST ``/feishu/react``，为消息添加机器人表情（``emoji`` 为开放平台 emoji_type，如 ``Typing``）。""" body: dict[str, Any] = { "action": "add", "message_id": message_id, "emoji": emoji, } if account_id: body["account_id"] = account_id return await self._http.post_json("/feishu/react", body) async def remove_bot_reactions_for_emoji( self, message_id: str, emoji: str, *, account_id: str | None = None, ) -> dict[str, Any]: """移除该消息上本机器人添加的指定 ``emoji`` 表情（适配器 ``action: remove``）。""" body: dict[str, Any] = { "action": "remove", "message_id": message_id, "emoji": emoji, } if account_id: body["account_id"] = account_id return await self._http.post_json("/feishu/react", body)