kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(feishu): 清理冗余代码并补全飞书工具集实现

Browse Source 
1.  移除多个文件中的空行、冗余导入
2.  修复文件末尾缺少换行符的问题
3.  新增并补全飞书多类工具API实现：
    - 多维表格：更新、删除记录，列出视图
    - 文档：更新、追加、删除块
    - 云文档：上传、下载文件
    - 群组：创建、添加成员、更新信息、创建公告
    - 目录：重构用户部门缓存逻辑
4.  优化消息发送、回复、转发等API的错误处理和逻辑
5.  新增消息列表查询、已读状态查询等功能
This commit is contained in:
Kris 2026-05-13 16:07:59 +08:00
parent 9ab904c5bd
commit 69b7139863
29 changed files with 1251 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/package/yuxi/channels/adapters/feishu/accounts.py
View File

@ -187,4 +187,4 @@ class FeishuAccountManager:
if account is None:
return "unknown: no account configured"
status = "active" if account.enabled else "disabled"
return f"{account.name}: {status} (platform={account.platform})"
return f"{account.name}: {status} (platform={account.platform})"

431
backend/package/yuxi/channels/adapters/feishu/adapter.py
View File

@ -8,11 +8,12 @@ from typing import Any, ClassVar
from yuxi.channels.base import BaseChannelAdapter
from yuxi.channels.capabilities import ChannelCapabilities
from yuxi.channels.meta import ChannelMeta
from yuxi.channels.exceptions import (
ChannelAuthenticationError,
ChannelConnectionError,
)
from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError
from yuxi.channels.meta import ChannelMeta
from yuxi.channels.models import (
ChannelMessage,
ChannelResponse,
@ -26,41 +27,66 @@ from yuxi.channels.registry import register_builtin_adapter
from yuxi.utils.datetime_utils import utc_now_naive
from yuxi.utils.logging_config import logger
from .cards import TEXT_CHUNK_LIMIT
from .accounts import FeishuAccountManager
from .approval import FeishuApprovalAdapter
from .card_action import decode_card_action, validate_card_context
from .cards import TEXT_CHUNK_LIMIT
from .chat_cache import ChatNameCache
from .client import FeishuClientCache
from .commands import build_synthetic_message, parse_bot_menu_event
from .comment_handler import handle_comment_event
from .dedup import FeishuDedupStore
from .directory import list_groups as _list_groups_feishu, list_peers as _list_peers_feishu
from .directory import list_groups as _list_groups_feishu
from .directory import list_peers as _list_peers_feishu
from .dynamic_agent import FeishuDynamicAgentConfig, FeishuDynamicAgentManager
from .card_action import decode_card_action, validate_card_context
from .formatter import format_outbound
from .media import (
download_media as _download_media,
)
from .media import (
upload_file,
upload_image,
)
from .normalizer import normalize_inbound
from .pins import create_pin as _create_pin, list_pins as _list_pins, remove_pin as _remove_pin
from .oauth import TokenBackoffManager
from .pins import create_pin as _create_pin
from .pins import list_pins as _list_pins
from .pins import remove_pin as _remove_pin
from .reactions import (
clear_all_bot_reactions,
)
from .reactions import (
list_reactions as _list_reactions,
)
from .reactions import (
remove_reaction as _remove_reaction,
)
from .reply_dispatcher import dispatch_render, extract_urls
from .secret_resolver import resolve_secret
from .send import (
forward_message as _forward_message,
)
from .send import (
get_message_read_status as _get_message_read_status,
)
from .send import (
is_local_image_path,
read_message as _read_message,
reply_message as _reply_message,
send_card,
send_reaction as _send_reaction,
send_text,
update_card_message,
)
from .send import (
list_messages as _list_messages,
)
from .send import (
read_message as _read_message,
)
from .send import (
reply_message as _reply_message,
)
from .send import (
send_reaction as _send_reaction,
)
from .sequential import FeishuSequentialQueue
from .session import CHAT_ID_PREFIX_DM, CHAT_ID_PREFIX_GROUP
from .stream import STREAM_START_BACKOFF_S, FeishuStreamingSession
@ -96,6 +122,7 @@ class FeishuAdapter(BaseChannelAdapter):
edit=True,
unsend=True,
reply=True,
forward=True,
media=True,
threads=True,
pin=True,
@ -131,6 +158,7 @@ class FeishuAdapter(BaseChannelAdapter):
self._ws_tasks: dict[str, asyncio.Task] = {}
self._account_clients: dict[str, Any] = {}
self._multi_account_mode: bool = False
self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60, channel_id="feishu")
self._token_refresh_task: asyncio.Task | None = None
self._connected_at: float | None = None
self._token_expire_at: float = 0
@ -164,56 +192,28 @@ class FeishuAdapter(BaseChannelAdapter):
self._group_session_scope: str = "group"
self._reactions_enabled: bool = True
self._tool_enabled: dict[str, bool] = {}
self._401_backoff_until: float = 0
self._401_retry_count: int = 0
self._401_max_retries: int = 3
self._401_base_delay: float = 1.0
self._401_max_delay: float = 30.0
self._401_backoff = TokenBackoffManager(max_retries=3, base_delay=1.0, max_delay=30.0)
self._token_refresh_lock = asyncio.Lock()
async def _handle_401_backoff(self) -> None:
now = time.monotonic()
if self._401_backoff_until > now:
wait = self._401_backoff_until - now
logger.warning(f"[Feishu] 401 backoff active, waiting {wait:.1f}s")
await asyncio.sleep(wait)
await self._401_backoff.wait_if_backing_off()
async def _on_401_error(self) -> bool:
self._401_retry_count += 1
if self._401_retry_count > self._401_max_retries:
logger.error(f"[Feishu] 401 retry count exceeded ({self._401_max_retries})")
if not self._401_backoff.record_error():
return False
delay = min(self._401_base_delay * (2 ** (self._401_retry_count - 1)), self._401_max_delay)
self._401_backoff_until = time.monotonic() + delay
logger.warning(
f"[Feishu] 401 error, refreshing token and retrying in {delay:.1f}s (attempt {self._401_retry_count})"
)
async with self._token_refresh_lock:
if self._token_expire_at > time.time() + self._token_refresh_margin:
return True
try:
await self._refresh_token()
except Exception as e:
logger.error(f"[Feishu] Token refresh after 401 failed: {e}")
return False
return True
try:
return await self._refresh_token()
except Exception as e:
logger.error(f"[Feishu] Token refresh after 401 failed: {e}")
return False
def _reset_401_backoff(self) -> None:
if self._401_retry_count > 0:
logger.info("[Feishu] 401 backoff reset after successful request")
self._401_retry_count = 0
self._401_backoff_until = 0
async def _call_with_401_retry(self, callable_fn, *args, **kwargs) -> Any:
result = await callable_fn(*args, **kwargs)
if isinstance(result, DeliveryResult) and result.error_code == "auth_expired":
if await self._on_401_error():
result = await callable_fn(*args, **kwargs)
if isinstance(result, DeliveryResult) and not result.success:
return result
else:
return result
self._reset_401_backoff()
return result
self._401_backoff.reset()
async def connect(self) -> None:
if self._status == ChannelStatus.CONNECTED:
@ -285,7 +285,15 @@ class FeishuAdapter(BaseChannelAdapter):
if not token_resp.success():
raise ChannelAuthenticationError(f"Token acquisition failed: {token_resp.msg}")
expire = token_resp.data.get("expire", 0) if hasattr(token_resp, "data") else 0
data = token_resp.data if hasattr(token_resp, "data") else {}
token_value = data.get("tenant_access_token", "")
if not token_value:
raise ChannelAuthenticationError("Token acquisition succeeded but returned empty token value")
expire = data.get("expire", 0)
if expire <= 0:
raise ChannelAuthenticationError(f"Token acquisition returned invalid expire: {expire}")
self._token_expire_at = time.time() + expire
logger.info(f"[Feishu] Token acquired, expire in {expire}s")
@ -367,78 +375,91 @@ class FeishuAdapter(BaseChannelAdapter):
await self._handle_401_backoff()
payload = self.format_outbound(response)
chat_id = response.identity.channel_chat_id
chat_type = (
response.identity.chat_type.value if isinstance(response.identity.chat_type, ChatType) else "private"
)
reply_to_msg_id = response.identity.metadata.get("reply_to_msg_id", "") if response.identity.metadata else ""
root_id = response.identity.metadata.get("root_id", "") if response.identity.metadata else ""
if reply_to_msg_id:
self._stream_context[chat_id] = {
"reply_to_msg_id": reply_to_msg_id,
"root_id": root_id or reply_to_msg_id,
}
content = payload.get("content", "")
buttons = payload.get("buttons")
thread_id = payload.get("thread_id")
template = payload.get("template")
tone = payload.get("tone")
use_post_format = payload.get("use_post_format", False)
if image_path := is_local_image_path(content):
try:
with open(image_path, "rb") as f:
image_data = f.read()
except OSError as e:
return DeliveryResult(success=False, error=f"Failed to read image file: {e}")
image_key = await upload_image(self._lark_client, image_data)
result = await self._send_image_msg(chat_id, image_key)
return self._check_401_result(result)
if chat_type in ("group", "thread") and reply_to_msg_id:
result = await _reply_message(
self._lark_client,
reply_to_msg_id,
content,
buttons=buttons,
thread_id=thread_id or (reply_to_msg_id if self._reply_in_thread else None),
use_post_format=use_post_format,
async def _do_send() -> DeliveryResult:
payload = self.format_outbound(response)
chat_id = response.identity.channel_chat_id
chat_type = (
response.identity.chat_type.value if isinstance(response.identity.chat_type, ChatType) else "private"
)
return self._check_401_result(result)
identity_meta = response.identity.metadata or {}
reply_to_msg_id = identity_meta.get("reply_to_msg_id", "")
root_id = identity_meta.get("root_id", "")
render_mode = dispatch_render(content, buttons=buttons, render_mode=self._render_mode)
if render_mode == "card":
url_unfurl = extract_urls(content)
result = await send_card(
self._lark_client,
chat_id,
content,
chat_type=chat_type,
buttons=buttons,
thread_id=thread_id,
template=template,
tone=tone,
url_unfurl=url_unfurl if url_unfurl else None,
)
else:
result = await send_text(
self._lark_client,
chat_id,
content,
chat_type=chat_type,
thread_id=thread_id,
use_post_format=use_post_format,
)
return self._check_401_result(result)
if reply_to_msg_id:
self._stream_context[chat_id] = {
"reply_to_msg_id": reply_to_msg_id,
"root_id": root_id or reply_to_msg_id,
}
content = payload.get("content", "")
buttons = payload.get("buttons")
thread_id = payload.get("thread_id")
template = payload.get("template")
tone = payload.get("tone")
use_post_format = payload.get("use_post_format", False)
if image_path := is_local_image_path(content):
try:
with open(image_path, "rb") as f:
image_data = f.read()
except OSError as e:
raise ChannelConnectionError(f"Failed to read image file: {e}")
image_key = await upload_image(self._lark_client, image_data)
result = await self._send_image_msg(chat_id, image_key)
return result
if chat_type in ("group", "thread") and reply_to_msg_id:
result = await _reply_message(
self._lark_client,
reply_to_msg_id,
content,
buttons=buttons,
thread_id=thread_id or (reply_to_msg_id if self._reply_in_thread else None),
use_post_format=use_post_format,
chat_type=chat_type,
chat_id=chat_id,
)
return result
render_mode = dispatch_render(content, buttons=buttons, render_mode=self._render_mode)
if render_mode == "card":
url_unfurl = extract_urls(content)
result = await send_card(
self._lark_client,
chat_id,
content,
chat_type=chat_type,
buttons=buttons,
thread_id=thread_id,
template=template,
tone=tone,
url_unfurl=url_unfurl if url_unfurl else None,
)
else:
result = await send_text(
self._lark_client,
chat_id,
content,
chat_type=chat_type,
thread_id=thread_id,
use_post_format=use_post_format,
)
return result
try:
result = await self._circuit_breaker.call(_do_send)
except CircuitBreakerOpenError:
return DeliveryResult(success=False, error="Circuit breaker open")
def _check_401_result(self, result: DeliveryResult) -> DeliveryResult:
if not result.success and result.error_code == "auth_expired":
asyncio.create_task(self._on_401_error())
if await self._on_401_error():
try:
result = await self._circuit_breaker.call(_do_send)
except CircuitBreakerOpenError:
return DeliveryResult(success=False, error="Circuit breaker open")
else:
self._reset_401_backoff()
return result
async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult:
@ -511,6 +532,55 @@ class FeishuAdapter(BaseChannelAdapter):
return {}
return await _read_message(self._lark_client, msg_id)
async def list_messages(
self,
chat_id: str,
page_token: str = "",
page_size: int = 20,
start_time: str = "",
end_time: str = "",
) -> dict[str, Any]:
if not self._lark_client:
return {"messages": [], "has_more": False, "page_token": ""}
return await _list_messages(
self._lark_client,
chat_id,
page_token=page_token,
page_size=page_size,
start_time=start_time,
end_time=end_time,
)
async def forward_message(
self,
message_id: str,
target_chat_id: str,
target_type: str = "chat",
) -> DeliveryResult:
if not self._lark_client:
return DeliveryResult(success=False, error="Not connected")
return await _forward_message(
self._lark_client,
message_id,
target_chat_id,
target_type=target_type,
)
async def get_message_read_status(
self,
message_id: str,
page_token: str = "",
page_size: int = 20,
) -> dict[str, Any]:
if not self._lark_client:
return {"users": [], "has_more": False, "page_token": ""}
return await _get_message_read_status(
self._lark_client,
message_id,
page_token=page_token,
page_size=page_size,
)
async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult:
if not self._lark_client:
return DeliveryResult(success=False, error="Not connected")
@ -633,6 +703,8 @@ class FeishuAdapter(BaseChannelAdapter):
reply_to_msg_id,
content,
thread_id=root_id,
chat_type=chat_type,
chat_id=receive_id,
)
return await send_card(
self._lark_client,
@ -1044,35 +1116,54 @@ class FeishuAdapter(BaseChannelAdapter):
while self._status == ChannelStatus.CONNECTED:
remaining = self._token_expire_at - time.time()
if remaining <= 0:
logger.warning("[Feishu] Token already expired, skipping refresh")
break
delay = max(remaining - self._token_refresh_margin, 10)
logger.debug(f"[Feishu] Next token refresh in {delay:.0f}s")
logger.warning("[Feishu] Token expired, attempting immediate refresh")
else:
delay = max(remaining - self._token_refresh_margin, 10)
logger.debug(f"[Feishu] Next token refresh in {delay:.0f}s")
try:
await asyncio.sleep(delay)
except asyncio.CancelledError:
logger.info("[Feishu] Token refresh loop cancelled")
return
try:
await asyncio.sleep(delay)
await self._refresh_token()
if not await self._refresh_token():
logger.warning("[Feishu] Token refresh failed, retrying in 30s")
await asyncio.sleep(30)
except asyncio.CancelledError:
logger.info("[Feishu] Token refresh loop cancelled")
return
except Exception as e:
logger.error(f"[Feishu] Token refresh failed: {e}")
logger.error(f"[Feishu] Token refresh unexpected error: {e}, retrying in 30s")
await asyncio.sleep(30)
async def _refresh_token(self) -> None:
async def _refresh_token(self) -> bool:
if not self._lark_client:
return
return False
try:
resp = await asyncio.to_thread(self._lark_client.auth.tenant_access_token_internal)
except Exception as e:
logger.error(f"[Feishu] Token refresh call failed: {e}")
return
return False
if not resp.success():
logger.error(f"[Feishu] Token refresh failed: {resp.msg}")
return
return False
data = resp.data if hasattr(resp, "data") else {}
token_value = data.get("tenant_access_token", "")
if not token_value:
logger.error("[Feishu] Token refresh returned empty token value")
return False
expire = data.get("expire", 0)
if expire <= 0:
logger.error(f"[Feishu] Token refresh returned invalid expire: {expire}")
return False
expire = resp.data.get("expire", 0) if hasattr(resp, "data") else 0
self._token_expire_at = time.time() + expire
logger.info(f"[Feishu] Token refreshed, expire in {expire}s")
return True
@staticmethod
def _inject_event_loop_proxy() -> None:
@ -1278,3 +1369,87 @@ class FeishuAdapter(BaseChannelAdapter):
"open_id": token_data.get("open_id", ""),
"name": token_data.get("name", ""),
}
def list_account_ids(self) -> list[str]:
return self._accounts.list_account_ids()
def resolve_account(self, account_id: str | None = None) -> dict | None:
account = self._accounts.resolve_account(account_id)
if account is None:
return None
return {"account_id": account.account_id, "label": account.label, "enabled": account.enabled}
def default_account_id(self) -> str:
return self._accounts.default_account_id or ""
async def add_account(self, account_id: str, config: dict[str, Any]) -> bool:
if account_id in self._accounts.accounts:
logger.warning(f"[Feishu] Account '{account_id}' already exists")
return False
from .accounts import FeishuAccount
new_account = FeishuAccount(
account_id=account_id,
label=config.get("label", account_id),
app_id=config.get("app_id", ""),
app_secret=config.get("app_secret", ""),
verification_token=config.get("verification_token", ""),
encrypt_key=config.get("encrypt_key", ""),
bot_open_id=config.get("bot_open_id", ""),
enabled=True,
config=config,
)
self._accounts.accounts[account_id] = new_account
if not self._accounts.default_account_id:
self._accounts.default_account_id = account_id
logger.info(f"[Feishu] Added account '{account_id}'")
return True
async def remove_account(self, account_id: str) -> bool:
return self._accounts.delete_account(account_id)
async def delete_account(self, account_id: str) -> bool:
return self._accounts.delete_account(account_id)
async def update_account(self, account_id: str, config: dict[str, Any]) -> bool:
account = self._accounts.accounts.get(account_id)
if account is None:
logger.warning(f"[Feishu] Account '{account_id}' not found")
return False
account.label = config.get("label", account.label)
account.app_id = config.get("app_id", account.app_id)
account.app_secret = config.get("app_secret", account.app_secret)
account.verification_token = config.get("verification_token", account.verification_token)
account.encrypt_key = config.get("encrypt_key", account.encrypt_key)
account.bot_open_id = config.get("bot_open_id", account.bot_open_id)
account.enabled = config.get("enabled", account.enabled)
account.config = config
logger.info(f"[Feishu] Updated account '{account_id}'")
return True
def build_channel_summary(self) -> dict[str, Any]:
enabled = self._accounts.list_enabled()
return {
"channel_id": self.channel_id,
"channel_type": self.channel_type,
"account_count": len(self._accounts.accounts),
"enabled_count": len(enabled),
"connected": self._status == ChannelStatus.CONNECTED,
}
def build_account_snapshot(self, account_id: str) -> dict[str, Any] | None:
account = self._accounts.accounts.get(account_id)
if account is None:
return None
return {
"account_id": account.account_id,
"label": account.label,
"enabled": account.enabled,
"has_app_id": bool(account.app_id),
"has_secret": bool(account.app_secret),
"has_verification_token": bool(account.verification_token),
}

