Lark/飞书 机器人 Python 脚手架

Lark 最大的优点,就是有类似于 Telegram 的「机器人」机制。机器人可以推送通知,可以通过按钮、指令,处理一些简单的东西,比如预约会议室,确认服务上线,等等。

到目前为止,我自己已经使用 Python 写了三个机器人了(OCR 小助手、定时自动换头像和标注队列助手),经历了内部同事自己封装的 SDK,也用过开放平台官方的 SDK,踩了不少坑,也在内部群里看到很多人踩同样的坑。单独水成好几篇文章没太大意思,那就写成一个小集合,顺带做个基于 Flask 的脚手架代码,放 Github 上。为了尽可能多地演示功能,可能代码不是特别精简。

任务

示例代码中,将实现下面几个基本功能:

  1. 对话消息
    1. 用户给机器人发送一个数字 N,机器人返回随机 N 段文字
    2. 用户给机器人发送两个数字 N 和 M,机器人在 M 秒种后随机返回 N 段文字
    3. 用户给机器人发送一个数字 N,并有一个参数 --card,机器人使用卡片返回随机 N 段文字
    4. 用户给机器人发送两个数字 N 和 M,并有一个参数 --card,机器人在 M 秒种后使用卡片返回随机 N 段文字
  2. 指令消息(/ 开头的消息)
    1. 在「对话消息」前面加入 /get 指令。其余参数和作用
  3. 卡片消息
    1. 用户点击卡片上的「更换一批」,机器人更新卡片上的 N 段文字
    2. 用户点击卡片上的「M 秒钟后更换一批」,机器人更新卡片上的 N 段文字

开放平台配置

申请权限

需要申请的权限不多

配置回调

把服务起起来,得到 IP+端口。

代码编写

Git 在这里:https://github.com/jinyu121/lark_bot_demo

异步处理文字消息

这里要先说一下整个处理流程,也就是说,从「用户给机器人发消息」,到「用户看到机器人返回的消息」,中间都发生了什么。

  1. 用户给机器人发消息
  2. 消息到达 Lark/飞书服务器,服务器将消息转发/投递到你的应用(叫做「事件回调」)
  3. 你的应用处理消息,并给 Lark/飞书服务器返回一个 200 消息,表示「消息已收到」
  4. 在「你的应用处理消息」的过程中,给用户发了一条或多条消息

好了,这就算完了。注意重点,「你的应用给飞书/Lark 服务器返回一个 200 消息」表示「这个消息我收到了」,这事儿就算完了,Lark/飞书服务器并没有要求你必须「再发一个消息告诉用户什么什么东西」。

那么,用户的感觉是「我给机器人发送消息,机器人响应了我」,这个是怎么做到的呢?注意看「你的应用处理消息」,这个「处理」可能包含各种东西,例如打 Log,例如查数据库,再例如,给用户发消息。对的,「给用户发消息」只是「处理消息」过程中的一个可选项,是机器人「主动」做的,并且可以做许多次。

有了上面的理解,我们就可以写最最基本的代码了。最基本的代码可以见 SDK 的官方 Demo。这个 Demo 写的是「收到消息回调,显示消息内容」,不涉及「回复消息」。

必须在 1 秒内返回结果:将实际处理放到线程里

Lark/飞书有个比较奇怪的设定:用户给机器人发消息,机器人应用收到消息后,必须在 1 秒内返回一个 200 响应,表示「我收到消息了」。如果你的处理比较简单,例如只是查个数据库,更改个状态啥的,那么 1 秒钟内绝对能处理得完。但如果你的处理比较重、比较耗时,例如文件下载、图片过识别模型、处理视频,那么 1 秒钟是不够用的。这个时候,如果你超时了,没有在 1 秒种内返回个 200 确认,那么 Lark/飞书服务器就会认为「消息投递失败」,于是会在 5 秒、5 分钟、1 小时、6 小时后分别再尝试投递一次消息。但是呢,都说了处理消息很耗时,同一个消息处理,这次 20 秒才能处理完,下次不可能 1 秒就处理完呀,于是当服务器再次投递消息的时候,你依旧会超时。这样不仅你的应用会重复处理,如果处理过程中有「给用户发消息」的操作的话,用户也会多次收到消息,体验不好(想象下晚上 1 点收到消息的抓狂)。

这个时候,我们可以这样做:收到 Lark/飞书服务器转发来的消息后,新开一个线程用于慢慢处理消息,同时主线程直接返回 200,让 Lark/飞书服务器认为「投递消息成功」,避免消息再次投递。而且因为 Lark/飞书服务器需要的只是一个 200 的状态确认,不强制要求于用户交互(不强制要求要「回应」用户消息),所以我们可以将「确认收到消息」和「实际处理」分开。

这里一切都看起来「理所当然」,但是很快就会遇到一个问题:在新开的处理线程里面无法获取到 current app,导致一大堆事儿干不了。这里我们使用一个叫 flask-thread 的包来替代 Python 原生的 thread。这个包是对原生 thread 包的很薄的一层封装,主要干的事儿就是把 flask 的 app_stack 给传到了线程里,这样线程里面就能用 current_app 了。

消息滤重:Redis

上面说了,「如果在 1 秒内没返回结果,那么会 Lark/飞书服务器会延迟一会儿再重复投递消息」。在将消息处理放到线程里后,99.999% 的情况下,都可以在 1 秒内返回个 200 确认出来,同时线程里面也将该处理的东西都处理了,该通知的用户也都发了消息。但是,你自己这里万事 OK 了,Lark/飞书服务器也是会抽风的啊,比如网络出现了什么岔子,导致 Lark/飞书服务器没收到这个 200 确认,咋办?总不能老老实实再处理一遍,再通知一遍用户?

