google_ai_studio/job_video_processing.py

import time
from concurrent.futures import ThreadPoolExecutor

from loguru import logger

from video_processing.video_processing import VideoProcessing

max_workers = 1  # 最大线程数


def video_ai_task_start():
    redis_task_list = ['task:video_ai_recommend', 'task:video_ai_top']
    futures = []  # 用来存储任务的 Future 对象
    task_index = 0
    with ThreadPoolExecutor( max_workers=max_workers ) as executor:
        while True:
            futures = [f for f in futures if not f.done()]
            if len(futures) < max_workers:
                try:
                    # 提交新任务
                    future = executor.submit( process_video_ai, redis_task_list[task_index] )
                    futures.append( future )
                    task_index += 1
                    if task_index >= len( redis_task_list ):
                        task_index = 0
                    time.sleep( 1 )
                except Exception as e:
                    logger.error( f"异常信息: {e}" )
                    time.sleep(1)
            else:
                time.sleep(1)


def process_video_ai(redis_task):
    try:
        logger.info( f"开始执行任务: {redis_task}" )
        video_processor = VideoProcessing()
        video_processor.get_video( redis_task )
        logger.success( f"执行完成: {redis_task}" )
        time.sleep( 1 )  # 模拟处理时间
    except Exception as e:
        logger.error( f"处理任务时出现异常: {e}" )
        time.sleep( 1 )  # 出现异常时，等待5秒再处理


if __name__ == '__main__':
    video_ai_task_start()