16
backend/package/yuxi/channels/adapters/feishu/api.py
View File

@ -14,8 +14,8 @@ FEISHU_BUSINESS_ERROR_CODES: dict[int, str] = {
99991664: "请求频繁，触发限流",
99991668: "app_ticket 无效",
99991669: "app_access_token 已过期",
230001: "频率限制",
230002: "请求过于频繁",
230001: "频率限制",
230002: "请求过于频繁",
}
MAX_RETRIES = 3
@ -103,9 +103,7 @@ class FeishuApiClient:
return f"{base_msg} ({detail})" if detail else base_msg
return None
async def _request_with_retry(
self, method: str, path: str, body: dict | None = None, **params: Any
) -> dict:
async def _request_with_retry(self, method: str, path: str, body: dict | None = None, **params: Any) -> dict:
last_error: Exception | None = None
for attempt in range(self._max_retries + 1):
@ -132,7 +130,7 @@ class FeishuApiClient:
continue
if resp.status_code in RETRYABLE_STATUS_CODES and attempt < self._max_retries:
wait = RETRY_BACKOFF_BASE * (2 ** attempt)
wait = RETRY_BACKOFF_BASE * (2**attempt)
await asyncio.sleep(wait)
continue
@ -156,7 +154,7 @@ class FeishuApiClient:
except (httpx.TimeoutException, httpx.NetworkError) as e:
last_error = e
if attempt < self._max_retries:
wait = RETRY_BACKOFF_BASE * (2 ** attempt)
wait = RETRY_BACKOFF_BASE * (2**attempt)
await asyncio.sleep(wait)
continue
raise RuntimeError(f"Network error: {e}") from e
@ -167,7 +165,7 @@ class FeishuApiClient:
except Exception as e:
last_error = e
if attempt < self._max_retries:
await asyncio.sleep(RETRY_BACKOFF_BASE * (2 ** attempt))
await asyncio.sleep(RETRY_BACKOFF_BASE * (2**attempt))
continue
raise
@ -183,4 +181,4 @@ class FeishuApiClient:
return await self._request_with_retry("PATCH", path, body=body)
async def delete(self, path: str) -> dict:
return await self._request_with_retry("DELETE", path)
return await self._request_with_retry("DELETE", path)

