google_ai_studio/job_redis_data.py

import datetime
import time

import schedule

from common.redis import install_video_data

def bot_video_ai_top():
    """当日头部"""
    dt = datetime.datetime.now().strftime('%Y%m%d')
    print(f"开始执行头部{dt}")
    redis_task = 'task:video_ai_top'
    table_name = 'content_ai_tag_return_top_merge'
    install_video_data(dt, redis_task, table_name)



def bot_video_ai_recommend():
    """新推荐"""
    dt = datetime.datetime.now().strftime('%Y%m%d%H')
    print(f"开始执行新推荐{dt}")
    redis_task = 'task:video_ai_recommend'
    table_name = 'content_ai_tag_recommend'
    install_video_data(dt, redis_task, table_name)



def schedule_tasks():
    schedule.every().hour.at(":22").do(bot_video_ai_recommend)
    schedule.every().day.at("01:25").do(bot_video_ai_top)


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