video_decode/decode_task/decodeTask.py

from ast import Dict
import json
import re
from loguru import logger
import sys
import time
from utils.sync_mysql_help import mysql
from typing import List
from datetime import datetime
from utils.params import DecodeWorkflowParam
from src.workflows.decode_workflow import DecodeWorkflow


from src.workflows.what_deconstruction_workflow import WhatDeconstructionWorkflow


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



   


def invoke_decode_workflow(task_params: Dict[DecodeWorkflowParam]):
    """主函数"""

    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_record 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['decode_result'], tasks['task_status']


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

    if not task:
        logger.info("任务列表为空")
        return
    else:
        sql = "UPDATE decode_record SET task_status = 1 WHERE task_id = %s"
        mysql.execute(sql, (task_id,))     
        
        # 获取任务结果
    try:
        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": '任务异常'}