2
backend/package/yuxi/channels/adapters/feishu/app_owner.py
View File

@ -72,4 +72,4 @@ async def ensure_owner_access(
return True
except Exception as e:
logger.warning("[FeishuAppOwner] Failed to grant owner access: %s", e)
return False
return False

51
backend/package/yuxi/channels/adapters/feishu/cards.py
View File

@ -68,10 +68,7 @@ def _build_plain_text(content: str) -> dict[str, Any]:
def _build_options(options: list[dict[str, str]]) -> list[dict[str, Any]]:
return [
{"text": _build_plain_text(opt.get("text", "")), "value": opt.get("value", "")}
for opt in options
]
return [{"text": _build_plain_text(opt.get("text", "")), "value": opt.get("value", "")} for opt in options]
def make_select_static(
@ -224,7 +221,7 @@ def _truncate_card_content(content: str, limit: int = CARD_CONTENT_LIMIT) -> str
if len(content) <= limit:
return content
suffix = "\n\n...(内容过长已截断)"
return content[:limit - len(suffix)] + suffix
return content[: limit - len(suffix)] + suffix
def build_feishu_card(
@ -242,14 +239,17 @@ def build_feishu_card(
context_text: str | None = None,
dividers: int = 0,
selectors: list[dict[str, Any]] | None = None,
visible_to_operator: bool = False,
) -> dict[str, Any]:
elements: list[dict[str, Any]] = []
if context_text:
elements.append({
"tag": "note",
"elements": [_build_plain_text(context_text)],
})
elements.append(
{
"tag": "note",
"elements": [_build_plain_text(context_text)],
}
)
for _ in range(dividers):
elements.append({"tag": "hr"})
@ -288,12 +288,14 @@ def build_feishu_card(
actions: list[dict[str, Any]] = []
for btn in buttons:
btn_type = resolve_button_type(btn.get("style"))
actions.append({
"tag": "button",
"text": _build_plain_text(btn.get("text", "")),
"type": btn_type,
"value": {"action": btn.get("action", "")},
})
actions.append(
{
"tag": "button",
"text": _build_plain_text(btn.get("text", "")),
"type": btn_type,
"value": {"action": btn.get("action", "")},
}
)
elements.append({"tag": "action", "actions": actions})
if url_unfurl:
@ -301,18 +303,25 @@ def build_feishu_card(
elements.append({"tag": "markdown", "content": f"[🔗 {url}]({url})"})
if note:
elements.append({
"tag": "note",
"elements": [_build_plain_text(note)],
})
elements.append(
{
"tag": "note",
"elements": [_build_plain_text(note)],
}
)
header_title = f"{title} (回复中...)" if streaming else title
color = resolve_template_color(template, tone)
return {
card: dict[str, Any] = {
"header": {
"title": _build_plain_text(header_title),
"template": color,
},
"elements": elements,
}
}
if visible_to_operator:
card["config"] = {"update_multi": False}
return card

2
backend/package/yuxi/channels/adapters/feishu/chat_cache.py
View File

@ -47,4 +47,4 @@ class ChatNameCache:
async def clear(self) -> None:
async with self._lock:
self._cache.clear()
self._cache.clear()

6
backend/package/yuxi/channels/adapters/feishu/dedup.py
View File

@ -1,20 +1,18 @@
from __future__ import annotations
import hashlib
import json
import os
import time
from collections import OrderedDict
from typing import Any
import hashlib
DEFAULT_MAX_ENTRIES = 10000
DEFAULT_TTL_S = 300
PERSIST_BATCH_INTERVAL_S = 5
class FeishuDedupStore:
def __init__(
self,
max_entries: int = DEFAULT_MAX_ENTRIES,
@ -138,4 +136,4 @@ class FeishuDedupStore:
if loaded > 0:
self._dirty = False
self._last_persist_at = now
self._last_persist_at = now

90
backend/package/yuxi/channels/adapters/feishu/directory.py
View File

@ -12,10 +12,10 @@ CACHE_TTL_DEPARTMENTS_S = 600
class FeishuDirectoryClient:
def __init__(self, api_client: Any, page_size: int = PAGE_SIZE_DEFAULT):
self._api = api_client
self._page_size = min(page_size, PAGE_SIZE_MAX)
self._lock = asyncio.Lock()
self._user_cache: dict[str, tuple[dict, float]] = {}
self._user_list_cache: list[dict] = []
self._user_list_age: float = 0
@ -25,23 +25,28 @@ class FeishuDirectoryClient:
async def get_user(self, user_id: str, *, user_id_type: str = "open_id") -> dict | None:
cache_key = f"{user_id_type}:{user_id}"
cached = self._user_cache.get(cache_key)
if cached is not None:
data, ts = cached
if time.monotonic() - ts < CACHE_TTL_USERS_S:
return data
del self._user_cache[cache_key]
async with self._lock:
cached = self._user_cache.get(cache_key)
if cached is not None:
data, ts = cached
if time.monotonic() - ts < CACHE_TTL_USERS_S:
return data
path = f"/open-apis/contact/v3/users/{user_id}"
try:
resp = await self._api.get(path, user_id_type=user_id_type)
user = resp.get("data", {}).get("user", {})
if user:
self._user_cache[cache_key] = (user, time.monotonic())
return user
except Exception:
return None
async with self._lock:
if user:
self._user_cache[cache_key] = (user, time.monotonic())
else:
self._user_cache.pop(cache_key, None)
return user
async def list_users(
self,
*,
@ -58,16 +63,18 @@ class FeishuDirectoryClient:
params["page_token"] = page_token
if not department_id and not page_token:
if time.monotonic() - self._user_list_age < CACHE_TTL_USERS_S and self._user_list_cache:
return {"items": self._user_list_cache, "has_more": False, "page_token": ""}
async with self._lock:
if time.monotonic() - self._user_list_age < CACHE_TTL_USERS_S and self._user_list_cache:
return {"items": list(self._user_list_cache), "has_more": False, "page_token": ""}
resp = await self._api.get(path, **params)
data = resp.get("data", {})
items = data.get("items", [])
if not department_id and not page_token:
self._user_list_cache = items
self._user_list_age = time.monotonic()
async with self._lock:
self._user_list_cache = items
self._user_list_age = time.monotonic()
return {
"items": items,
@ -77,23 +84,28 @@ class FeishuDirectoryClient:
async def get_department(self, department_id: str) -> dict | None:
cache_key = department_id
cached = self._dept_cache.get(cache_key)
if cached is not None:
data, ts = cached
if time.monotonic() - ts < CACHE_TTL_DEPARTMENTS_S:
return data
del self._dept_cache[cache_key]
async with self._lock:
cached = self._dept_cache.get(cache_key)
if cached is not None:
data, ts = cached
if time.monotonic() - ts < CACHE_TTL_DEPARTMENTS_S:
return data
path = f"/open-apis/contact/v3/departments/{department_id}"
try:
resp = await self._api.get(path)
dept = resp.get("data", {}).get("department", {})
if dept:
self._dept_cache[cache_key] = (dept, time.monotonic())
return dept
except Exception:
return None
async with self._lock:
if dept:
self._dept_cache[cache_key] = (dept, time.monotonic())
else:
self._dept_cache.pop(cache_key, None)
return dept
async def list_departments(
self,
*,
@ -110,16 +122,18 @@ class FeishuDirectoryClient:
params["page_token"] = page_token
if not parent_department_id and not page_token:
if time.monotonic() - self._dept_list_age < CACHE_TTL_DEPARTMENTS_S and self._dept_list_cache:
return {"items": self._dept_list_cache, "has_more": False, "page_token": ""}
async with self._lock:
if time.monotonic() - self._dept_list_age < CACHE_TTL_DEPARTMENTS_S and self._dept_list_cache:
return {"items": list(self._dept_list_cache), "has_more": False, "page_token": ""}
resp = await self._api.get(path, **params)
data = resp.get("data", {})
items = data.get("items", [])
if not parent_department_id and not page_token:
self._dept_list_cache = items
self._dept_list_age = time.monotonic()
async with self._lock:
self._dept_list_cache = items
self._dept_list_age = time.monotonic()
return {
"items": items,
@ -145,10 +159,12 @@ async def list_groups(client: Any) -> list[dict]:
items = []
data = resp.data
for group in getattr(data, "items", []) or []:
items.append({
"id": getattr(group, "chat_id", ""),
"name": getattr(group, "name", ""),
})
items.append(
{
"id": getattr(group, "chat_id", ""),
"name": getattr(group, "name", ""),
}
)
return items
@ -159,8 +175,10 @@ async def list_peers(client: Any) -> list[dict]:
items = []
data = resp.data
for user in getattr(data, "items", []) or []:
items.append({
"id": getattr(user, "open_id", ""),
"name": getattr(user, "name", ""),
})
return items
items.append(
{
"id": getattr(user, "open_id", ""),
"name": getattr(user, "name", ""),
}
)
return items

5
backend/package/yuxi/channels/adapters/feishu/formatter.py
View File

@ -4,7 +4,6 @@ from typing import Any
from .cards import (
TEXT_CHUNK_LIMIT,
CARD_CONTENT_LIMIT,
build_feishu_card,
build_feishu_post_content,
build_feishu_text_content,
@ -73,7 +72,7 @@ def _truncate_text(content: str, limit: int = TEXT_CHUNK_LIMIT) -> str:
if len(content) <= limit:
return content
suffix = "\n\n...(内容过长已截断，请查看完整的消息记录)"
return content[:limit - len(suffix)] + suffix
return content[: limit - len(suffix)] + suffix
def make_error_card(title: str, detail: str) -> dict[str, Any]:
@ -95,4 +94,4 @@ def format_outbound(content: str, **_kwargs: Any) -> dict[str, Any]:
for key in ("buttons", "thread_id"):
if isinstance(metadata, dict) and key in metadata:
result[key] = metadata[key]
return result
return result

78
backend/package/yuxi/channels/adapters/feishu/media.py
View File

@ -146,41 +146,77 @@ async def upload_file(client: Any, file_data: bytes, filename: str, file_type: s
async def download_media(client: Any, message_id: str, file_key: str, file_type: str) -> bytes:
resp = await _download_with_fallback(client, message_id, file_key, file_type)
resp = await _download_with_fallback(client, message_id, file_key, file_type, max_retries=3)
if resp is not None and resp.success():
return resp.file.read()
if file_type == "file":
logger.info("[FeishuMedia] File download failed, trying media type fallback for %s", file_key)
resp = await _download_with_fallback(client, message_id, file_key, "media")
resp = await _download_with_fallback(client, message_id, file_key, "media", max_retries=2)
if resp is not None and resp.success():
return resp.file.read()
raise DeliveryFailedError(f"Media download failed: file_key={file_key}")
async def _download_with_fallback(client: Any, message_id: str, file_key: str, file_type: str) -> Any:
async def _download_with_fallback(
client: Any, message_id: str, file_key: str, file_type: str, max_retries: int = 1
) -> Any:
import lark_oapi
try:
request = (
lark_oapi.api.im.v1.GetMessageResourceRequest.builder()
.message_id(message_id)
.file_key(file_key)
.type(file_type)
.build()
)
for attempt in range(max_retries):
try:
request = (
lark_oapi.api.im.v1.GetMessageResourceRequest.builder()
.message_id(message_id)
.file_key(file_key)
.type(file_type)
.build()
)
resp = client.im.v1.message_resource.get(request)
http_status = getattr(resp, "http_status", 0) or getattr(resp, "status_code", 0)
if resp.success():
return resp
if http_status == 502:
logger.warning("[FeishuMedia] 502 error downloading type=%s for %s", file_type, file_key)
return None
except Exception as e:
logger.warning("[FeishuMedia] Download error (type=%s): %s", file_type, e)
return None
resp = client.im.v1.message_resource.get(request)
http_status = getattr(resp, "http_status", 0) or getattr(resp, "status_code", 0)
if resp.success():
return resp
if http_status == 502:
logger.warning(
"[FeishuMedia] 502 error downloading type=%s for %s (attempt %d/%d)",
file_type,
file_key,
attempt + 1,
max_retries,
)
if attempt < max_retries - 1:
await asyncio.sleep(0.5 * (attempt + 1))
continue
elif http_status >= 500:
logger.warning(
"[FeishuMedia] %d error downloading type=%s for %s (attempt %d/%d)",
http_status,
file_type,
file_key,
attempt + 1,
max_retries,
)
if attempt < max_retries - 1:
await asyncio.sleep(1.0 * (attempt + 1))
continue
return None
except Exception as e:
logger.warning(
"[FeishuMedia] Download error (type=%s, attempt %d/%d): %s",
file_type,
attempt + 1,
max_retries,
e,
)
if attempt < max_retries - 1:
await asyncio.sleep(0.5 * (attempt + 1))
return None
async def _get_tenant_token(client: Any) -> str:

20
backend/package/yuxi/channels/adapters/feishu/normalizer.py
View File

@ -372,8 +372,9 @@ def extract_attachments(event: dict) -> list[Attachment]:
file_key = parsed.get("file_key", "")
if image_key:
result.append(
Attachment(type="image", file_id=_encode_file_id(message_id, image_key, "image"),
metadata={"role": "cover"})
Attachment(
type="image", file_id=_encode_file_id(message_id, image_key, "image"), metadata={"role": "cover"}
)
)
if file_key:
result.append(Attachment(type="video", file_id=_encode_file_id(message_id, file_key, "file")))
@ -387,11 +388,18 @@ def extract_attachments(event: dict) -> list[Attachment]:
item_message_id = item.get("message_id", message_id)
item_content = _parse_content(item.get("content", ""))
if item_type == "image" and item_content.get("image_key"):
result.append(Attachment(type="image", file_id=_encode_file_id(item_message_id, item_content["image_key"], "image")))
result.append(
Attachment(
type="image", file_id=_encode_file_id(item_message_id, item_content["image_key"], "image")
)
)
elif item_type == "file" and item_content.get("file_key"):
result.append(
Attachment(type="file", file_id=_encode_file_id(item_message_id, item_content["file_key"], "file"),
filename=item_content.get("file_name", ""))
Attachment(
type="file",
file_id=_encode_file_id(item_message_id, item_content["file_key"], "file"),
filename=item_content.get("file_name", ""),
)
)
return result
@ -481,4 +489,4 @@ def normalize_inbound(
attachments=attachments,
mentions=mentions,
metadata=metadata,
)
)

45
backend/package/yuxi/channels/adapters/feishu/oauth.py
View File

@ -8,11 +8,51 @@ from yuxi.utils.logging_config import logger
TOKEN_CACHE_TTL_S = 3600
TOKEN_REFRESH_MARGIN_S = 300
DEFAULT_FEISHU_DOMAIN = "https://open.feishu.cn"
DEFAULT_LARK_DOMAIN = "https://open.larksuite.com"
class TokenBackoffManager:
"""共享的 401/token 退避重试管理器"""
def __init__(self, max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 30.0):
self._backoff_until: float = 0
self._retry_count: int = 0
self._max_retries = max_retries
self._base_delay = base_delay
self._max_delay = max_delay
@property
def is_backing_off(self) -> bool:
return self._backoff_until > time.monotonic()
async def wait_if_backing_off(self) -> None:
import asyncio
now = time.monotonic()
if self._backoff_until > now:
wait = self._backoff_until - now
logger.warning(f"[TokenBackoff] Backoff active, waiting {wait:.1f}s")
await asyncio.sleep(wait)
def record_error(self) -> bool:
self._retry_count += 1
if self._retry_count > self._max_retries:
logger.error(f"[TokenBackoff] Retry count exceeded ({self._max_retries})")
return False
delay = min(self._base_delay * (2 ** (self._retry_count - 1)), self._max_delay)
self._backoff_until = time.monotonic() + delay
logger.warning(f"[TokenBackoff] Error recorded, retrying in {delay:.1f}s (attempt {self._retry_count})")
return True
def reset(self) -> None:
if self._retry_count > 0:
logger.info("[TokenBackoff] Backoff reset after success")
self._retry_count = 0
self._backoff_until = 0
class FeishuOAuthClient:
def __init__(self, app_id: str, app_secret: str, redirect_uri: str = "", domain: str = ""):
self._app_id = app_id
@ -125,7 +165,6 @@ class FeishuOAuthClient:
class FeishuDeviceCodeClient:
def __init__(self, app_id: str, app_secret: str, domain: str = ""):
self._app_id = app_id
self._app_secret = app_secret
@ -194,4 +233,4 @@ class FeishuDeviceCodeClient:
return None
def clear_cache(self) -> None:
self._token_cache.clear()
self._token_cache.clear()

21
backend/package/yuxi/channels/adapters/feishu/policy.py
View File

@ -19,14 +19,11 @@ class PolicyAccessTracker:
entry = (time.monotonic(), resource_id, allowed)
self._access_log[key].append(entry)
if len(self._access_log[key]) > self.MAX_HISTORY_PER_KEY:
self._access_log[key] = self._access_log[key][-self.MAX_HISTORY_PER_KEY:]
self._access_log[key] = self._access_log[key][-self.MAX_HISTORY_PER_KEY :]
def get_history(self, key: str) -> list[dict[str, Any]]:
entries = self._access_log.get(key, [])
return [
{"timestamp": ts, "resource_id": rid, "allowed": allowed}
for ts, rid, allowed in entries
]
return [{"timestamp": ts, "resource_id": rid, "allowed": allowed} for ts, rid, allowed in entries]
def count_recent(self, key: str, window_s: float = 60) -> tuple[int, int]:
entries = self._access_log.get(key, [])
@ -43,13 +40,14 @@ class PolicyAccessTracker:
class FeishuPolicyMatcher:
def __init__(self, config: dict[str, Any]):
self._group_policy = config.get("groupPolicy", config.get("group_policy", "allowlist"))
self._dm_policy = config.get("dmPolicy", config.get("dm_policy", "pairing"))
self._allowlist = set(config.get("allowFrom", config.get("allowlist", [])))
self._blocklist = set(config.get("blockFrom", config.get("blocklist", [])))
self._deny_message = config.get("denyMessage", "对不起，您没有权限使用此机器人。")
self._global_require_mention = config.get("requireMention", True)
self._groups_config: dict[str, dict[str, Any]] = config.get("groups", {})
self._tracker = PolicyAccessTracker()
def check_chat_access(self, chat_id: str, chat_type: str) -> tuple[bool, str]:
@ -81,6 +79,15 @@ class FeishuPolicyMatcher:
self._tracker.record_access(policy, chat_id, False)
return False, self._deny_message
def resolve_require_mention(self, chat_id: str) -> bool:
group_cfg = self._groups_config.get(chat_id, {})
if "requireMention" in group_cfg:
return bool(group_cfg["requireMention"])
return self._global_require_mention
def resolve_group_config(self, chat_id: str) -> dict[str, Any]:
return self._groups_config.get(chat_id, {})
def get_access_history(self, policy_type: str = "") -> list[dict[str, Any]]:
if policy_type:
return self._tracker.get_history(policy_type)
@ -101,4 +108,4 @@ class FeishuPolicyMatcher:
return bool(re.fullmatch(pattern.replace("*", ".*").replace("?", "."), value))
except re.error:
return False
return False
return False

1
backend/package/yuxi/channels/adapters/feishu/reasoning.py
View File

@ -2,7 +2,6 @@ from __future__ import annotations
from typing import Any
REASONING_WRAPPER_START = "🤔 推理中..."
REASONING_WRAPPER_END = "✅ 推理完成"

2
backend/package/yuxi/channels/adapters/feishu/secret_resolver.py
View File

@ -90,4 +90,4 @@ def _exec_secret(command: str) -> str:
return ""
_secrets_cache: dict[str, tuple[str, float]] = {}
_secrets_cache: dict[str, tuple[str, float]] = {}

2
backend/package/yuxi/channels/adapters/feishu/security_audit.py
View File

@ -228,4 +228,4 @@ def collect_feishu_security_audit_findings(config: dict[str, Any]) -> list[dict]
"suggestion": w.suggestion,
}
)
return findings
return findings

200
backend/package/yuxi/channels/adapters/feishu/send.py
View File

@ -15,13 +15,32 @@ try:
import lark_oapi
HAS_LARK_SDK = True
_LarkAccessDenied = lark_oapi.exception.AccessDeniedException
_LarkInvalidArgs = lark_oapi.exception.InvalidArgsException
except ImportError:
HAS_LARK_SDK = False
lark_oapi = None # type: ignore
class _DummyLarkError(BaseException):
pass
_LarkAccessDenied = _DummyLarkError
_LarkInvalidArgs = _DummyLarkError
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp"}
CHUNK_STRATEGIES = ["auto", "zero", "newline", "sentence"]
REPLY_DEGRADE_CODES = {230011, 231003}
# Feishu 401 错误分类: 消息中包含这些关键词的才是 token 过期，需要刷新
_TOKEN_EXPIRED_KEYWORDS = ("tenant_access_token", "app_access_token", "token expire", "token invalid")
def _is_token_expired_error(code: int, msg: str) -> bool:
if code != 401:
return False
msg_lower = msg.lower()
return any(kw in msg_lower for kw in _TOKEN_EXPIRED_KEYWORDS)
def _resolve_receive_id_type(chat_type: str) -> str:
if chat_type in ("direct", "private", "p2p"):
@ -40,6 +59,8 @@ async def reply_message(
thread_id: str | None = None,
use_post_format: bool = False,
silent: bool = False,
chat_type: str | None = None,
chat_id: str | None = None,
) -> DeliveryResult:
if not HAS_LARK_SDK:
return DeliveryResult(success=False, error="lark-oapi SDK not available")
@ -75,15 +96,19 @@ async def reply_message(
logger.info(f"[Feishu] Reply target revoked (code={error_code}), falling back to direct send")
return await send_text(
client,
message_id,
chat_id if chat_id else message_id,
content,
chat_type="private",
chat_type=chat_type or "private",
thread_id=thread_id,
)
return DeliveryResult(success=False, error=f"Feishu API error: {resp.msg}")
except AttributeError as e:
return DeliveryResult(success=False, error=f"SDK error: {e}")
except _LarkAccessDenied as e:
return DeliveryResult(success=False, error=f"Access denied: {e}")
except _LarkInvalidArgs as e:
return DeliveryResult(success=False, error=f"Invalid args: {e}")
except Exception as e:
logger.error(f"Feishu reply_message failed: {e}")
return DeliveryResult(success=False, error=str(e))
@ -120,11 +145,15 @@ async def send_text(
return DeliveryResult(success=True, message_id=resp.data.get("message_id", ""))
code = getattr(resp, "code", -1)
msg = getattr(resp, "msg", str(resp))
if code == 401:
if _is_token_expired_error(code, msg):
return DeliveryResult(success=False, error=f"Auth failed: {msg}", error_code="auth_expired")
return DeliveryResult(success=False, error=f"Feishu API error: {msg}")
return DeliveryResult(success=False, error=f"Send failed: {msg}")
except AttributeError as e:
return DeliveryResult(success=False, error=f"SDK error: {e}")
except _LarkAccessDenied as e:
return DeliveryResult(success=False, error=f"Access denied: {e}")
except _LarkInvalidArgs as e:
return DeliveryResult(success=False, error=f"Invalid args: {e}")
except Exception as e:
logger.error(f"Feishu send_text failed: {e}")
return DeliveryResult(success=False, error=str(e))
@ -159,6 +188,68 @@ async def read_message(client: Any, message_id: str) -> dict:
return {}
async def list_messages(
client: Any,
chat_id: str,
*,
page_token: str = "",
page_size: int = 20,
start_time: str = "",
end_time: str = "",
) -> dict[str, Any]:
if not HAS_LARK_SDK:
return {"messages": [], "has_more": False, "page_token": ""}
try:
request_builder = (
lark_oapi.api.im.v1.ListMessageRequest.builder()
.container_id_type("chat")
.container_id(chat_id)
.page_size(min(page_size, 50))
)
if page_token:
request_builder.page_token(page_token)
if start_time:
request_builder.start_time(start_time)
if end_time:
request_builder.end_time(end_time)
request = request_builder.build()
resp = await client.im.v1.message.list(request)
if not resp.success():
raise RuntimeError(f"获取消息列表失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
items = data.get("items", [])
messages = [
{
"message_id": item.get("message_id", ""),
"msg_type": item.get("msg_type", ""),
"chat_id": item.get("chat_id", ""),
"chat_type": item.get("chat_type", ""),
"content": item.get("body", {}).get("content", ""),
"root_id": item.get("root_id", ""),
"parent_id": item.get("parent_id", ""),
"sender_id": item.get("sender", {}).get("id", ""),
"create_time": item.get("create_time", ""),
"mentions": item.get("mentions", []),
}
for item in items
]
return {
"messages": messages,
"has_more": data.get("has_more", False),
"page_token": data.get("page_token", ""),
}
except _LarkAccessDenied as e:
raise RuntimeError(f"权限不足: {e}") from e
except _LarkInvalidArgs as e:
raise RuntimeError(f"参数错误: {e}") from e
except Exception as e:
raise RuntimeError(f"获取消息列表失败: {e}") from e
async def send_card(
client: Any,
receive_id: str,
@ -202,11 +293,15 @@ async def send_card(
return DeliveryResult(success=True, message_id=resp.data.get("message_id", ""))
code = getattr(resp, "code", -1)
msg = getattr(resp, "msg", str(resp))
if code == 401:
if _is_token_expired_error(code, msg):
return DeliveryResult(success=False, error=f"Auth failed: {msg}", error_code="auth_expired")
return DeliveryResult(success=False, error=f"Feishu API error: {msg}")
except AttributeError as e:
return DeliveryResult(success=False, error=f"SDK error: {e}")
except _LarkAccessDenied as e:
return DeliveryResult(success=False, error=f"Access denied: {e}")
except _LarkInvalidArgs as e:
return DeliveryResult(success=False, error=f"Invalid args: {e}")
except Exception as e:
logger.error(f"Feishu send_card failed: {e}")
return DeliveryResult(success=False, error=str(e))
@ -243,6 +338,10 @@ async def send_reaction(
success=False,
error="SDK version incompatible: CreateMessageReactionRequestReactionType not found",
)
except _LarkAccessDenied as e:
return DeliveryResult(success=False, error=f"Access denied: {e}")
except _LarkInvalidArgs as e:
return DeliveryResult(success=False, error=f"Invalid args: {e}")
except Exception as e:
logger.error(f"Feishu send_reaction failed: {e}")
return DeliveryResult(success=False, error=str(e))
@ -264,6 +363,7 @@ def _make_create_msg_request(
lark_oapi.api.im.v1.CreateMessageRequest.builder()
.receive_id_type(receive_id_type)
.request_body(body_builder.build())
.silent(silent)
)
return req_builder.build()
@ -301,9 +401,97 @@ async def update_card_message(client: Any, message_id: str, content: str) -> Del
resp = await client.im.v1.message.patch(request)
if resp.success():
return DeliveryResult(success=True, message_id=message_id)
code = getattr(resp, "code", -1)
getattr(resp, "code", -1)
msg = getattr(resp, "msg", str(resp))
return DeliveryResult(success=False, error=f"Feishu API error: {msg}")
except _LarkAccessDenied as e:
return DeliveryResult(success=False, error=f"Access denied: {e}")
except _LarkInvalidArgs as e:
return DeliveryResult(success=False, error=f"Invalid args: {e}")
except Exception as e:
logger.error(f"Feishu update_card_message failed: {e}")
return DeliveryResult(success=False, error=str(e))
async def forward_message(
client: Any,
message_id: str,
target_chat_id: str,
*,
target_type: str = "chat",
) -> DeliveryResult:
if not HAS_LARK_SDK:
return DeliveryResult(success=False, error="lark-oapi SDK not available")
try:
request_body = lark_oapi.api.im.v1.ForwardMessageRequestBody.builder().receive_id(target_chat_id).build()
request = (
lark_oapi.api.im.v1.ForwardMessageRequest.builder()
.message_id(message_id)
.receive_id_type(target_type)
.request_body(request_body)
.build()
)
resp = await client.im.v1.message.forward(request)
if resp.success():
new_msg_id = resp.data.get("message_id", "") if hasattr(resp, "data") else ""
return DeliveryResult(success=True, message_id=new_msg_id)
code = getattr(resp, "code", -1)
msg = getattr(resp, "msg", str(resp))
if _is_token_expired_error(code, msg):
return DeliveryResult(success=False, error=f"Auth failed: {msg}", error_code="auth_expired")
return DeliveryResult(success=False, error=f"Feishu API error: {msg}")
except _LarkAccessDenied as e:
return DeliveryResult(success=False, error=f"Access denied: {e}")
except _LarkInvalidArgs as e:
return DeliveryResult(success=False, error=f"Invalid args: {e}")
except Exception as e:
logger.error(f"Feishu forward_message failed: {e}")
return DeliveryResult(success=False, error=str(e))
async def get_message_read_status(
client: Any,
message_id: str,
*,
page_token: str = "",
page_size: int = 20,
) -> dict[str, Any]:
if not HAS_LARK_SDK:
return {"users": [], "has_more": False, "page_token": ""}
try:
request_builder = (
lark_oapi.api.im.v1.ListReadUserRequest.builder().message_id(message_id).page_size(min(page_size, 100))
)
if page_token:
request_builder.page_token(page_token)
request = request_builder.build()
resp = await client.im.v1.message.read_user.list(request)
if not resp.success():
raise RuntimeError(f"获取已读状态失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
items = data.get("items", [])
users = [
{
"user_id": item.get("user_id", ""),
"timestamp": item.get("timestamp", ""),
}
for item in items
]
return {
"users": users,
"has_more": data.get("has_more", False),
"page_token": data.get("page_token", ""),
}
except _LarkAccessDenied as e:
raise RuntimeError(f"权限不足: {e}") from e
except _LarkInvalidArgs as e:
raise RuntimeError(f"参数错误: {e}") from e
except Exception as e:
raise RuntimeError(f"获取已读状态失败: {e}") from e

3
backend/package/yuxi/channels/adapters/feishu/sent_cache.py
View File

@ -4,7 +4,6 @@ import asyncio
import time
from typing import Any
DEFAULT_MAX_ENTRIES = 1000
DEFAULT_TTL_S = 600
@ -61,4 +60,4 @@ class FeishuSentCache:
async def clear(self) -> None:
async with self._lock:
self._cache.clear()
self._timestamps.clear()
self._timestamps.clear()

7
backend/package/yuxi/channels/adapters/feishu/sequential.py
View File

@ -2,7 +2,8 @@ from __future__ import annotations
import asyncio
from collections import defaultdict
from typing import Any, Coroutine
from collections.abc import Coroutine
from typing import Any
_SEQUENTIAL_TIMEOUT_S = 300
@ -17,7 +18,7 @@ class FeishuSequentialQueue:
lock = self._locks[key]
try:
await asyncio.wait_for(lock.acquire(), timeout=self._timeout_s)
except asyncio.TimeoutError:
except TimeoutError:
raise RuntimeError(f"[Sequential] Timeout acquiring lock for key '{key}'")
def release(self, key: str) -> None:
@ -44,4 +45,4 @@ class FeishuSequentialQueue:
for lock in self._locks.values():
if lock.locked():
lock.release()
self._locks.clear()
self._locks.clear()

2
backend/package/yuxi/channels/adapters/feishu/session.py
View File

@ -181,4 +181,4 @@ class SessionPersistence:
for sid in expired:
self._data.pop(sid, None)
if expired:
self._save()
self._save()

2
backend/package/yuxi/channels/adapters/feishu/thread_bindings.py
View File

@ -84,4 +84,4 @@ class ThreadBindingManager:
session_id = self._bindings.pop(thread_id, None)
if session_id:
self._reverse.pop(session_id, None)
self._timestamps.pop(thread_id, None)
self._timestamps.pop(thread_id, None)

71
backend/package/yuxi/channels/adapters/feishu/tools/__init__.py
View File

@ -0,0 +1,71 @@
from __future__ import annotations
from ._audit import audit_tool_call
from ._validate import validate_params
from .bitable import (
create_bitable_record,
delete_bitable_record,
get_bitable_fields,
list_bitable_records,
list_bitable_tables,
list_bitable_views,
update_bitable_record,
)
from .calendar import get_calendar_event, list_calendar_events, list_calendars
from .chat import (
add_chat_members,
create_chat,
create_chat_announcement,
get_chat_members,
search_chat_by_name,
update_chat,
)
from .directory import search_users
from .doc import (
append_doc_block,
create_doc,
delete_doc_block,
get_doc_content,
list_docs,
update_doc_block,
)
from .drive import (
create_folder,
download_drive_file,
get_file_detail,
list_drive_files,
upload_drive_file,
)
__all__ = [
"validate_params",
"audit_tool_call",
"list_bitable_tables",
"get_bitable_fields",
"list_bitable_records",
"create_bitable_record",
"update_bitable_record",
"delete_bitable_record",
"list_bitable_views",
"create_doc",
"get_doc_content",
"list_docs",
"update_doc_block",
"append_doc_block",
"delete_doc_block",
"list_drive_files",
"get_file_detail",
"create_folder",
"upload_drive_file",
"download_drive_file",
"get_chat_members",
"search_chat_by_name",
"create_chat",
"add_chat_members",
"update_chat",
"create_chat_announcement",
"search_users",
"list_calendar_events",
"get_calendar_event",
"list_calendars",
]

1
backend/package/yuxi/channels/adapters/feishu/tools/_audit.py
View File

@ -1,7 +1,6 @@
from __future__ import annotations
import logging
import time
from typing import Any
audit_logger = logging.getLogger("feishu.tools.audit")

23
backend/package/yuxi/channels/adapters/feishu/tools/_validate.py
View File

@ -1,25 +1,34 @@
from __future__ import annotations
from collections.abc import Callable
from functools import wraps
from typing import Any, Callable
from typing import Any
REQUIRED_PARAMS: dict[str, list[str]] = {
"list_bitable_tables": ["app_token"],
"get_bitable_fields": ["app_token", "table_id"],
"list_bitable_records": ["app_token", "table_id"],
"create_bitable_record": ["app_token", "table_id", "fields"],
"update_bitable_record": ["app_token", "table_id", "record_id", "fields"],
"delete_bitable_record": ["app_token", "table_id", "record_id"],
"list_bitable_views": ["app_token", "table_id"],
"create_doc": ["title"],
"get_doc_content": ["document_id"],
"update_doc_block": ["document_id", "block_id", "content"],
"append_doc_block": ["document_id", "parent_block_id", "content"],
"delete_doc_block": ["document_id", "block_id"],
"create_folder": ["name"],
"list_file_members": ["file_token"],
"add_file_permission": ["file_token", "member_type", "member_id"],
"remove_file_permission": ["file_token", "member_type", "member_id"],
"transfer_file_owner": ["file_token", "member_type", "member_id"],
"get_space_node_tree": ["space_id"],
"get_node_detail": ["node_token"],
"upload_drive_file": ["folder_token", "file_path"],
"download_drive_file": ["file_token", "save_path"],
"search_users": ["query"],
"get_chat_members": ["chat_id"],
"search_chat_by_name": ["name"],
"create_chat": ["name"],
"add_chat_members": ["chat_id", "user_ids"],
"update_chat": ["chat_id"],
"create_chat_announcement": ["chat_id", "content"],
"get_file_detail": ["file_token"],
"get_calendar_event": ["calendar_id", "event_id"],
}

98
backend/package/yuxi/channels/adapters/feishu/tools/bitable.py
View File

@ -149,3 +149,101 @@ async def create_bitable_record(
}
except Exception as e:
raise RuntimeError(f"创建记录失败: {e}") from e
@validate_params
async def update_bitable_record(
client: Any,
app_token: str,
table_id: str,
record_id: str,
fields: dict[str, Any],
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
request_body = lark_oapi.api.bitable.v1.UpdateAppTableRecordRequestBody.builder().fields(fields).build()
request = (
lark_oapi.api.bitable.v1.UpdateAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.record_id(record_id)
.request_body(request_body)
.build()
)
resp = await client.bitable.v1.app_table_record.update(request)
if not resp.success():
raise RuntimeError(f"更新记录失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
record = data.get("record", {}) or data
return {
"record_id": record.get("record_id", record_id),
"fields": record.get("fields", fields),
}
except Exception as e:
raise RuntimeError(f"更新记录失败: {e}") from e
@validate_params
async def delete_bitable_record(
client: Any,
app_token: str,
table_id: str,
record_id: str,
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
request = (
lark_oapi.api.bitable.v1.DeleteAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.record_id(record_id)
.build()
)
resp = await client.bitable.v1.app_table_record.delete(request)
if not resp.success():
raise RuntimeError(f"删除记录失败: {resp.msg}")
return {
"deleted": True,
"record_id": record_id,
"app_token": app_token,
"table_id": table_id,
}
except Exception as e:
raise RuntimeError(f"删除记录失败: {e}") from e
@validate_params
async def list_bitable_views(
client: Any,
app_token: str,
table_id: str,
) -> list[dict[str, Any]]:
if not HAS_LARK_SDK or not client:
return []
try:
request = (
lark_oapi.api.bitable.v1.ListAppTableViewRequest.builder().app_token(app_token).table_id(table_id).build()
)
resp = await client.bitable.v1.app_table_view.list(request)
if not resp.success():
raise RuntimeError(f"获取视图列表失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
items = data.get("items", [])
return [
{
"view_id": item.get("view_id", ""),
"view_name": item.get("view_name", ""),
"view_type": item.get("view_type", ""),
}
for item in items
]
except Exception as e:
raise RuntimeError(f"获取视图列表失败: {e}") from e

125
backend/package/yuxi/channels/adapters/feishu/tools/chat.py
View File

@ -94,3 +94,128 @@ async def search_chat_by_name(client: Any, name: str) -> list[dict[str, Any]]:
raise RuntimeError(f"搜索群组失败: {e}") from e
return chats
@validate_params
async def create_chat(
client: Any,
name: str,
description: str = "",
user_ids: list[str] | None = None,
chat_type: str = "private",
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
builder = lark_oapi.api.im.v1.CreateChatRequestBody.builder().name(name)
if description:
builder.description(description)
if chat_type:
builder.chat_type(chat_type)
if user_ids:
builder.user_ids(user_ids)
request_body = builder.build()
request = lark_oapi.api.im.v1.CreateChatRequest.builder().request_body(request_body).build()
resp = await client.im.v1.chat.create(request)
if not resp.success():
raise RuntimeError(f"创建群组失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
return {
"chat_id": data.get("chat_id", ""),
"name": data.get("name", name),
"description": data.get("description", description),
}
except Exception as e:
raise RuntimeError(f"创建群组失败: {e}") from e
@validate_params
async def add_chat_members(
client: Any,
chat_id: str,
user_ids: list[str],
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
request_body = lark_oapi.api.im.v1.CreateChatMembersRequestBody.builder().id_list(user_ids).build()
request = (
lark_oapi.api.im.v1.CreateChatMembersRequest.builder().chat_id(chat_id).request_body(request_body).build()
)
resp = await client.im.v1.chat_members.create(request)
if not resp.success():
raise RuntimeError(f"添加群成员失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
invalid_ids = data.get("invalid_id_list", []) or []
return {
"chat_id": chat_id,
"added_count": len(user_ids) - len(invalid_ids),
"invalid_ids": invalid_ids,
}
except Exception as e:
raise RuntimeError(f"添加群成员失败: {e}") from e
@validate_params
async def update_chat(
client: Any,
chat_id: str,
name: str = "",
description: str = "",
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
builder = lark_oapi.api.im.v1.UpdateChatRequestBody.builder()
if name:
builder.name(name)
if description:
builder.description(description)
request_body = builder.build()
request = lark_oapi.api.im.v1.UpdateChatRequest.builder().chat_id(chat_id).request_body(request_body).build()
resp = await client.im.v1.chat.update(request)
if not resp.success():
raise RuntimeError(f"更新群组失败: {resp.msg}")
return {
"chat_id": chat_id,
"updated": True,
}
except Exception as e:
raise RuntimeError(f"更新群组失败: {e}") from e
@validate_params
async def create_chat_announcement(
client: Any,
chat_id: str,
content: str,
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
request_body = lark_oapi.api.im.v1.CreateChatAnnouncementRequestBody.builder().content(content).build()
request = (
lark_oapi.api.im.v1.CreateChatAnnouncementRequest.builder()
.chat_id(chat_id)
.request_body(request_body)
.build()
)
resp = await client.im.v1.chat.announcement.create(request)
if not resp.success():
raise RuntimeError(f"创建群公告失败: {resp.msg}")
return {
"chat_id": chat_id,
"created": True,
}
except Exception as e:
raise RuntimeError(f"创建群公告失败: {e}") from e

132
backend/package/yuxi/channels/adapters/feishu/tools/doc.py
View File

@ -124,3 +124,135 @@ def _extract_doc_blocks(blocks: list) -> list[dict[str, Any]]:
result.append(extracted)
return result
@validate_params
async def update_doc_block(
client: Any,
document_id: str,
block_id: str,
content: str,
block_type: str = "text",
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
text_elements = [{"text_run": {"content": content}}]
block_data = {
"block_id": block_id,
block_type: {"elements": text_elements},
}
if block_type in ("heading1", "heading2", "heading3", "heading4", "heading5"):
block_data[block_type]["style"] = {}
request_body = (
lark_oapi.api.docx.v1.UpdateDocumentBlockRequestBody.builder()
.update_document_block_request(
lark_oapi.api.docx.v1.UpdateDocumentBlockRequestBlock.builder().replace_block(block_data).build()
)
.build()
)
request = (
lark_oapi.api.docx.v1.PatchDocumentBlockRequest.builder()
.document_id(document_id)
.block_id(block_id)
.request_body(request_body)
.build()
)
resp = await client.docx.v1.document_block.patch(request)
if not resp.success():
raise RuntimeError(f"更新文档块失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
block = data.get("block", {}) or data
return {
"block_id": block.get("block_id", block_id),
"document_id": document_id,
}
except Exception as e:
raise RuntimeError(f"更新文档块失败: {e}") from e
@validate_params
async def append_doc_block(
client: Any,
document_id: str,
parent_block_id: str,
content: str,
block_type: str = "text",
index: int = -1,
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
text_elements = [{"text_run": {"content": content}}]
children: list[dict[str, Any]] = [
{
"block_type": block_type,
block_type: {"elements": text_elements},
}
]
request_body = (
lark_oapi.api.docx.v1.CreateDocumentBlockChildrenRequestBody.builder()
.children(children)
.index(index if index >= 0 else -1)
.build()
)
request = (
lark_oapi.api.docx.v1.CreateDocumentBlockChildrenRequest.builder()
.document_id(document_id)
.block_id(parent_block_id)
.request_body(request_body)
.build()
)
resp = await client.docx.v1.document_block_children.create(request)
if not resp.success():
raise RuntimeError(f"追加文档块失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
children_resp = data.get("children", [])
block = children_resp[0] if children_resp else data
return {
"block_id": block.get("block_id", ""),
"parent_block_id": parent_block_id,
"document_id": document_id,
}
except Exception as e:
raise RuntimeError(f"追加文档块失败: {e}") from e
@validate_params
async def delete_doc_block(
client: Any,
document_id: str,
block_id: str,
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
try:
request = (
lark_oapi.api.docx.v1.DeleteDocumentBlockRequest.builder()
.document_id(document_id)
.block_id(block_id)
.build()
)
resp = await client.docx.v1.document_block.delete(request)
if not resp.success():
raise RuntimeError(f"删除文档块失败: {resp.msg}")
return {
"deleted": True,
"block_id": block_id,
"document_id": document_id,
}
except Exception as e:
raise RuntimeError(f"删除文档块失败: {e}") from e

78
backend/package/yuxi/channels/adapters/feishu/tools/drive.py
View File

@ -113,3 +113,81 @@ async def create_folder(client: Any, name: str, parent_token: str = "") -> dict[
}
except Exception as e:
raise RuntimeError(f"创建文件夹失败: {e}") from e
@validate_params
async def upload_drive_file(
client: Any,
folder_token: str,
file_path: str,
file_name: str = "",
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
import os
try:
if not file_name:
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path)
with open(file_path, "rb") as f:
if file_size <= 20 * 1024 * 1024:
resp = await client.drive.v1.media.upload(
folder_token=folder_token,
file_name=file_name,
file=f,
)
else:
resp = await client.drive.v1.media.upload_all(
folder_token=folder_token,
file_name=file_name,
file=f,
)
if not resp.success():
raise RuntimeError(f"上传文件失败: {resp.msg}")
data = resp.data if hasattr(resp, "data") else {}
return {
"token": data.get("file_token", ""),
"name": file_name,
"size": file_size,
}
except Exception as e:
raise RuntimeError(f"上传文件失败: {e}") from e
@validate_params
async def download_drive_file(
client: Any,
file_token: str,
save_path: str,
) -> dict[str, Any]:
if not HAS_LARK_SDK or not client:
raise RuntimeError("SDK 不可用")
import os
try:
request = lark_oapi.api.drive.v1.DownloadFileRequest.builder().file_token(file_token).build()
resp = await client.drive.v1.file.download(request)
if not resp.success():
raise RuntimeError(f"下载文件失败: {resp.msg}")
content = resp.file.read() if hasattr(resp, "file") else resp.read()
os.makedirs(os.path.dirname(save_path) or ".", exist_ok=True)
with open(save_path, "wb") as f:
f.write(content)
return {
"token": file_token,
"saved_path": save_path,
"size": len(content),
}
except Exception as e:
raise RuntimeError(f"下载文件失败: {e}") from e

1
backend/package/yuxi/channels/adapters/feishu/typing.py
View File

@ -4,7 +4,6 @@ import logging
import time
from dataclasses import dataclass
logger = logging.getLogger(__name__)
try: