使用python写一个简单的到期事件钉钉提醒功能

前言：学习python第3天

需求：简单的事件提醒功能

版本：python3.9、mysql5.7

1、现在mysql建一个表event_remind

CREATE TABLE `event_remind` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `project_name` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '所属项目',
  `type` varchar(16) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '类型：domain/SSL/',
  `title` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '事项',
  `expire_date` datetime DEFAULT NULL COMMENT '到期时间',
  `ding_talk_group` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '钉钉群',
  `ding_talk_token` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '钉钉token',
  `remark` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '备注',
  `send_num` int(11) DEFAULT NULL COMMENT '发送计数',
  `send_last_time` datetime DEFAULT NULL COMMENT '最近发送时间',
  `status` varchar(16) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '状态',
  `test_flag` tinyint(4) DEFAULT NULL COMMENT '测试',
  `period` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '周期（common=普通，every_month=每月)',
  `repeat_interval` int(11) DEFAULT NULL COMMENT '重复周期(结合period的值，1=1号）',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='运维事项提醒';

关键查询到期事件sql

# 查询大于7天且30天内的的事件在周日提醒一次， 小于7天的事件则每天提醒一次。
select 
		id,
		project_name,
		title, 
		expire_date, 
		date_format(expire_date, '%Y-%m-%d') expire_date_format,
		status,
		ding_talk_token,
		DATEDIFF(expire_date,now()) diff_days,
		test_flag,
		period,
		repeat_interval,
		# 类型为common默认提醒，every_month类型需匹配repeat_interval日期才提醒，1提醒，-1不提醒
		(case when period = 'common' then 1
			    when period = 'every_month' and repeat_interval = DATE_FORMAT(now(),'%e') then 1 
		else -1 end) match_flag
from np_event_remind  
where status = 'ON' 
and (case 
		when DATEDIFF(expire_date,now()) between 0 and 5 then 1
		when DATEDIFF(expire_date,now()) between 6 and 30 and WEEKDAY(now()) = 6 then 1
		else -1 end) = 1
and (case when period = 'common' then 1
			    when period = 'every_month' and repeat_interval = DATE_FORMAT(now(),'%e') then 1 
		else -1 end) = 1 ;

使用到的函数

# 星期函数
select 
		WEEK(now())     				          as '一年的第几周',
		WEEKDAY(now())  				       as '星期几(0=星期一)',
		MONTH(now())                   as '一年的第几月份',
		DATE_FORMAT(now(),'%e')  as '一个月的几号';

# 日期增加一个月
select adddate(now(), interval 1 month);

2、编写python脚本，event_remind.py文件

# -*- coding: utf-8 -*-
import datetime
import json

import pymysql.cursors
import requests

"""
功能说明：到期事件钉钉提醒功能
作者：CF
"""
connect = pymysql.Connect(
    host='127.0.0.1',
    port=3306,
    user='root',
    passwd='',
    db='',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

# 返回datetime格式：eg：2019-12-07 20:38:35.82816, 格式化：now.strftime('%Y-%m-%d')
now = datetime.datetime.now()
cursor = connect.cursor()


# 入口函数
def main():
    # 查询30天内即将到期的事项
    # 30天内且周日提醒，7天内每天提醒
    sql = '''
         select 
            id,
            project_name,
            title, 
            expire_date, 
            date_format(expire_date, '%Y-%m-%d') expire_date_format,
            status,
            ding_talk_token,
            DATEDIFF(expire_date,now()) diff_days,
            test_flag,
            period,
			repeat_interval
        from event_remind 
        where status = 'ON' 
        and (case 
            when DATEDIFF(expire_date,now()) between 0 and 5 then 1
            when DATEDIFF(expire_date,now()) between 6 and 30 and WEEKDAY(now()) = 6 then 1
            else -1 end) = 1
        and (case when period = 'common' then 1
			  when period = 'every_month' and repeat_interval = DATE_FORMAT(now(),'%e') then 1 
			else -1 end) = 1 
    '''

    # print("查询数据SQL：%s" % sql)
    cursor.execute(sql)
    result = cursor.fetchall()
    for x, event_remind in enumerate(result):
        print('{} - 序号: {}, id: {}, 事项: {}, 剩余日期：{}'.format(datetime.datetime.now(), x, event_remind['id'],
                                                            event_remind['title'],
                                                            event_remind['diff_days']))
        # 处理发送通知
        # period：common默认发送，every_month周期发
        notice(event_remind)
        # 关闭数据库连接
    connect.close()


def notice(event_remind):
    content = ""
    # 测试
    if event_remind['test_flag'] == 1:
        content = "通知测试"
    # 按周期
    elif event_remind['period'] == 'every_month':
        content = "{}-{}".format(event_remind['project_name'], event_remind['title'])
    # 默认
    elif event_remind['period'] == 'common':
        if event_remind['diff_days'] == 0:
            content = "{}-{}，今日到期。".format(event_remind['project_name'], event_remind['title'])
        elif event_remind['diff_days'] > 0:
            content = "{}-{}，到期日：{}，剩余时间：{}天。".format(event_remind['project_name'], event_remind['title'],
                                                      event_remind['expire_date_format'],
                                                      event_remind['diff_days'])
    # print(content)
    # 发送钉钉通知
    ding_talk_notice(event_remind['ding_talk_token'], content)
    # 记录发送，标记发送次数
    cursor.execute("update event_remind set send_num = send_num +1, send_last_time=now() where id ={}".format(
        event_remind['id']))
    cursor.execute("commit")


# 钉钉消息关键词发送(关键词10个)
# 官方文档：https://open.dingtalk.com/document/robots/custom-robot-access
def ding_talk_notice(dingding_access_token, content):
    if dingding_access_token == '':
        print("{}-发送钉钉消息异常，token为空".format(datetime.datetime.now()))
        return

    # print("测试钉钉发送消息-关键词方式或IP方式")
    # 钉钉机器人Webhook地址
    webhook_url = f"https://oapi.dingtalk.com/robot/send?access_token={dingding_access_token}"

    # 要发送的消息内容 （强制atAll,需要在钉钉端设置除管理员外都可以）
    message = {
        "msgtype": "text",
        "text": {
            "content": "[到期提醒]{}".format(content)
        },
        "at": {
            "isAtAll": True
        }
    }
    # 头信息，需要包含
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36',
        'Connection': 'keep-alive',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    }
    # 将消息转换为JSON格式并发送到钉钉机器人
    response = requests.post(webhook_url, data=json.dumps(message), headers=headers)
    if response.status_code != 200:
        print('发送异常,状态码:{}'.format(response.status_code))
        print('发送异常,返回内容: {}'.format(response.text))


if __name__ == '__main__':
    main()

3、设置crontab定时每天8点半执行

crontab -e

30 8 * * * /usr/bin/python3 /home/script/event_remind.py 2>&1 >> /var/log/script/event_remind.log

4、钉钉提醒

配置一个钉钉机器人， 用简单的关键词验证模式即可，比如配置一个提醒。 然后配置下webhook即可发送钉钉消息