当飞书外部群有人@我，机器人自动回复消息

需要开通的权限

{
  "scopes": {
    "tenant": [
      "contact:contact.base:readonly",
      "contact:user.base:readonly",
      "im:chat",
      "im:message.group_at_msg:readonly",
      "im:message.group_msg",
      "im:message.p2p_msg:readonly",
      "im:message:readonly",
      "im:message:send_as_bot"
    ],
    "user": [
      "contact:contact.base:readonly",
      "contact:user.base:readonly",
      "contact:user.employee_id:readonly",
      "im:chat",
      "im:message",
      "im:message.send_as_user",
      "im:message:readonly"
    ]
  }
}

订阅事件：

监听群消息

import lark_oapi as lark
from lark_oapi.ws import Client
import json
import requests
import time
import sys
import io
from datetime import datetime, timezone

# # ★ Windows 终端 UTF-8 乱码修复
# sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
# sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

# 你的凭证
APP_ID = "XXX"
APP_SECRET = "XXX"

# ★ 监听名单：只返回「外部群」且「@了名单中某人」的消息
#    名单里既可以填 open_id，也可以填显示名（如「黄金海岸」），两者都匹配
WATCH_USERS = [
    "小曼"  # TODO: 填入你要监听的人员 open_id 或名字
]

# ★ 消息转发服务器地址：命中筛选条件的消息会推送到这里
MY_MESSAGE_WEBHOOK_URL = "XXX"  # TODO: 填入你的服务器 webhook 地址

# 缓存 tenant_access_token
tenant_access_token = None
token_expire_time = 0

# 把消息内容推送到服务器
def push_all_event_info_to_webhook(event_json, WEBHOOK_URL):
    payload = {
        "data": event_json
    }
    res = requests.post(
        url=WEBHOOK_URL,
        json=payload,
        timeout=10
    )
    # print("推送 Webhook 返回：\n", res.text)

# 获取 tenant_access_token
def get_tenant_access_token():
    global tenant_access_token, token_expire_time

    if tenant_access_token and time.time() < token_expire_time:
        return tenant_access_token

    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    headers = {"Content-Type": "application/json"}
    data = {"app_id": APP_ID, "app_secret": APP_SECRET}

    try:
        response = requests.post(url, headers=headers, json=data)
        result = response.json()

        if result.get("code") == 0:
            tenant_access_token = result.get("tenant_access_token")
            token_expire_time = time.time() + result.get("expire", 7200) - 100
            return tenant_access_token
        else:
            print(f"获取token失败: {result}")
            return None
    except Exception as e:
        print(f"获取token异常: {e}")
        return None


# 缓存机器人 tenant_key（用于判断发送人是否跨租户）
bot_tenant_key_cache = None

def get_bot_tenant_key():
    global bot_tenant_key_cache
    if bot_tenant_key_cache:
        return bot_tenant_key_cache

    token = get_tenant_access_token()
    if not token:
        return None

    url = "https://open.feishu.cn/open-apis/bot/v3/info"
    headers = {"Authorization": f"Bearer {token}"}

    try:
        response = requests.get(url, headers=headers)
        result = response.json()
        if result.get("code") == 0:
            bot_tenant_key_cache = result.get("bot", {}).get("tenant_key")
            return bot_tenant_key_cache
        else:
            print(f"获取机器人信息失败: {result}")
            return None
    except Exception as e:
        print(f"获取机器人信息异常: {e}")
        return None


# 缓存群信息（区分内部群/外部群 + 群名称）
chat_info_cache = {}

# 缓存用户 open_id → {is_internal, name}
user_info_cache = {}

def get_user_info_by_open_id(open_id):
    """通过 contact API 查询用户信息，返回 {is_internal, name}。
    is_internal: True=内部, False=外部, None=未知/失败
    name: 内部用户返回真实姓名，外部联系人/失败返回 '无法显示外部联系人姓名'
    """
    external_name = "无法显示外部联系人姓名"
    if not open_id:
        return {"is_internal": None, "name": external_name}
    if open_id in user_info_cache:
        return user_info_cache[open_id]

    token = get_tenant_access_token()
    if not token:
        return {"is_internal": None, "name": external_name}

    url = f"https://open.feishu.cn/open-apis/contact/v3/users/{open_id}"
    headers = {"Authorization": f"Bearer {token}"}
    params = {"user_id_type": "open_id"}

    try:
        response = requests.get(url, headers=headers, params=params)
        result = response.json()

        if result.get("code") == 0:
            user_info = result.get("data", {}) or {}
            if "user" in user_info:
                user_info = user_info["user"]
            name = user_info.get("name", "未命名用户")
            info = {"is_internal": True, "name": name}
            user_info_cache[open_id] = info
            return info
        elif result.get("code") in (41050, 99992360):
            # 无权限或查无此人：说明不在当前企业通讯录，视为外部用户
            info = {"is_internal": False, "name": external_name}
            user_info_cache[open_id] = info
            return info
        else:
            print(f"查询用户失败: {result}")
            return {"is_internal": None, "name": external_name}
    except Exception as e:
        print(f"查询用户异常: {e}")
        return {"is_internal": None, "name": external_name}


