背景

公司赠品信息每年数量在3W+，飞书的多维表格查询数量不够，参考了影刀文章里面的（ “出道即颠峰“大佬写的文章 ）

1.功能介绍

1. 秒级精准查询：用户在飞书聊天框输入订单号，机器人通过长连接实时对接系统，3 秒内返回订单绑定的赠品信息（名称、规格、发放状态、物流单号）。
2. 异常清晰提示：若输入的订单号不存在，机器人将直接回复 “该订单不存在，请联系负责人处理”，无需用户额外追问。
3. 操作极简高效：无需跳转外部系统，在飞书私聊 / 群聊内直接输入即可查询，适配日常使用习惯，无需额外学习成本。
4. 安全稳定传输：订单号、赠品信息通过加密长连接传输，仅企业内部授权人员可查询，保障数据安全。

2.影刀配置

lark-oapi == 1.4.24
PyMySQL == 1.1.1

Mysql8.0安装教程

3.飞书配置

3.1进去飞书开发者平台（ 飞书开发者平台 ）

3.2代码长连接

import lark_oapi as lark

## P2ImMessageReceiveV1 为接收消息 v2.0；CustomizedEvent 内的 message 为接收消息 v1.0。

def do_p2_im_message_receive_v1(data: lark.im.v1.P2ImMessageReceiveV1) -> None:
    print(f'[ do_p2_im_message_receive_v1 access ], data: {lark.JSON.marshal(data, indent=4)}')
def do_message_event(data: lark.CustomizedEvent) -> None:
    print(f'[ do_customized_event access ], type: message, data: {lark.JSON.marshal(data, indent=4)}')
event_handler = lark.EventDispatcherHandler.builder("", "") \
    .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \
    .register_p1_customized_event("这里填入你要自定义订阅的 event 的 key，例如 out_approval", do_message_event) \
    .build()
def main():
    cli = lark.ws.Client("app_id", "app_Secret",
                         event_handler=event_handler,
                         log_level=lark.LogLevel.DEBUG)
    cli.start()
if __name__ == "__main__":
    main()

如果出现这个（表示长连接成功了）

3.3配置长连接（配置长连接会出现权限的问题，按照提示打开即可）

4.代码实例

import requests
import json
import pymysql
from lark_oapi import EventDispatcherHandler, ws, JSON, im, LogLevel
from xbot import print


class FeishuConfig:
    '''飞书API的配置信息
    配置说明：需从飞书开放平台「企业自建应用」中获取对应信息
    '''
    # 飞书应用唯一标识（必填）：开放平台应用详情页的「App ID」
    APP_ID = ''
    # 飞书应用密钥（必填）：开放平台应用详情页的「App Secret」
    APP_SECRET = ''


class DatabaseConfig:
    '''数据库配置信息
    配置说明：需填写目标MySQL数据库的连接信息及数据表名称
    '''
    # 数据库服务器地址（必填）：本地库填localhost，远程库填IP地址（示例：192.168.1.100）
    HOST = ''
    # 数据库端口（必填）：MySQL默认端口为3306，需填写数字类型（示例：3306）
    PORT = 3306
    # 数据库登录账号（必填）：访问数据库的用户名（示例：root）
    USER = ''
    # 数据库登录密码（必填）：访问数据库的密码（示例：123456）
    PASSWORD = ''
    # 数据库名称（必填）：要查询的赠品信息所在的数据库名（示例：zengpinku）
    DB_NAME = ''
    # 数据库字符集（固定）：适配中文的字符集，无需修改
    CHARSET = 'utf8mb4'
    # 数据表名称（必填）：要查询的赠品信息表名（示例：zaze赠品信息）
    TABLE_NAME = ''

class DatabaseApi:
    '''数据库操作类'''

    @staticmethod
    def query_data(query_key):
        '''根据订单号查询数据库中zaze赠品信息表的数据'''
        conn = None
        cursor = None
        try:
            # 建立数据库连接
            conn = pymysql.connect(
                host=DatabaseConfig.HOST,
                port=DatabaseConfig.PORT,
                user=DatabaseConfig.USER,
                password=DatabaseConfig.PASSWORD,
                db=DatabaseConfig.DB_NAME,
                charset=DatabaseConfig.CHARSET
            )

            # 创建游标
            cursor = conn.cursor(pymysql.cursors.DictCursor)

            # 按订单号精确查询（核心改动2：使用配置中的表名）
            sql = f"SELECT * FROM `{DatabaseConfig.TABLE_NAME}` WHERE 订单号 = %s"
            cursor.execute(sql, (query_key,))

            # 获取查询结果
            results = cursor.fetchall()
            return results

        except Exception as e:
            print(f"数据库查询错误-联系赠品负责人更新-赠品数据库: {str(e)}")
            return None  # 数据库错误时返回None
        finally:
            # 关闭连接
            if cursor:
                cursor.close()
            if conn:
                conn.close()


