top_automatic_traceability/job_top_hour_data_redis.py

import time

import schedule
from loguru import logger

from common.redis import insert_job_data


def jab_top_recommend():
    """获取每日每小时top前100数据"""
    try:
        logger.info(f"开始获取每日每小时top前100数据")
        redis_task = 'task:top_all_data'
        top_count = insert_job_data(redis_task)
        logger.info(f"获取每日每小时top前100数据共{top_count}条")
        return
    except Exception as e:
        logger.error(f"获取每日每小时top前100数据异常,异常信息{e}")
        return
def schedule_tasks():
    schedule.every().hour.at(":25").do(jab_top_recommend)

if __name__ == "__main__":
    # jab_top_recommend()
    schedule_tasks()  # 调用任务调度函数
    while True:
        schedule.run_pending()
        time.sleep(1)  # 每秒钟检查一次