调用钉钉接口实现机器人推送消息

调用钉钉接口实现机器人推送消息

文章目录

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进行代码实现，代码整体由三个对象类组成：

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

    # 更新SQL
    def updata_one(self, sql):
        try:
            self.cur.execute(sql)  # 执行sql
            self.conn.commit()  # 增删改操作完数据库后，需要执行提交操作
        except:
            # 发生错误时回滚
            self.conn.rollback()
        self.conn.close()  # 记得关闭数据库连接

    # 插入SQL
    def insert_one(self, sql):
        try:
            self.cur.execute(sql)  # 执行sql
            self.conn.commit()  # 增删改操作完数据库后，需要执行提交操作
        except:
            # 发生错误时回滚
            self.conn.rollback()
        self.conn.close()

    # 删除sql
    def delete_one(self, sql):
        try:
            self.cur.execute(sql)  # 执行sql
            self.conn.commit()  # 增删改操作完数据库后，需要执行提交操作
        except:
            # 发生错误时回滚
            self.conn.rollback()
        self.conn.close()

3.2 钉钉自定义机器人对象类

import json
import requests

# 类对象二: 钉钉机器人API的调度
class My_WHRobot:

    # 钉钉机器人消息API
    def 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()拼接成长string
            arr = arr + unit + '\n'  # 把每条记录拼接成text
        print(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)

来源：Phil_xian