flask_apscheduler实现定时推送飞书消息

2023-11-10 22:20
文章标签 实现 apscheduler 推送 消息 flask 定时

本文主要是介绍flask_apscheduler实现定时推送飞书消息，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

需求场景：

  实现一个flask服务，通过接口控制一个定时任务任务（对酒店订房情况进行检查）的开启和停止。要求定时任务完成后，可以通过飞书机器人推送任务完成的消息。

展现效果：

  1. 启动定时任务
    在这里插入图片描述

  2. 关闭定时任务
    在这里插入图片描述

  3. 飞书推送消息
    在这里插入图片描述

代码实现：

  1. 项目结构：
    在这里插入图片描述

  2. 业务代码：

    1. 定时任务。先通过schedule模块实现基础的定时任务业务代码。

      # -*- coding:UTF-8 -*-"""@ProjectName  : @FileName     : schedule_monitor_task@Description  : @Time         : 2023/9/18 9:21@Author       : Qredsun"""
import os
import time
from datetime import datetime
from datetime import timedeltaimport schedule
import functools
from utils.webhook import send_messagedef read_check_list_from_excel(file):try:logger.info("业务代码")except Exception as e:logger.error(f'程序异常：{e}')finally:prompt_message = f'**酒店订房检查已完成**\n' \f'检查时间：{begin_time} - {end_time} \n' \f'检查报告已生成： {save_path}'send_message(Environment.WEBHOOK_URL, Environment.WEBHOOK_SECRET, prompt_message)# todo  定时任务配置
def catch_exceptions(cancel_on_failure = False):def catch_exceptions_decorator(job_func):@functools.wraps(job_func)def wrapper(*args, **kwargs):try:return job_func(*args, **kwargs)except:import tracebacktraceback.format_exc()if cancel_on_failure:return schedule.CancelJobreturn wrapperreturn catch_exceptions_decorator# 异常捕获方法的使用
@catch_exceptions(cancel_on_failure=False)
def task_job(file_path):# 定时任务read_check_list_from_excel(file_path)file_path = r'../data/订房检查任务.xlsx'
schedule.every().day.at("09:00").do(task_job, file_path)
schedule.run_all()while True:schedule.run_pending()  # 运行所有可以运行的任务time.sleep(60 * 30)

    2. 飞书消息推送功能实现

      # -*- coding:UTF-8 -*-"""@ProjectName  : @FileName     : webhook@Description  : 飞书消息推送@Time         : 2023/9/17 13:36@Author       : Qredsun"""
import base64
import hashlib
import hmac
import json
from datetime import datetimeimport requestsfrom utils.env_manager import Environment, loggerWEBHOOK_URL = '机器人推送地址'
WEBHOOK_SECRET = '机器人密码'def gen_sign(secret, timestamp):# 拼接时间戳以及签名校验string_to_sign = '{}\n{}'.format(timestamp, secret)# 使用 HMAC-SHA256 进行加密hmac_code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()# 对结果进行 base64 编码sign = base64.b64encode(hmac_code).decode('utf-8')return signdef send_message(WEBHOOK_URL, WEBHOOK_SECRET, MSG):timestamp = int(datetime.now().timestamp())sign = gen_sign(WEBHOOK_SECRET, timestamp)params = {"timestamp": timestamp,"sign"     : sign,"msg_type" : "interactive","card"     : {"config"   : {"wide_screen_mode": True},"elements" : [{"tag"    : "markdown","content": f"<at id=all></at> \n "f"{MSG}"},{"tag"    : "action","actions": [{"tag"      : "button","text"     : {"tag"    : "plain_text","content": "跳转至订房检查"},"type"     : "primary","multi_url": {"url"        : Environment.CALL_BACK_URL,"android_url": "","ios_url"    : "","pc_url"     : ""}}]}],"header"   : {"template": "blue","title"   : {"content": "订房检查异常提示","tag"    : "plain_text"}},"card_link": {"url"        : "","pc_url"     : "","android_url": "","ios_url"    : ""}},}resp = requests.post(WEBHOOK_URL, json=params)resp.raise_for_status()result = resp.json()if result.get("code") and result.get("code") != 0:logger.error(f'飞书机器人消息 ： {MSG} 发送失败：{resp.text}')else:logger.debug(f'飞书机器人消息 ： {MSG} 发送成功')

    3. flask_apscheduler替换schedule

      app.schedule_job.add_job(id="check_room", func='s_app:read_check_list_from_excel', **{"args"   : (Environment.SRC_FILE,),# 'trigger': 'interval',  # 指定 定时任务的类型# 'seconds': 5  # 运行的间隔时间# 每天九点开始执行'trigger': 'cron','day'    : '*','hour'   : '09','minute' : '00','second' : '00'
})
app.schedule_job.start()  # 启动任务列表

    4. flask服务中启动、停止接口实现

      from flask import Flaskfrom utils.env_manager import Environment, logger