def get_chat_info(chat_id):
    """获取群详情：返回 {name, external} 等字段，失败返回 None"""
    if not chat_id:
        return None
    if chat_id in chat_info_cache:
        return chat_info_cache[chat_id]

    token = get_tenant_access_token()
    if not token:
        return None

    url = f"https://open.feishu.cn/open-apis/im/v1/chats/{chat_id}"
    headers = {"Authorization": f"Bearer {token}"}

    try:
        response = requests.get(url, headers=headers)
        result = response.json()
        if result.get("code") == 0:
            chat_info_cache[chat_id] = result.get("data", {})
            return chat_info_cache[chat_id]
        else:
            print(f"获取群信息失败: {result}")
            return None
    except Exception as e:
        print(f"获取群信息异常: {e}")
        return None


# 处理普通消息
def on_recv_msg(event):
    # 把 SDK 事件对象转成 dict，方便取字段
    event_dict = json.loads(lark.JSON.marshal(event))
    event_data = event_dict.get("event", {})
    sender_data = event_data.get("sender", {})
    msg_data = event_data.get("message", {})

    sender_open_id = sender_data.get("sender_id", {}).get("open_id", "")
    chat_id = msg_data.get("chat_id", "")
    chat_type_raw = msg_data.get("chat_type", "")

    # ---------- 第 1 步：私聊直接跳过（不请求任何接口） ----------
    if chat_type_raw == "p2p":
        return

    # ---------- 第 2 步：先查群详情，不是外部群直接跳过 ----------
    # get_chat_info 有缓存，同一群只请求一次
    chat_info = get_chat_info(chat_id)
    is_external = chat_info.get("external") if chat_info else None
    if is_external is not True:
        return
    chat_name = chat_info.get("name", "")

    # ---------- 第 3 步：提取被 @ 的人，并判断是否在监听名单 ----------
    content = json.loads(msg_data.get("content", "{}"))
    original_content = content.get("text", "")

    mentioned_users = []
    for mention in msg_data.get("mentions", []):
        mention_open_id = mention.get("id", {}).get("open_id", "")
        mention_name = mention.get("name", "")
        mention_key = mention.get("key", "")

        mentioned_users.append({
            "name": mention_name,
            "open_id": mention_open_id
        })

        # 把 original_content 里的 @_user_1 占位符替换成 @中文名
        if mention_key and mention_name:
            original_content = original_content.replace(mention_key, f"@{mention_name}")

    mentioned_ids = {m["open_id"] for m in mentioned_users}
    mentioned_names = {m["name"] for m in mentioned_users}
    watched_matched = [u for u in WATCH_USERS if u in mentioned_ids or u in mentioned_names]
    if not watched_matched:
        return  # 没 @ 指定人员，跳过，不请求通讯录接口

    # ---------- 第 4 步：到这里才查询发送人详细信息 ----------
    user_info = get_user_info_by_open_id(sender_open_id)
    sender_name = user_info["name"]
    if user_info["is_internal"] is True:
        sender_scope = "内部用户"
    elif user_info["is_internal"] is False:
        sender_scope = "外部用户"
    else:
        sender_scope = "未知"

    # 时间戳：飞书给的是毫秒级 Unix 时间戳，转成 ISO 8601 UTC
    create_time = msg_data.get("create_time")
    if create_time:
        try:
            dt = datetime.fromtimestamp(int(create_time) / 1000, tz=timezone.utc)
            timestamp = dt.strftime("%Y-%m-%dT%H:%M:%SZ")
        except Exception:
            timestamp = str(create_time)
    else:
        timestamp = ""

    output = {
        "original_content": original_content,
        "mentioned_users": mentioned_users,
        "sender_name": sender_name,
        "sender_open_id": sender_open_id,
        "sender_scope": sender_scope,
        "chat_type": "外部群",
        "chat_name": chat_name,
        "chat_id": chat_id,
        "timestamp": timestamp
    }

    print(json.dumps(output, ensure_ascii=False, indent=4))

    # 命中筛选条件后，推送到服务器（推送失败不影响主流程）
    if MY_MESSAGE_WEBHOOK_URL and MY_MESSAGE_WEBHOOK_URL != "xxx":
        try:
            push_all_event_info_to_webhook(output, MY_MESSAGE_WEBHOOK_URL)
            print("推送 Webhook 成功")
        except Exception as e:
            print(f"推送 Webhook 失败: {e}")

