"""Feishu/Lark channel implementation using lark-oapi SDK with WebSocket long connection.""" import asyncio import json import os import re import threading from collections import OrderedDict from pathlib import Path from typing import Any from loguru import logger from nanobot.bus.events import OutboundMessage from nanobot.bus.queue import MessageBus from nanobot.channels.base import BaseChannel from nanobot.config.paths import get_media_dir from nanobot.config.schema import FeishuConfig import importlib.util FEISHU_AVAILABLE = importlib.util.find_spec("lark_oapi") is not None # Message type display mapping MSG_TYPE_MAP = { "image": "[image]", "audio": "[audio]", "file": "[file]", "sticker": "[sticker]", } def _extract_share_card_content(content_json: dict, msg_type: str) -> str: """Extract text representation from share cards and interactive messages.""" parts = [] if msg_type == "share_chat": parts.append(f"[shared chat: {content_json.get('chat_id', '')}]") elif msg_type == "share_user": parts.append(f"[shared user: {content_json.get('user_id', '')}]") elif msg_type == "interactive": parts.extend(_extract_interactive_content(content_json)) elif msg_type == "share_calendar_event": parts.append(f"[shared calendar event: {content_json.get('event_key', '')}]") elif msg_type == "system": parts.append("[system message]") elif msg_type == "merge_forward": parts.append("[merged forward messages]") return "\n".join(parts) if parts else f"[{msg_type}]" def _extract_interactive_content(content: dict) -> list[str]: """Recursively extract text and links from interactive card content.""" parts = [] if isinstance(content, str): try: content = json.loads(content) except (json.JSONDecodeError, TypeError): return [content] if content.strip() else [] if not isinstance(content, dict): return parts if "title" in content: title = content["title"] if isinstance(title, dict): title_content = title.get("content", "") or title.get("text", "") if title_content: parts.append(f"title: {title_content}") elif isinstance(title, str): parts.append(f"title: {title}") for elements in content.get("elements", []) if isinstance(content.get("elements"), list) else []: for element in elements: parts.extend(_extract_element_content(element)) card = content.get("card", {}) if card: parts.extend(_extract_interactive_content(card)) header = content.get("header", {}) if header: header_title = header.get("title", {}) if isinstance(header_title, dict): header_text = header_title.get("content", "") or header_title.get("text", "") if header_text: parts.append(f"title: {header_text}") return parts def _extract_element_content(element: dict) -> list[str]: """Extract content from a single card element.""" parts = [] if not isinstance(element, dict): return parts tag = element.get("tag", "") if tag in ("markdown", "lark_md"): content = element.get("content", "") if content: parts.append(content) elif tag == "div": text = element.get("text", {}) if isinstance(text, dict): text_content = text.get("content", "") or text.get("text", "") if text_content: parts.append(text_content) elif isinstance(text, str): parts.append(text) for field in element.get("fields", []): if isinstance(field, dict): field_text = field.get("text", {}) if isinstance(field_text, dict): c = field_text.get("content", "") if c: parts.append(c) elif tag == "a": href = element.get("href", "") text = element.get("text", "") if href: parts.append(f"link: {href}") if text: parts.append(text) elif tag == "button": text = element.get("text", {}) if isinstance(text, dict): c = text.get("content", "") if c: parts.append(c) url = element.get("url", "") or element.get("multi_url", {}).get("url", "") if url: parts.append(f"link: {url}") elif tag == "img": alt = element.get("alt", {}) parts.append(alt.get("content", "[image]") if isinstance(alt, dict) else "[image]") elif tag == "note": for ne in element.get("elements", []): parts.extend(_extract_element_content(ne)) elif tag == "column_set": for col in element.get("columns", []): for ce in col.get("elements", []): parts.extend(_extract_element_content(ce)) elif tag == "plain_text": content = element.get("content", "") if content: parts.append(content) else: for ne in element.get("elements", []): parts.extend(_extract_element_content(ne)) return parts def _extract_post_content(content_json: dict) -> tuple[str, list[str]]: """Extract text and image keys from Feishu post (rich text) message. Handles three payload shapes: - Direct: {"title": "...", "content": [[...]]} - Localized: {"zh_cn": {"title": "...", "content": [...]}} - Wrapped: {"post": {"zh_cn": {"title": "...", "content": [...]}}} """ def _parse_block(block: dict) -> tuple[str | None, list[str]]: if not isinstance(block, dict) or not isinstance(block.get("content"), list): return None, [] texts, images = [], [] if title := block.get("title"): texts.append(title) for row in block["content"]: if not isinstance(row, list): continue for el in row: if not isinstance(el, dict): continue tag = el.get("tag") if tag in ("text", "a"): texts.append(el.get("text", "")) elif tag == "at": texts.append(f"@{el.get('user_name', 'user')}") elif tag == "img" and (key := el.get("image_key")): images.append(key) return (" ".join(texts).strip() or None), images # Unwrap optional {"post": ...} envelope root = content_json if isinstance(root, dict) and isinstance(root.get("post"), dict): root = root["post"] if not isinstance(root, dict): return "", [] # Direct format if "content" in root: text, imgs = _parse_block(root) if text or imgs: return text or "", imgs # Localized: prefer known locales, then fall back to any dict child for key in ("zh_cn", "en_us", "ja_jp"): if key in root: text, imgs = _parse_block(root[key]) if text or imgs: return text or "", imgs for val in root.values(): if isinstance(val, dict): text, imgs = _parse_block(val) if text or imgs: return text or "", imgs return "", [] def _extract_post_text(content_json: dict) -> str: """Extract plain text from Feishu post (rich text) message content. Legacy wrapper for _extract_post_content, returns only text. """ text, _ = _extract_post_content(content_json) return text class FeishuChannel(BaseChannel): """ Feishu/Lark channel using WebSocket long connection. Uses WebSocket to receive events - no public IP or webhook required. Requires: - App ID and App Secret from Feishu Open Platform - Bot capability enabled - Event subscription enabled (im.message.receive_v1) """ name = "feishu" def __init__(self, config: FeishuConfig, bus: MessageBus, groq_api_key: str = ""): super().__init__(config, bus) self.config: FeishuConfig = config self.groq_api_key = groq_api_key self._client: Any = None self._ws_client: Any = None self._ws_thread: threading.Thread | None = None self._processed_message_ids: OrderedDict[str, None] = OrderedDict() # Ordered dedup cache self._loop: asyncio.AbstractEventLoop | None = None @staticmethod def _register_optional_event(builder: Any, method_name: str, handler: Any) -> Any: """Register an event handler only when the SDK supports it.""" method = getattr(builder, method_name, None) return method(handler) if callable(method) else builder async def start(self) -> None: """Start the Feishu bot with WebSocket long connection.""" if not FEISHU_AVAILABLE: logger.error("Feishu SDK not installed. Run: pip install lark-oapi") return if not self.config.app_id or not self.config.app_secret: logger.error("Feishu app_id and app_secret not configured") return import lark_oapi as lark self._running = True self._loop = asyncio.get_running_loop() # Create Lark client for sending messages self._client = lark.Client.builder() \ .app_id(self.config.app_id) \ .app_secret(self.config.app_secret) \ .log_level(lark.LogLevel.INFO) \ .build() builder = lark.EventDispatcherHandler.builder( self.config.encrypt_key or "", self.config.verification_token or "", ).register_p2_im_message_receive_v1( self._on_message_sync ) builder = self._register_optional_event( builder, "register_p2_im_message_reaction_created_v1", self._on_reaction_created ) builder = self._register_optional_event( builder, "register_p2_im_message_message_read_v1", self._on_message_read ) builder = self._register_optional_event( builder, "register_p2_im_chat_access_event_bot_p2p_chat_entered_v1", self._on_bot_p2p_chat_entered, ) event_handler = builder.build() # Create WebSocket client for long connection self._ws_client = lark.ws.Client( self.config.app_id, self.config.app_secret, event_handler=event_handler, log_level=lark.LogLevel.INFO ) # Start WebSocket client in a separate thread with reconnect loop. # A dedicated event loop is created for this thread so that lark_oapi's # module-level `loop = asyncio.get_event_loop()` picks up an idle loop # instead of the already-running main asyncio loop, which would cause # "This event loop is already running" errors. def run_ws(): import time import lark_oapi.ws.client as _lark_ws_client ws_loop = asyncio.new_event_loop() asyncio.set_event_loop(ws_loop) # Patch the module-level loop used by lark's ws Client.start() _lark_ws_client.loop = ws_loop try: while self._running: try: self._ws_client.start() except Exception as e: logger.warning("Feishu WebSocket error: {}", e) if self._running: time.sleep(5) finally: ws_loop.close() self._ws_thread = threading.Thread(target=run_ws, daemon=True) self._ws_thread.start() logger.info("Feishu bot started with WebSocket long connection") logger.info("No public IP required - using WebSocket to receive events") # Keep running until stopped while self._running: await asyncio.sleep(1) async def stop(self) -> None: """ Stop the Feishu bot. Notice: lark.ws.Client does not expose stop method， simply exiting the program will close the client. Reference: https://github.com/larksuite/oapi-sdk-python/blob/v2_main/lark_oapi/ws/client.py#L86 """ self._running = False logger.info("Feishu bot stopped") def _add_reaction_sync(self, message_id: str, emoji_type: str) -> None: """Sync helper for adding reaction (runs in thread pool).""" from lark_oapi.api.im.v1 import CreateMessageReactionRequest, CreateMessageReactionRequestBody, Emoji try: request = CreateMessageReactionRequest.builder() \ .message_id(message_id) \ .request_body( CreateMessageReactionRequestBody.builder() .reaction_type(Emoji.builder().emoji_type(emoji_type).build()) .build() ).build() response = self._client.im.v1.message_reaction.create(request) if not response.success(): logger.warning("Failed to add reaction: code={}, msg={}", response.code, response.msg) else: logger.debug("Added {} reaction to message {}", emoji_type, message_id) except Exception as e: logger.warning("Error adding reaction: {}", e) async def _add_reaction(self, message_id: str, emoji_type: str = "THUMBSUP") -> None: """ Add a reaction emoji to a message (non-blocking). Common emoji types: THUMBSUP, OK, EYES, DONE, OnIt, HEART """ if not self._client: return loop = asyncio.get_running_loop() await loop.run_in_executor(None, self._add_reaction_sync, message_id, emoji_type) # Regex to match markdown tables (header + separator + data rows) _TABLE_RE = re.compile( r"((?:^[ \t]*\|.+\|[ \t]*\n)(?:^[ \t]*\|[-:\s|]+\|[ \t]*\n)(?:^[ \t]*\|.+\|[ \t]*\n?)+)", re.MULTILINE, ) _HEADING_RE = re.compile(r"^(#{1,6})\s+(.+)$", re.MULTILINE) _CODE_BLOCK_RE = re.compile(r"(```[\s\S]*?```)", re.MULTILINE) @staticmethod def _parse_md_table(table_text: str) -> dict | None: """Parse a markdown table into a Feishu table element.""" lines = [_line.strip() for _line in table_text.strip().split("\n") if _line.strip()] if len(lines) < 3: return None def split(_line: str) -> list[str]: return [c.strip() for c in _line.strip("|").split("|")] headers = split(lines[0]) rows = [split(_line) for _line in lines[2:]] columns = [{"tag": "column", "name": f"c{i}", "display_name": h, "width": "auto"} for i, h in enumerate(headers)] return { "tag": "table", "page_size": len(rows) + 1, "columns": columns, "rows": [{f"c{i}": r[i] if i < len(r) else "" for i in range(len(headers))} for r in rows], } def _build_card_elements(self, content: str) -> list[dict]: """Split content into div/markdown + table elements for Feishu card.""" elements, last_end = [], 0 for m in self._TABLE_RE.finditer(content): before = content[last_end:m.start()] if before.strip(): elements.extend(self._split_headings(before)) elements.append(self._parse_md_table(m.group(1)) or {"tag": "markdown", "content": m.group(1)}) last_end = m.end() remaining = content[last_end:] if remaining.strip(): elements.extend(self._split_headings(remaining)) return elements or [{"tag": "markdown", "content": content}] @staticmethod def _split_elements_by_table_limit(elements: list[dict], max_tables: int = 1) -> list[list[dict]]: """Split card elements into groups with at most *max_tables* table elements each. Feishu cards have a hard limit of one table per card (API error 11310). When the rendered content contains multiple markdown tables each table is placed in a separate card message so every table reaches the user. """ if not elements: return [[]] groups: list[list[dict]] = [] current: list[dict] = [] table_count = 0 for el in elements: if el.get("tag") == "table": if table_count >= max_tables: if current: groups.append(current) current = [] table_count = 0 current.append(el) table_count += 1 else: current.append(el) if current: groups.append(current) return groups or [[]] def _split_headings(self, content: str) -> list[dict]: """Split content by headings, converting headings to div elements.""" protected = content code_blocks = [] for m in self._CODE_BLOCK_RE.finditer(content): code_blocks.append(m.group(1)) protected = protected.replace(m.group(1), f"\x00CODE{len(code_blocks)-1}\x00", 1) elements = [] last_end = 0 for m in self._HEADING_RE.finditer(protected): before = protected[last_end:m.start()].strip() if before: elements.append({"tag": "markdown", "content": before}) text = m.group(2).strip() elements.append({ "tag": "div", "text": { "tag": "lark_md", "content": f"**{text}**", }, }) last_end = m.end() remaining = protected[last_end:].strip() if remaining: elements.append({"tag": "markdown", "content": remaining}) for i, cb in enumerate(code_blocks): for el in elements: if el.get("tag") == "markdown": el["content"] = el["content"].replace(f"\x00CODE{i}\x00", cb) return elements or [{"tag": "markdown", "content": content}] # ── Smart format detection ────────────────────────────────────────── # Patterns that indicate "complex" markdown needing card rendering _COMPLEX_MD_RE = re.compile( r"```" # fenced code block r"|^\|.+\|.*\n\s*\|[-:\s|]+\|" # markdown table (header + separator) r"|^#{1,6}\s+" # headings , re.MULTILINE, ) # Simple markdown patterns (bold, italic, strikethrough) _SIMPLE_MD_RE = re.compile( r"\*\*.+?\*\*" # **bold** r"|__.+?__" # __bold__ r"|(? str: """Determine the optimal Feishu message format for *content*. Returns one of: - ``"text"`` – plain text, short and no markdown - ``"post"`` – rich text (links only, moderate length) - ``"interactive"`` – card with full markdown rendering """ stripped = content.strip() # Complex markdown (code blocks, tables, headings) → always card if cls._COMPLEX_MD_RE.search(stripped): return "interactive" # Long content → card (better readability with card layout) if len(stripped) > cls._POST_MAX_LEN: return "interactive" # Has bold/italic/strikethrough → card (post format can't render these) if cls._SIMPLE_MD_RE.search(stripped): return "interactive" # Has list items → card (post format can't render list bullets well) if cls._LIST_RE.search(stripped) or cls._OLIST_RE.search(stripped): return "interactive" # Has links → post format (supports tags) if cls._MD_LINK_RE.search(stripped): return "post" # Short plain text → text format if len(stripped) <= cls._TEXT_MAX_LEN: return "text" # Medium plain text without any formatting → post format return "post" @classmethod def _markdown_to_post(cls, content: str) -> str: """Convert markdown content to Feishu post message JSON. Handles links ``[text](url)`` as ``a`` tags; everything else as ``text`` tags. Each line becomes a paragraph (row) in the post body. """ lines = content.strip().split("\n") paragraphs: list[list[dict]] = [] for line in lines: elements: list[dict] = [] last_end = 0 for m in cls._MD_LINK_RE.finditer(line): # Text before this link before = line[last_end:m.start()] if before: elements.append({"tag": "text", "text": before}) elements.append({ "tag": "a", "text": m.group(1), "href": m.group(2), }) last_end = m.end() # Remaining text after last link remaining = line[last_end:] if remaining: elements.append({"tag": "text", "text": remaining}) # Empty line → empty paragraph for spacing if not elements: elements.append({"tag": "text", "text": ""}) paragraphs.append(elements) post_body = { "zh_cn": { "content": paragraphs, } } return json.dumps(post_body, ensure_ascii=False) _IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".ico", ".tiff", ".tif"} _AUDIO_EXTS = {".opus"} _VIDEO_EXTS = {".mp4", ".mov", ".avi"} _FILE_TYPE_MAP = { ".opus": "opus", ".mp4": "mp4", ".pdf": "pdf", ".doc": "doc", ".docx": "doc", ".xls": "xls", ".xlsx": "xls", ".ppt": "ppt", ".pptx": "ppt", } def _upload_image_sync(self, file_path: str) -> str | None: """Upload an image to Feishu and return the image_key.""" from lark_oapi.api.im.v1 import CreateImageRequest, CreateImageRequestBody try: with open(file_path, "rb") as f: request = CreateImageRequest.builder() \ .request_body( CreateImageRequestBody.builder() .image_type("message") .image(f) .build() ).build() response = self._client.im.v1.image.create(request) if response.success(): image_key = response.data.image_key logger.debug("Uploaded image {}: {}", os.path.basename(file_path), image_key) return image_key else: logger.error("Failed to upload image: code={}, msg={}", response.code, response.msg) return None except Exception as e: logger.error("Error uploading image {}: {}", file_path, e) return None def _upload_file_sync(self, file_path: str) -> str | None: """Upload a file to Feishu and return the file_key.""" from lark_oapi.api.im.v1 import CreateFileRequest, CreateFileRequestBody ext = os.path.splitext(file_path)[1].lower() file_type = self._FILE_TYPE_MAP.get(ext, "stream") file_name = os.path.basename(file_path) try: with open(file_path, "rb") as f: request = CreateFileRequest.builder() \ .request_body( CreateFileRequestBody.builder() .file_type(file_type) .file_name(file_name) .file(f) .build() ).build() response = self._client.im.v1.file.create(request) if response.success(): file_key = response.data.file_key logger.debug("Uploaded file {}: {}", file_name, file_key) return file_key else: logger.error("Failed to upload file: code={}, msg={}", response.code, response.msg) return None except Exception as e: logger.error("Error uploading file {}: {}", file_path, e) return None def _download_image_sync(self, message_id: str, image_key: str) -> tuple[bytes | None, str | None]: """Download an image from Feishu message by message_id and image_key.""" from lark_oapi.api.im.v1 import GetMessageResourceRequest try: request = GetMessageResourceRequest.builder() \ .message_id(message_id) \ .file_key(image_key) \ .type("image") \ .build() response = self._client.im.v1.message_resource.get(request) if response.success(): file_data = response.file # GetMessageResourceRequest returns BytesIO, need to read bytes if hasattr(file_data, 'read'): file_data = file_data.read() return file_data, response.file_name else: logger.error("Failed to download image: code={}, msg={}", response.code, response.msg) return None, None except Exception as e: logger.error("Error downloading image {}: {}", image_key, e) return None, None def _download_file_sync( self, message_id: str, file_key: str, resource_type: str = "file" ) -> tuple[bytes | None, str | None]: """Download a file/audio/media from a Feishu message by message_id and file_key.""" from lark_oapi.api.im.v1 import GetMessageResourceRequest # Feishu API only accepts 'image' or 'file' as type parameter # Convert 'audio' to 'file' for API compatibility if resource_type == "audio": resource_type = "file" try: request = ( GetMessageResourceRequest.builder() .message_id(message_id) .file_key(file_key) .type(resource_type) .build() ) response = self._client.im.v1.message_resource.get(request) if response.success(): file_data = response.file if hasattr(file_data, "read"): file_data = file_data.read() return file_data, response.file_name else: logger.error("Failed to download {}: code={}, msg={}", resource_type, response.code, response.msg) return None, None except Exception: logger.exception("Error downloading {} {}", resource_type, file_key) return None, None async def _download_and_save_media( self, msg_type: str, content_json: dict, message_id: str | None = None ) -> tuple[str | None, str]: """ Download media from Feishu and save to local disk. Returns: (file_path, content_text) - file_path is None if download failed """ loop = asyncio.get_running_loop() media_dir = get_media_dir("feishu") data, filename = None, None if msg_type == "image": image_key = content_json.get("image_key") if image_key and message_id: data, filename = await loop.run_in_executor( None, self._download_image_sync, message_id, image_key ) if not filename: filename = f"{image_key[:16]}.jpg" elif msg_type in ("audio", "file", "media"): file_key = content_json.get("file_key") if file_key and message_id: data, filename = await loop.run_in_executor( None, self._download_file_sync, message_id, file_key, msg_type ) if not filename: filename = file_key[:16] if msg_type == "audio" and not filename.endswith(".opus"): filename = f"{filename}.opus" if data and filename: file_path = media_dir / filename file_path.write_bytes(data) logger.debug("Downloaded {} to {}", msg_type, file_path) return str(file_path), f"[{msg_type}: {filename}]" return None, f"[{msg_type}: download failed]" def _send_message_sync(self, receive_id_type: str, receive_id: str, msg_type: str, content: str) -> bool: """Send a single message (text/image/file/interactive) synchronously.""" from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody try: request = CreateMessageRequest.builder() \ .receive_id_type(receive_id_type) \ .request_body( CreateMessageRequestBody.builder() .receive_id(receive_id) .msg_type(msg_type) .content(content) .build() ).build() response = self._client.im.v1.message.create(request) if not response.success(): logger.error( "Failed to send Feishu {} message: code={}, msg={}, log_id={}", msg_type, response.code, response.msg, response.get_log_id() ) return False logger.debug("Feishu {} message sent to {}", msg_type, receive_id) return True except Exception as e: logger.error("Error sending Feishu {} message: {}", msg_type, e) return False async def send(self, msg: OutboundMessage) -> None: """Send a message through Feishu, including media (images/files) if present.""" if not self._client: logger.warning("Feishu client not initialized") return try: receive_id_type = "chat_id" if msg.chat_id.startswith("oc_") else "open_id" loop = asyncio.get_running_loop() for file_path in msg.media: if not os.path.isfile(file_path): logger.warning("Media file not found: {}", file_path) continue ext = os.path.splitext(file_path)[1].lower() if ext in self._IMAGE_EXTS: key = await loop.run_in_executor(None, self._upload_image_sync, file_path) if key: await loop.run_in_executor( None, self._send_message_sync, receive_id_type, msg.chat_id, "image", json.dumps({"image_key": key}, ensure_ascii=False), ) else: key = await loop.run_in_executor(None, self._upload_file_sync, file_path) if key: # Use msg_type "media" for audio/video so users can play inline; # "file" for everything else (documents, archives, etc.) if ext in self._AUDIO_EXTS or ext in self._VIDEO_EXTS: media_type = "media" else: media_type = "file" await loop.run_in_executor( None, self._send_message_sync, receive_id_type, msg.chat_id, media_type, json.dumps({"file_key": key}, ensure_ascii=False), ) if msg.content and msg.content.strip(): fmt = self._detect_msg_format(msg.content) if fmt == "text": # Short plain text – send as simple text message text_body = json.dumps({"text": msg.content.strip()}, ensure_ascii=False) await loop.run_in_executor( None, self._send_message_sync, receive_id_type, msg.chat_id, "text", text_body, ) elif fmt == "post": # Medium content with links – send as rich-text post post_body = self._markdown_to_post(msg.content) await loop.run_in_executor( None, self._send_message_sync, receive_id_type, msg.chat_id, "post", post_body, ) else: # Complex / long content – send as interactive card elements = self._build_card_elements(msg.content) for chunk in self._split_elements_by_table_limit(elements): card = {"config": {"wide_screen_mode": True}, "elements": chunk} await loop.run_in_executor( None, self._send_message_sync, receive_id_type, msg.chat_id, "interactive", json.dumps(card, ensure_ascii=False), ) except Exception as e: logger.error("Error sending Feishu message: {}", e) def _on_message_sync(self, data: Any) -> None: """ Sync handler for incoming messages (called from WebSocket thread). Schedules async handling in the main event loop. """ if self._loop and self._loop.is_running(): asyncio.run_coroutine_threadsafe(self._on_message(data), self._loop) async def _on_message(self, data: Any) -> None: """Handle incoming message from Feishu.""" try: event = data.event message = event.message sender = event.sender # Deduplication check message_id = message.message_id if message_id in self._processed_message_ids: return self._processed_message_ids[message_id] = None # Trim cache while len(self._processed_message_ids) > 1000: self._processed_message_ids.popitem(last=False) # Skip bot messages if sender.sender_type == "bot": return sender_id = sender.sender_id.open_id if sender.sender_id else "unknown" chat_id = message.chat_id chat_type = message.chat_type msg_type = message.message_type # Add reaction await self._add_reaction(message_id, self.config.react_emoji) # Parse content content_parts = [] media_paths = [] try: content_json = json.loads(message.content) if message.content else {} except json.JSONDecodeError: content_json = {} if msg_type == "text": text = content_json.get("text", "") if text: content_parts.append(text) elif msg_type == "post": text, image_keys = _extract_post_content(content_json) if text: content_parts.append(text) # Download images embedded in post for img_key in image_keys: file_path, content_text = await self._download_and_save_media( "image", {"image_key": img_key}, message_id ) if file_path: media_paths.append(file_path) content_parts.append(content_text) elif msg_type in ("image", "audio", "file", "media"): file_path, content_text = await self._download_and_save_media(msg_type, content_json, message_id) if file_path: media_paths.append(file_path) # Transcribe audio using Groq Whisper if msg_type == "audio" and file_path and self.groq_api_key: try: from nanobot.providers.transcription import GroqTranscriptionProvider transcriber = GroqTranscriptionProvider(api_key=self.groq_api_key) transcription = await transcriber.transcribe(file_path) if transcription: content_text = f"[transcription: {transcription}]" except Exception as e: logger.warning("Failed to transcribe audio: {}", e) content_parts.append(content_text) elif msg_type in ("share_chat", "share_user", "interactive", "share_calendar_event", "system", "merge_forward"): # Handle share cards and interactive messages text = _extract_share_card_content(content_json, msg_type) if text: content_parts.append(text) else: content_parts.append(MSG_TYPE_MAP.get(msg_type, f"[{msg_type}]")) content = "\n".join(content_parts) if content_parts else "" if not content and not media_paths: return # Forward to message bus reply_to = chat_id if chat_type == "group" else sender_id await self._handle_message( sender_id=sender_id, chat_id=reply_to, content=content, media=media_paths, metadata={ "message_id": message_id, "chat_type": chat_type, "msg_type": msg_type, } ) except Exception as e: logger.error("Error processing Feishu message: {}", e) def _on_reaction_created(self, data: Any) -> None: """Ignore reaction events so they do not generate SDK noise.""" pass def _on_message_read(self, data: Any) -> None: """Ignore read events so they do not generate SDK noise.""" pass def _on_bot_p2p_chat_entered(self, data: Any) -> None: """Ignore p2p-enter events when a user opens a bot chat.""" logger.debug("Bot entered p2p chat (user opened chat window)") pass