一、飞书长链接
简介
开发者通过集成飞书 SDK 与开放平台建立一条 WebSocket 全双工通道，当有事件回调发生时，开放平台会通过该通道向开发者发送消息。
与传统的 Webhook 模式相比，长连接模式大大降低了接入成本，将原先 1 周左右的开发周期降低到 5 分钟。具体优势如下：
-
测试阶段无需使用内网穿透工具，通过长连接模式在本地开发环境中即可接收事件回调；
-
只在建连时进行鉴权，后续事件推送均为明文数据，无需开发者再处理解密和验签逻辑；
-
只需保证运行环境具备访问公网能力即可，无需提供公网 IP 或域名；
-
无需部署防火墙和配置白名单。
接入限制
每个应用最多建立 50 个连接（每初始化一个 client 就是一个连接）
网页配置
-
事件与回调，订阅方式更换为：使用长链接接收事件
-
回调配置，订阅方式更换为：使用长链接接收事件
二、SDK 接入
Github 地址：https://github.com/larksuite/oapi-sdk-python
python 安装 SDK：pip install lark-oapi 1.4.0
以下只简单介绍几个常用的方法，具体可以在飞书开放平台上查看，https://open.feishu.cn/
我这里先在 main.py 中单独起了一个线程用来执行长链接，如果你不单独启线程会阻塞你自身功能代码。
# main.py
if __name__ == '__main__':
# 启动飞书 WebSocket 客户端线程
start_client_thread()
run("main:app", host=host, reload=True, port=port)lark_client.py具体实现 start_client_thread 中内容
在下面代码中只需要将 APP_ID 和APP_SECRET改成你自己的即可
"""
@Author: Jason
@FileName: lark_client.py
@DateTime: 2024/11/17 22:06
@SoftWare: PyCharm
"""
import asyncio
import threading
import lark_oapi as lark
from backend.core import settings
from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTrigger, P2CardActionTriggerResponse
from backend.core.lark.lark_card import handle_callback
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)}')
def do_card_action_trigger(data: P2CardActionTrigger) -> P2CardActionTriggerResponse:
datas = lark.JSON.marshal(data)
print(datas)
asyncio.create_task(handle_callback(datas))
resp = {
"toast": {
"type": "info",
"content": "功能已触发成功！"
}
}
return P2CardActionTriggerResponse(resp)
# 飞书卡片回调
event_handler = lark.EventDispatcherHandler.builder("", "") \
.register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \
.register_p1_customized_event("message", do_message_event) \
.register_p2_card_action_trigger(do_card_action_trigger) \
.build()
def start_lark_client():
cli = lark.ws.Client(
settings.LarkInfo["APP_ID"],
settings.LarkInfo["APP_SECRET"],
event_handler=event_handler,
log_level=lark.LogLevel.DEBUG
)
cli.start()
def start_client_thread():
threading.Thread(target=start_lark_client, daemon=True).start()三、案例介绍
卡片回调
do_card_action_trigger
这是由于飞书要求触发事件后需要在三秒内返回结果, 但是我们支出功能在三秒内是肯定无法完成的, 所以在这里我只告诉用户你已经触发了这个功能, 然后异步执行这个函数 handle_callback 去进行内部处理
定义飞书类
每次发送消息或者更新消息都需要 Token 值，这里直接封装一个类直接调用即可
class LarkClient:
def __init__(self, app_id=settings.LarkInfo["APP_ID"], app_secret=settings.LarkInfo["APP_SECRET"],
log_level=lark.LogLevel.DEBUG):
self.client = lark.Client.builder() \
.app_id(app_id) \
.app_secret(app_secret) \
.log_level(log_level) \
.build()
def __getattr__(self, item):
# 当尝试获取 LarkClient 类不存在的属性时，会转发到 self.client
return getattr(self.client, item)获取用户 ID
通过手机号码获取到用户 ID
定义后直接使用：LarkClient().batch_get_id(mobiles=mobiles)即可完成调用，非常方便快捷。
def batch_get_id(self, mobiles, user_id_type="user_id"):
# 获取用户 user ID
request = BatchGetIdUserRequest.builder() \
.user_id_type(user_id_type) \
.request_body(BatchGetIdUserRequestBody.builder()
.mobiles(mobiles)
.build()) \
.build()
response = self.client.contact.v3.user.batch_get_id(request)
if not response.success():
lark.logger.error(
f"batch_get_id failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")
return None
return response发送消息
获取到用户的 ID 后，调用
create_message，即可给用户发送对应消息实例：
LarkClient().create_message(lark_id, data)
def create_message(self, receive_id, content, receive_id_type="user_id"):
"""
创建新的消息
:param receive_id:
:param content:
:param receive_id_type:
:return:
"""
request = CreateMessageRequest.builder() \
.receive_id_type(receive_id_type) \
.request_body(CreateMessageRequestBody.builder()
.receive_id(receive_id)
.msg_type("interactive")
.content(content)
.build()) \
.build()
response = self.client.im.v1.message.create(request)
if not response.success():
lark.logger.error(
f"create_message failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}")
return None
return response四、卡片消息
在飞书中，卡片消息 因其简洁、美观而成为使用频率最高的消息类型之一。使用卡片消息非常便捷：首先定义好模板，然后在创建消息时，直接填入 卡片模板 ID和相应的参数即可。
卡片搭建网址：https://open.feishu.cn/cardkit?from=op_develop_app
进入卡片搭建后，可以新建一个卡片，从组件中拖入到右侧，然后开始设计编辑内容。最后保存发布后就可以进行调用了。
发送卡片消息
可以看到上图中 富文本 中有两个变量，我们在传递可以不同的变量，具体实现可看：
content_value = {
"type": "template",
"uuid": str(uuid.uuid4()),
"data": {
"template_id": "AAq8******",
"template_version_name": "",
"template_variable": {
"fs_number": fs_number,
"cw_number": cw_number,
"env": env,
"time": Public_fun.get_now_time_format()
}
}
}
data = json.dumps(content_value, ensure_ascii=False)
LarkClient().create_message(lark_id, data)更新卡片消息
飞书中 更新卡片消息 的原理是通过覆盖之前发送的消息来实现的，而不是对卡片信息进行局部更新。这是一个常见且实用的功能。
async def update_order_info_tips(fs_number: str, dk_number: str, environment: int, message_id: str, order_msg:str):
# 更新未录单卡片消息
content_value = {
"type": "template",
"uuid": str(uuid.uuid4()),
"data": {
"template_id": LarkConfig.card["orders_pay_tip"],
"template_version_name": "",
"template_variable": {
"fs_number": fs_number,
"dk_number": dk_number,
"env_name": settings.env[environment]["zh_name"],
"order_msg": order_msg
}
}
}
data = json.dumps(content_value, ensure_ascii=False)
lark_fun = LarkClient().patch_message(data, message_id)这里也是需要填入更新后的卡片 ID template_id，非常方便，其他都是一些自定义变量。最要注意的是message_id，这个是原消息的 ID，传入了这个才会更新你指定的消息。
五、坑点
长链接仅支持 protobuf3 版本
在使用 mysql-connector-python 时遇到了问题，因为该库需要 Protobuf 版本大于 4.21.1，而这与飞书 SDK 冲突。
经过在 Github 和 Google 上查找大量资料未果后，最终在官方群得知：由于飞书长连接依赖于字节的长连接基础设施，字节未升级其组件，因此飞书也无法支持更高版本的 Protobuf。
最后，我只能卸载 mysql-connector-python