from utils.schedule_monitor_task import read_check_list_from_excel
from flask_apscheduler import APSchedulerapp = Flask(__name__)
app.schedule_job = APScheduler()@app.route('/', methods=['get'])
def hello_world():return '酒店订房检查服务!'@app.route('/stop', methods=['get'])
def stop_job():if not app.schedule_job.get_jobs():return '没有正在执行的订房检查任务'else:app.schedule_job.remove_all_jobs()return '终止订房检查任务'class Config(object):DEBUG = True  # flask 调试模式"""flask_apscheduler 配置"""SCHEDULER_API_ENABLED = True  # 开放APISCHEDULER_TIMEZONE = 'Asia/Shanghai'  # 使用上海时间@app.route('/start', methods=['get'])
def start_job():if app.schedule_job.get_jobs():return '订房检查任务已启动'else:app.schedule_job.add_job(id="check_room", func='s_app:read_check_list_from_excel', **{"args"   : (Environment.SRC_FILE,)# 每天九点开始执行'trigger': 'cron','day'    : '*','hour'   : '09','minute' : '00','second' : '00'})return '开始订房检查任务'app.config.from_object(Config)
app.schedule_job.init_app(app)  # 把任务列表放入 flask
app.schedule_job.start()  # 启动任务列表
app.run()

ps：

  • 机器人推送接口的配置：
    在这里插入图片描述

  • 在flask的配置中将SCHEDULER_API_ENABLED设置为True，服务启动后自动加载flask_apscheduler提供的API接口：
    1. /scheduler [GET] > 获取服务基本信息
    2. /scheduler/jobs [POST json job data] > 添加新的任务
    3. /scheduler/jobs/<job_id> [GET] > 根据job_id返回任务的详细信息
    4. /scheduler/jobs [GET] > 返回所有任务的信息
    5. /scheduler/jobs/<job_id> [DELETE] > 删除任务
    6. /scheduler/jobs/<job_id> [PATCH json job data] > 更新一个已经存在的任务
    7. /scheduler/jobs/<job_id>/pause [POST] > 暂停一个任务并返回任务的信息
    8. /scheduler/jobs/<job_id>/resume [POST] > 重新启动一个任务并返回任务信息
    9. /scheduler/jobs/<job_id>/run [POST] > 启动一个任务并返回任务的信息
    在这里插入图片描述

  • 在实现定时任务的启动和关闭时，并没有直接flask_apscheduler使用自带的接口，而是通过flask_apscheduler提供的定时任务管理方法实现。还有下面一些方法可参考使用：

    1. scheduler.start() 开始任务
    2. scheduler.shutdown() 停止任务
    3. scheduler.pause() 暂停所有任务
    4. scheduler.resume() 开启任务
    5. scheduler.add_listener(<callback function>,<event>) 添加监听事件
    6. scheduler.remove_listener(<callback function>) 去除监听事件
    7. scheduler.add_job(<id>,<function>, **kwargs) 添加job
    8. scheduler.remove_job(<id>, **<jobstore>) 删除job
    9. scheduler.remove_all_jobs(**<jobstore>) 删除所有定时任务
    10. scheduler.get_job(<id>,**<jobstore>) 获取job信息
    11. scheduler.modify_job(<id>,**<jobstore>, **kwargs) 修改job
    12. scheduler.pause_job(<id>, **<jobstore>) 暂停job
    13. scheduler.resume_job(<id>, **<jobstore>) 恢复job
    14. scheduler.run_job(<id>, **<jobstore>) 启动job
    15. scheduler.authenticate(<function>) 验证

这篇关于flask_apscheduler实现定时推送飞书消息的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/385652。

相关文章

hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

利用康拓展开将一个排列映射成一个自然数，然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu
阅读更多...
【C++】_list常用方法解析及模拟实现

【C++】_list常用方法解析及模拟实现

相信自己的力量，只要对自己始终保持信心，尽自己最大努力去完成任何事，就算事情最终结果是失败了，努力了也不留遗憾。💓💓💓 目录   ✨说在前面 🍋知识点一：什么是list？ •🌰1.list的定义 •🌰2.list的基本特性 •🌰3.常用接口介绍 🍋知识点二：list常用接口 •🌰1.默认成员函数 🔥构造函数（⭐） 🔥析构函数 •🌰2.list对象
阅读更多...
【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯，先赞后看哦~🎈🎈 🏆 作者简介：景天科技苑 🏆《头衔》：大厂架构师，华为云开发者社区专家博主，阿里云开发者社区专家博主，CSDN全栈领域优质创作者，掘金优秀博主，51CTO博客专家等。 🏆《博客》：Python全栈，前后端开发，小程序开发，人工智能，js逆向，App逆向，网络系统安全，数据分析，Django，fastapi
阅读更多...
让树莓派智能语音助手实现定时提醒功能

让树莓派智能语音助手实现定时提醒功能

最初的时候是想直接在rasa 的chatbot上实现，因为rasa本身是带有remindschedule模块的。不过经过一番折腾后，忽然发现，chatbot上实现的定时，语音助手不一定会有响应。因为，我目前语音助手的代码设置了长时间无应答会结束对话，这样一来，chatbot定时提醒的触发就不会被语音助手获悉。那怎么让语音助手也具有定时提醒功能呢？ 我最后选择的方法是用threading.Time
阅读更多...
Android实现任意版本设置默认的锁屏壁纸和桌面壁纸（两张壁纸可不一致）

Android实现任意版本设置默认的锁屏壁纸和桌面壁纸（两张壁纸可不一致）

客户有些需求需要设置默认壁纸和锁屏壁纸  在默认情况下 这两个壁纸是相同的  如果需要默认的锁屏壁纸和桌面壁纸不一样 需要额外修改 Android13实现 替换默认桌面壁纸： 将图片文件替换frameworks/base/core/res/res/drawable-nodpi/default_wallpaper.*  （注意不能是bmp格式） 替换默认锁屏壁纸： 将图片资源放入vendo
阅读更多...
C#实战|大乐透选号器[6]：实现实时显示已选择的红蓝球数量

C#实战|大乐透选号器[6]：实现实时显示已选择的红蓝球数量

哈喽，你好啊，我是雷工。 关于大乐透选号器在前面已经记录了5篇笔记，这是第6篇； 接下来实现实时显示当前选中红球数量，蓝球数量； 以下为练习笔记。 01 效果演示 当选择和取消选择红球或蓝球时，在对应的位置显示实时已选择的红球、蓝球的数量； 02 标签名称 分别设置Label标签名称为：lblRedCount、lblBlueCount
阅读更多...
Kubernetes PodSecurityPolicy：PSP能实现的5种主要安全策略

Kubernetes PodSecurityPolicy：PSP能实现的5种主要安全策略

Kubernetes PodSecurityPolicy：PSP能实现的5种主要安全策略 1. 特权模式限制2. 宿主机资源隔离3. 用户和组管理4. 权限提升控制5. SELinux配置 💖The Begin💖点点关注，收藏不迷路💖 Kubernetes的PodSecurityPolicy（PSP）是一个关键的安全特性，它在Pod创建之前实施安全策略，确保P
阅读更多...
工厂ERP管理系统实现源码（JAVA）

工厂ERP管理系统实现源码（JAVA）

工厂进销存管理系统是一个集采购管理、仓库管理、生产管理和销售管理于一体的综合解决方案。该系统旨在帮助企业优化流程、提高效率、降低成本，并实时掌握各环节的运营状况。 在采购管理方面，系统能够处理采购订单、供应商管理和采购入库等流程，确保采购过程的透明和高效。仓库管理方面，实现库存的精准管理，包括入库、出库、盘点等操作，确保库存数据的准确性和实时性。 生产管理模块则涵盖了生产计划制定、物料需求计划、
阅读更多...
C++——stack、queue的实现及deque的介绍

C++——stack、queue的实现及deque的介绍

目录 1.stack与queue的实现 1.1stack的实现  1.2 queue的实现 2.重温vector、list、stack、queue的介绍 2.1 STL标准库中stack和queue的底层结构  3.deque的简单介绍 3.1为什么选择deque作为stack和queue的底层默认容器  3.2 STL中对stack与queue的模拟实现 ①stack模拟实现
阅读更多...
基于51单片机的自动转向修复系统的设计与实现

基于51单片机的自动转向修复系统的设计与实现

文章目录 前言资料获取设计介绍功能介绍设计清单具体实现截图参考文献设计获取 前言 💗博主介绍：✌全网粉丝10W+,CSDN特邀作者、博客专家、CSDN新星计划导师，一名热衷于单片机技术探索与分享的博主、专注于 精通51/STM32/MSP430/AVR等单片机设计 主要对象是咱们电子相关专业的大学生，希望您们都共创辉煌！✌💗 👇🏻 精彩专栏 推荐订阅👇🏻 单片机
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2