ForcePilot/backend/package/yuxi/channel/extensions/feishu/mentions.py
Kris 5e91bb9985 feat(feishu): 新增飞书渠道完整插件实现
新增飞书渠道插件,包含基础配置、事件处理、消息收发、工具调用、权限管理等完整功能模块,支持WebSocket长连接、卡片消息、流式响应、群管理、审批通知等能力
2026-05-21 10:46:42 +08:00

58 lines
1.7 KiB
Python

from __future__ import annotations
import logging
import re
from yuxi.channel.extensions.feishu.utils import is_broadcast_mention
logger = logging.getLogger(__name__)
def extract_mentions(content: dict) -> list[dict]:
mentions = content.get("mentions", []) or []
result = []
for m in mentions:
if isinstance(m, dict):
mention_key = m.get("key", "")
if not is_broadcast_mention(mention_key):
mention_id = m.get("id", {})
if isinstance(mention_id, dict):
open_id = mention_id.get("open_id", "")
if open_id:
result.append({"open_id": open_id, "name": m.get("name", ""), "key": mention_key})
else:
result.append({"open_id": "", "name": m.get("name", ""), "key": mention_key})
return result
def parse_at_tags(text: str) -> list[str]:
return re.findall(r'<at\s+user_id="([^"]+)"\s*>(.*?)</at>', text)
def format_mentioned_list(mentions: list[dict]) -> str:
parts = []
for m in mentions:
open_id = m.get("open_id", "")
name = m.get("name", open_id)
parts.append(f"@{name}({open_id})")
return ", ".join(parts) if parts else ""
def check_bot_mentioned(
content: dict,
bot_open_id: str,
require_mention: bool = True,
) -> bool:
if not require_mention:
return True
mentions = content.get("mentions", []) or []
for m in mentions:
if isinstance(m, dict):
mention_id = m.get("id", {})
if isinstance(mention_id, dict):
if mention_id.get("open_id") == bot_open_id:
return True
elif mention_id == bot_open_id:
return True
return False