# 初始化事件处理器
handler = (
    lark.EventDispatcherHandler.builder("", "")
    .register_p2_im_message_receive_v1(on_recv_msg)
    .build()
)

# 建立长连接
client = Client(APP_ID, APP_SECRET, event_handler=handler)
print("正在连接飞书...")
client.start()

发送消息到群

# 使用此指令前，请确保安装必要的Python库，此指令仅使用Python标准库，无需额外安装。

import json
import urllib.request
import urllib.parse

from typing import *
try:
    from xbot.app.logging import trace as print
except:
    from xbot import print


def feishu_auto_reply(app_id, app_secret, reply_content, sender_open_id, chat_id=None):
    """
    title: 飞书忙碌自动回复
    description: 向指定的飞书用户或群聊发送带 @ 提及的自动回复消息。若提供 % chat_id % 则回复至群聊，否则回复至 % sender_open_id % 的私聊窗口。
    inputs:
        - app_id (str): 飞书应用的 App ID，eg: "cli_xxx"
        - app_secret (str): 飞书应用的 App Secret密匙，eg: "xxx"
        - reply_content (str): 回复的文本内容，eg: "我在忙碌中，稍后回复您"
        - sender_open_id (str): 发送人的 Open ID（用于 @ 提及），eg: "ou_xxx"
        - chat_id (str): 群聊的 Chat ID。如果不为空，则消息发送到该群聊；如果为空，则发送私聊消息给发送人，eg: "oc_xxx"
    outputs:
        - response (dict): 飞书接口返回的原始响应结果，eg: {"code": 0, "msg": "ok", "data": {}}
    """

    def _request(method, url, token=None, params=None, payload=None):
        """统一的 JSON 请求封装（标准库实现）。"""
        if params:
            url += "?" + urllib.parse.urlencode(params)
        data = json.dumps(payload).encode("utf-8") if payload is not None else None
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Content-Type", "application/json; charset=utf-8")
        if token:
            req.add_header("Authorization", "Bearer " + token)
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8")
            raise RuntimeError(f"飞书 API 请求失败 HTTP {e.code}: {body}")
        except Exception as e:
            raise RuntimeError(f"网络请求异常: {str(e)}")

    def _get_tenant_access_token(app_id, app_secret):
        """获取 tenant_access_token（应用身份）。"""
        base_url = "https://open.feishu.cn/open-apis"
        r = _request(
            "POST",
            f"{base_url}/auth/v3/tenant_access_token/internal",
            payload={"app_id": app_id, "app_secret": app_secret},
        )
        if r.get("code") != 0:
            raise RuntimeError("获取飞书 tenant_access_token 失败: " + json.dumps(r, ensure_ascii=False))
        return r["tenant_access_token"]

    def _build_text_with_at(message, mention_id):
        """构造带 @ 提及的文本消息内容。"""
        text = f'<at user_id="{mention_id}"></at> {message}'
        return json.dumps({"text": text}, ensure_ascii=False)

    # 1. 检查必要输入有效性
    if not app_id or not app_secret:
        raise ValueError("参数 app_id 和 app_secret 不能为空")
    
    if not sender_open_id:
        raise ValueError("参数 sender_open_id 不能为空，它是 @ 提及的必要条件")

    # 2. 获取 Token
    token = _get_tenant_access_token(app_id, app_secret)

    # 3. 构造消息内容
    content = _build_text_with_at(reply_content, sender_open_id)

    # 4. 确定接收对象类型
    # 如果提供了 chat_id，则视为群聊回复；否则视为私聊回复
    if chat_id and chat_id.strip():
        receive_id_type = "chat_id"
        receive_id = chat_id.strip()
    else:
        receive_id_type = "open_id"
        receive_id = sender_open_id

    # 5. 调用发送接口
    base_url = "https://open.feishu.cn/open-apis"
    result = _request(
        "POST",
        f"{base_url}/im/v1/messages",
        token=token,
        params={"receive_id_type": receive_id_type},
        payload={
            "receive_id": receive_id,
            "msg_type": "text",
            "content": content
        },
    )

    if result.get("code") != 0:
        raise RuntimeError("飞书消息发送失败：" + json.dumps(result, ensure_ascii=False))

    return result