class FeishuApi:
    '''FeishuApi类用于处理与飞书API的交互'''
    TOKEN_URL = 'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal'
    REPLY_MESSAGE_URL_TEMPLATE = 'https://open.feishu.cn/open-apis/im/v1/messages/{message_id}/reply'
    HEADERS = {'Content-Type': 'application/json; charset=utf-8'}

    def __init__(self):
        self.session = requests.Session()
        self.token = self.get_token()

    def get_token(self):
        '''获取飞书API的访问令牌'''
        data = {'app_id': FeishuConfig.APP_ID, 'app_secret': FeishuConfig.APP_SECRET}
        try:
            response = self.session.post(self.TOKEN_URL, headers=self.HEADERS, json=data)
            response.raise_for_status()
            token = response.json().get('tenant_access_token')
            if not token:
                raise ValueError("获取token失败，返回结果中无tenant_access_token")
            return token
        except Exception as e:
            print(f"获取token错误: {str(e)}，返回内容: {response.text if 'response' in locals() else '无响应'}")
            raise

    def reply_message(self, message_id, user_id, message):
        '''回复飞书群聊消息'''
        url = self.REPLY_MESSAGE_URL_TEMPLATE.format(message_id=message_id)
        try:
            content = json.dumps({
                "text": f'<at user_id="{user_id}"></at> {message}'
            }, ensure_ascii=False)

            data = {
                "content": content,
                "msg_type": "text"
            }
            headers = {'Authorization': 'Bearer ' + self.token, **self.HEADERS}
            response = self.session.post(url, headers=headers, json=data)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(
                f"回复消息失败: {str(e)}，URL: {url}，请求内容: {data}，返回内容: {response.text if 'response' in locals() else '无响应'}")
            raise


def handle_p2_im_message(data: im.v1.P2ImMessageReceiveV1):
    '''处理接收到的个人或群聊消息'''
    try:
        data_dict = json.loads(JSON.marshal(data))
        message_id = data_dict["event"]["message"]["message_id"]
        content_json = json.loads(data.event.message.content)
        content = content_json.get("text", "").strip()
        
        # 移除@机器人前缀
        if "@_user_1" in content:
            content = content.replace("@_user_1", "").strip()

        # 获取用户ID
        sender_id = data_dict["event"]["sender"]["sender_id"]
        user_id = sender_id.get("open_id") or sender_id.get("user_id") or sender_id.get("union_id")

        print(f"收到消息：订单号={content}，用户ID={user_id}，消息ID={message_id}")

        # 数据库查询逻辑
        if content:
            db_results = DatabaseApi.query_data(content)
            print(f"数据库查询结果：{db_results}")

            # 处理查询结果
            if db_results is None:  # 数据库查询错误
                reply_msg = "订单号不存在，请联系赠品负责人！！！"
            elif len(db_results) == 0:  # 订单号不存在
                reply_msg = "订单号不存在，请联系赠品负责人！！！"
            else:  # 有查询结果
                result_str = ""
                for item in db_results:
                    row_str = "\n".join([f"{k}: {v}" for k, v in item.items()])
                    result_str += f"【订单 {item.get('订单号')} 信息】\n{row_str}\n\n"
                reply_msg = f"查询到订单号为「{content}」的信息：\n{result_str}"
        else:
            reply_msg = "请输入订单号进行查询"

        # 发送回复
        feishu = FeishuApi()
        feishu.reply_message(
            message_id=message_id,
            user_id=user_id,
            message=reply_msg
        )
    except Exception as e:
        print(f"处理消息失败: {str(e)}")


def main():
    '''启动飞书长连接 WebSocket客户端'''
    event_handler = EventDispatcherHandler.builder("", "") \
        .register_p2_im_message_receive_v1(handle_p2_im_message) \
        .build()

    cli = ws.Client(FeishuConfig.APP_ID, FeishuConfig.APP_SECRET, event_handler=event_handler, log_level=LogLevel.DEBUG)
    cli.start()


main()

注意！！！

需要自己配置飞书机器人和数据库的信息

5.执行效果