调用钉钉接口实现机器人推送消息
文章目录
- 调用钉钉接口实现机器人推送消息
- 0 内容简介
- 1 钉钉群设置
- 2 创建自定机器人
- 3 API开发
- 3.1 RDS(MySQL)数据库对象类
- 3.2 钉钉自定义机器人对象类
- 3.3 时间调度与执行主类
0 内容简介
钉钉群机器人是一个高级扩展功能，只要有一个钉钉账号，就可以使用它。它可以将第三方信息聚合到钉钉群中，实现信息自动同步。支持Webhook协议的自定义接入，通过发明者量化机器人，将提醒、报警等信息聚合到钉钉群中。支持文本（text）、链接（link）、markdown三种消息格式，五种消息类型。同一条信息还可以同时发送至多个钉钉群。
本篇通过调用钉钉群接口实现机器人推送消息，以文本（text）消息格式为实例进行开发，功能实现：从RDS(MySQL)数据库中读取需求消息，并配置时间调度，如一天内需求范围内每10分钟读取一次数据库信息，并发送至钉钉群内的指定接收人。
文章借鉴：https://zhuanlan.zhihu.com/p/244858570
【操作流程】：钉钉群设置 --> 创建自定义机器人 --> API开发
1 钉钉群设置
首先，创建钉钉群。我们以PC端钉钉为例，首先点击左上方“+”号发起群聊，如果只想自己接受消息，可以随便拉两个人再踢出去，填写群名称：如“WebHook_机器人”，群类型选择普通群即可。
2 创建自定机器人
钉钉群每创建一个自定义机器人都会产生唯一的Hook地址，我们称为WebHook地址，通过向该WebHook地址推送消息，钉钉群就会收到消息。
3 API开发
调用钉钉接口实现机器人推送消息，这里使用Python-3.10.3进行代码实现，代码整体由三个对象类组成：
- OperateMysql：RDS(MySQL)数据库对象类
- My_WHRobot：钉钉自定义机器人对象类
- main：时间调度与执行主类
3.1 RDS(MySQL)数据库对象类
import pymysql# 类对象一: 数据库连接与增删改查
class OperateMysql:# 数据库连接与操作模块def __init__(self):# 创建一个连接数据库的对象self.conn = pymysql.connect(host='rm-...mysql.rds.aliyuncs.com', # 连接的数据库服务器主机名port=3310, # 数据库端口号user='root', # 数据库登录用户名passwd='*******',db='database', # 数据库名称charset='utf8' # 连接编码)# 使用cursor()方法创建一个游标对象，用于操作数据库self.cur = self.conn.cursor()# 查询一条数据def search_one(self, sql):self.cur.execute(sql)# result = self.cur.fetchone() # 使用 fetchone()方法获取单条数据.只显示一行结果result = self.cur.fetchall() # 显示所有结果return result# 更新SQLdef updata_one(self, sql):try:self.cur.execute(sql) # 执行sqlself.conn.commit() # 增删改操作完数据库后，需要执行提交操作except:# 发生错误时回滚self.conn.rollback()self.conn.close() # 记得关闭数据库连接# 插入SQLdef insert_one(self, sql):try:self.cur.execute(sql) # 执行sqlself.conn.commit() # 增删改操作完数据库后，需要执行提交操作except:# 发生错误时回滚self.conn.rollback()self.conn.close()# 删除sqldef delete_one(self, sql):try:self.cur.execute(sql) # 执行sqlself.conn.commit() # 增删改操作完数据库后，需要执行提交操作except:# 发生错误时回滚self.conn.rollback()self.conn.close()
3.2 钉钉自定义机器人对象类
import json
import requests# 类对象二: 钉钉机器人API的调度
class My_WHRobot:# 钉钉机器人消息APIdef msg(self, text):headers = {'Content-Type': 'application/json;charset=utf-8'}# 只需更改这一项，将双引号内容替换为刚才复制的Webhook地址token = "***********************************"api_url = f"https://oapi.dingtalk.com/robot/send?access_token={token}"json_text = {"msgtype": "text","at": {"atMobiles": ["177*****620"],"isAtAll": False # false : 仅仅@ "at" 对应的的人},"text": {"content": "123：" + text}}print(requests.post(api_url, json.dumps(json_text), headers=headers).content)
3.3 时间调度与执行主类
import schedule
from datetime import time
from OperateMysql import *
from My_WHRobot import *# 类对象继承三: 钉钉机器人API的调度
class Schedule_Robot(OperateMysql, My_WHRobot):# 钉钉机器人消息API调度def main(self, sql):robot_mysql = OperateMysql() # 启动Mysql对象类robot_webhook = My_WHRobot() # 启动My_WHRobot对象类res = robot_mysql.search_one(sql)# tuple结果转换成text的过程代码arr = '每2分钟更新 \n'for row in res:ROW = map(str, row) # 把每条记录row中的元素map成str类型unit = ' '.join(ROW) # 每条记录.join()拼接成长stringarr = arr + unit + '\n' # 把每条记录拼接成textprint(arr)# 创建时间调度器-周期调度，设置调度周期、调度结束时间schedule.every(60*2).seconds.until(time(22, 52, 59)).do(robot_webhook.msg, text=arr) # 今天 21:59:59 停止while True:schedule.run_pending()# 运行
if __name__ == '__main__':sql = "SELECT" \" e.empno as 员工号, " \" e.ename as 员工姓名, " \" d.dname as 部门 " \"from emp e left join dept d " \"on e.deptno = d.deptno " \"limit 4 " \";"myrobot = Schedule_Robot()myrobot.main(sql)