所以 Lark/飞书在转发消息的时候,都会带上消息 id,我们可以利用消息 id 进行消息去重。在处理完成后,我们可以把消息 id 记录到数据库中,再来消息的时候就先检查下数据库,如果消息 id 在数据库中,那么说明这个是重复消息,不用处理,只管回个 200 敷衍确认下就好了;如果不在数据库里,那么该干啥就干啥。

这里推荐使用「带过期时间」的数据库(例如,redis)来做这个事儿。一个是因为这种记录性质的东西本身不是啥重要内容,永久存储的话划不来;另一个是,Lark/飞书人家也说了,超过 6 个小时后就不会再投递了,也就是说这个 id 过了 6 个小时后就完全没用了,永久存储的话没意义。所以,使用 redis,设置个最多 12 小时的过期时间,就能满足要求了。

子线程需要带上请求环境

写起来都挺简单的。但是写不久就会遇到一个问题:这个线程只有「应用」环境,没有「请求」环境。所以如果只是纯 Python 的操作,例如打打 Log,整理整理数据库啥的还好说,但是如果想使用 url_for 等东西,就会报错,或者结果不符合预期,比如报「没有 BasePath」啥啥的,不能生成相对 URL,只能生成绝对 URL,这个对灵活部署十分不友好。

所以解决方法也很简单,带上请求环境,或者模拟个请求环境,不就好了么。Flask 里面提供了这样的函数(copy_current_request_context),直接用就是了,但是我自己在使用的时候,就「这句话应该写在哪儿」纠结了好几个早上,最终发现应该写在生成线程的地方。

from flask import copy_current_request_context
from flaskthreads import AppContextThread
AppContextThread(target=copy_current_request_context(some_function)).start())

同步处理卡片消息

这里也先说一下,从用户和卡片交互,到用户收到「与卡片交互完成」,中间发生了啥:

  1. 用户点击卡片上的可交互元素(例如,按钮,下拉选框等),将交互的「值」传给 Lark/飞书服务器
  2. Lark/飞书服务器将交互的「值」转发给你的应用
  3. 你的应用进行处理,并给 Lark/飞书服务器返回交互结果(例如卡片内容更新)
  4. Lark/飞书服务器将交互结果返回给用户,改变用户端可以看到的卡片

可以看出来,这里的处理方式和上面处理消息的方式不同:上面是「随便回个 200,确认下收到消息就完事儿了」,这里是「一定要返回个处理结果」。并且,由于是交互,所以成功就是成功,失败就是失败(失败后用户可以自己重试),Lark/飞书服务器不会自动重新投递消息。

最基本的代码见 SDK 官方 Demo。这个 Demo 写的是「收到消息卡片回调,更新卡片内容」。

参数传递

如果只是「展示消息卡片」,那么不需要传递啥东西。

如果需要「交互」,那么就需要考虑一下了。如果用户点击卡片上的按钮后,回传给自己服务的永远都是纯纯的「用户点击了按钮」,那么我完全不知道它按了啥。所以用户点击卡片上的东西之后,肯定是需要回传一些特殊的东西,我才能对不同卡片进行不同处理。

在文档里面,我们可以看到,如果是按钮,那么会有 value 这个东西,也就是说,点击按钮后,value 将被回传给我们的应用。如果是 SelectMenu 和 Overflow,它的值有两部分,一部分是固定的 value,另一部分是每个选项的 value,在回传的时候,会回传固定部分的 value 和你选择的东西。

OK,那么我们应该如何进行路由呢?最简单的办法是,把路由信息写在 value 里面,然后我们的程序解析 value,得到路由终端。比如,我使用 catrgory 来标记「使用哪个类进行处理」,使用 action 来标记「使用类中的哪个方法进行处理」。至于其他固定参数,我固定写在 data 里。

必须在 3 秒内返回结果:异步更新卡片

由于是同步更新,所以「等待时间」不能太长,否则体验不好。Lark/飞书要求的是在 3 秒内处理完,并返回结果。一般来说这个时间是凑合够用的:卡片一般用在「确认」场景,做的重活、耗时的活儿比较少。

如果是这样的话,那么我们就可以开始写代码了。

当然,如果你能预料到你的卡片干的活比较重、比较耗时,不保证在 3 秒之内能完成处理,那么也可以仿照上面处理消息的套路,先快速返回一个「200 消息确认+不更新卡片」,然后单独开一个线程处理消息,处理完成后,使用「异步更新卡片」更新交互的卡片。

最后是个吐槽

嗯,Lark/飞书的开放平台和 SDK 确实不咋好用,Lark/飞书本身也抛弃了之前那么漂亮的设计,改成了现在这个比较奇怪的模样,连 Logo 也变成了异形扳手。另外 Lark 目前越做越大,越来越「大而全」,我不知道是不是个好事儿。功能越来越多的话,就像在逛杂货铺,真的是啥功能都有,但是「能一击解决我的痛点」的功能,越来越少了。但总之,要比国内某个蓝色软件和绿色软件好用得多得多。

然后是,Lark/飞书 Python SDK 的 Demo 的「重复感」太强了,想找啥东西都有种「从三叶草丛(重复的东西)里面找四叶草(不重复的东西)」的感觉。两个 Demo,里面只有两三行不一样,作者的想法是「随便打开一个文件,东西都是完整的」,但我并不认为这是个好想法。

然后,有几个词我一直不理解为啥要这样叫,比如「免登」,比如特色功能「Reaction」是不是翻译成「回应」会更好一点(目前是「表情回复」,我觉得怪怪的)。

最后,Go 版本的 SDK,可以试试这个

发表回复

您的电子邮箱地址不会被公开。