video_decode/decode_task/decodeTask.py

from loguru import logger
import sys
import time
from utils.sync_mysql_help import mysql
from typing import Dict, Any

from utils.params import DecodeWorkflowParam
from src.workflows.decode_workflow import DecodeWorkflow




logger.add(sink=sys.stderr, level="ERROR", backtrace=True, diagnose=True)



   


def invoke_decode_workflow(task_params: Dict[str, Any]):
    """主函数"""

    result = DecodeWorkflow(task_params)
    if result:
            return result
    else:
        print(f"❌ 保存结果失败，但将继续处理")
        return None


def get_decode_result_by_id(task_id:str):
    sql = "SELECT * FROM decode_videos WHERE task_id = %s AND task_status = 2 "
    tasks = mysql.fetchone(sql, (task_id,))
    if not tasks:
        logger.info(f"task_id = {task_id} , 任务不存在")
        return None
    return tasks['result'], tasks['status']


def  decode_task_status_handler():
    # 从数据库中获取任务，每次获取一个
    sql = "SELECT * FROM decode_record WHERE task_status = 0 "
    """json"""
    task = mysql.fetchone(sql)
   

    if not task:
        logger.info("任务列表为空")
        return
    else:
        task_id = task['task_id']
        sql = "UPDATE decode_record SET task_status = 1 WHERE task_id = %s"
        mysql.execute(sql, (task_id,))     
        
        # 获取任务结果
    try:
        task_id = task['task_id']
        video_url = task['video_url']
        video_id = task['video_id']
        task_params = {'task_id':task_id, 'video_id':video_id, 'video_url':video_url}
        logger.info(f"task_id = {task_id} , video_id = {video_id}")
        task_create_timestamp = task['create_timestamp']
        current_timestamp = int(time.time() * 1000)

        decode_result = invoke_decode_workflow(task_params)

        if decode_result:
            # 更新任务状态为2，任务完成
            sql = "UPDATE decode_record SET task_status = 2, WHERE task_id = %s"
            mysql.execute(sql, (task_id))
            logger.info(f"task_id = {task_id} , decode_result = {decode_result}")

           
        else:
            if current_timestamp - task_create_timestamp > 1000 * 60 * 60:
                sql = "UPDATE decode_record SET task_status = 3, WHERE task_id = %s"
                mysql.execute(sql, (task_id))
                logger.info(f"task_id = {task_id} ,任务状态异常")
    except Exception as e:
        logger.error(f"task_id = {task_id} , error = {e}")
        sql = "UPDATE decode_record SET task_status = 3, WHERE task_id = %s"
        mysql.execute(sql,  (task_id))
        logger.info(f"task_id = {task_id} ,任务异常")
        raise {"task_id": task_id, "error": '任务异常'}