钉钉定时推送提醒

为了能及时收到重要的事件提醒，这里介绍一种利用python定时推送提醒的方法。设计思路就是使用python读取mysql表，通过钉钉群推送当天到期的事件提醒。

配置过程

• 第1步，创建钉钉群聊（比如 测试群），添加机器人 选择 自定义（通过Webhook接入自定义服务），保存 Webhook地址

机器人名字：事件通知
接收群组：测试群
消息推送：开启
Webhook：https://oapi.dingtalk.com/robot/send?access_token=3da7216e1c31362935f55c5ccfae23b3a307a799c4eed13ec560ec3a6a5b832c
安全设置：
    自定义关键词：
    加签：
    IP地址(端)：124.79.149.45   #.允许向该群推送消息的服务器ip地址,请根据实际修改

• 第2步，测试一下钉钉消息推送

url=https://oapi.dingtalk.com/robot/send?access_token=3da7216e1c31362935f55c5ccfae23b3a307a799c4eed13ec560ec3a6a5b832c
curl $url -H 'Content-Type: application/json' -d '{"msgtype": "text", "text": {"content": "钉钉机器人群消息测试"}}'

• 第3步，登录mysql实例，示例 192.168.31.101

mysql -h192.168.31.101 -uroot -p

• 第4步，执行如下脚本，完成mysql数据初始化

#.建库建表
CREATE DATABASE IF NOT EXISTS it;
DROP TABLE IF EXISTS it.notice;
CREATE TABLE IF NOT EXISTS it.notice (
  id int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT 'PK',
  notice_date date NOT NULL COMMENT '事件到期日期',
  notice_type nvarchar(10) NOT NULL DEFAULT '' comment '事件类型',
  notice_body nvarchar(40) NOT NULL DEFAULT '' comment '事件内容',
  webhook varchar(128) NOT NULL DEFAULT '' comment '发送token',
  insert_time timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '插入时间',
  update_time timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间',
  status tinyint(1) NULL DEFAULT 0 COMMENT '状态,0未发送/1已发送',
  PRIMARY KEY pk_id(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='事件提醒表';

#.初始化数据(使用同一个token,即所有消息发给同一个钉钉群)
insert into it.notice(notice_date,notice_type,notice_body) values('2024-06-26','财务','今天该缴房贷');
insert into it.notice(notice_date,notice_type,notice_body) values('2024-08-01','节假日','今天是建军节');
insert into it.notice(notice_date,notice_type,notice_body) values('2024-09-10','工作','今天xxx.com证书到期');
insert into it.notice(notice_date,notice_type,notice_body) values('2024-11-10','纪念日','今天是你的生日');
update it.notice set webhook = '3da7216e1c31362935f55c5ccfae23b3a307a799c4eed13ec560ec3a6a5b832c';

#.创建账号
CREATE USER 'user_dingding'@'%' IDENTIFIED BY 'MXn896_VwTiek2_t';
GRANT SELECT,INSERT,UPDATE ON it.* TO 'user_dingding'@'%';
FLUSH PRIVILEGES;

• 第5步，安装 pymysql

注：Ubuntu 24.04 默认安装 Python 3.12.3（Ubuntu 20.04 开始弃用 Python 2），可通过 apt install python3-pymysql 来安装 pymysql。针对 Centos 7.9 系统的 Python 2.7.5 环境，可以执行 pip install pymysql requests

root@localhost:~# python3 --version
Python 3.12.3

root@localhost:~# apt install python3-pymysql
root@localhost:~# python3 -c "import pymysql; print('ok')"
ok

• 第6步，编写python脚本，并保存为 /usr/local/dingding.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import pymysql
import json 
import time
import logging
import requests
import sys

#.请根据实际情况修改如下参数
mysqlhost="192.168.31.101"
mysqluser="user_dingding"
mysqlpassword="MXn896_VwTiek2_t" 
mysqlport=3306
mysqldatabase="it"
mysqlcharset='utf8'

def ding_send(taoken,smscontent):
    webhook = "https://oapi.dingtalk.com/robot/send?access_token="+str(taoken)
    headers = {'Content-Type': 'application/json;charset=utf-8'}
    data = { "msgtype": "text", "text": { "content": smscontent } }
    r = requests.post(webhook,data = json.dumps(data),headers=headers)
    #return r.text

def mysql_alarm():
    db = pymysql.connect(host=mysqlhost,user=mysqluser,passwd=mysqlpassword,port=mysqlport,database=mysqldatabase,charset=mysqlcharset)
    cursor = db.cursor(cursor =pymysql.cursors.DictCursor)

    sql1 = """SELECT id,notice_date,notice_body,webhook FROM notice WHERE notice_date='"""+time.strftime('%Y-%m-%d',time.localtime())+"""' and `status`=0"""
    try: 
        cursor.execute(sql1)
        result = cursor.fetchall()
        if len(result)>0: 
            for row in result:
                ding_send(row['webhook'],str(row['notice_date'])+" , "+row['notice_body'])
                sql2 = """UPDATE notice SET `status`=1,update_time='"""+time.strftime('%Y-%m-%d %H:%M:%S',time.localtime())+"""' WHERE id="""+str(row['id']) 
                try:
                    cursor.execute(sql2)
                    db.commit()
                except:
                    db.rollback()

    except Exception as ex:
        print("Error: unable to fecth data: ",ex)

    cursor.close()
    db.close()

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

if __name__ == '__main__':
    mysql_alarm()

• 第7步，测试一下，效果如下图：

python3 /usr/local/dingding.py

• 第8步，部署定时任务，示例 每天 09:01 执行

crontab -l | grep dingding || echo "01 9 * * * python3 /usr/local/dingding.py" >> /var/spool/cron/`whoami`