前言：
此刻看官们对“python周期函数代码”大概比较关切，大家都需要剖析一些“python周期函数代码”的相关内容。那么小编同时在网络上网罗了一些有关“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_flagfrom 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) = 1and (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 datetimeimport jsonimport pymysql.cursorsimport 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个)# 官方文档： ding_talk_notice(dingding_access_token, content): if dingding_access_token == '': print("{}-发送钉钉消息异常，token为空".format(datetime.datetime.now())) return # print("测试钉钉发送消息-关键词方式或IP方式") # 钉钉机器人Webhook地址 webhook_url = f"{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 -e30 8 * * * /usr/bin/python3 /home/script/event_remind.py 2>&1 >> /var/log/script/event_remind.log
4、钉钉提醒
配置一个钉钉机器人， 用简单的关键词验证模式即可，比如配置一个提醒。 然后配置下webhook即可发送钉钉消息
webhook
钉钉提醒
版权声明：
本站文章均来自互联网搜集，如有侵犯您的权益，请联系我们删除，谢谢。
标签： #python周期函数代码
