作为一名技术向作者,平时写写脚本、做做数据采集是常事。
但是等到要采集的数据逐渐变多起来,我们就没有充足的时间每天查看它们的执行结果了,但是有些数据是不可重现的,一旦错过就无法再采集,有时一个脚本报错停止好几天之后才发现异常,而中间的数据已经无法补回。
所以我们需要一个消息服务,在我们的脚本成功运行或者报错时将信息发送到手机上,便于我们掌握脚本的运行状态。
盘点我曾用过的方案
首先是阿里云的短信推送,但价格算不上香,一条 0.045 元,如果有五个脚本,每个月就要花费 6.75 元,如果要推送给更多人,成本还会增加,对于预算敏感的人来说,并不是好的选择。
而且阿里云的短信服务配置相对复杂,模板需要审核,而且有字数限制,推送渠道为短信也让它被忽略的概率大大提高。
接下来是 Server 酱,配置简单,免费可用,不需要过多审核,消息可以自定义,但免费版每天只能推送三条消息,渠道选择虽然多样,但有些需要配置企业微信,有些要软件后台常驻,有些可能被随时关停,限制有点多。
我在一年之前用过一段时间飞书,对于团队管理而言,这是一个不错的工具,但今天我们用到的是它的 API 接口,用来实现一个属于自己的消息推送服务。
操作流程
创建机器人
下载并注册飞书略过。
如果没有绑定邮箱,请先进行绑定。
进入飞书开放平台,点击“创建应用”,应用名称和描述根据自己需求填写即可,注意图标一定要上传,否则后面会出现报错。
image在“企业自建应用”中找到我们刚刚创建的 Bot,点击进入:
image进入“应用功能” > “机器人”,启用机器人:
image进入“权限管理”,添加“获取与发送单聊、群组消息”权限:
image进入“应用发布” > “版本管理与发布”,发布一个新版本,并提交审核申请,这里的版本号和更新说明是必填项。
image-20210912222528945在飞书消息页面,“开发者小助手”会提示你有新的应用等待审核,不用理会,因为你注册的飞书是个人版,审核需要升级为团队,虽然不收费,但操作较为复杂。
调试接口
查阅飞书开放平台相关文档,我们可以找到发送消息接口的调用方式。我们的需求不需要消息卡片,所以使用的是老版本接口,新版本发送消息卡片的接口需要更多配置,有兴趣的小伙伴可以自行研究。
在编写程序之前,我们需要确认接口的可用性。回到应用页面,选择“凭证与基础信息”,复制 App ID 和 App Secret 备用。
image打开 API 鉴权机制文档,点击“尝试一下”,在左侧填入App ID 和 App Secret,点击“发起调用”。
image复制返回的 tenant_access_token 备用。
image-20210912224229604可以使用任何能发送网络请求的工具对接口进行调试,这里我使用 VS Code 的 Thunder Client 插件,请求配置如下:
请求网址: https://open.feishu.cn/open-apis/message/v4/send/
请求方式:POST
需要两个 Headers,一个是之前获取的 tenant_access_token,注意要加固定前缀“Bearer ”;另一个是固定字符串,根据开发文档填写即可。
image请求体格式是 Json,第一个参数可以填 open_id、user_id、email 或者 chat_id,这里为了简化流程使用 email。
msg_type 是固定值“text”。
“content”项里有一个子项“text”,里面填入要发送的文本。
image完成填写后点击发送,返回信息类似这样:
image-20210912230129119code 为 0 代表发送成功,返回的 message_id 我们不需要用到,msg 为“ok”。
同时,你将在飞书中看到你发送的消息:
image至此,你已经使用这个接口成功发送了一次消息。
但每次都需要获取鉴权信息、设置请求头和请求体未免有些麻烦,接下来我们使用代码实现这一流程。
代码实现
我们使用 Python 的 Requests 库进行开发,完整代码如下:
import requests
request_url = "https://open.feishu.cn/open-apis/message/v4/send/"
headers = {
"Authorization": "Bearer t-e9214fc4*************************76957a5",
"Content-Type": "application/json; charset=utf-8"
}
data = {
"email": "yehaowei20060411@qq.com",
"msg_type": "text",
"content": {
"text": "测试消息"
}
}
req = requests.post(request_url, headers=headers, json=data)
if req.status_code != 200 or req.json()["code"] != 0:
raise Exception(req.json()["code"])
else:
print("发送成功!")
print("message_id:", req.json()["data"]["message_id"])
运行代码,输出“发送成功”和 message_id。
为了便于使用,我们将其封装为一个函数,并增加自动获取鉴权信息的逻辑,代码如下:
import requests
def SendByFeishu(app_id, app_secret, email, text):
headers = {
"Content-Type": "application/json; charset=utf-8"
}
data_to_get_token = {
"app_id": app_id,
"app_secret": app_secret
}
headers["Authorization"] = "Bearer " + requests.post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
headers=headers, json=data_to_get_token).json()["tenant_access_token"]
data_to_send_message = {
"email": email,
"msg_type": "text",
"content": {
"text": text
}
}
req = requests.post("https://open.feishu.cn/open-apis/message/v4/send/", headers=headers, json=data_to_send_message)
if req.json()["code"] != 0:
raise Exception("消息发送失败,错误码:" + str(req.json()["code"]))
else:
return req.json()["data"]["message_id"]
调用该函数,返回 message_id 即为发送成功。
SendByFeishu("cli_a1b4*******8d00c", "FvUtRVSA********************LW00t", "yehaowei20060411@qq.com", "测试消息")
至此,我们成功使用飞书完成了脚本运行状态推送服务的搭建。
网友评论