"""
Feishu/Lark platform adapter.
Supports:
- WebSocket long connection and Webhook transport
- Direct-message and group @mention-gated text receive/send
- Inbound image/file/audio/media caching
- Gateway allowlist integration via FEISHU_ALLOWED_USERS
- Persistent dedup state across restarts
- Per-chat serial message processing (matches openclaw createChatQueue)
- Persistent ACK emoji reaction on inbound messages
- Reaction events routed as synthetic text events (matches openclaw)
- Interactive card button-click events routed as synthetic COMMAND events
- Webhook anomaly tracking (matches openclaw createWebhookAnomalyTracker)
- Verification token validation as second auth layer (matches openclaw)
"""
from __future__ import annotations
import asyncio
import hashlib
import hmac
import json
import logging
import mimetypes
import os
import re
import threading
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Dict, List, Optional
# aiohttp/websockets are independent optional deps — import outside lark_oapi
# so they remain available for tests and webhook mode even if lark_oapi is missing.
try:
import aiohttp
from aiohttp import web
except ImportError:
aiohttp = None # type: ignore[assignment]
web = None # type: ignore[assignment]
try:
import websockets
except ImportError:
websockets = None # type: ignore[assignment]
try:
import lark_oapi as lark
from lark_oapi.api.application.v6 import GetApplicationRequest
from lark_oapi.api.im.v1 import (
CreateFileRequest,
CreateFileRequestBody,
CreateImageRequest,
CreateImageRequestBody,
CreateMessageRequest,
CreateMessageRequestBody,
GetChatRequest,
GetMessageRequest,
GetImageRequest,
GetMessageResourceRequest,
P2ImMessageMessageReadV1,
ReplyMessageRequest,
ReplyMessageRequestBody,
UpdateMessageRequest,
UpdateMessageRequestBody,
)
from lark_oapi.core.const import FEISHU_DOMAIN, LARK_DOMAIN
from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse
from lark_oapi.event.dispatcher_handler import EventDispatcherHandler
from lark_oapi.ws import Client as FeishuWSClient
FEISHU_AVAILABLE = True
except ImportError:
FEISHU_AVAILABLE = False
lark = None # type: ignore[assignment]
P2CardActionTriggerResponse = None # type: ignore[assignment]
EventDispatcherHandler = None # type: ignore[assignment]
FeishuWSClient = None # type: ignore[assignment]
FEISHU_DOMAIN = None # type: ignore[assignment]
LARK_DOMAIN = None # type: ignore[assignment]
FEISHU_WEBSOCKET_AVAILABLE = websockets is not None
FEISHU_WEBHOOK_AVAILABLE = aiohttp is not None
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
BasePlatformAdapter,
MessageEvent,
MessageType,
SendResult,
SUPPORTED_DOCUMENT_TYPES,
cache_document_from_bytes,
cache_image_from_url,
cache_audio_from_bytes,
cache_image_from_bytes,
)
from gateway.status import acquire_scoped_lock, release_scoped_lock
from hermes_constants import get_hermes_home
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Regex patterns
# ---------------------------------------------------------------------------
_MARKDOWN_HINT_RE = re.compile(
r"(^#{1,6}\s)|(^\s*[-*]\s)|(^\s*\d+\.\s)|(^\s*---+\s*$)|(```)|(`[^`\n]+`)|(\*\*[^*\n].+?\*\*)|(~~[^~\n].+?~~)|(.+?)|(\*[^*\n]+\*)|(\[[^\]]+\]\([^)]+\))|(^>\s)",
re.MULTILINE,
)
_MARKDOWN_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
_MENTION_RE = re.compile(r"@_user_\d+")
_MULTISPACE_RE = re.compile(r"[ \t]{2,}")
_POST_CONTENT_INVALID_RE = re.compile(r"content format of the post type is incorrect", re.IGNORECASE)
# ---------------------------------------------------------------------------
# Media type sets and upload constants
# ---------------------------------------------------------------------------
_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}
_AUDIO_EXTENSIONS = {".ogg", ".mp3", ".wav", ".m4a", ".aac", ".flac", ".opus", ".webm"}
_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".3gp"}
_DOCUMENT_MIME_TO_EXT = {mime: ext for ext, mime in SUPPORTED_DOCUMENT_TYPES.items()}
_FEISHU_IMAGE_UPLOAD_TYPE = "message"
_FEISHU_FILE_UPLOAD_TYPE = "stream"
_FEISHU_OPUS_UPLOAD_EXTENSIONS = {".ogg", ".opus"}
_FEISHU_MEDIA_UPLOAD_EXTENSIONS = {".mp4", ".mov", ".avi", ".m4v"}
_FEISHU_DOC_UPLOAD_TYPES = {
".pdf": "pdf",
".doc": "doc",
".docx": "doc",
".xls": "xls",
".xlsx": "xls",
".ppt": "ppt",
".pptx": "ppt",
}
# ---------------------------------------------------------------------------
# Connection, retry and batching tuning
# ---------------------------------------------------------------------------
_MAX_TEXT_INJECT_BYTES = 100 * 1024
_FEISHU_CONNECT_ATTEMPTS = 3
_FEISHU_SEND_ATTEMPTS = 3
_FEISHU_APP_LOCK_SCOPE = "feishu-app-id"
_DEFAULT_TEXT_BATCH_DELAY_SECONDS = 0.6
_DEFAULT_TEXT_BATCH_MAX_MESSAGES = 8
_DEFAULT_TEXT_BATCH_MAX_CHARS = 4000
_DEFAULT_MEDIA_BATCH_DELAY_SECONDS = 0.8
_DEFAULT_DEDUP_CACHE_SIZE = 2048
_DEFAULT_WEBHOOK_HOST = "127.0.0.1"
_DEFAULT_WEBHOOK_PORT = 8765
_DEFAULT_WEBHOOK_PATH = "/feishu/webhook"
# ---------------------------------------------------------------------------
# TTL, rate-limit and webhook security constants
# ---------------------------------------------------------------------------
_FEISHU_DEDUP_TTL_SECONDS = 24 * 60 * 60 # 24 hours — matches openclaw
_FEISHU_SENDER_NAME_TTL_SECONDS = 10 * 60 # 10 minutes sender-name cache
_FEISHU_WEBHOOK_MAX_BODY_BYTES = 1 * 1024 * 1024 # 1 MB body limit
_FEISHU_WEBHOOK_RATE_WINDOW_SECONDS = 60 # sliding window for rate limiter
_FEISHU_WEBHOOK_RATE_LIMIT_MAX = 120 # max requests per window per IP — matches openclaw
_FEISHU_WEBHOOK_RATE_MAX_KEYS = 4096 # max tracked keys (prevents unbounded growth)
_FEISHU_WEBHOOK_BODY_TIMEOUT_SECONDS = 30 # max seconds to read request body
_FEISHU_WEBHOOK_ANOMALY_THRESHOLD = 25 # consecutive error responses before WARNING log
_FEISHU_WEBHOOK_ANOMALY_TTL_SECONDS = 6 * 60 * 60 # anomaly tracker TTL (6 hours) — matches openclaw
_FEISHU_CARD_ACTION_DEDUP_TTL_SECONDS = 15 * 60 # card action token dedup window (15 min)
_FEISHU_BOT_MSG_TRACK_SIZE = 512 # LRU size for tracking sent message IDs
_FEISHU_REPLY_FALLBACK_CODES = frozenset({230011, 231003}) # reply target withdrawn/missing → create fallback
_FEISHU_ACK_EMOJI = "OK"
# ---------------------------------------------------------------------------
# Fallback display strings
# ---------------------------------------------------------------------------
FALLBACK_POST_TEXT = "[Rich text message]"
FALLBACK_FORWARD_TEXT = "[Merged forward message]"
FALLBACK_SHARE_CHAT_TEXT = "[Shared chat]"
FALLBACK_INTERACTIVE_TEXT = "[Interactive message]"
FALLBACK_IMAGE_TEXT = "[Image]"
FALLBACK_ATTACHMENT_TEXT = "[Attachment]"
# ---------------------------------------------------------------------------
# Post/card parsing helpers
# ---------------------------------------------------------------------------
_PREFERRED_LOCALES = ("zh_cn", "en_us")
_MARKDOWN_SPECIAL_CHARS_RE = re.compile(r"([\\`*_{}\[\]()#+\-!|>~])")
_MENTION_PLACEHOLDER_RE = re.compile(r"@_user_\d+")
_WHITESPACE_RE = re.compile(r"\s+")
_SUPPORTED_CARD_TEXT_KEYS = (
"title",
"text",
"content",
"label",
"value",
"name",
"summary",
"subtitle",
"description",
"placeholder",
"hint",
)
_SKIP_TEXT_KEYS = {
"tag",
"type",
"msg_type",
"message_type",
"chat_id",
"open_chat_id",
"share_chat_id",
"file_key",
"image_key",
"user_id",
"open_id",
"union_id",
"url",
"href",
"link",
"token",
"template",
"locale",
}
@dataclass(frozen=True)
class FeishuPostMediaRef:
file_key: str
file_name: str = ""
resource_type: str = "file"
@dataclass(frozen=True)
class FeishuPostParseResult:
text_content: str
image_keys: List[str] = field(default_factory=list)
media_refs: List[FeishuPostMediaRef] = field(default_factory=list)
mentioned_ids: List[str] = field(default_factory=list)
@dataclass(frozen=True)
class FeishuNormalizedMessage:
raw_type: str
text_content: str
preferred_message_type: str = "text"
image_keys: List[str] = field(default_factory=list)
media_refs: List[FeishuPostMediaRef] = field(default_factory=list)
mentioned_ids: List[str] = field(default_factory=list)
relation_kind: str = "plain"
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass(frozen=True)
class FeishuAdapterSettings:
app_id: str
app_secret: str
domain_name: str
connection_mode: str
encrypt_key: str
verification_token: str
group_policy: str
allowed_group_users: frozenset[str]
bot_open_id: str
bot_user_id: str
bot_name: str
dedup_cache_size: int
text_batch_delay_seconds: float
text_batch_max_messages: int
text_batch_max_chars: int
media_batch_delay_seconds: float
webhook_host: str
webhook_port: int
webhook_path: str
@dataclass
class FeishuBatchState:
events: Dict[str, MessageEvent] = field(default_factory=dict)
tasks: Dict[str, asyncio.Task] = field(default_factory=dict)
counts: Dict[str, int] = field(default_factory=dict)
# ---------------------------------------------------------------------------
# Markdown rendering helpers
# ---------------------------------------------------------------------------
def _escape_markdown_text(text: str) -> str:
return _MARKDOWN_SPECIAL_CHARS_RE.sub(r"\\\1", text)
def _to_boolean(value: Any) -> bool:
return value is True or value == 1 or value == "true"
def _is_style_enabled(style: Dict[str, Any] | None, key: str) -> bool:
if not style:
return False
return _to_boolean(style.get(key))
def _wrap_inline_code(text: str) -> str:
max_run = max([0, *[len(run) for run in re.findall(r"`+", text)]])
fence = "`" * (max_run + 1)
body = f" {text} " if text.startswith("`") or text.endswith("`") else text
return f"{fence}{body}{fence}"
def _sanitize_fence_language(language: str) -> str:
return language.strip().replace("\n", " ").replace("\r", " ")
def _render_text_element(element: Dict[str, Any]) -> str:
text = str(element.get("text", "") or "")
style = element.get("style")
style_dict = style if isinstance(style, dict) else None
if _is_style_enabled(style_dict, "code"):
return _wrap_inline_code(text)
rendered = _escape_markdown_text(text)
if not rendered:
return ""
if _is_style_enabled(style_dict, "bold"):
rendered = f"**{rendered}**"
if _is_style_enabled(style_dict, "italic"):
rendered = f"*{rendered}*"
if _is_style_enabled(style_dict, "underline"):
rendered = f"{rendered}"
if _is_style_enabled(style_dict, "strikethrough"):
rendered = f"~~{rendered}~~"
return rendered
def _render_code_block_element(element: Dict[str, Any]) -> str:
language = _sanitize_fence_language(
str(element.get("language", "") or "") or str(element.get("lang", "") or "")
)
code = (
str(element.get("text", "") or "") or str(element.get("content", "") or "")
).replace("\r\n", "\n")
trailing_newline = "" if code.endswith("\n") else "\n"
return f"```{language}\n{code}{trailing_newline}```"
def _strip_markdown_to_plain_text(text: str) -> str:
plain = text.replace("\r\n", "\n")
plain = _MARKDOWN_LINK_RE.sub(lambda m: f"{m.group(1)} ({m.group(2).strip()})", plain)
plain = re.sub(r"^#{1,6}\s+", "", plain, flags=re.MULTILINE)
plain = re.sub(r"^>\s?", "", plain, flags=re.MULTILINE)
plain = re.sub(r"^\s*---+\s*$", "---", plain, flags=re.MULTILINE)
plain = re.sub(r"```(?:[^\n]*\n)?([\s\S]*?)```", lambda m: m.group(1).strip("\n"), plain)
plain = re.sub(r"`([^`\n]+)`", r"\1", plain)
plain = re.sub(r"\*\*([^*\n]+)\*\*", r"\1", plain)
plain = re.sub(r"\*([^*\n]+)\*", r"\1", plain)
plain = re.sub(r"~~([^~\n]+)~~", r"\1", plain)
plain = re.sub(r"([\s\S]*?)", r"\1", plain)
plain = re.sub(r"\n{3,}", "\n\n", plain)
return plain.strip()
# ---------------------------------------------------------------------------
# Post payload builders and parsers
# ---------------------------------------------------------------------------
def _build_markdown_post_payload(content: str) -> str:
return json.dumps(
{
"zh_cn": {
"content": [
[
{
"tag": "md",
"text": content,
}
]
],
}
},
ensure_ascii=False,
)
def parse_feishu_post_content(raw_content: str) -> FeishuPostParseResult:
try:
parsed = json.loads(raw_content) if raw_content else {}
except json.JSONDecodeError:
return FeishuPostParseResult(text_content=FALLBACK_POST_TEXT)
return parse_feishu_post_payload(parsed)
def parse_feishu_post_payload(payload: Any) -> FeishuPostParseResult:
resolved = _resolve_post_payload(payload)
if not resolved:
return FeishuPostParseResult(text_content=FALLBACK_POST_TEXT)
image_keys: List[str] = []
media_refs: List[FeishuPostMediaRef] = []
mentioned_ids: List[str] = []
parts: List[str] = []
title = _normalize_feishu_text(str(resolved.get("title", "")).strip())
if title:
parts.append(title)
for row in resolved.get("content", []) or []:
if not isinstance(row, list):
continue
row_text = _normalize_feishu_text(
"".join(_render_post_element(item, image_keys, media_refs, mentioned_ids) for item in row)
)
if row_text:
parts.append(row_text)
return FeishuPostParseResult(
text_content="\n".join(parts).strip() or FALLBACK_POST_TEXT,
image_keys=image_keys,
media_refs=media_refs,
mentioned_ids=mentioned_ids,
)
def _resolve_post_payload(payload: Any) -> Dict[str, Any]:
direct = _to_post_payload(payload)
if direct:
return direct
if not isinstance(payload, dict):
return {}
wrapped = payload.get("post")
wrapped_direct = _resolve_locale_payload(wrapped)
if wrapped_direct:
return wrapped_direct
return _resolve_locale_payload(payload)
def _resolve_locale_payload(payload: Any) -> Dict[str, Any]:
direct = _to_post_payload(payload)
if direct:
return direct
if not isinstance(payload, dict):
return {}
for key in _PREFERRED_LOCALES:
candidate = _to_post_payload(payload.get(key))
if candidate:
return candidate
for value in payload.values():
candidate = _to_post_payload(value)
if candidate:
return candidate
return {}
def _to_post_payload(candidate: Any) -> Dict[str, Any]:
if not isinstance(candidate, dict):
return {}
content = candidate.get("content")
if not isinstance(content, list):
return {}
return {
"title": str(candidate.get("title", "") or ""),
"content": content,
}
def _render_post_element(
element: Any,
image_keys: List[str],
media_refs: List[FeishuPostMediaRef],
mentioned_ids: List[str],
) -> str:
if isinstance(element, str):
return element
if not isinstance(element, dict):
return ""
tag = str(element.get("tag", "")).strip().lower()
if tag == "text":
return _render_text_element(element)
if tag == "a":
href = str(element.get("href", "")).strip()
label = str(element.get("text", href) or "").strip()
if not label:
return ""
escaped_label = _escape_markdown_text(label)
return f"[{escaped_label}]({href})" if href else escaped_label
if tag == "at":
mentioned_id = (
str(element.get("open_id", "")).strip()
or str(element.get("user_id", "")).strip()
)
if mentioned_id and mentioned_id not in mentioned_ids:
mentioned_ids.append(mentioned_id)
display_name = (
str(element.get("user_name", "")).strip()
or str(element.get("name", "")).strip()
or str(element.get("text", "")).strip()
or mentioned_id
)
return f"@{_escape_markdown_text(display_name)}" if display_name else "@"
if tag in {"img", "image"}:
image_key = str(element.get("image_key", "")).strip()
if image_key and image_key not in image_keys:
image_keys.append(image_key)
alt = str(element.get("text", "")).strip() or str(element.get("alt", "")).strip()
return f"[Image: {alt}]" if alt else "[Image]"
if tag in {"media", "file", "audio", "video"}:
file_key = str(element.get("file_key", "")).strip()
file_name = (
str(element.get("file_name", "")).strip()
or str(element.get("title", "")).strip()
or str(element.get("text", "")).strip()
)
if file_key:
media_refs.append(
FeishuPostMediaRef(
file_key=file_key,
file_name=file_name,
resource_type=tag if tag in {"audio", "video"} else "file",
)
)
return f"[Attachment: {file_name}]" if file_name else "[Attachment]"
if tag in {"emotion", "emoji"}:
label = str(element.get("text", "")).strip() or str(element.get("emoji_type", "")).strip()
return f":{_escape_markdown_text(label)}:" if label else "[Emoji]"
if tag == "br":
return "\n"
if tag in {"hr", "divider"}:
return "\n\n---\n\n"
if tag == "code":
code = str(element.get("text", "") or "") or str(element.get("content", "") or "")
return _wrap_inline_code(code) if code else ""
if tag in {"code_block", "pre"}:
return _render_code_block_element(element)
nested_parts: List[str] = []
for key in ("text", "title", "content", "children", "elements"):
value = element.get(key)
extracted = _render_nested_post(value, image_keys, media_refs, mentioned_ids)
if extracted:
nested_parts.append(extracted)
return " ".join(part for part in nested_parts if part)
def _render_nested_post(
value: Any,
image_keys: List[str],
media_refs: List[FeishuPostMediaRef],
mentioned_ids: List[str],
) -> str:
if isinstance(value, str):
return _escape_markdown_text(value)
if isinstance(value, list):
return " ".join(
part
for item in value
for part in [_render_nested_post(item, image_keys, media_refs, mentioned_ids)]
if part
)
if isinstance(value, dict):
direct = _render_post_element(value, image_keys, media_refs, mentioned_ids)
if direct:
return direct
return " ".join(
part
for item in value.values()
for part in [_render_nested_post(item, image_keys, media_refs, mentioned_ids)]
if part
)
return ""
# ---------------------------------------------------------------------------
# Message normalization
# ---------------------------------------------------------------------------
def normalize_feishu_message(*, message_type: str, raw_content: str) -> FeishuNormalizedMessage:
normalized_type = str(message_type or "").strip().lower()
payload = _load_feishu_payload(raw_content)
if normalized_type == "text":
return FeishuNormalizedMessage(
raw_type=normalized_type,
text_content=_normalize_feishu_text(str(payload.get("text", "") or "")),
)
if normalized_type == "post":
parsed_post = parse_feishu_post_payload(payload)
return FeishuNormalizedMessage(
raw_type=normalized_type,
text_content=parsed_post.text_content,
image_keys=list(parsed_post.image_keys),
media_refs=list(parsed_post.media_refs),
mentioned_ids=list(parsed_post.mentioned_ids),
relation_kind="post",
)
if normalized_type == "image":
image_key = str(payload.get("image_key", "") or "").strip()
alt_text = _normalize_feishu_text(
str(payload.get("text", "") or "")
or str(payload.get("alt", "") or "")
or FALLBACK_IMAGE_TEXT
)
return FeishuNormalizedMessage(
raw_type=normalized_type,
text_content=alt_text if alt_text != FALLBACK_IMAGE_TEXT else "",
preferred_message_type="photo",
image_keys=[image_key] if image_key else [],
relation_kind="image",
)
if normalized_type in {"file", "audio", "media"}:
media_ref = _build_media_ref_from_payload(payload, resource_type=normalized_type)
placeholder = _attachment_placeholder(media_ref.file_name)
return FeishuNormalizedMessage(
raw_type=normalized_type,
text_content="",
preferred_message_type="audio" if normalized_type == "audio" else "document",
media_refs=[media_ref] if media_ref.file_key else [],
relation_kind=normalized_type,
metadata={"placeholder_text": placeholder},
)
if normalized_type == "merge_forward":
return _normalize_merge_forward_message(payload)
if normalized_type == "share_chat":
return _normalize_share_chat_message(payload)
if normalized_type in {"interactive", "card"}:
return _normalize_interactive_message(normalized_type, payload)
return FeishuNormalizedMessage(raw_type=normalized_type, text_content="")
def _load_feishu_payload(raw_content: str) -> Dict[str, Any]:
try:
parsed = json.loads(raw_content) if raw_content else {}
except json.JSONDecodeError:
return {"text": raw_content}
return parsed if isinstance(parsed, dict) else {"content": parsed}
def _normalize_merge_forward_message(payload: Dict[str, Any]) -> FeishuNormalizedMessage:
title = _first_non_empty_text(
payload.get("title"),
payload.get("summary"),
payload.get("preview"),
_find_first_text(payload, keys=("title", "summary", "preview", "description")),
)
entries = _collect_forward_entries(payload)
lines: List[str] = []
if title:
lines.append(title)
lines.extend(entries[:8])
text_content = "\n".join(lines).strip() or FALLBACK_FORWARD_TEXT
return FeishuNormalizedMessage(
raw_type="merge_forward",
text_content=text_content,
relation_kind="merge_forward",
metadata={"entry_count": len(entries), "title": title},
)
def _normalize_share_chat_message(payload: Dict[str, Any]) -> FeishuNormalizedMessage:
chat_name = _first_non_empty_text(
payload.get("chat_name"),
payload.get("name"),
payload.get("title"),
_find_first_text(payload, keys=("chat_name", "name", "title")),
)
share_id = _first_non_empty_text(
payload.get("chat_id"),
payload.get("open_chat_id"),
payload.get("share_chat_id"),
)
lines = []
if chat_name:
lines.append(f"Shared chat: {chat_name}")
else:
lines.append(FALLBACK_SHARE_CHAT_TEXT)
if share_id:
lines.append(f"Chat ID: {share_id}")
text_content = "\n".join(lines)
return FeishuNormalizedMessage(
raw_type="share_chat",
text_content=text_content,
relation_kind="share_chat",
metadata={"chat_id": share_id, "chat_name": chat_name},
)
def _normalize_interactive_message(message_type: str, payload: Dict[str, Any]) -> FeishuNormalizedMessage:
card_payload = payload.get("card") if isinstance(payload.get("card"), dict) else payload
title = _first_non_empty_text(
_find_header_title(card_payload),
payload.get("title"),
_find_first_text(card_payload, keys=("title", "summary", "subtitle")),
)
body_lines = _collect_card_lines(card_payload)
actions = _collect_action_labels(card_payload)
lines: List[str] = []
if title:
lines.append(title)
for line in body_lines:
if line != title:
lines.append(line)
if actions:
lines.append(f"Actions: {', '.join(actions)}")
text_content = "\n".join(lines[:12]).strip() or FALLBACK_INTERACTIVE_TEXT
return FeishuNormalizedMessage(
raw_type=message_type,
text_content=text_content,
relation_kind="interactive",
metadata={"title": title, "actions": actions},
)
# ---------------------------------------------------------------------------
# Content extraction utilities (card / forward / text walking)
# ---------------------------------------------------------------------------
def _collect_forward_entries(payload: Dict[str, Any]) -> List[str]:
candidates: List[Any] = []
for key in ("messages", "items", "message_list", "records", "content"):
value = payload.get(key)
if isinstance(value, list):
candidates.extend(value)
entries: List[str] = []
for item in candidates:
if not isinstance(item, dict):
text = _normalize_feishu_text(str(item or ""))
if text:
entries.append(f"- {text}")
continue
sender = _first_non_empty_text(
item.get("sender_name"),
item.get("user_name"),
item.get("sender"),
item.get("name"),
)
nested_type = str(item.get("message_type", "") or item.get("msg_type", "")).strip().lower()
if nested_type == "post":
body = parse_feishu_post_payload(item.get("content") or item).text_content
else:
body = _first_non_empty_text(
item.get("text"),
item.get("summary"),
item.get("preview"),
item.get("content"),
_find_first_text(item, keys=("text", "content", "summary", "preview", "title")),
)
body = _normalize_feishu_text(body)
if sender and body:
entries.append(f"- {sender}: {body}")
elif body:
entries.append(f"- {body}")
return _unique_lines(entries)
def _collect_card_lines(payload: Any) -> List[str]:
lines = _collect_text_segments(payload, in_rich_block=False)
normalized = [_normalize_feishu_text(line) for line in lines]
return _unique_lines([line for line in normalized if line])
def _collect_action_labels(payload: Any) -> List[str]:
labels: List[str] = []
for item in _walk_nodes(payload):
if not isinstance(item, dict):
continue
tag = str(item.get("tag", "") or item.get("type", "")).strip().lower()
if tag not in {"button", "select_static", "overflow", "date_picker", "picker"}:
continue
label = _first_non_empty_text(
item.get("text"),
item.get("name"),
item.get("value"),
_find_first_text(item, keys=("text", "content", "name", "value")),
)
if label:
labels.append(label)
return _unique_lines(labels)
def _collect_text_segments(value: Any, *, in_rich_block: bool) -> List[str]:
if isinstance(value, str):
return [_normalize_feishu_text(value)] if in_rich_block else []
if isinstance(value, list):
segments: List[str] = []
for item in value:
segments.extend(_collect_text_segments(item, in_rich_block=in_rich_block))
return segments
if not isinstance(value, dict):
return []
tag = str(value.get("tag", "") or value.get("type", "")).strip().lower()
next_in_rich_block = in_rich_block or tag in {
"plain_text",
"lark_md",
"markdown",
"note",
"div",
"column_set",
"column",
"action",
"button",
"select_static",
"date_picker",
}
segments: List[str] = []
for key in _SUPPORTED_CARD_TEXT_KEYS:
item = value.get(key)
if isinstance(item, str) and next_in_rich_block:
normalized = _normalize_feishu_text(item)
if normalized:
segments.append(normalized)
for key, item in value.items():
if key in _SKIP_TEXT_KEYS:
continue
segments.extend(_collect_text_segments(item, in_rich_block=next_in_rich_block))
return segments
def _build_media_ref_from_payload(payload: Dict[str, Any], *, resource_type: str) -> FeishuPostMediaRef:
file_key = str(payload.get("file_key", "") or "").strip()
file_name = _first_non_empty_text(
payload.get("file_name"),
payload.get("title"),
payload.get("text"),
)
effective_type = resource_type if resource_type in {"audio", "video"} else "file"
return FeishuPostMediaRef(file_key=file_key, file_name=file_name, resource_type=effective_type)
def _attachment_placeholder(file_name: str) -> str:
normalized_name = _normalize_feishu_text(file_name)
return f"[Attachment: {normalized_name}]" if normalized_name else FALLBACK_ATTACHMENT_TEXT
def _find_header_title(payload: Any) -> str:
if not isinstance(payload, dict):
return ""
header = payload.get("header")
if not isinstance(header, dict):
return ""
title = header.get("title")
if isinstance(title, dict):
return _first_non_empty_text(title.get("content"), title.get("text"), title.get("name"))
return _normalize_feishu_text(str(title or ""))
def _find_first_text(payload: Any, *, keys: tuple[str, ...]) -> str:
for node in _walk_nodes(payload):
if not isinstance(node, dict):
continue
for key in keys:
value = node.get(key)
if isinstance(value, str):
normalized = _normalize_feishu_text(value)
if normalized:
return normalized
return ""
def _walk_nodes(value: Any):
if isinstance(value, dict):
yield value
for item in value.values():
yield from _walk_nodes(item)
elif isinstance(value, list):
for item in value:
yield from _walk_nodes(item)
def _first_non_empty_text(*values: Any) -> str:
for value in values:
if isinstance(value, str):
normalized = _normalize_feishu_text(value)
if normalized:
return normalized
elif value is not None and not isinstance(value, (dict, list)):
normalized = _normalize_feishu_text(str(value))
if normalized:
return normalized
return ""
# ---------------------------------------------------------------------------
# General text utilities
# ---------------------------------------------------------------------------
def _normalize_feishu_text(text: str) -> str:
cleaned = _MENTION_PLACEHOLDER_RE.sub(" ", text or "")
cleaned = cleaned.replace("\r\n", "\n").replace("\r", "\n")
cleaned = "\n".join(_WHITESPACE_RE.sub(" ", line).strip() for line in cleaned.split("\n"))
cleaned = "\n".join(line for line in cleaned.split("\n") if line)
cleaned = _MULTISPACE_RE.sub(" ", cleaned)
return cleaned.strip()
def _unique_lines(lines: List[str]) -> List[str]:
seen: set[str] = set()
unique: List[str] = []
for line in lines:
if not line or line in seen:
continue
seen.add(line)
unique.append(line)
return unique
def _run_official_feishu_ws_client(ws_client: Any) -> None:
"""Run the official Lark WS client in its own thread-local event loop."""
import lark_oapi.ws.client as ws_client_module
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
ws_client_module.loop = loop
ws_client.start()
def check_feishu_requirements() -> bool:
"""Check if Feishu/Lark dependencies are available."""
return FEISHU_AVAILABLE
class FeishuAdapter(BasePlatformAdapter):
"""Feishu/Lark bot adapter."""
MAX_MESSAGE_LENGTH = 8000
# =========================================================================
# Lifecycle — init / settings / connect / disconnect
# =========================================================================
def __init__(self, config: PlatformConfig):
super().__init__(config, Platform.FEISHU)
self._settings = self._load_settings(config.extra or {})
self._apply_settings(self._settings)
self._client: Optional[Any] = None
self._ws_client: Optional[Any] = None
self._ws_future: Optional[asyncio.Future] = None
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._webhook_runner: Optional[Any] = None
self._webhook_site: Optional[Any] = None
self._event_handler = self._build_event_handler()
self._seen_message_ids: Dict[str, float] = {} # message_id → seen_at (time.time())
self._seen_message_order: List[str] = []
self._dedup_state_path = get_hermes_home() / "feishu_seen_message_ids.json"
self._dedup_lock = threading.Lock()
self._sender_name_cache: Dict[str, tuple[str, float]] = {} # sender_id → (name, expire_at)
self._webhook_rate_counts: Dict[str, tuple[int, float]] = {} # rate_key → (count, window_start)
self._webhook_anomaly_counts: Dict[str, tuple[int, str, float]] = {} # ip → (count, last_status, first_seen)
self._card_action_tokens: Dict[str, float] = {} # token → first_seen_time
self._chat_locks: Dict[str, asyncio.Lock] = {} # chat_id → lock (per-chat serial processing)
self._sent_message_ids_to_chat: Dict[str, str] = {} # message_id → chat_id (for reaction routing)
self._sent_message_id_order: List[str] = [] # LRU order for _sent_message_ids_to_chat
self._chat_info_cache: Dict[str, Dict[str, Any]] = {}
self._message_text_cache: Dict[str, Optional[str]] = {}
self._app_lock_identity: Optional[str] = None
self._text_batch_state = FeishuBatchState()
self._pending_text_batches = self._text_batch_state.events
self._pending_text_batch_tasks = self._text_batch_state.tasks
self._pending_text_batch_counts = self._text_batch_state.counts
self._media_batch_state = FeishuBatchState()
self._pending_media_batches = self._media_batch_state.events
self._pending_media_batch_tasks = self._media_batch_state.tasks
self._load_seen_message_ids()
@staticmethod
def _load_settings(extra: Dict[str, Any]) -> FeishuAdapterSettings:
return FeishuAdapterSettings(
app_id=str(extra.get("app_id") or os.getenv("FEISHU_APP_ID", "")).strip(),
app_secret=str(extra.get("app_secret") or os.getenv("FEISHU_APP_SECRET", "")).strip(),
domain_name=str(extra.get("domain") or os.getenv("FEISHU_DOMAIN", "feishu")).strip().lower(),
connection_mode=str(
extra.get("connection_mode") or os.getenv("FEISHU_CONNECTION_MODE", "websocket")
).strip().lower(),
encrypt_key=os.getenv("FEISHU_ENCRYPT_KEY", "").strip(),
verification_token=os.getenv("FEISHU_VERIFICATION_TOKEN", "").strip(),
group_policy=os.getenv("FEISHU_GROUP_POLICY", "allowlist").strip().lower(),
allowed_group_users=frozenset(
item.strip()
for item in os.getenv("FEISHU_ALLOWED_USERS", "").split(",")
if item.strip()
),
bot_open_id=os.getenv("FEISHU_BOT_OPEN_ID", "").strip(),
bot_user_id=os.getenv("FEISHU_BOT_USER_ID", "").strip(),
bot_name=os.getenv("FEISHU_BOT_NAME", "").strip(),
dedup_cache_size=max(
32,
int(os.getenv("HERMES_FEISHU_DEDUP_CACHE_SIZE", str(_DEFAULT_DEDUP_CACHE_SIZE))),
),
text_batch_delay_seconds=float(
os.getenv("HERMES_FEISHU_TEXT_BATCH_DELAY_SECONDS", str(_DEFAULT_TEXT_BATCH_DELAY_SECONDS))
),
text_batch_max_messages=max(
1,
int(os.getenv("HERMES_FEISHU_TEXT_BATCH_MAX_MESSAGES", str(_DEFAULT_TEXT_BATCH_MAX_MESSAGES))),
),
text_batch_max_chars=max(
1,
int(os.getenv("HERMES_FEISHU_TEXT_BATCH_MAX_CHARS", str(_DEFAULT_TEXT_BATCH_MAX_CHARS))),
),
media_batch_delay_seconds=float(
os.getenv("HERMES_FEISHU_MEDIA_BATCH_DELAY_SECONDS", str(_DEFAULT_MEDIA_BATCH_DELAY_SECONDS))
),
webhook_host=str(
extra.get("webhook_host") or os.getenv("FEISHU_WEBHOOK_HOST", _DEFAULT_WEBHOOK_HOST)
).strip(),
webhook_port=int(
extra.get("webhook_port") or os.getenv("FEISHU_WEBHOOK_PORT", str(_DEFAULT_WEBHOOK_PORT))
),
webhook_path=(
str(extra.get("webhook_path") or os.getenv("FEISHU_WEBHOOK_PATH", _DEFAULT_WEBHOOK_PATH)).strip()
or _DEFAULT_WEBHOOK_PATH
),
)
def _apply_settings(self, settings: FeishuAdapterSettings) -> None:
self._app_id = settings.app_id
self._app_secret = settings.app_secret
self._domain_name = settings.domain_name
self._connection_mode = settings.connection_mode
self._encrypt_key = settings.encrypt_key
self._verification_token = settings.verification_token
self._group_policy = settings.group_policy
self._allowed_group_users = set(settings.allowed_group_users)
self._bot_open_id = settings.bot_open_id
self._bot_user_id = settings.bot_user_id
self._bot_name = settings.bot_name
self._dedup_cache_size = settings.dedup_cache_size
self._text_batch_delay_seconds = settings.text_batch_delay_seconds
self._text_batch_max_messages = settings.text_batch_max_messages
self._text_batch_max_chars = settings.text_batch_max_chars
self._media_batch_delay_seconds = settings.media_batch_delay_seconds
self._webhook_host = settings.webhook_host
self._webhook_port = settings.webhook_port
self._webhook_path = settings.webhook_path
def _build_event_handler(self) -> Any:
if EventDispatcherHandler is None:
return None
return (
EventDispatcherHandler.builder(
self._encrypt_key,
self._verification_token,
)
.register_p2_im_message_message_read_v1(self._on_message_read_event)
.register_p2_im_message_receive_v1(self._on_message_event)
.register_p2_im_message_reaction_created_v1(
lambda data: self._on_reaction_event("im.message.reaction.created_v1", data)
)
.register_p2_im_message_reaction_deleted_v1(
lambda data: self._on_reaction_event("im.message.reaction.deleted_v1", data)
)
.register_p2_card_action_trigger(self._on_card_action_trigger)
.build()
)
async def connect(self) -> bool:
"""Connect to Feishu/Lark."""
if not FEISHU_AVAILABLE:
logger.error("[Feishu] lark-oapi not installed")
return False
if not self._app_id or not self._app_secret:
logger.error("[Feishu] FEISHU_APP_ID or FEISHU_APP_SECRET not set")
return False
if self._connection_mode not in {"websocket", "webhook"}:
logger.error(
"[Feishu] Unsupported FEISHU_CONNECTION_MODE=%s. Supported modes: websocket, webhook.",
self._connection_mode,
)
return False
try:
self._app_lock_identity = self._app_id
acquired, existing = acquire_scoped_lock(
_FEISHU_APP_LOCK_SCOPE,
self._app_lock_identity,
metadata={"platform": self.platform.value},
)
if not acquired:
owner_pid = existing.get("pid") if isinstance(existing, dict) else None
message = (
"Another local Hermes gateway is already using this Feishu app_id"
+ (f" (PID {owner_pid})." if owner_pid else ".")
+ " Stop the other gateway before starting a second Feishu websocket client."
)
logger.error("[Feishu] %s", message)
self._set_fatal_error("feishu_app_lock", message, retryable=False)
return False
self._loop = asyncio.get_running_loop()
await self._connect_with_retry()
self._mark_connected()
logger.info("[Feishu] Connected in %s mode (%s)", self._connection_mode, self._domain_name)
return True
except Exception as exc:
await self._release_app_lock()
message = f"Feishu startup failed: {exc}"
self._set_fatal_error("feishu_connect_error", message, retryable=True)
logger.error("[Feishu] Failed to connect: %s", exc, exc_info=True)
return False
async def disconnect(self) -> None:
"""Disconnect from Feishu/Lark."""
self._running = False
await self._cancel_pending_tasks(self._pending_text_batch_tasks)
await self._cancel_pending_tasks(self._pending_media_batch_tasks)
self._reset_batch_buffers()
self._disable_websocket_auto_reconnect()
await self._stop_webhook_server()
self._ws_future = None
self._loop = None
self._persist_seen_message_ids()
await self._release_app_lock()
self._mark_disconnected()
logger.info("[Feishu] Disconnected")
async def _cancel_pending_tasks(self, tasks: Dict[str, asyncio.Task]) -> None:
pending = [task for task in tasks.values() if task and not task.done()]
for task in pending:
task.cancel()
if pending:
await asyncio.gather(*pending, return_exceptions=True)
tasks.clear()
def _reset_batch_buffers(self) -> None:
self._pending_text_batches.clear()
self._pending_text_batch_counts.clear()
self._pending_media_batches.clear()
def _disable_websocket_auto_reconnect(self) -> None:
if self._ws_client is None:
return
try:
setattr(self._ws_client, "_auto_reconnect", False)
except Exception:
pass
finally:
self._ws_client = None
async def _stop_webhook_server(self) -> None:
if self._webhook_runner is None:
return
try:
await self._webhook_runner.cleanup()
finally:
self._webhook_runner = None
self._webhook_site = None
# =========================================================================
# Outbound — send / edit / send_image / send_voice / …
# =========================================================================
async def send(
self,
chat_id: str,
content: str,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Send a Feishu message."""
if not self._client:
return SendResult(success=False, error="Not connected")
formatted = self.format_message(content)
chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH)
last_response = None
try:
for chunk in chunks:
msg_type, payload = self._build_outbound_payload(chunk)
try:
response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type=msg_type,
payload=payload,
reply_to=reply_to,
metadata=metadata,
)
except Exception as exc:
if msg_type != "post" or not _POST_CONTENT_INVALID_RE.search(str(exc)):
raise
logger.warning("[Feishu] Invalid post payload rejected by API; falling back to plain text")
response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type="text",
payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
reply_to=reply_to,
metadata=metadata,
)
if (
msg_type == "post"
and not self._response_succeeded(response)
and _POST_CONTENT_INVALID_RE.search(str(getattr(response, "msg", "") or ""))
):
logger.warning("[Feishu] Post payload rejected by API response; falling back to plain text")
response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type="text",
payload=json.dumps({"text": _strip_markdown_to_plain_text(chunk)}, ensure_ascii=False),
reply_to=reply_to,
metadata=metadata,
)
last_response = response
return self._finalize_send_result(last_response, "send failed")
except Exception as exc:
logger.error("[Feishu] Send error: %s", exc, exc_info=True)
return SendResult(success=False, error=str(exc))
async def edit_message(
self,
chat_id: str,
message_id: str,
content: str,
) -> SendResult:
"""Edit a previously sent Feishu text/post message."""
if not self._client:
return SendResult(success=False, error="Not connected")
try:
msg_type, payload = self._build_outbound_payload(content)
body = self._build_update_message_body(msg_type=msg_type, content=payload)
request = self._build_update_message_request(message_id=message_id, request_body=body)
response = await asyncio.to_thread(self._client.im.v1.message.update, request)
result = self._finalize_send_result(response, "update failed")
if not result.success and msg_type == "post" and _POST_CONTENT_INVALID_RE.search(result.error or ""):
logger.warning("[Feishu] Invalid post update payload rejected by API; falling back to plain text")
fallback_body = self._build_update_message_body(
msg_type="text",
content=json.dumps({"text": _strip_markdown_to_plain_text(content)}, ensure_ascii=False),
)
fallback_request = self._build_update_message_request(message_id=message_id, request_body=fallback_body)
fallback_response = await asyncio.to_thread(self._client.im.v1.message.update, fallback_request)
result = self._finalize_send_result(fallback_response, "update failed")
if result.success:
result.message_id = message_id
return result
except Exception as exc:
logger.error("[Feishu] Failed to edit message %s: %s", message_id, exc, exc_info=True)
return SendResult(success=False, error=str(exc))
async def send_voice(
self,
chat_id: str,
audio_path: str,
caption: Optional[str] = None,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs,
) -> SendResult:
"""Send audio to Feishu as a file attachment plus optional caption."""
return await self._send_uploaded_file_message(
chat_id=chat_id,
file_path=audio_path,
reply_to=reply_to,
metadata=metadata,
caption=caption,
outbound_message_type="audio",
)
async def send_document(
self,
chat_id: str,
file_path: str,
caption: Optional[str] = None,
file_name: Optional[str] = None,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs,
) -> SendResult:
"""Send a document/file attachment to Feishu."""
return await self._send_uploaded_file_message(
chat_id=chat_id,
file_path=file_path,
reply_to=reply_to,
metadata=metadata,
caption=caption,
file_name=file_name,
)
async def send_video(
self,
chat_id: str,
video_path: str,
caption: Optional[str] = None,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs,
) -> SendResult:
"""Send a video file to Feishu."""
return await self._send_uploaded_file_message(
chat_id=chat_id,
file_path=video_path,
reply_to=reply_to,
metadata=metadata,
caption=caption,
outbound_message_type="media",
)
async def send_image_file(
self,
chat_id: str,
image_path: str,
caption: Optional[str] = None,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
**kwargs,
) -> SendResult:
"""Send a local image file to Feishu."""
if not self._client:
return SendResult(success=False, error="Not connected")
if not os.path.exists(image_path):
return SendResult(success=False, error=f"Image file not found: {image_path}")
try:
with open(image_path, "rb") as image_file:
body = self._build_image_upload_body(
image_type=_FEISHU_IMAGE_UPLOAD_TYPE,
image=image_file,
)
request = self._build_image_upload_request(body)
upload_response = await asyncio.to_thread(self._client.im.v1.image.create, request)
image_key = self._extract_response_field(upload_response, "image_key")
if not image_key:
return self._response_error_result(
upload_response,
default_message="image upload failed",
override_error="Feishu image upload missing image_key",
)
if caption:
post_payload = self._build_media_post_payload(
caption=caption,
media_tag={"tag": "img", "image_key": image_key},
)
message_response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type="post",
payload=post_payload,
reply_to=reply_to,
metadata=metadata,
)
else:
message_response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type="image",
payload=json.dumps({"image_key": image_key}, ensure_ascii=False),
reply_to=reply_to,
metadata=metadata,
)
return self._finalize_send_result(message_response, "image send failed")
except Exception as exc:
logger.error("[Feishu] Failed to send image %s: %s", image_path, exc, exc_info=True)
return SendResult(success=False, error=str(exc))
async def send_typing(self, chat_id: str, metadata=None) -> None:
"""Feishu bot API does not expose a typing indicator."""
return None
async def send_image(
self,
chat_id: str,
image_url: str,
caption: Optional[str] = None,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Download a remote image then send it through the native Feishu image flow."""
try:
image_path = await self._download_remote_image(image_url)
except Exception as exc:
logger.error("[Feishu] Failed to download image %s: %s", image_url, exc, exc_info=True)
return await super().send_image(
chat_id=chat_id,
image_url=image_url,
caption=caption,
reply_to=reply_to,
metadata=metadata,
)
return await self.send_image_file(
chat_id=chat_id,
image_path=image_path,
caption=caption,
reply_to=reply_to,
metadata=metadata,
)
async def send_animation(
self,
chat_id: str,
animation_url: str,
caption: Optional[str] = None,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Feishu has no native GIF bubble; degrade to a downloadable file."""
try:
file_path, file_name = await self._download_remote_document(
animation_url,
default_ext=".gif",
preferred_name="animation.gif",
)
except Exception as exc:
logger.error("[Feishu] Failed to download animation %s: %s", animation_url, exc, exc_info=True)
return await super().send_animation(
chat_id=chat_id,
animation_url=animation_url,
caption=caption,
reply_to=reply_to,
metadata=metadata,
)
degraded_caption = f"[GIF downgraded to file]\n{caption}" if caption else "[GIF downgraded to file]"
return await self.send_document(
chat_id=chat_id,
file_path=file_path,
file_name=file_name,
caption=degraded_caption,
reply_to=reply_to,
metadata=metadata,
)
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
"""Return real chat metadata from Feishu when available."""
fallback = {
"chat_id": chat_id,
"name": chat_id,
"type": "dm",
}
if not self._client:
return fallback
cached = self._chat_info_cache.get(chat_id)
if cached is not None:
return dict(cached)
try:
request = self._build_get_chat_request(chat_id)
response = await asyncio.to_thread(self._client.im.v1.chat.get, request)
if not response or getattr(response, "success", lambda: False)() is False:
code = getattr(response, "code", "unknown")
msg = getattr(response, "msg", "chat lookup failed")
logger.warning("[Feishu] Failed to get chat info for %s: [%s] %s", chat_id, code, msg)
return fallback
data = getattr(response, "data", None)
raw_chat_type = str(getattr(data, "chat_type", "") or "").strip().lower()
info = {
"chat_id": chat_id,
"name": str(getattr(data, "name", None) or chat_id),
"type": self._map_chat_type(raw_chat_type),
"raw_type": raw_chat_type or None,
}
self._chat_info_cache[chat_id] = info
return dict(info)
except Exception:
logger.warning("[Feishu] Failed to get chat info for %s", chat_id, exc_info=True)
return fallback
def format_message(self, content: str) -> str:
"""Feishu text messages are plain text by default."""
return content.strip()
# =========================================================================
# Inbound event handlers
# =========================================================================
def _on_message_event(self, data: Any) -> None:
"""Normalize Feishu inbound events into MessageEvent."""
if self._loop is None:
logger.warning("[Feishu] Dropping inbound message before adapter loop is ready")
return
future = asyncio.run_coroutine_threadsafe(
self._handle_message_event_data(data),
self._loop,
)
future.add_done_callback(self._log_background_failure)
async def _handle_message_event_data(self, data: Any) -> None:
"""Shared inbound message handling for websocket and webhook transports."""
event = getattr(data, "event", None)
message = getattr(event, "message", None)
sender = getattr(event, "sender", None)
sender_id = getattr(sender, "sender_id", None)
if not message or not sender_id:
logger.debug("[Feishu] Dropping malformed inbound event: missing message or sender_id")
return
message_id = getattr(message, "message_id", None)
if not message_id or self._is_duplicate(message_id):
logger.debug("[Feishu] Dropping duplicate/missing message_id: %s", message_id)
return
if getattr(sender, "sender_type", "") == "bot":
logger.debug("[Feishu] Dropping bot-originated event: %s", message_id)
return
chat_type = getattr(message, "chat_type", "p2p")
if chat_type != "p2p" and not self._should_accept_group_message(message, sender_id):
logger.debug("[Feishu] Dropping group message that failed mention/policy gate: %s", message_id)
return
await self._process_inbound_message(
data=data,
message=message,
sender_id=sender_id,
chat_type=chat_type,
message_id=message_id,
)
def _on_message_read_event(self, data: P2ImMessageMessageReadV1) -> None:
"""Ignore read-receipt events that Hermes does not act on."""
event = getattr(data, "event", None)
message = getattr(event, "message", None)
message_id = getattr(message, "message_id", None) or ""
logger.debug("[Feishu] Ignoring message_read event: %s", message_id)
def _on_bot_added_to_chat(self, data: Any) -> None:
"""Handle bot being added to a group chat."""
event = getattr(data, "event", None)
chat_id = str(getattr(event, "chat_id", "") or "")
logger.info("[Feishu] Bot added to chat: %s", chat_id)
self._chat_info_cache.pop(chat_id, None)
def _on_bot_removed_from_chat(self, data: Any) -> None:
"""Handle bot being removed from a group chat."""
event = getattr(data, "event", None)
chat_id = str(getattr(event, "chat_id", "") or "")
logger.info("[Feishu] Bot removed from chat: %s", chat_id)
self._chat_info_cache.pop(chat_id, None)
def _on_reaction_event(self, event_type: str, data: Any) -> None:
"""Route user reactions on bot messages as synthetic text events."""
event = getattr(data, "event", None)
message_id = str(getattr(event, "message_id", "") or "")
operator_type = str(getattr(event, "operator_type", "") or "")
reaction_type_obj = getattr(event, "reaction_type", None)
emoji_type = str(getattr(reaction_type_obj, "emoji_type", "") or "")
action = "added" if "created" in event_type else "removed"
logger.debug(
"[Feishu] Reaction %s on message %s (operator_type=%s, emoji=%s)",
action,
message_id,
operator_type,
emoji_type,
)
# Only process reactions from real users. Ignore app/bot-generated reactions
# and Hermes' own ACK emoji to avoid feedback loops.
if (
operator_type in {"bot", "app"}
or emoji_type == _FEISHU_ACK_EMOJI
or not message_id
or self._loop is None
):
return
future = asyncio.run_coroutine_threadsafe(
self._handle_reaction_event(event_type, data),
self._loop,
)
future.add_done_callback(self._log_background_failure)
def _on_card_action_trigger(self, data: Any) -> Any:
"""Schedule Feishu card actions on the adapter loop and acknowledge immediately."""
if self._loop is None:
logger.warning("[Feishu] Dropping card action before adapter loop is ready")
else:
future = asyncio.run_coroutine_threadsafe(
self._handle_card_action_event(data),
self._loop,
)
future.add_done_callback(self._log_background_failure)
if P2CardActionTriggerResponse is None:
return None
return P2CardActionTriggerResponse()
async def _handle_reaction_event(self, event_type: str, data: Any) -> None:
"""Fetch the reacted-to message; if it was sent by this bot, emit a synthetic text event."""
if not self._client:
return
event = getattr(data, "event", None)
message_id = str(getattr(event, "message_id", "") or "")
if not message_id:
return
# Fetch the target message to verify it was sent by us and to obtain chat context.
try:
request = self._build_get_message_request(message_id)
response = await asyncio.to_thread(self._client.im.v1.message.get, request)
if not response or not getattr(response, "success", lambda: False)():
return
items = getattr(getattr(response, "data", None), "items", None) or []
msg = items[0] if items else None
if not msg:
return
sender = getattr(msg, "sender", None)
sender_type = str(getattr(sender, "sender_type", "") or "").lower()
if sender_type != "app":
return # only route reactions on our own bot messages
chat_id = str(getattr(msg, "chat_id", "") or "")
chat_type_raw = str(getattr(msg, "chat_type", "p2p") or "p2p")
if not chat_id:
return
except Exception:
logger.debug("[Feishu] Failed to fetch message for reaction routing", exc_info=True)
return
user_id_obj = getattr(event, "user_id", None)
reaction_type_obj = getattr(event, "reaction_type", None)
emoji_type = str(getattr(reaction_type_obj, "emoji_type", "") or "UNKNOWN")
action = "added" if "created" in event_type else "removed"
synthetic_text = f"reaction:{action}:{emoji_type}"
sender_profile = await self._resolve_sender_profile(user_id_obj)
chat_info = await self.get_chat_info(chat_id)
source = self.build_source(
chat_id=chat_id,
chat_name=chat_info.get("name") or chat_id or "Feishu Chat",
chat_type=self._resolve_source_chat_type(chat_info=chat_info, event_chat_type=chat_type_raw),
user_id=sender_profile["user_id"],
user_name=sender_profile["user_name"],
thread_id=None,
user_id_alt=sender_profile["user_id_alt"],
)
synthetic_event = MessageEvent(
text=synthetic_text,
message_type=MessageType.TEXT,
source=source,
raw_message=data,
message_id=message_id,
timestamp=datetime.now(),
)
logger.info("[Feishu] Routing reaction %s:%s on bot message %s as synthetic event", action, emoji_type, message_id)
await self._handle_message_with_guards(synthetic_event)
def _is_card_action_duplicate(self, token: str) -> bool:
"""Return True if this card action token was already processed within the dedup window."""
now = time.time()
# Prune expired tokens lazily each call.
expired = [t for t, ts in self._card_action_tokens.items() if now - ts > _FEISHU_CARD_ACTION_DEDUP_TTL_SECONDS]
for t in expired:
del self._card_action_tokens[t]
if token in self._card_action_tokens:
return True
self._card_action_tokens[token] = now
return False
async def _handle_card_action_event(self, data: Any) -> None:
"""Route Feishu interactive card button clicks as synthetic COMMAND events."""
event = getattr(data, "event", None)
token = str(getattr(event, "token", "") or "")
if token and self._is_card_action_duplicate(token):
logger.debug("[Feishu] Dropping duplicate card action token: %s", token)
return
context = getattr(event, "context", None)
chat_id = str(getattr(context, "open_chat_id", "") or "")
operator = getattr(event, "operator", None)
open_id = str(getattr(operator, "open_id", "") or "")
if not chat_id or not open_id:
logger.debug("[Feishu] Card action missing chat_id or operator open_id, dropping")
return
action = getattr(event, "action", None)
action_tag = str(getattr(action, "tag", "") or "button")
action_value = getattr(action, "value", {}) or {}
synthetic_text = f"/card {action_tag}"
if action_value:
try:
synthetic_text += f" {json.dumps(action_value, ensure_ascii=False)}"
except Exception:
pass
sender_id = SimpleNamespace(open_id=open_id, user_id=None, union_id=None)
sender_profile = await self._resolve_sender_profile(sender_id)
chat_info = await self.get_chat_info(chat_id)
source = self.build_source(
chat_id=chat_id,
chat_name=chat_info.get("name") or chat_id or "Feishu Chat",
chat_type=self._resolve_source_chat_type(chat_info=chat_info, event_chat_type="group"),
user_id=sender_profile["user_id"],
user_name=sender_profile["user_name"],
thread_id=None,
user_id_alt=sender_profile["user_id_alt"],
)
synthetic_event = MessageEvent(
text=synthetic_text,
message_type=MessageType.COMMAND,
source=source,
raw_message=data,
message_id=token or str(uuid.uuid4()),
timestamp=datetime.now(),
)
logger.info("[Feishu] Routing card action %r from %s in %s as synthetic command", action_tag, open_id, chat_id)
await self._handle_message_with_guards(synthetic_event)
# =========================================================================
# Per-chat serialization and typing indicator
# =========================================================================
def _get_chat_lock(self, chat_id: str) -> asyncio.Lock:
"""Return (creating if needed) the per-chat asyncio.Lock for serial message processing."""
lock = self._chat_locks.get(chat_id)
if lock is None:
lock = asyncio.Lock()
self._chat_locks[chat_id] = lock
return lock
async def _handle_message_with_guards(self, event: MessageEvent) -> None:
"""Dispatch a single event through the agent pipeline with per-chat serialization
and a persistent ACK emoji reaction before processing starts.
- Per-chat lock: ensures messages in the same chat are processed one at a time
(matches openclaw's createChatQueue serial queue behaviour).
- ACK indicator: adds a CHECK reaction to the triggering message before handing
off to the agent and leaves it in place as a receipt marker.
"""
chat_id = getattr(event.source, "chat_id", "") or "" if event.source else ""
chat_lock = self._get_chat_lock(chat_id)
async with chat_lock:
message_id = event.message_id
if message_id:
await self._add_ack_reaction(message_id)
await self.handle_message(event)
async def _add_ack_reaction(self, message_id: str) -> Optional[str]:
"""Add a persistent ACK emoji reaction to signal the message was received."""
if not self._client or not message_id:
return None
try:
from lark_oapi.api.im.v1 import ( # lazy import — keeps optional dep optional
CreateMessageReactionRequest,
CreateMessageReactionRequestBody,
)
body = (
CreateMessageReactionRequestBody.builder()
.reaction_type({"emoji_type": _FEISHU_ACK_EMOJI})
.build()
)
request = (
CreateMessageReactionRequest.builder()
.message_id(message_id)
.request_body(body)
.build()
)
response = await asyncio.to_thread(self._client.im.v1.message_reaction.create, request)
if response and getattr(response, "success", lambda: False)():
data = getattr(response, "data", None)
return getattr(data, "reaction_id", None)
logger.warning(
"[Feishu] Failed to add ack reaction to %s: code=%s msg=%s",
message_id,
getattr(response, "code", None),
getattr(response, "msg", None),
)
except Exception:
logger.warning("[Feishu] Failed to add ack reaction to %s", message_id, exc_info=True)
return None
# =========================================================================
# Webhook server and security
# =========================================================================
def _record_webhook_anomaly(self, remote_ip: str, status: str) -> None:
"""Increment the anomaly counter for remote_ip and emit a WARNING every threshold hits.
Mirrors openclaw's createWebhookAnomalyTracker: TTL 6 hours, log every 25 consecutive
error responses from the same IP.
"""
now = time.time()
entry = self._webhook_anomaly_counts.get(remote_ip)
if entry is not None:
count, _last_status, first_seen = entry
if now - first_seen < _FEISHU_WEBHOOK_ANOMALY_TTL_SECONDS:
count += 1
if count % _FEISHU_WEBHOOK_ANOMALY_THRESHOLD == 0:
logger.warning(
"[Feishu] Webhook anomaly: %d consecutive error responses (%s) from %s "
"over the last %.0fs",
count,
status,
remote_ip,
now - first_seen,
)
self._webhook_anomaly_counts[remote_ip] = (count, status, first_seen)
return
# Either first occurrence or TTL expired — start fresh.
self._webhook_anomaly_counts[remote_ip] = (1, status, now)
def _clear_webhook_anomaly(self, remote_ip: str) -> None:
"""Reset the anomaly counter for remote_ip after a successful request."""
self._webhook_anomaly_counts.pop(remote_ip, None)
# =========================================================================
# Inbound processing pipeline
# =========================================================================
async def _process_inbound_message(
self,
*,
data: Any,
message: Any,
sender_id: Any,
chat_type: str,
message_id: str,
) -> None:
text, inbound_type, media_urls, media_types = await self._extract_message_content(message)
if inbound_type == MessageType.TEXT and not text and not media_urls:
logger.debug("[Feishu] Ignoring unsupported or empty message type: %s", getattr(message, "message_type", ""))
return
if inbound_type == MessageType.TEXT and text.startswith("/"):
inbound_type = MessageType.COMMAND
reply_to_message_id = (
getattr(message, "parent_id", None)
or getattr(message, "upper_message_id", None)
or None
)
reply_to_text = await self._fetch_message_text(reply_to_message_id) if reply_to_message_id else None
logger.info(
"[Feishu] Inbound %s message received: id=%s type=%s chat_id=%s text=%r media=%d",
"dm" if chat_type == "p2p" else "group",
message_id,
inbound_type.value,
getattr(message, "chat_id", "") or "",
text[:120],
len(media_urls),
)
chat_id = getattr(message, "chat_id", "") or ""
chat_info = await self.get_chat_info(chat_id)
sender_profile = await self._resolve_sender_profile(sender_id)
source = self.build_source(
chat_id=chat_id,
chat_name=chat_info.get("name") or chat_id or "Feishu Chat",
chat_type=self._resolve_source_chat_type(chat_info=chat_info, event_chat_type=chat_type),
user_id=sender_profile["user_id"],
user_name=sender_profile["user_name"],
thread_id=getattr(message, "thread_id", None) or None,
user_id_alt=sender_profile["user_id_alt"],
)
normalized = MessageEvent(
text=text,
message_type=inbound_type,
source=source,
raw_message=data,
message_id=message_id,
media_urls=media_urls,
media_types=media_types,
reply_to_message_id=reply_to_message_id,
reply_to_text=reply_to_text,
timestamp=datetime.now(),
)
await self._dispatch_inbound_event(normalized)
async def _dispatch_inbound_event(self, event: MessageEvent) -> None:
"""Apply Feishu-specific burst protection before entering the base adapter."""
if event.message_type == MessageType.TEXT and not event.is_command():
await self._enqueue_text_event(event)
return
if self._should_batch_media_event(event):
await self._enqueue_media_event(event)
return
await self._handle_message_with_guards(event)
# =========================================================================
# Media batching
# =========================================================================
def _should_batch_media_event(self, event: MessageEvent) -> bool:
return bool(
event.media_urls
and event.message_type in {MessageType.PHOTO, MessageType.VIDEO, MessageType.DOCUMENT, MessageType.AUDIO}
)
def _media_batch_key(self, event: MessageEvent) -> str:
from gateway.session import build_session_key
session_key = build_session_key(
event.source,
group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
)
return f"{session_key}:media:{event.message_type.value}"
@staticmethod
def _media_batch_is_compatible(existing: MessageEvent, incoming: MessageEvent) -> bool:
return (
existing.message_type == incoming.message_type
and existing.reply_to_message_id == incoming.reply_to_message_id
and existing.reply_to_text == incoming.reply_to_text
and existing.source.thread_id == incoming.source.thread_id
)
async def _enqueue_media_event(self, event: MessageEvent) -> None:
key = self._media_batch_key(event)
existing = self._pending_media_batches.get(key)
if existing is None:
self._pending_media_batches[key] = event
self._schedule_media_batch_flush(key)
return
if not self._media_batch_is_compatible(existing, event):
await self._flush_media_batch_now(key)
self._pending_media_batches[key] = event
self._schedule_media_batch_flush(key)
return
existing.media_urls.extend(event.media_urls)
existing.media_types.extend(event.media_types)
if event.text:
if not existing.text:
existing.text = event.text
elif event.text not in existing.text.split("\n\n"):
existing.text = f"{existing.text}\n\n{event.text}"
existing.timestamp = event.timestamp
if event.message_id:
existing.message_id = event.message_id
self._schedule_media_batch_flush(key)
def _schedule_media_batch_flush(self, key: str) -> None:
self._reschedule_batch_task(
self._pending_media_batch_tasks,
key,
self._flush_media_batch,
)
async def _flush_media_batch(self, key: str) -> None:
current_task = asyncio.current_task()
try:
await asyncio.sleep(self._media_batch_delay_seconds)
await self._flush_media_batch_now(key)
finally:
if self._pending_media_batch_tasks.get(key) is current_task:
self._pending_media_batch_tasks.pop(key, None)
async def _flush_media_batch_now(self, key: str) -> None:
event = self._pending_media_batches.pop(key, None)
if not event:
return
logger.info(
"[Feishu] Flushing media batch %s with %d attachment(s)",
key,
len(event.media_urls),
)
await self._handle_message_with_guards(event)
async def _download_remote_image(self, image_url: str) -> str:
ext = self._guess_remote_extension(image_url, default=".jpg")
return await cache_image_from_url(image_url, ext=ext)
async def _download_remote_document(
self,
file_url: str,
*,
default_ext: str,
preferred_name: str,
) -> tuple[str, str]:
import httpx
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
response = await client.get(
file_url,
headers={
"User-Agent": "Mozilla/5.0 (compatible; HermesAgent/1.0)",
"Accept": "*/*",
},
)
response.raise_for_status()
filename = self._derive_remote_filename(
file_url,
content_type=str(response.headers.get("Content-Type", "")),
default_name=preferred_name,
default_ext=default_ext,
)
cached_path = cache_document_from_bytes(response.content, filename)
return cached_path, filename
@staticmethod
def _guess_remote_extension(url: str, *, default: str) -> str:
ext = Path((url or "").split("?", 1)[0]).suffix.lower()
return ext if ext in (_IMAGE_EXTENSIONS | _AUDIO_EXTENSIONS | _VIDEO_EXTENSIONS | set(SUPPORTED_DOCUMENT_TYPES)) else default
@staticmethod
def _derive_remote_filename(file_url: str, *, content_type: str, default_name: str, default_ext: str) -> str:
candidate = Path((file_url or "").split("?", 1)[0]).name or default_name
ext = Path(candidate).suffix.lower()
if not ext:
guessed = mimetypes.guess_extension((content_type or "").split(";", 1)[0].strip().lower() or "") or default_ext
candidate = f"{candidate}{guessed}"
return candidate
@staticmethod
def _namespace_from_mapping(value: Any) -> Any:
if isinstance(value, dict):
return SimpleNamespace(**{key: FeishuAdapter._namespace_from_mapping(item) for key, item in value.items()})
if isinstance(value, list):
return [FeishuAdapter._namespace_from_mapping(item) for item in value]
return value
async def _handle_webhook_request(self, request: Any) -> Any:
remote_ip = (getattr(request, "remote", None) or "unknown")
# Rate limiting — composite key: app_id:path:remote_ip (matches openclaw key structure).
rate_key = f"{self._app_id}:{self._webhook_path}:{remote_ip}"
if not self._check_webhook_rate_limit(rate_key):
logger.warning("[Feishu] Webhook rate limit exceeded for %s", remote_ip)
self._record_webhook_anomaly(remote_ip, "429")
return web.Response(status=429, text="Too Many Requests")
# Content-Type guard — Feishu always sends application/json.
headers = getattr(request, "headers", {}) or {}
content_type = str(headers.get("Content-Type", "") or "").split(";")[0].strip().lower()
if content_type and content_type != "application/json":
logger.warning("[Feishu] Webhook rejected: unexpected Content-Type %r from %s", content_type, remote_ip)
self._record_webhook_anomaly(remote_ip, "415")
return web.Response(status=415, text="Unsupported Media Type")
# Body size guard — reject early via Content-Length when present.
content_length = getattr(request, "content_length", None)
if content_length is not None and content_length > _FEISHU_WEBHOOK_MAX_BODY_BYTES:
logger.warning("[Feishu] Webhook body too large (%d bytes) from %s", content_length, remote_ip)
self._record_webhook_anomaly(remote_ip, "413")
return web.Response(status=413, text="Request body too large")
try:
body_bytes: bytes = await asyncio.wait_for(
request.read(),
timeout=_FEISHU_WEBHOOK_BODY_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
logger.warning("[Feishu] Webhook body read timed out after %ds from %s", _FEISHU_WEBHOOK_BODY_TIMEOUT_SECONDS, remote_ip)
self._record_webhook_anomaly(remote_ip, "408")
return web.Response(status=408, text="Request Timeout")
except Exception:
self._record_webhook_anomaly(remote_ip, "400")
return web.json_response({"code": 400, "msg": "failed to read body"}, status=400)
if len(body_bytes) > _FEISHU_WEBHOOK_MAX_BODY_BYTES:
logger.warning("[Feishu] Webhook body exceeds limit (%d bytes) from %s", len(body_bytes), remote_ip)
self._record_webhook_anomaly(remote_ip, "413")
return web.Response(status=413, text="Request body too large")
try:
payload = json.loads(body_bytes.decode("utf-8"))
except (json.JSONDecodeError, UnicodeDecodeError):
self._record_webhook_anomaly(remote_ip, "400")
return web.json_response({"code": 400, "msg": "invalid json"}, status=400)
# URL verification challenge — respond before other checks so that Feishu's
# subscription setup works even before encrypt_key is wired.
if payload.get("type") == "url_verification":
return web.json_response({"challenge": payload.get("challenge", "")})
# Verification token check — second layer of defence beyond signature (matches openclaw).
if self._verification_token:
header = payload.get("header") or {}
incoming_token = str(header.get("token") or payload.get("token") or "")
if not incoming_token or not hmac.compare_digest(incoming_token, self._verification_token):
logger.warning("[Feishu] Webhook rejected: invalid verification token from %s", remote_ip)
self._record_webhook_anomaly(remote_ip, "401-token")
return web.Response(status=401, text="Invalid verification token")
# Timing-safe signature verification (only enforced when encrypt_key is set).
if self._encrypt_key and not self._is_webhook_signature_valid(request.headers, body_bytes):
logger.warning("[Feishu] Webhook rejected: invalid signature from %s", remote_ip)
self._record_webhook_anomaly(remote_ip, "401-sig")
return web.Response(status=401, text="Invalid signature")
if payload.get("encrypt"):
logger.error("[Feishu] Encrypted webhook payloads are not supported by Hermes webhook mode")
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)
event_type = str((payload.get("header") or {}).get("event_type") or "")
data = self._namespace_from_mapping(payload)
if event_type == "im.message.receive_v1":
await self._handle_message_event_data(data)
elif event_type == "im.message.message_read_v1":
self._on_message_read_event(data)
elif event_type == "im.chat.member.bot.added_v1":
self._on_bot_added_to_chat(data)
elif event_type == "im.chat.member.bot.deleted_v1":
self._on_bot_removed_from_chat(data)
elif event_type in ("im.message.reaction.created_v1", "im.message.reaction.deleted_v1"):
self._on_reaction_event(event_type, data)
elif event_type == "card.action.trigger":
asyncio.ensure_future(self._handle_card_action_event(data))
else:
logger.debug("[Feishu] Ignoring webhook event type: %s", event_type or "unknown")
return web.json_response({"code": 0, "msg": "ok"})
def _is_webhook_signature_valid(self, headers: Any, body_bytes: bytes) -> bool:
"""Verify Feishu webhook signature using timing-safe comparison.
Feishu signature algorithm:
SHA256(timestamp + nonce + encrypt_key + body_string)
Headers checked: x-lark-request-timestamp, x-lark-request-nonce, x-lark-signature.
"""
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
try:
body_str = body_bytes.decode("utf-8", errors="replace")
content = f"{timestamp}{nonce}{self._encrypt_key}{body_str}"
computed = hashlib.sha256(content.encode("utf-8")).hexdigest()
return hmac.compare_digest(computed, signature)
except Exception:
logger.debug("[Feishu] Signature verification raised an exception", exc_info=True)
return False
def _check_webhook_rate_limit(self, rate_key: str) -> bool:
"""Return False when the composite rate_key has exceeded _FEISHU_WEBHOOK_RATE_LIMIT_MAX.
The rate_key is composed as "{app_id}:{path}:{remote_ip}" — matching openclaw's key
structure so the limit is scoped to a specific (account, endpoint, IP) triple rather
than a bare IP, which causes fewer false-positive denials in multi-tenant setups.
The tracking dict is capped at _FEISHU_WEBHOOK_RATE_MAX_KEYS entries to prevent unbounded
memory growth. Stale (expired) entries are pruned when the cap is reached.
"""
now = time.time()
# Fast path: existing entry within the current window.
entry = self._webhook_rate_counts.get(rate_key)
if entry is not None:
count, window_start = entry
if now - window_start < _FEISHU_WEBHOOK_RATE_WINDOW_SECONDS:
if count >= _FEISHU_WEBHOOK_RATE_LIMIT_MAX:
return False
self._webhook_rate_counts[rate_key] = (count + 1, window_start)
return True
# New window for an existing key, or a brand-new key — prune stale entries first.
if len(self._webhook_rate_counts) >= _FEISHU_WEBHOOK_RATE_MAX_KEYS:
stale_keys = [
k for k, (_, ws) in self._webhook_rate_counts.items()
if now - ws >= _FEISHU_WEBHOOK_RATE_WINDOW_SECONDS
]
for k in stale_keys:
del self._webhook_rate_counts[k]
# If still at capacity after pruning, allow through without tracking.
if rate_key not in self._webhook_rate_counts and len(self._webhook_rate_counts) >= _FEISHU_WEBHOOK_RATE_MAX_KEYS:
return True
self._webhook_rate_counts[rate_key] = (1, now)
return True
# =========================================================================
# Text batching
# =========================================================================
def _text_batch_key(self, event: MessageEvent) -> str:
"""Return the session-scoped key used for Feishu text aggregation."""
from gateway.session import build_session_key
return build_session_key(
event.source,
group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
)
@staticmethod
def _text_batch_is_compatible(existing: MessageEvent, incoming: MessageEvent) -> bool:
"""Only merge text events when reply/thread context is identical."""
return (
existing.reply_to_message_id == incoming.reply_to_message_id
and existing.reply_to_text == incoming.reply_to_text
and existing.source.thread_id == incoming.source.thread_id
)
async def _enqueue_text_event(self, event: MessageEvent) -> None:
"""Debounce rapid Feishu text bursts into a single MessageEvent."""
key = self._text_batch_key(event)
existing = self._pending_text_batches.get(key)
if existing is None:
self._pending_text_batches[key] = event
self._pending_text_batch_counts[key] = 1
self._schedule_text_batch_flush(key)
return
if not self._text_batch_is_compatible(existing, event):
await self._flush_text_batch_now(key)
self._pending_text_batches[key] = event
self._pending_text_batch_counts[key] = 1
self._schedule_text_batch_flush(key)
return
existing_count = self._pending_text_batch_counts.get(key, 1)
next_count = existing_count + 1
appended_text = event.text or ""
next_text = f"{existing.text}\n{appended_text}" if existing.text and appended_text else (existing.text or appended_text)
if next_count > self._text_batch_max_messages or len(next_text) > self._text_batch_max_chars:
await self._flush_text_batch_now(key)
self._pending_text_batches[key] = event
self._pending_text_batch_counts[key] = 1
self._schedule_text_batch_flush(key)
return
existing.text = next_text
existing.timestamp = event.timestamp
if event.message_id:
existing.message_id = event.message_id
self._pending_text_batch_counts[key] = next_count
self._schedule_text_batch_flush(key)
def _schedule_text_batch_flush(self, key: str) -> None:
"""Reset the debounce timer for a pending Feishu text batch."""
self._reschedule_batch_task(
self._pending_text_batch_tasks,
key,
self._flush_text_batch,
)
@staticmethod
def _reschedule_batch_task(
task_map: Dict[str, asyncio.Task],
key: str,
flush_fn: Any,
) -> None:
prior_task = task_map.get(key)
if prior_task and not prior_task.done():
prior_task.cancel()
task_map[key] = asyncio.create_task(flush_fn(key))
async def _flush_text_batch(self, key: str) -> None:
"""Flush a pending text batch after the quiet period."""
current_task = asyncio.current_task()
try:
await asyncio.sleep(self._text_batch_delay_seconds)
await self._flush_text_batch_now(key)
finally:
if self._pending_text_batch_tasks.get(key) is current_task:
self._pending_text_batch_tasks.pop(key, None)
async def _flush_text_batch_now(self, key: str) -> None:
"""Dispatch the current text batch immediately."""
event = self._pending_text_batches.pop(key, None)
self._pending_text_batch_counts.pop(key, None)
if not event:
return
logger.info(
"[Feishu] Flushing text batch %s (%d chars)",
key,
len(event.text or ""),
)
await self._handle_message_with_guards(event)
# =========================================================================
# Message content extraction and resource download
# =========================================================================
async def _extract_message_content(self, message: Any) -> tuple[str, MessageType, List[str], List[str]]:
"""Extract text and cached media from a normalized Feishu message."""
raw_content = getattr(message, "content", "") or ""
raw_type = getattr(message, "message_type", "") or ""
message_id = str(getattr(message, "message_id", "") or "")
logger.info("[Feishu] Received raw message type=%s message_id=%s", raw_type, message_id)
normalized = normalize_feishu_message(message_type=raw_type, raw_content=raw_content)
media_urls, media_types = await self._download_feishu_message_resources(
message_id=message_id,
normalized=normalized,
)
inbound_type = self._resolve_normalized_message_type(normalized, media_types)
text = normalized.text_content
if (
inbound_type in {MessageType.DOCUMENT, MessageType.AUDIO, MessageType.VIDEO, MessageType.PHOTO}
and len(media_urls) == 1
and normalized.preferred_message_type in {"document", "audio"}
):
injected = await self._maybe_extract_text_document(media_urls[0], media_types[0])
if injected:
text = injected
return text, inbound_type, media_urls, media_types
async def _download_feishu_message_resources(
self,
*,
message_id: str,
normalized: FeishuNormalizedMessage,
) -> tuple[List[str], List[str]]:
media_urls: List[str] = []
media_types: List[str] = []
for image_key in normalized.image_keys:
cached_path, media_type = await self._download_feishu_image(
message_id=message_id,
image_key=image_key,
)
if cached_path:
media_urls.append(cached_path)
media_types.append(media_type)
for media_ref in normalized.media_refs:
cached_path, media_type = await self._download_feishu_message_resource(
message_id=message_id,
file_key=media_ref.file_key,
resource_type=media_ref.resource_type,
fallback_filename=media_ref.file_name,
)
if cached_path:
media_urls.append(cached_path)
media_types.append(media_type)
return media_urls, media_types
@staticmethod
def _resolve_media_message_type(media_type: str, *, default: MessageType) -> MessageType:
normalized = (media_type or "").lower()
if normalized.startswith("image/"):
return MessageType.PHOTO
if normalized.startswith("audio/"):
return MessageType.AUDIO
if normalized.startswith("video/"):
return MessageType.VIDEO
return default
def _resolve_normalized_message_type(
self,
normalized: FeishuNormalizedMessage,
media_types: List[str],
) -> MessageType:
preferred = normalized.preferred_message_type
if preferred == "photo":
return self._resolve_media_message_type(media_types[0] if media_types else "", default=MessageType.PHOTO)
if preferred == "audio":
return self._resolve_media_message_type(media_types[0] if media_types else "", default=MessageType.AUDIO)
if preferred == "document":
return self._resolve_media_message_type(media_types[0] if media_types else "", default=MessageType.DOCUMENT)
return MessageType.TEXT
def _normalize_inbound_text(self, text: str) -> str:
"""Strip Feishu mention placeholders from inbound text."""
text = _MENTION_RE.sub(" ", text or "")
text = _MULTISPACE_RE.sub(" ", text)
return text.strip()
async def _maybe_extract_text_document(self, cached_path: str, media_type: str) -> str:
if not cached_path or not media_type.startswith("text/"):
return ""
try:
if os.path.getsize(cached_path) > _MAX_TEXT_INJECT_BYTES:
return ""
ext = Path(cached_path).suffix.lower()
if ext not in {".txt", ".md"} and media_type not in {"text/plain", "text/markdown"}:
return ""
content = Path(cached_path).read_text(encoding="utf-8")
display_name = self._display_name_from_cached_path(cached_path)
return f"[Content of {display_name}]:\n{content}"
except (OSError, UnicodeDecodeError):
logger.warning("[Feishu] Failed to inject text document content from %s", cached_path, exc_info=True)
return ""
async def _download_feishu_image(self, *, message_id: str, image_key: str) -> tuple[str, str]:
if not self._client or not message_id:
return "", ""
try:
request = self._build_message_resource_request(
message_id=message_id,
file_key=image_key,
resource_type="image",
)
response = await asyncio.to_thread(self._client.im.v1.message_resource.get, request)
if not response or not response.success():
logger.warning(
"[Feishu] Failed to download image %s: %s %s",
image_key,
getattr(response, "code", "unknown"),
getattr(response, "msg", "request failed"),
)
return "", ""
raw_bytes = self._read_binary_response(response)
if not raw_bytes:
return "", ""
content_type = self._get_response_header(response, "Content-Type")
filename = getattr(response, "file_name", None) or f"{image_key}.jpg"
ext = self._guess_extension(filename, content_type, ".jpg", allowed=_IMAGE_EXTENSIONS)
cached_path = cache_image_from_bytes(raw_bytes, ext=ext)
media_type = self._normalize_media_type(content_type, default=self._default_image_media_type(ext))
return cached_path, media_type
except Exception:
logger.warning("[Feishu] Failed to cache image resource %s", image_key, exc_info=True)
return "", ""
async def _download_feishu_message_resource(
self,
*,
message_id: str,
file_key: str,
resource_type: str,
fallback_filename: str,
) -> tuple[str, str]:
if not self._client or not message_id:
return "", ""
request_types = [resource_type]
if resource_type in {"audio", "media"}:
request_types.append("file")
for request_type in request_types:
try:
request = self._build_message_resource_request(
message_id=message_id,
file_key=file_key,
resource_type=request_type,
)
response = await asyncio.to_thread(self._client.im.v1.message_resource.get, request)
if not response or not response.success():
logger.debug(
"[Feishu] Resource download failed for %s/%s via type=%s: %s %s",
message_id,
file_key,
request_type,
getattr(response, "code", "unknown"),
getattr(response, "msg", "request failed"),
)
continue
raw_bytes = self._read_binary_response(response)
if not raw_bytes:
continue
content_type = self._get_response_header(response, "Content-Type")
response_filename = getattr(response, "file_name", None) or ""
filename = response_filename or fallback_filename or f"{request_type}_{file_key}"
media_type = self._normalize_media_type(
content_type,
default=self._guess_media_type_from_filename(filename),
)
if media_type.startswith("image/"):
ext = self._guess_extension(filename, content_type, ".jpg", allowed=_IMAGE_EXTENSIONS)
cached_path = cache_image_from_bytes(raw_bytes, ext=ext)
logger.info("[Feishu] Cached message image resource at %s", cached_path)
return cached_path, media_type or self._default_image_media_type(ext)
if request_type == "audio" or media_type.startswith("audio/"):
ext = self._guess_extension(filename, content_type, ".ogg", allowed=_AUDIO_EXTENSIONS)
cached_path = cache_audio_from_bytes(raw_bytes, ext=ext)
logger.info("[Feishu] Cached message audio resource at %s", cached_path)
return cached_path, (media_type or f"audio/{ext.lstrip('.') or 'ogg'}")
if media_type.startswith("video/"):
if not Path(filename).suffix:
filename = f"{filename}.mp4"
cached_path = cache_document_from_bytes(raw_bytes, filename)
logger.info("[Feishu] Cached message video resource at %s", cached_path)
return cached_path, media_type
if not Path(filename).suffix and media_type in _DOCUMENT_MIME_TO_EXT:
filename = f"{filename}{_DOCUMENT_MIME_TO_EXT[media_type]}"
cached_path = cache_document_from_bytes(raw_bytes, filename)
logger.info("[Feishu] Cached message document resource at %s", cached_path)
return cached_path, (media_type or self._guess_document_media_type(filename))
except Exception:
logger.warning(
"[Feishu] Failed to cache message resource %s/%s",
message_id,
file_key,
exc_info=True,
)
return "", ""
# =========================================================================
# Static helpers — extension / media-type guessing
# =========================================================================
@staticmethod
def _read_binary_response(response: Any) -> bytes:
file_obj = getattr(response, "file", None)
if file_obj is None:
return b""
if hasattr(file_obj, "getvalue"):
return bytes(file_obj.getvalue())
return bytes(file_obj.read())
@staticmethod
def _get_response_header(response: Any, name: str) -> str:
raw = getattr(response, "raw", None)
headers = getattr(raw, "headers", {}) or {}
return str(headers.get(name, headers.get(name.lower(), "")) or "").split(";", 1)[0].strip().lower()
@staticmethod
def _guess_extension(filename: str, content_type: str, default: str, *, allowed: set[str]) -> str:
ext = Path(filename or "").suffix.lower()
if ext in allowed:
return ext
guessed = mimetypes.guess_extension((content_type or "").split(";", 1)[0].strip().lower() or "")
if guessed in allowed:
return guessed
return default
@staticmethod
def _normalize_media_type(content_type: str, *, default: str) -> str:
normalized = (content_type or "").split(";", 1)[0].strip().lower()
return normalized or default
@staticmethod
def _guess_document_media_type(filename: str) -> str:
ext = Path(filename or "").suffix.lower()
return SUPPORTED_DOCUMENT_TYPES.get(ext, mimetypes.guess_type(filename or "")[0] or "application/octet-stream")
@staticmethod
def _display_name_from_cached_path(path: str) -> str:
basename = os.path.basename(path)
parts = basename.split("_", 2)
display_name = parts[2] if len(parts) >= 3 else basename
return re.sub(r"[^\w.\- ]", "_", display_name)
@staticmethod
def _guess_media_type_from_filename(filename: str) -> str:
guessed = (mimetypes.guess_type(filename or "")[0] or "").lower()
if guessed:
return guessed
ext = Path(filename or "").suffix.lower()
if ext in _VIDEO_EXTENSIONS:
return f"video/{ext.lstrip('.')}"
if ext in _AUDIO_EXTENSIONS:
return f"audio/{ext.lstrip('.')}"
if ext in _IMAGE_EXTENSIONS:
return FeishuAdapter._default_image_media_type(ext)
return ""
@staticmethod
def _map_chat_type(raw_chat_type: str) -> str:
normalized = (raw_chat_type or "").strip().lower()
if normalized == "p2p":
return "dm"
if "topic" in normalized or "thread" in normalized or "forum" in normalized:
return "forum"
if normalized == "group":
return "group"
return "dm"
@staticmethod
def _resolve_source_chat_type(*, chat_info: Dict[str, Any], event_chat_type: str) -> str:
resolved = str(chat_info.get("type") or "").strip().lower()
if resolved in {"group", "forum"}:
return resolved
if event_chat_type == "p2p":
return "dm"
return "group"
async def _resolve_sender_profile(self, sender_id: Any) -> Dict[str, Optional[str]]:
open_id = getattr(sender_id, "open_id", None) or None
user_id = getattr(sender_id, "user_id", None) or None
union_id = getattr(sender_id, "union_id", None) or None
primary_id = open_id or user_id
display_name = await self._resolve_sender_name_from_api(primary_id or union_id)
return {
"user_id": primary_id,
"user_name": display_name,
"user_id_alt": union_id,
}
async def _resolve_sender_name_from_api(self, sender_id: Optional[str]) -> Optional[str]:
"""Fetch the sender's display name from the Feishu contact API with a 10-minute cache.
ID-type detection mirrors openclaw: ou_ → open_id, on_ → union_id, else user_id.
Failures are silently suppressed; the message pipeline must not block on name resolution.
"""
if not sender_id or not self._client:
return None
trimmed = sender_id.strip()
if not trimmed:
return None
now = time.time()
cached = self._sender_name_cache.get(trimmed)
if cached is not None:
name, expire_at = cached
if now < expire_at:
return name
try:
from lark_oapi.api.contact.v3 import GetUserRequest # lazy import
if trimmed.startswith("ou_"):
id_type = "open_id"
elif trimmed.startswith("on_"):
id_type = "union_id"
else:
id_type = "user_id"
request = GetUserRequest.builder().user_id(trimmed).user_id_type(id_type).build()
response = await asyncio.to_thread(self._client.contact.v3.user.get, request)
if not response or not response.success():
return None
user = getattr(getattr(response, "data", None), "user", None)
name = (
getattr(user, "name", None)
or getattr(user, "display_name", None)
or getattr(user, "nickname", None)
or getattr(user, "en_name", None)
)
if name and isinstance(name, str):
name = name.strip()
if name:
self._sender_name_cache[trimmed] = (name, now + _FEISHU_SENDER_NAME_TTL_SECONDS)
return name
except Exception:
logger.debug("[Feishu] Failed to resolve sender name for %s", sender_id, exc_info=True)
return None
async def _fetch_message_text(self, message_id: str) -> Optional[str]:
if not self._client or not message_id:
return None
if message_id in self._message_text_cache:
return self._message_text_cache[message_id]
try:
request = self._build_get_message_request(message_id)
response = await asyncio.to_thread(self._client.im.v1.message.get, request)
if not response or getattr(response, "success", lambda: False)() is False:
code = getattr(response, "code", "unknown")
msg = getattr(response, "msg", "message lookup failed")
logger.warning("[Feishu] Failed to fetch parent message %s: [%s] %s", message_id, code, msg)
return None
items = getattr(getattr(response, "data", None), "items", None) or []
parent = items[0] if items else None
body = getattr(parent, "body", None)
msg_type = getattr(parent, "msg_type", "") or ""
raw_content = getattr(body, "content", "") or ""
text = self._extract_text_from_raw_content(msg_type=msg_type, raw_content=raw_content)
self._message_text_cache[message_id] = text
return text
except Exception:
logger.warning("[Feishu] Failed to fetch parent message %s", message_id, exc_info=True)
return None
def _extract_text_from_raw_content(self, *, msg_type: str, raw_content: str) -> Optional[str]:
normalized = normalize_feishu_message(message_type=msg_type, raw_content=raw_content)
if normalized.text_content:
return normalized.text_content
placeholder = normalized.metadata.get("placeholder_text") if isinstance(normalized.metadata, dict) else None
return str(placeholder).strip() or None
@staticmethod
def _default_image_media_type(ext: str) -> str:
normalized_ext = (ext or "").lower()
if normalized_ext in {".jpg", ".jpeg"}:
return "image/jpeg"
return f"image/{normalized_ext.lstrip('.') or 'jpeg'}"
@staticmethod
def _log_background_failure(future: Any) -> None:
try:
future.result()
except Exception:
logger.exception("[Feishu] Background inbound processing failed")
# =========================================================================
# Group policy and mention gating
# =========================================================================
def _allow_group_message(self, sender_id: Any) -> bool:
"""Current group policy gate for non-DM traffic."""
if self._group_policy == "disabled":
return False
sender_open_id = getattr(sender_id, "open_id", None) or getattr(sender_id, "user_id", None)
if self._group_policy == "open":
return True
return bool(sender_open_id and sender_open_id in self._allowed_group_users)
def _should_accept_group_message(self, message: Any, sender_id: Any) -> bool:
"""Require an explicit @mention before group messages enter the agent."""
if not self._allow_group_message(sender_id):
return False
# @_all is Feishu's @everyone placeholder — always route to the bot.
raw_content = getattr(message, "content", "") or ""
if "@_all" in raw_content:
return True
mentions = getattr(message, "mentions", None) or []
if mentions:
return self._message_mentions_bot(mentions)
normalized = normalize_feishu_message(
message_type=getattr(message, "message_type", "") or "",
raw_content=raw_content,
)
if normalized.mentioned_ids:
return self._post_mentions_bot(normalized.mentioned_ids)
return False
def _message_mentions_bot(self, mentions: List[Any]) -> bool:
"""Check whether any mention targets the configured or inferred bot identity."""
for mention in mentions:
mention_id = getattr(mention, "id", None)
mention_open_id = getattr(mention_id, "open_id", None)
mention_user_id = getattr(mention_id, "user_id", None)
mention_name = (getattr(mention, "name", None) or "").strip()
if self._bot_open_id and mention_open_id == self._bot_open_id:
return True
if self._bot_user_id and mention_user_id == self._bot_user_id:
return True
if self._bot_name and mention_name == self._bot_name:
return True
return False
def _post_mentions_bot(self, mentioned_ids: List[str]) -> bool:
if not mentioned_ids:
return False
if self._bot_open_id and self._bot_open_id in mentioned_ids:
return True
if self._bot_user_id and self._bot_user_id in mentioned_ids:
return True
return False
async def _hydrate_bot_identity(self) -> None:
"""Best-effort discovery of bot identity for precise group mention gating."""
if not self._client:
return
if any((self._bot_open_id, self._bot_user_id, self._bot_name)):
return
try:
request = self._build_get_application_request(app_id=self._app_id, lang="en_us")
response = await asyncio.to_thread(self._client.application.v6.application.get, request)
if not response or not response.success():
code = getattr(response, "code", None)
if code == 99991672:
logger.warning(
"[Feishu] Unable to hydrate bot identity from application info. "
"Grant admin:app.info:readonly or application:application:self_manage "
"so group @mention gating can resolve the bot name precisely."
)
return
app = getattr(getattr(response, "data", None), "app", None)
app_name = (getattr(app, "app_name", None) or "").strip()
if app_name:
self._bot_name = app_name
except Exception:
logger.debug("[Feishu] Failed to hydrate bot identity", exc_info=True)
# =========================================================================
# Deduplication — seen message ID cache (persistent)
# =========================================================================
def _load_seen_message_ids(self) -> None:
try:
payload = json.loads(self._dedup_state_path.read_text(encoding="utf-8"))
except FileNotFoundError:
return
except (OSError, json.JSONDecodeError):
logger.warning("[Feishu] Failed to load persisted dedup state from %s", self._dedup_state_path, exc_info=True)
return
seen_data = payload.get("message_ids", {}) if isinstance(payload, dict) else {}
now = time.time()
ttl = _FEISHU_DEDUP_TTL_SECONDS
# Backward-compat: old format stored a plain list of IDs (no timestamps).
if isinstance(seen_data, list):
entries: Dict[str, float] = {str(item).strip(): 0.0 for item in seen_data if str(item).strip()}
elif isinstance(seen_data, dict):
entries = {k: float(v) for k, v in seen_data.items() if isinstance(k, str) and k.strip()}
else:
return
# Filter out TTL-expired entries (entries saved with ts=0.0 are treated as immortal
# for one migration cycle to avoid nuking old data on first upgrade).
valid: Dict[str, float] = {
msg_id: ts for msg_id, ts in entries.items()
if ts == 0.0 or ttl <= 0 or now - ts < ttl
}
# Apply size cap; keep the most recently seen IDs.
sorted_ids = sorted(valid, key=lambda k: valid[k], reverse=True)[:self._dedup_cache_size]
self._seen_message_order = list(reversed(sorted_ids))
self._seen_message_ids = {k: valid[k] for k in sorted_ids}
def _persist_seen_message_ids(self) -> None:
try:
self._dedup_state_path.parent.mkdir(parents=True, exist_ok=True)
recent = self._seen_message_order[-self._dedup_cache_size:]
# Save as {msg_id: timestamp} so TTL filtering works across restarts.
payload = {"message_ids": {k: self._seen_message_ids[k] for k in recent if k in self._seen_message_ids}}
self._dedup_state_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
except OSError:
logger.warning("[Feishu] Failed to persist dedup state to %s", self._dedup_state_path, exc_info=True)
def _is_duplicate(self, message_id: str) -> bool:
now = time.time()
ttl = _FEISHU_DEDUP_TTL_SECONDS
with self._dedup_lock:
seen_at = self._seen_message_ids.get(message_id)
if seen_at is not None and (ttl <= 0 or now - seen_at < ttl):
return True
# Record with current wall-clock timestamp so TTL works across restarts.
self._seen_message_ids[message_id] = now
self._seen_message_order.append(message_id)
while len(self._seen_message_order) > self._dedup_cache_size:
stale = self._seen_message_order.pop(0)
self._seen_message_ids.pop(stale, None)
self._persist_seen_message_ids()
return False
# =========================================================================
# Outbound payload construction and send pipeline
# =========================================================================
def _build_outbound_payload(self, content: str) -> tuple[str, str]:
if _MARKDOWN_HINT_RE.search(content):
return "post", _build_markdown_post_payload(content)
text_payload = {"text": content}
return "text", json.dumps(text_payload, ensure_ascii=False)
async def _send_uploaded_file_message(
self,
*,
chat_id: str,
file_path: str,
reply_to: Optional[str],
metadata: Optional[Dict[str, Any]],
caption: Optional[str] = None,
file_name: Optional[str] = None,
outbound_message_type: str = "file",
) -> SendResult:
if not self._client:
return SendResult(success=False, error="Not connected")
if not os.path.exists(file_path):
return SendResult(success=False, error=f"File not found: {file_path}")
display_name = file_name or os.path.basename(file_path)
upload_file_type, resolved_message_type = self._resolve_outbound_file_routing(
file_path=display_name,
requested_message_type=outbound_message_type,
)
try:
with open(file_path, "rb") as file_obj:
body = self._build_file_upload_body(
file_type=upload_file_type,
file_name=display_name,
file=file_obj,
)
request = self._build_file_upload_request(body)
upload_response = await asyncio.to_thread(self._client.im.v1.file.create, request)
file_key = self._extract_response_field(upload_response, "file_key")
if not file_key:
return self._response_error_result(
upload_response,
default_message="file upload failed",
override_error="Feishu file upload missing file_key",
)
if caption:
media_tag = {
"tag": "media",
"file_key": file_key,
"file_name": display_name,
}
message_response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type="post",
payload=self._build_media_post_payload(caption=caption, media_tag=media_tag),
reply_to=reply_to,
metadata=metadata,
)
else:
message_response = await self._feishu_send_with_retry(
chat_id=chat_id,
msg_type=resolved_message_type,
payload=json.dumps({"file_key": file_key}, ensure_ascii=False),
reply_to=reply_to,
metadata=metadata,
)
return self._finalize_send_result(message_response, "file send failed")
except Exception as exc:
logger.error("[Feishu] Failed to send file %s: %s", file_path, exc, exc_info=True)
return SendResult(success=False, error=str(exc))
async def _send_raw_message(
self,
*,
chat_id: str,
msg_type: str,
payload: str,
reply_to: Optional[str],
metadata: Optional[Dict[str, Any]],
) -> Any:
reply_in_thread = bool((metadata or {}).get("thread_id"))
if reply_to:
body = self._build_reply_message_body(
content=payload,
msg_type=msg_type,
reply_in_thread=reply_in_thread,
uuid_value=str(uuid.uuid4()),
)
request = self._build_reply_message_request(reply_to, body)
return await asyncio.to_thread(self._client.im.v1.message.reply, request)
body = self._build_create_message_body(
receive_id=chat_id,
msg_type=msg_type,
content=payload,
uuid_value=str(uuid.uuid4()),
)
request = self._build_create_message_request("chat_id", body)
return await asyncio.to_thread(self._client.im.v1.message.create, request)
@staticmethod
def _response_succeeded(response: Any) -> bool:
return bool(response and getattr(response, "success", lambda: False)())
@staticmethod
def _extract_response_field(response: Any, field_name: str) -> Any:
if not FeishuAdapter._response_succeeded(response):
return None
data = getattr(response, "data", None)
return getattr(data, field_name, None) if data else None
def _response_error_result(
self,
response: Any,
*,
default_message: str,
override_error: Optional[str] = None,
) -> SendResult:
if override_error:
return SendResult(success=False, error=override_error, raw_response=response)
code = getattr(response, "code", "unknown")
msg = getattr(response, "msg", default_message)
return SendResult(success=False, error=f"[{code}] {msg}", raw_response=response)
def _finalize_send_result(self, response: Any, default_message: str) -> SendResult:
if not self._response_succeeded(response):
return self._response_error_result(response, default_message=default_message)
return SendResult(
success=True,
message_id=self._extract_response_field(response, "message_id"),
raw_response=response,
)
# =========================================================================
# Connection internals — websocket / webhook setup
# =========================================================================
async def _connect_with_retry(self) -> None:
for attempt in range(_FEISHU_CONNECT_ATTEMPTS):
try:
if self._connection_mode == "websocket":
await self._connect_websocket()
else:
await self._connect_webhook()
return
except Exception as exc:
self._running = False
self._disable_websocket_auto_reconnect()
self._ws_future = None
await self._stop_webhook_server()
if attempt >= _FEISHU_CONNECT_ATTEMPTS - 1:
raise
wait_seconds = 2 ** attempt
logger.warning(
"[Feishu] Connect attempt %d/%d failed; retrying in %ds: %s",
attempt + 1,
_FEISHU_CONNECT_ATTEMPTS,
wait_seconds,
exc,
)
await asyncio.sleep(wait_seconds)
async def _connect_websocket(self) -> None:
if not FEISHU_WEBSOCKET_AVAILABLE:
raise RuntimeError("websockets not installed; websocket mode unavailable")
domain = FEISHU_DOMAIN if self._domain_name != "lark" else LARK_DOMAIN
self._client = self._build_lark_client(domain)
await self._hydrate_bot_identity()
self._ws_client = FeishuWSClient(
app_id=self._app_id,
app_secret=self._app_secret,
log_level=lark.LogLevel.INFO,
event_handler=self._event_handler,
domain=domain,
)
self._ws_future = self._loop.run_in_executor(
None,
_run_official_feishu_ws_client,
self._ws_client,
)
async def _connect_webhook(self) -> None:
if not FEISHU_WEBHOOK_AVAILABLE:
raise RuntimeError("aiohttp not installed; webhook mode unavailable")
domain = FEISHU_DOMAIN if self._domain_name != "lark" else LARK_DOMAIN
self._client = self._build_lark_client(domain)
await self._hydrate_bot_identity()
app = web.Application()
app.router.add_post(self._webhook_path, self._handle_webhook_request)
self._webhook_runner = web.AppRunner(app)
await self._webhook_runner.setup()
self._webhook_site = web.TCPSite(self._webhook_runner, self._webhook_host, self._webhook_port)
await self._webhook_site.start()
def _build_lark_client(self, domain: Any) -> Any:
return (
lark.Client.builder()
.app_id(self._app_id)
.app_secret(self._app_secret)
.domain(domain)
.log_level(lark.LogLevel.WARNING)
.build()
)
async def _feishu_send_with_retry(
self,
*,
chat_id: str,
msg_type: str,
payload: str,
reply_to: Optional[str],
metadata: Optional[Dict[str, Any]],
) -> Any:
last_error: Optional[Exception] = None
active_reply_to = reply_to
for attempt in range(_FEISHU_SEND_ATTEMPTS):
try:
response = await self._send_raw_message(
chat_id=chat_id,
msg_type=msg_type,
payload=payload,
reply_to=active_reply_to,
metadata=metadata,
)
# If replying to a message failed because it was withdrawn or not found,
# fall back to posting a new message directly to the chat.
if active_reply_to and not self._response_succeeded(response):
code = getattr(response, "code", None)
if code in _FEISHU_REPLY_FALLBACK_CODES:
logger.warning(
"[Feishu] Reply to %s failed (code %s — message withdrawn/missing); "
"falling back to new message in chat %s",
active_reply_to,
code,
chat_id,
)
active_reply_to = None
response = await self._send_raw_message(
chat_id=chat_id,
msg_type=msg_type,
payload=payload,
reply_to=None,
metadata=metadata,
)
return response
except Exception as exc:
last_error = exc
if msg_type == "post" and _POST_CONTENT_INVALID_RE.search(str(exc)):
raise
if attempt >= _FEISHU_SEND_ATTEMPTS - 1:
raise
wait_seconds = 2 ** attempt
logger.warning(
"[Feishu] Send attempt %d/%d failed for chat %s; retrying in %ds: %s",
attempt + 1,
_FEISHU_SEND_ATTEMPTS,
chat_id,
wait_seconds,
exc,
)
await asyncio.sleep(wait_seconds)
raise last_error or RuntimeError("Feishu send failed")
async def _release_app_lock(self) -> None:
if not self._app_lock_identity:
return
try:
release_scoped_lock(_FEISHU_APP_LOCK_SCOPE, self._app_lock_identity)
except Exception as exc:
logger.warning("[Feishu] Failed to release app lock: %s", exc, exc_info=True)
finally:
self._app_lock_identity = None
# =========================================================================
# Lark API request builders
# =========================================================================
@staticmethod
def _build_get_chat_request(chat_id: str) -> Any:
if "GetChatRequest" in globals():
return GetChatRequest.builder().chat_id(chat_id).build()
return SimpleNamespace(chat_id=chat_id)
@staticmethod
def _build_get_message_request(message_id: str) -> Any:
if "GetMessageRequest" in globals():
return GetMessageRequest.builder().message_id(message_id).build()
return SimpleNamespace(message_id=message_id)
@staticmethod
def _build_message_resource_request(*, message_id: str, file_key: str, resource_type: str) -> Any:
if "GetMessageResourceRequest" in globals():
return (
GetMessageResourceRequest.builder()
.message_id(message_id)
.file_key(file_key)
.type(resource_type)
.build()
)
return SimpleNamespace(message_id=message_id, file_key=file_key, type=resource_type)
@staticmethod
def _build_get_application_request(*, app_id: str, lang: str) -> Any:
if "GetApplicationRequest" in globals():
return (
GetApplicationRequest.builder()
.app_id(app_id)
.lang(lang)
.build()
)
return SimpleNamespace(app_id=app_id, lang=lang)
@staticmethod
def _build_reply_message_body(*, content: str, msg_type: str, reply_in_thread: bool, uuid_value: str) -> Any:
if "ReplyMessageRequestBody" in globals():
return (
ReplyMessageRequestBody.builder()
.content(content)
.msg_type(msg_type)
.reply_in_thread(reply_in_thread)
.uuid(uuid_value)
.build()
)
return SimpleNamespace(
content=content,
msg_type=msg_type,
reply_in_thread=reply_in_thread,
uuid=uuid_value,
)
@staticmethod
def _build_reply_message_request(message_id: str, request_body: Any) -> Any:
if "ReplyMessageRequest" in globals():
return (
ReplyMessageRequest.builder()
.message_id(message_id)
.request_body(request_body)
.build()
)
return SimpleNamespace(message_id=message_id, request_body=request_body)
@staticmethod
def _build_update_message_body(*, msg_type: str, content: str) -> Any:
if "UpdateMessageRequestBody" in globals():
return (
UpdateMessageRequestBody.builder()
.msg_type(msg_type)
.content(content)
.build()
)
return SimpleNamespace(msg_type=msg_type, content=content)
@staticmethod
def _build_update_message_request(message_id: str, request_body: Any) -> Any:
if "UpdateMessageRequest" in globals():
return (
UpdateMessageRequest.builder()
.message_id(message_id)
.request_body(request_body)
.build()
)
return SimpleNamespace(message_id=message_id, request_body=request_body)
@staticmethod
def _build_create_message_body(*, receive_id: str, msg_type: str, content: str, uuid_value: str) -> Any:
if "CreateMessageRequestBody" in globals():
return (
CreateMessageRequestBody.builder()
.receive_id(receive_id)
.msg_type(msg_type)
.content(content)
.uuid(uuid_value)
.build()
)
return SimpleNamespace(
receive_id=receive_id,
msg_type=msg_type,
content=content,
uuid=uuid_value,
)
@staticmethod
def _build_create_message_request(receive_id_type: str, request_body: Any) -> Any:
if "CreateMessageRequest" in globals():
return (
CreateMessageRequest.builder()
.receive_id_type(receive_id_type)
.request_body(request_body)
.build()
)
return SimpleNamespace(receive_id_type=receive_id_type, request_body=request_body)
@staticmethod
def _build_image_upload_body(*, image_type: str, image: Any) -> Any:
if "CreateImageRequestBody" in globals():
return (
CreateImageRequestBody.builder()
.image_type(image_type)
.image(image)
.build()
)
return SimpleNamespace(image_type=image_type, image=image)
@staticmethod
def _build_image_upload_request(request_body: Any) -> Any:
if "CreateImageRequest" in globals():
return CreateImageRequest.builder().request_body(request_body).build()
return SimpleNamespace(request_body=request_body)
@staticmethod
def _build_file_upload_body(*, file_type: str, file_name: str, file: Any) -> Any:
if "CreateFileRequestBody" in globals():
return (
CreateFileRequestBody.builder()
.file_type(file_type)
.file_name(file_name)
.file(file)
.build()
)
return SimpleNamespace(file_type=file_type, file_name=file_name, file=file)
@staticmethod
def _build_file_upload_request(request_body: Any) -> Any:
if "CreateFileRequest" in globals():
return CreateFileRequest.builder().request_body(request_body).build()
return SimpleNamespace(request_body=request_body)
def _build_post_payload(self, content: str) -> str:
return _build_markdown_post_payload(content)
def _build_media_post_payload(self, *, caption: str, media_tag: Dict[str, str]) -> str:
payload = json.loads(self._build_post_payload(caption))
content = payload.setdefault("zh_cn", {}).setdefault("content", [])
content.append([media_tag])
return json.dumps(payload, ensure_ascii=False)
@staticmethod
def _resolve_outbound_file_routing(
*,
file_path: str,
requested_message_type: str,
) -> tuple[str, str]:
ext = Path(file_path).suffix.lower()
if ext in _FEISHU_OPUS_UPLOAD_EXTENSIONS:
return "opus", "audio"
if ext in _FEISHU_MEDIA_UPLOAD_EXTENSIONS:
return "mp4", "media"
if ext in _FEISHU_DOC_UPLOAD_TYPES:
return _FEISHU_DOC_UPLOAD_TYPES[ext], "file"
if requested_message_type == "file":
return _FEISHU_FILE_UPLOAD_TYPE, "file"
return _FEISHU_FILE_UPLOAD_TYPE, "file"