qidao123.com ToB IT社区-企服评测·应用市场»论坛 › 软件与程序人生 › DevOps与敏捷开发 › flask_apscheduler实现定时推送飞书消息

flask_apscheduler实现定时推送飞书消息

[复制链接]
发表于 2026-2-2 20:33:05 | 显示全部楼层 |阅读模式

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

×
需求场景：

  实现一个flask服务，通过接口控制一个定时任务任务（对旅店订房情况举行查抄）的开启和克制。要求定时任务完成后，可以通过飞书呆板人推送任务完成的消息。
显现效果：


  • 启动定时任务

  • 关闭定时任务

  • 飞书推送消息

代码实现：


  • 项目结构：

  • 业务代码：

    • 定时任务。先通过schedule模块实现底子的定时任务业务代码。
      1. # -*- coding:UTF-8 -*-
      3. """
      4. @ProjectName  :
      5. @FileName     : schedule_monitor_task
      6. @Description  :
      7. @Time         : 2023/9/18 9:21
      8. @Author       : Qredsun
      9. """
      10. import os
      11. import time
      12. from datetime import datetime
      13. from datetime import timedelta
      15. import schedule
      16. import functools
      17. from utils.webhook import send_message
      19. def read_check_list_from_excel(file):
      20.     try:
      21.        logger.info("业务代码")
      22.     except Exception as e:
      23.         logger.error(f'程序异常：{e}')
      24.     finally:
      25.         prompt_message = f'**酒店订房检查已完成**\n' \
      26.                          f'检查时间：{begin_time} - {end_time} \n' \
      27.                          f'检查报告已生成： {save_path}'
      28.         send_message(Environment.WEBHOOK_URL, Environment.WEBHOOK_SECRET, prompt_message)
      30. # todo  定时任务配置
      31. def catch_exceptions(cancel_on_failure = False):
      32.     def catch_exceptions_decorator(job_func):
      33.         @functools.wraps(job_func)
      34.         def wrapper(*args, **kwargs):
      35.             try:
      36.                 return job_func(*args, **kwargs)
      37.             except:
      38.                 import traceback
      39.                 traceback.format_exc()
      40.                 if cancel_on_failure:
      41.                     return schedule.CancelJob
      43.         return wrapper
      45.     return catch_exceptions_decorator
      47. # 异常捕获方法的使用
      48. @catch_exceptions(cancel_on_failure=False)
      49. def task_job(file_path):
      50.     # 定时任务
      51.     read_check_list_from_excel(file_path)
      53. file_path = r'../data/订房检查任务.xlsx'
      54. schedule.every().day.at("09:00").do(task_job, file_path)
      55. schedule.run_all()
      57. while True:
      58.     schedule.run_pending()  # 运行所有可以运行的任务
      59.     time.sleep(60 * 30)
      复制代码
    • 飞书消息推送功能实现
      1. # -*- coding:UTF-8 -*-
      3. """
      4. @ProjectName  :
      5. @FileName     : webhook
      6. @Description  : 飞书消息推送
      7. @Time         : 2023/9/17 13:36
      8. @Author       : Qredsun
      9. """
      10. import base64
      11. import hashlib
      12. import hmac
      13. import json
      14. from datetime import datetime
      16. import requests
      18. from utils.env_manager import Environment, logger
      20. WEBHOOK_URL = '机器人推送地址'
      21. WEBHOOK_SECRET = '机器人密码'
      23. def gen_sign(secret, timestamp):
      24.     # 拼接时间戳以及签名校验
      25.     string_to_sign = '{}\n{}'.format(timestamp, secret)
      27.     # 使用 HMAC-SHA256 进行加密
      28.     hmac_code = hmac.new(
      29.             string_to_sign.encode("utf-8"), digestmod=hashlib.sha256
      30.     ).digest()
      32.     # 对结果进行 base64 编码
      33.     sign = base64.b64encode(hmac_code).decode('utf-8')
      35.     return sign
      37. def send_message(WEBHOOK_URL, WEBHOOK_SECRET, MSG):
      38.     timestamp = int(datetime.now().timestamp())
      39.     sign = gen_sign(WEBHOOK_SECRET, timestamp)
      40.     params = {
      41.         "timestamp": timestamp,
      42.         "sign"     : sign,
      43.         "msg_type" : "interactive",
      44.         "card"     : {
      45.             "config"   : {
      46.                 "wide_screen_mode": True
      47.             },
      48.             "elements" : [
      49.                 {
      50.                     "tag"    : "markdown",
      51.                     "content": f"<at id=all></at> \n "
      52.                                f"{MSG}"
      53.                 },
      54.                 {
      55.                     "tag"    : "action",
      56.                     "actions": [
      57.                         {
      58.                             "tag"      : "button",
      59.                             "text"     : {
      60.                                 "tag"    : "plain_text",
      61.                                 "content": "跳转至订房检查"
      62.                             },
      63.                             "type"     : "primary",
      64.                             "multi_url": {
      65.                                 "url"        : Environment.CALL_BACK_URL,
      66.                                 "android_url": "",
      67.                                 "ios_url"    : "",
      68.                                 "pc_url"     : ""
      69.                             }
      70.                         }
      71.                     ]
      72.                 }
      73.             ],
      74.             "header"   : {
      75.                 "template": "blue",
      76.                 "title"   : {
      77.                     "content": "订房检查异常提示",
      78.                     "tag"    : "plain_text"
      79.                 }
      80.             },
      81.             "card_link": {
      82.                 "url"        : "",
      83.                 "pc_url"     : "",
      84.                 "android_url": "",
      85.                 "ios_url"    : ""
      86.             }
      87.         },
      88.     }
      90.     resp = requests.post(WEBHOOK_URL, json=params)
      91.     resp.raise_for_status()
      92.     result = resp.json()
      93.     if result.get("code") and result.get("code") != 0:
      94.         logger.error(f'飞书机器人消息 ： {MSG} 发送失败：{resp.text}')
      95.     else:
      96.         logger.debug(f'飞书机器人消息 ： {MSG} 发送成功')
      复制代码
    • flask_apscheduler更换schedule
      1. app.schedule_job.add_job(id="check_room", func='s_app:read_check_list_from_excel', **{
      2.     "args"   : (Environment.SRC_FILE,),
      3.     # 'trigger': 'interval',  # 指定 定时任务的类型
      4.     # 'seconds': 5  # 运行的间隔时间
      6.     # 每天九点开始执行
      7.     'trigger': 'cron',
      8.     'day'    : '*',
      9.     'hour'   : '09',
      10.     'minute' : '00',
      11.     'second' : '00'
      12. })
      13. app.schedule_job.start()  # 启动任务列表
      复制代码
    • flask服务中启动、克制接口实现
      1. from flask import Flask
      3. from utils.env_manager import Environment, logger
      4. from utils.schedule_monitor_task import read_check_list_from_excel
      5. from flask_apscheduler import APScheduler
      7. app = Flask(__name__)
      8. app.schedule_job = APScheduler()
      10. @app.route('/', methods=['get'])
      11. def hello_world():
      12.     return '酒店订房检查服务!'
      14. @app.route('/stop', methods=['get'])
      15. def stop_job():
      16.     if not app.schedule_job.get_jobs():
      17.         return '没有正在执行的订房检查任务'
      18.     else:
      19.         app.schedule_job.remove_all_jobs()
      20.         return '终止订房检查任务'
      22. class Config(object):
      23.     DEBUG = True  # flask 调试模式
      24.     """flask_apscheduler 配置"""
      25.     SCHEDULER_API_ENABLED = True  # 开放API
      26.     SCHEDULER_TIMEZONE = 'Asia/Shanghai'  # 使用上海时间
      28. @app.route('/start', methods=['get'])
      29. def start_job():
      30.     if app.schedule_job.get_jobs():
      31.         return '订房检查任务已启动'
      32.     else:
      33.         app.schedule_job.add_job(id="check_room", func='s_app:read_check_list_from_excel', **{
      34.             "args"   : (Environment.SRC_FILE,)
      35.             # 每天九点开始执行
      36.             'trigger': 'cron',
      37.             'day'    : '*',
      38.             'hour'   : '09',
      39.             'minute' : '00',
      40.             'second' : '00'
      41.         })
      42.         return '开始订房检查任务'
      44. app.config.from_object(Config)
      45. app.schedule_job.init_app(app)  # 把任务列表放入 flask
      46. app.schedule_job.start()  # 启动任务列表
      47. 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提供的定时任务管理方法实现。尚有下面一些方法可参考利用：

    • scheduler.start()  开始任务
    • scheduler.shutdown() 克制任务
    • scheduler.pause()  停息全部任务
    • scheduler.resume() 开启任务
    • scheduler.add_listener(<callback function>,<event>) 添加监听变乱
    • scheduler.remove_listener(<callback function>) 去除监听变乱
    • scheduler.add_job(<id>,<function>, **kwargs)  添加job
    • scheduler.remove_job(<id>, **<jobstore>) 删除job
    • scheduler.remove_all_jobs(**<jobstore>)  删除全部定时任务
    • scheduler.get_job(<id>,**<jobstore>) 获取job信息
    • scheduler.modify_job(<id>,**<jobstore>, **kwargs) 修改job
    • scheduler.pause_job(<id>, **<jobstore>) 停息job
    • scheduler.resume_job(<id>, **<jobstore>)  规复job
    • scheduler.run_job(<id>, **<jobstore>) 启动job
    • scheduler.authenticate(<function>)  验证
回复

使用道具 举报

返回列表

饭宝
+ 我要发帖

登录后关闭弹窗

登录参与点评抽奖  加入IT实名职场社区
去登录
快速回复 返回顶部 返回列表