lark实际就是飞书的国际版，目前使用了lark机器人，常用的有两种机器人，一种是普通机器人，另外一种就是具有对话能力的机器人，实际第二种机器人就是第一种机器人增加了一些能力而已，当然还有其他能力也可以添加给机器人，但是我们常用到的就是这两种。
1.普通机器人
没有对话功能，你可以通过添加的机器人提供的Webhook 地址向机器人推送信息，例如我们经常需要的就是报警到lark，那么我们就可以在集群上的报警通过这个webhook推送到lark上，这样就能及时关注发生的问题。使用方法大致如下，webhook地址和secret需要在机器人配置中获取。
from typing import Tuple, Optional, Any
import requests
import hashlib
import base64
import hmac
import time
import logging
import os
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(filename)s[line:%(lineno)d] %(funcName)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
WEBHOOK_URL = "https://open.larksuite.com/open-apis/bot/v2/hook/xxx"
SECRET = "xxx"
#生成签名
def gen_sign(secret: str) -> tuple[str, str]:
timestamp = str(int(time.time()))
string_to_sign = f"{timestamp}\n{secret}"
hmac_code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()
sign = base64.b64encode(hmac_code).decode('utf-8')
return timestamp, sign
#推送信息到Lark
def push_to_lark(content: str, message: str, columns: list[dict[str, str | int]]) -> None:
timestamp, sign = gen_sign(SECRET)
payload = {
"timestamp": timestamp,
"sign": sign,
"msg_type": "interactive",
"card": {
"schema": "2.0",
"config": {"wide_screen_mode": True},
"body": {
"elements": [
{
"tag": "div",
"text": {
"content": "<at id=all></at> {}".format(message),
"tag": "lark_md"
}
},
{
"tag": "table",
"columns": columns,
"rows": content
}
]
}
}
}
try:
ret = requests.post(WEBHOOK_URL, json=payload).json()
logging.info("推送结果, {}".format(ret))
except Exception as e:
logging.error("额推送失败, {}".format(e))
return None
#业务操作
def stat_count(fa_list: list) -> None:
content = []
succ_total, fail_total = 0, 0
for idx,uuid in enumerate(fa_list):
try:
...... 省略业务代码
succ_total += succ_number
fail_total += fail_number
if int(fail_number) != 0:
if fail_number >= FAILURE_NUMBER or fail_number/(succ_number+fail_number)*100 >= FAILURE_RATE:
content.append({
"col1" : fail_number,
"col2" : succ_number,
"col3" : f"{fail_number/(succ_number+fail_number)*100:.2f}%" if succ_number + fail_number > 0 else "0.00%"
})
except Exception as e:
logging.error("模型数据解析异常: {}，接口返回结构异常".format(e), exc_info=True)
if fail_total == 0:
total_fail_rate = "0%"
else:
total_fail_rate = "{:.2f}%".format(fail_total*100/(succ_total+fail_total))
message = "1小时内平台模型总请求数为{}，请求失败数为{}，总失败率为{}".format(succ_total+fail_total, fail_total, total_fail_rate)
columns = [
{"name": "col1", "display_name": "失败数", "data_type": "text"},
{"name": "col2", "display_name": "成功数", "data_type": "text"},
{"name": "col3", "display_name": "错误率", "data_type": "text"}
]
push_to_lark(content, message, columns)2.对话机器人
这种就是具有对话功能，我们经常用到的就是需要查询一些信息，如果有经常需要查的信息可以通过机器人来实现，避免分配这些人一些平台登录权限或者是数据权限。对话机器人稍微繁琐一点，需要去开放平台申请机器人的能力，并添加这些能力。
2.1.创建自建应用与配置
首先去开放平台：https://open.larksuite.com/，然后进入开发者后台，创建企业自建应用。然后去申请机器人权限
机器人权限一般至少要以下几种
获取单聊、群组消息：im:message
更新消息：im:message:update
获取与发送单聊、群组消息：im:message
获取群组中用户@机器人消息：im:message.group_at_msg:readonly
以应用的身份发消息：im:message:send_as_bot
时间与回调里面可以根据自身情况选择，如果可以暴露服务接口给机器人那就选这个，如果是内网无暴露服务那么可以选择长连接的方式，然后这里就配置基本完成。
2.2.实现机器对话
主要是通过在群里@机器人然后让触发代码里面事件帮我们做一些工作，代码如下，例如@它做巡检或者分析，然后把结果回复到Lark消息群里。首先需要单独安装lark_oapi
pip install lark_oapi然后我的部分代码如下
from typing import Optional, List
from collections import deque
from urllib.parse import urlparse
import lark_oapi as lark
import dns.resolver
import requests
import logging
import time
import json
import math
import statistics
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(filename)s[line:%(lineno)d] %(funcName)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
APP_ID = "cli_aaa10e90c0391eea"
APP_SECRET = "BxqTpb5TKI0kNiwiUOKpfpYpj4LoCc85"
DOMAIN = "https://open.larksuite.com"
HANDLED_EVENT_IDS = deque(maxlen=500)
def send_message_templete(sender_id: str, cmd: str, field_list: list) -> dict:
card = {
"config": {
"wide_screen_mode": True
},
"header": {
"template": "blue",
"title": {"tag": "plain_text", "content": f"{cmd}查询结果"}
},
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": f"<at id={sender_id}></at> 你查询的{cmd}结果如下: "
}
},
{"tag": "hr"},
{
"tag": "div",
"fields": field_list
}
]
}
return card
def error_message_templete(sender_id: str, tips: str = "您的请求未查询到任何结果！") -> dict:
error_card = {
"config": {
"wide_screen_mode": True
},
"header": {
"template": "blue",
"title": {"tag": "plain_text", "content": "查询结果"}
},
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": f"<at id={sender_id}></at> {tips}"
}
}
]
}
return error_card
def help_message_templete(sender_id: str) -> dict:
help_card = {
"config": {
"wide_screen_mode": True
},
"header": {
"template": "blue",
"title": {"tag": "plain_text", "content": "查询机器人命令帮助"}
},
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": f"<at id={sender_id}></at> 支持以下查询指令，请@机器人发送对应命令"
}
},
{"tag": "hr"},
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": """**1. 任务巡检**
命令格式：`task param`
示例：`task check`
功能：巡检服务器"""
}
},
{"tag": "hr"},
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": """**2. 数据统计分析**
命令格式：`task param`
示例：`task analysis`
功能：分析当前日志数据"""
}
}
]
}
return help_card
#接收事件
def do_p2_im_message_receive_v1(data: lark.im.v1.P2ImMessageReceiveV1) -> None:
timestamp = int(time.time()*1000)
chat_id = data.event.message.chat_id
sender_id = data.event.sender.sender_id.open_id
event_id = data.header.event_id
if event_id in HANDLED_EVENT_IDS:
logging.info(f"重复事件已跳过 event_id:{event_id}")
return
HANDLED_EVENT_IDS.append(event_id)
content = json.loads(data.event.message.content)
parts = content["text"].split(" ")
cmd = parts[1].lower()
help_message = help_message_templete(sender_id)
error_message = error_message_templete(sender_id)
#@所有人的消息直接跳过
if parts[0] == "@_all":
return
param = parts[2]
if cmd == "check":
......
field_dict = [
["ID", id],
["Status", status],
["ISP", isp]
]
elif cmd == "analysis":
......
else:
send_reply_message(error_message, chat_id)
return
fields = []
for d in field_dict:
fields.append({
"tag": "div",
"text": {"tag": "lark_md", "content": "**{}**\n{}".format(d[0], d[1])}
})
card = send_message_templete(sender_id, cmd, fields)
send_reply_message(card, chat_id)
return None
#注册事件
event_handler = lark.EventDispatcherHandler.builder("", "") \
.register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \
.build()
def main() -> None:
# 构建 client Build client
cli = lark.ws.Client(APP_ID, APP_SECRET,
event_handler=event_handler, log_level=lark.LogLevel.DEBUG,
domain=DOMAIN)
# 建立长连接 Establish persistent connection
cli.start()
if __name__ == "__main__":
main()
上述代码中需要注意的是，长连接中如果我们在做后端的的查询分析比较耗时的时候出触发多次事件导致对话会返回多次机器人回复，我目前的做法是将事件ID加入到双端队列中，如果存在队列重复事件直接返回空，不做任何回复，另外有还有种做法就是直接反馈一个“请稍等，正常查询”，然后再起一个线程去做真正的耗时查询，直到返回结果再发送到群里的聊天对话。
3.机器人发布
上述代码撰写好了以后我们还需要去发布下这个机器人，在开发者平台上找到版本管理与发布进行发布即可。
发布以后就可以在Lark中添加机器人时找到你发布的机器人了，注意这个机器人要加到群里才能实现对话，不要单独去和它对话。
发布以后我们可以将上述撰写的代码进行部署然后进行测试。
内容版权声明：除非注明，否则皆为本站原创文章。
评论列表