from __future__ import annotations import asyncio import contextlib import hashlib import hmac import json import threading import time from collections import OrderedDict, deque from pathlib import Path from typing import Any, Optional from aiohttp import web from openspace.communication.adapters.base import BaseChannelAdapter from openspace.communication.attachment_cache import AttachmentCache from openspace.communication.config import FeishuConfig from openspace.communication.policy import is_authorized from openspace.communication.types import ( AttachmentKind, ChannelMessage, ChannelPlatform, ChannelSource, SendResult, ) from openspace.utils.logging import Logger logger = Logger.get_logger(__name__) _FEISHU_WEBHOOK_MAX_BODY_BYTES = 1 * 1024 * 1024 _FEISHU_WEBHOOK_READ_TIMEOUT_SECONDS = 30 _FEISHU_WEBHOOK_RATE_WINDOW_SECONDS = 60 _FEISHU_WEBHOOK_RATE_LIMIT_MAX = 120 _FEISHU_WEBHOOK_RATE_MAX_KEYS = 4096 _FEISHU_WEBHOOK_ANOMALY_TTL_SECONDS = 6 * 60 * 60 _FEISHU_DEDUP_CACHE_SIZE = 2048 _FEISHU_DEDUP_TTL_SECONDS = 24 * 60 * 60 try: import lark_oapi as lark from lark_oapi.api.im.v1 import ( CreateMessageRequest, CreateMessageRequestBody, GetMessageRequest, GetMessageResourceRequest, ReplyMessageRequest, ReplyMessageRequestBody, ) from lark_oapi.core.const import FEISHU_DOMAIN, LARK_DOMAIN FEISHU_AVAILABLE = True except ImportError: FEISHU_AVAILABLE = False lark = None # type: ignore[assignment] CreateMessageRequest = None # type: ignore[assignment] CreateMessageRequestBody = None # type: ignore[assignment] GetMessageRequest = None # type: ignore[assignment] GetMessageResourceRequest = None # type: ignore[assignment] ReplyMessageRequest = None # type: ignore[assignment] ReplyMessageRequestBody = None # type: ignore[assignment] FEISHU_DOMAIN = None # type: ignore[assignment] LARK_DOMAIN = None # type: ignore[assignment] class FeishuAdapter(BaseChannelAdapter): MAX_MESSAGE_LENGTH = 8000 _REPLY_CONTEXT_MAX_LEN = 200 def __init__( self, config: FeishuConfig, attachment_cache: AttachmentCache, *, runtime_dir: Optional[Path] = None, ): super().__init__(ChannelPlatform.FEISHU) self.config = config self.attachment_cache = attachment_cache self.runtime_dir = ( Path(runtime_dir).expanduser().resolve() if runtime_dir is not None else attachment_cache.base_dir.parent.resolve() ) self._client: Any = None self._bot_open_id = config.bot_open_id self._loop: Optional[asyncio.AbstractEventLoop] = None self._ws_client: Any = None self._ws_thread: Optional[threading.Thread] = None self._running = False self._dedup_state_path = self.runtime_dir / "feishu_seen_message_ids.json" self._seen_message_ids: OrderedDict[str, float] = OrderedDict() self._recent_sent_message_ids: OrderedDict[str, None] = OrderedDict() self._rate_windows: dict[str, deque[float]] = {} self._webhook_anomalies: dict[str, tuple[int, str, float]] = {} self._dedup_dirty = False self._load_seen_message_ids() def register_http_routes(self, app: Any) -> None: if self.config.connection_mode == "webhook": app.router.add_post(self.config.webhook_path, self._handle_webhook) def validate_configuration(self) -> None: if self.config.connection_mode == "webhook" and not _optional_str(self.config.verification_token): raise ValueError("Feishu webhook mode requires verification_token") def get_lock_identity(self) -> Optional[tuple[str, str]]: app_id = _optional_str(self.config.app_id) if not app_id: return None return ("feishu-app", app_id) async def connect(self) -> bool: self.validate_configuration() if not FEISHU_AVAILABLE: logger.error("Feishu adapter requires lark-oapi") return False if not self.config.app_id or not self.config.app_secret: logger.error("Feishu adapter missing app_id/app_secret") return False domain = FEISHU_DOMAIN if self.config.domain != "lark" else LARK_DOMAIN self._client = ( lark.Client.builder() .app_id(self.config.app_id) .app_secret(self.config.app_secret) .domain(domain) .log_level(lark.LogLevel.WARNING) .build() ) self._running = True self._loop = asyncio.get_running_loop() if not self._bot_open_id: self._bot_open_id = await asyncio.to_thread(self._fetch_bot_open_id) if self.config.connection_mode == "websocket": self._start_websocket_client() self._connected = False else: self._connected = True logger.info( "Feishu adapter connected via %s mode", self.config.connection_mode, ) return True async def disconnect(self) -> None: self._running = False self._connected = False if self._ws_thread is not None and self._ws_thread.is_alive(): await asyncio.to_thread(self._ws_thread.join, 5) self._ws_thread = None self._persist_seen_message_ids() self._client = None async def send_text( self, chat_id: str, content: str, *, reply_to_message_id: Optional[str] = None, metadata: Optional[dict[str, Any]] = None, ) -> SendResult: if not self._client: return SendResult(success=False, error="Feishu client not initialized") last_message_id: Optional[str] = None for chunk in _split_text(content, self.MAX_MESSAGE_LENGTH): payload = json.dumps({"text": chunk}, ensure_ascii=False) if reply_to_message_id: body = ( ReplyMessageRequestBody.builder() .msg_type("text") .content(payload) .build() ) request = ( ReplyMessageRequest.builder() .message_id(reply_to_message_id) .request_body(body) .build() ) response = await asyncio.to_thread(self._client.im.v1.message.reply, request) else: body = ( CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type("text") .content(payload) .build() ) request = ( CreateMessageRequest.builder() .receive_id_type("chat_id") .request_body(body) .build() ) response = await asyncio.to_thread(self._client.im.v1.message.create, request) if not response.success(): return SendResult( success=False, error=f"[{response.code}] {response.msg}", raw_response=response, ) last_message_id = getattr(getattr(response, "data", None), "message_id", None) if last_message_id: self._remember_sent_message_id(last_message_id) return SendResult(success=True, message_id=last_message_id) async def _handle_webhook(self, request: web.Request) -> web.Response: remote_ip = _client_ip_from_request(request) rate_key = f"{self.config.app_id}:{self.config.webhook_path}:{remote_ip}" if not self._check_webhook_rate_limit(rate_key): self._record_webhook_anomaly(remote_ip, "429") return web.Response(status=429, text="Rate limit exceeded") content_length = request.content_length or 0 if content_length > _FEISHU_WEBHOOK_MAX_BODY_BYTES: self._record_webhook_anomaly(remote_ip, "413") return web.Response(status=413, text="Payload too large") try: async with asyncio.timeout(_FEISHU_WEBHOOK_READ_TIMEOUT_SECONDS): body_bytes = await request.read() except TimeoutError: self._record_webhook_anomaly(remote_ip, "408") return web.Response(status=408, text="Request timeout") if len(body_bytes) > _FEISHU_WEBHOOK_MAX_BODY_BYTES: self._record_webhook_anomaly(remote_ip, "413") return web.Response(status=413, text="Payload too large") try: payload = json.loads(body_bytes.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): self._record_webhook_anomaly(remote_ip, "400") return web.json_response({"code": 400, "msg": "invalid json"}, status=400) incoming_token = str((payload.get("header") or {}).get("token") or payload.get("token") or "") if not incoming_token or not hmac.compare_digest(incoming_token, self.config.verification_token or ""): self._record_webhook_anomaly(remote_ip, "401-token") return web.Response(status=401, text="Invalid verification token") if self.config.encrypt_key and not _is_webhook_signature_valid( encrypt_key=self.config.encrypt_key, headers=request.headers, body_bytes=body_bytes, ): self._record_webhook_anomaly(remote_ip, "401-signature") return web.Response(status=401, text="Invalid signature") if payload.get("encrypt"): self._record_webhook_anomaly(remote_ip, "400-encrypted") return web.json_response( {"code": 400, "msg": "encrypted webhook payloads are not supported"}, status=400, ) self._clear_webhook_anomaly(remote_ip) if payload.get("type") == "url_verification": return web.json_response({"challenge": payload.get("challenge", "")}) event_type = str((payload.get("header") or {}).get("event_type") or "") if event_type == "im.message.receive_v1": await self._handle_message_event(payload) return web.json_response({"code": 0, "msg": "ok"}) def _start_websocket_client(self) -> None: assert lark is not None handler = ( lark.EventDispatcherHandler.builder( self.config.encrypt_key or "", self.config.verification_token or "", ) .register_p2_im_message_receive_v1(self._on_message_sync) .build() ) self._ws_client = lark.ws.Client( self.config.app_id, self.config.app_secret, event_handler=handler, log_level=lark.LogLevel.INFO, ) def _run_ws_forever() -> None: import lark_oapi.ws.client as lark_ws_client ws_loop = asyncio.new_event_loop() asyncio.set_event_loop(ws_loop) lark_ws_client.loop = ws_loop try: while self._running: try: self._ws_client.start() except Exception as exc: self._connected = False logger.warning("Feishu WebSocket client error: %s", exc) if self._running: time.sleep(5) finally: self._connected = False ws_loop.close() self._ws_thread = threading.Thread( target=_run_ws_forever, daemon=True, name="openspace-feishu-ws", ) self._ws_thread.start() def _on_message_sync(self, data: Any) -> None: if not self._loop or not self._running: return self._connected = True asyncio.run_coroutine_threadsafe( self._handle_websocket_message(data), self._loop, ) async def _handle_message_event(self, payload: dict[str, Any]) -> None: normalized = await self._normalize_webhook_payload(payload) if normalized is not None: await self.dispatch_message(normalized) async def _handle_websocket_message(self, data: Any) -> None: normalized = await self._normalize_websocket_event(data) if normalized is not None: await self.dispatch_message(normalized) async def _normalize_webhook_payload(self, payload: dict[str, Any]) -> Optional[ChannelMessage]: event = payload.get("event") or {} message = event.get("message") or {} sender = event.get("sender") or {} sender_id = sender.get("sender_id") or {} if str(sender.get("sender_type", "")).lower() == "bot": return None return await self._normalize_inbound_message( message_id=_optional_str(message.get("message_id")), chat_id=_optional_str(message.get("chat_id")), chat_type=_optional_str(message.get("chat_type")) or "p2p", sender_uid=( _optional_str(sender_id.get("open_id")) or _optional_str(sender_id.get("user_id")) or _optional_str(sender_id.get("union_id")) ), sender_name=( _optional_str(sender_id.get("name")) or _optional_str(sender.get("sender_name")) ), thread_id=_optional_str(message.get("thread_id")), message_type=_optional_str(message.get("message_type")) or "", content=_safe_json_loads(message.get("content", "{}")), mentions=list(message.get("mentions") or []), reply_to_message_id=( _optional_str(message.get("parent_id")) or _optional_str(message.get("upper_message_id")) ), metadata={"webhook_payload": payload}, resolve_mentions=False, ) async def _normalize_websocket_event(self, data: Any) -> Optional[ChannelMessage]: event = getattr(data, "event", None) message = getattr(event, "message", None) sender = getattr(event, "sender", None) if message is None or sender is None: return None if str(getattr(sender, "sender_type", "")).lower() == "bot": return None sender_id = getattr(sender, "sender_id", None) return await self._normalize_inbound_message( message_id=_optional_str(getattr(message, "message_id", None)), chat_id=_optional_str(getattr(message, "chat_id", None)), chat_type=_optional_str(getattr(message, "chat_type", None)) or "p2p", sender_uid=( _optional_str(getattr(sender_id, "open_id", None)) or _optional_str(getattr(sender_id, "user_id", None)) or _optional_str(getattr(sender_id, "union_id", None)) ), sender_name=_optional_str(getattr(sender, "sender_name", None)), thread_id=_optional_str(getattr(message, "thread_id", None)), message_type=_optional_str(getattr(message, "message_type", None)) or "", content=_safe_json_loads(getattr(message, "content", "{}")), mentions=list(getattr(message, "mentions", None) or []), reply_to_message_id=( _optional_str(getattr(message, "parent_id", None)) or _optional_str(getattr(message, "upper_message_id", None)) ), metadata={"websocket_event": True}, resolve_mentions=True, ) async def _normalize_inbound_message( self, *, message_id: Optional[str], chat_id: Optional[str], chat_type: str, sender_uid: Optional[str], sender_name: Optional[str], thread_id: Optional[str], message_type: str, content: dict[str, Any], mentions: list[Any], reply_to_message_id: Optional[str], metadata: dict[str, Any], resolve_mentions: bool, ) -> Optional[ChannelMessage]: if not message_id or not chat_id: return None if self._is_message_seen(message_id): logger.debug("Skipping duplicate Feishu message %s", message_id) return None source = ChannelSource( platform=ChannelPlatform.FEISHU, chat_id=chat_id, chat_type="dm" if str(chat_type).lower() == "p2p" else "group", user_id=sender_uid, user_name=sender_name, chat_name=chat_id, thread_id=thread_id, ) session_key = _build_session_key_hint(source) normalized_type = str(message_type or "").strip().lower() mentions_bot = self._mentions_bot(mentions) text = "" if normalized_type == "text": text = str(content.get("text", "")).strip() if resolve_mentions: text = _resolve_mentions(text, mentions) elif normalized_type == "post": text = _extract_post_text(content) prefilter_message = ChannelMessage( source=source, text=text, message_id=message_id, reply_to_message_id=reply_to_message_id, mentions_bot=mentions_bot, metadata=metadata, ) if not self._passes_prefilter(prefilter_message): self._remember_message_seen(message_id) return None attachments = [] if normalized_type == "image": attachment = await self._download_attachment( session_key=session_key, message_id=message_id, file_key=str(content.get("image_key", "")).strip(), file_name=str(content.get("image_key", "image")).strip() + ".png", kind=AttachmentKind.IMAGE, resource_type="image", ) if attachment is not None: attachments.append(attachment) elif normalized_type == "file": attachment = await self._download_attachment( session_key=session_key, message_id=message_id, file_key=str(content.get("file_key", "")).strip(), file_name=str(content.get("file_name", "document")).strip(), kind=AttachmentKind.DOCUMENT, resource_type="file", ) if attachment is not None: attachments.append(attachment) prefilter_message.attachments = attachments prefilter_message.reply_to_text = await self._fetch_message_text(reply_to_message_id) self._remember_message_seen(message_id) return prefilter_message def _passes_prefilter(self, message: ChannelMessage) -> bool: if not is_authorized(message, self.config): logger.info("Rejected Feishu message from unauthorized user %s", message.source.user_id) return False if message.source.chat_type == "dm": return self.config.allow_dm if not self.config.allow_groups: return False if self.config.group_policy == "disabled": return False if self.config.group_policy == "mention_only": return message.mentions_bot if self.config.group_policy == "reply_or_mention": return message.mentions_bot or self._is_reply_to_recent_bot_message( message.reply_to_message_id ) return True def _is_reply_to_recent_bot_message(self, message_id: Optional[str]) -> bool: if not message_id: return False return message_id in self._recent_sent_message_ids def _remember_sent_message_id(self, message_id: str) -> None: self._recent_sent_message_ids.pop(message_id, None) self._recent_sent_message_ids[message_id] = None while len(self._recent_sent_message_ids) > _FEISHU_DEDUP_CACHE_SIZE: self._recent_sent_message_ids.popitem(last=False) async def _download_attachment( self, *, session_key: str, message_id: str, file_key: str, file_name: str, kind: AttachmentKind, resource_type: str, ): if not self._client or not file_key: return None request = ( GetMessageResourceRequest.builder() .message_id(message_id) .file_key(file_key) .type(resource_type) .build() ) response = await asyncio.to_thread(self._client.im.v1.message_resource.get, request) if not response.success(): logger.warning( "Failed to download Feishu attachment: code=%s msg=%s", response.code, response.msg, ) return None try: file_data = await asyncio.to_thread( _read_attachment_body, response.file, self.attachment_cache.max_attachment_bytes, ) except ValueError as exc: logger.warning( "Rejected Feishu attachment for session %s: %s", session_key, exc, ) return None return self.attachment_cache.save_bytes( session_key=session_key, data=file_data, filename=file_name, kind=kind, ) def _fetch_bot_open_id(self) -> Optional[str]: if not self._client or not lark: return None try: request = ( lark.BaseRequest.builder() .http_method(lark.HttpMethod.GET) .uri("/open-apis/bot/v3/info") .token_types({lark.AccessTokenType.APP}) .build() ) response = self._client.request(request) if not response.success(): logger.warning( "Failed to fetch Feishu bot info: code=%s msg=%s", response.code, response.msg, ) return None payload = json.loads(response.raw.content) bot = (payload.get("data") or payload).get("bot") or {} return _optional_str(bot.get("open_id")) except Exception as exc: logger.warning("Failed to resolve Feishu bot open_id: %s", exc) return None async def _fetch_message_text(self, message_id: Optional[str]) -> Optional[str]: if not self._client or not message_id or GetMessageRequest is None: return None request = GetMessageRequest.builder().message_id(message_id).build() try: response = await asyncio.to_thread(self._client.im.v1.message.get, request) except Exception as exc: logger.debug("Failed to fetch Feishu parent message %s: %s", message_id, exc) return None if not response.success(): return None data = getattr(response, "data", None) message_obj = None items = getattr(data, "items", None) if items: message_obj = items[0] elif data is not None: message_obj = getattr(data, "message", None) or data if message_obj is None: return None body = getattr(message_obj, "body", None) raw_content = getattr(body, "content", None) if body is not None else getattr(message_obj, "content", None) message_type = ( getattr(message_obj, "msg_type", None) or getattr(message_obj, "message_type", None) or "" ) content = _safe_json_loads(raw_content) text = "" if str(message_type).lower() == "text": text = str(content.get("text", "")).strip() elif str(message_type).lower() == "post": text = _extract_post_text(content) if not text: return None if len(text) > self._REPLY_CONTEXT_MAX_LEN: text = text[: self._REPLY_CONTEXT_MAX_LEN] + "..." return text def _mentions_bot(self, mentions: list[Any]) -> bool: if not mentions: return False if not self._bot_open_id: return False for mention in mentions: mention_id = (mention.get("id") or {}) if isinstance(mention, dict) else getattr(mention, "id", None) open_id = ( _optional_str(mention_id.get("open_id")) if isinstance(mention_id, dict) else _optional_str(getattr(mention_id, "open_id", None)) ) if open_id == self._bot_open_id: return True return False def _load_seen_message_ids(self) -> None: if not self._dedup_state_path.exists(): return try: payload = json.loads(self._dedup_state_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): logger.warning("Failed to load Feishu dedup cache from %s", self._dedup_state_path) return entries = payload.get("message_ids", {}) if isinstance(payload, dict) else {} now = time.time() valid: list[tuple[str, float]] = [] if isinstance(entries, dict): for message_id, seen_at in entries.items(): normalized_id = _optional_str(message_id) if not normalized_id: continue try: timestamp = float(seen_at) except (TypeError, ValueError): continue if now - timestamp <= _FEISHU_DEDUP_TTL_SECONDS: valid.append((normalized_id, timestamp)) for message_id, seen_at in sorted(valid, key=lambda item: item[1])[-_FEISHU_DEDUP_CACHE_SIZE:]: self._seen_message_ids[message_id] = seen_at def _persist_seen_message_ids(self) -> None: if not self._dedup_dirty: return self._dedup_state_path.parent.mkdir(parents=True, exist_ok=True) payload = {"message_ids": dict(self._seen_message_ids)} try: self._dedup_state_path.write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) except OSError: logger.warning("Failed to persist Feishu dedup cache to %s", self._dedup_state_path) return self._dedup_dirty = False def _is_message_seen(self, message_id: str) -> bool: now = time.time() self._prune_seen_message_ids(now) return message_id in self._seen_message_ids def _remember_message_seen(self, message_id: str) -> None: now = time.time() self._prune_seen_message_ids(now) if message_id in self._seen_message_ids: self._seen_message_ids.move_to_end(message_id) return self._seen_message_ids[message_id] = now self._seen_message_ids.move_to_end(message_id) while len(self._seen_message_ids) > _FEISHU_DEDUP_CACHE_SIZE: self._seen_message_ids.popitem(last=False) self._dedup_dirty = True self._persist_seen_message_ids() def _mark_message_seen(self, message_id: str) -> bool: if self._is_message_seen(message_id): return True self._remember_message_seen(message_id) return False def _prune_seen_message_ids(self, now: Optional[float] = None) -> None: current = now or time.time() stale = [ message_id for message_id, seen_at in self._seen_message_ids.items() if current - seen_at > _FEISHU_DEDUP_TTL_SECONDS ] for message_id in stale: self._seen_message_ids.pop(message_id, None) self._dedup_dirty = True def _check_webhook_rate_limit(self, rate_key: str) -> bool: now = time.time() window = self._rate_windows.get(rate_key) if window is None: if len(self._rate_windows) >= _FEISHU_WEBHOOK_RATE_MAX_KEYS: stale_keys = [ key for key, timestamps in self._rate_windows.items() if not timestamps or now - timestamps[-1] > _FEISHU_WEBHOOK_RATE_WINDOW_SECONDS ] for key in stale_keys: self._rate_windows.pop(key, None) if rate_key not in self._rate_windows and len(self._rate_windows) >= _FEISHU_WEBHOOK_RATE_MAX_KEYS: return False window = deque() self._rate_windows[rate_key] = window cutoff = now - _FEISHU_WEBHOOK_RATE_WINDOW_SECONDS while window and window[0] < cutoff: window.popleft() if len(window) >= _FEISHU_WEBHOOK_RATE_LIMIT_MAX: return False window.append(now) return True def _record_webhook_anomaly(self, remote_ip: str, status: str) -> None: now = time.time() current = self._webhook_anomalies.get(remote_ip) if current and now - current[2] < _FEISHU_WEBHOOK_ANOMALY_TTL_SECONDS: self._webhook_anomalies[remote_ip] = (current[0] + 1, status, current[2]) return self._webhook_anomalies[remote_ip] = (1, status, now) def _clear_webhook_anomaly(self, remote_ip: str) -> None: self._webhook_anomalies.pop(remote_ip, None) def _safe_json_loads(value: Any) -> dict[str, Any]: if isinstance(value, dict): return value try: return json.loads(str(value or "{}")) except json.JSONDecodeError: return {} def _optional_str(value: Any) -> Optional[str]: if value is None: return None value = str(value).strip() return value or None def _client_ip_from_request(request: web.Request) -> str: forwarded = str(request.headers.get("x-forwarded-for", "") or "").split(",")[0].strip() if forwarded: return forwarded peer = request.transport.get_extra_info("peername") if request.transport else None if isinstance(peer, tuple) and peer: return str(peer[0]) return request.remote or "unknown" def _resolve_mentions(text: str, mentions: list[Any]) -> str: if not text or not mentions: return text resolved = text for mention in mentions: key = _optional_str( mention.get("key") if isinstance(mention, dict) else getattr(mention, "key", None) ) if not key or key not in resolved: continue name = _optional_str( mention.get("name") if isinstance(mention, dict) else getattr(mention, "name", None) ) or "user" resolved = resolved.replace(key, f"@{name}") return resolved def _split_text(content: str, limit: int) -> list[str]: text = content.strip() if not text: return [""] if len(text) <= limit: return [text] chunks = [] remaining = text while remaining: chunk = remaining[:limit] if len(remaining) > limit: split_at = chunk.rfind("\n") if split_at < limit // 3: split_at = chunk.rfind(" ") if split_at >= limit // 3: chunk = chunk[:split_at] chunks.append(chunk.strip()) remaining = remaining[len(chunk):].lstrip() return [chunk for chunk in chunks if chunk] def _is_webhook_signature_valid(*, encrypt_key: str, headers: Any, body_bytes: bytes) -> bool: timestamp = str(headers.get("x-lark-request-timestamp", "") or "") nonce = str(headers.get("x-lark-request-nonce", "") or "") signature = str(headers.get("x-lark-signature", "") or "") if not timestamp or not nonce or not signature: return False content = f"{timestamp}{nonce}{encrypt_key}{body_bytes.decode('utf-8', errors='replace')}" expected = hashlib.sha256(content.encode("utf-8")).hexdigest() return hmac.compare_digest(signature, expected) def _build_session_key_hint(source: ChannelSource) -> str: parts = [source.platform.value, source.chat_id] if source.thread_id: parts.append(source.thread_id) return "__".join(part.replace("/", "_") for part in parts if part) def _extract_post_text(content: dict[str, Any]) -> str: texts: list[str] = [] def _walk(value: Any) -> None: if isinstance(value, dict): title = _optional_str(value.get("title")) if title: texts.append(title) tag = _optional_str(value.get("tag")) if tag == "at": user_name = _optional_str(value.get("user_name")) or "user" texts.append(f"@{user_name}") else: text = _optional_str(value.get("text")) if text: texts.append(text) for nested in value.values(): _walk(nested) elif isinstance(value, list): for item in value: _walk(item) _walk(content) deduped: list[str] = [] for text in texts: if text not in deduped: deduped.append(text) return "\n".join(deduped).strip() def _read_attachment_body(raw_file: Any, max_bytes: int) -> bytes: if raw_file is None: return b"" if isinstance(raw_file, bytes): data = raw_file elif isinstance(raw_file, bytearray): data = bytes(raw_file) elif hasattr(raw_file, "read"): chunks: list[bytes] = [] total = 0 try: while True: chunk = raw_file.read(65536) if not chunk: break if isinstance(chunk, str): chunk = chunk.encode("utf-8") elif isinstance(chunk, bytearray): chunk = bytes(chunk) elif not isinstance(chunk, bytes): chunk = bytes(chunk) total += len(chunk) if total > max_bytes: raise ValueError( f"attachment size {total} exceeds limit {max_bytes}" ) chunks.append(chunk) finally: close = getattr(raw_file, "close", None) if callable(close): close() data = b"".join(chunks) else: data = bytes(raw_file) if len(data) > max_bytes: raise ValueError(f"attachment size {len(data)} exceeds limit {max_bytes}") return data