video_decode/tasks/detail.py

from utils.sync_mysql_help import mysql
from utils.params import CapabilityEnum
from loguru import logger
import sys
import json
from typing import Optional, Dict, Any

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

# 任务状态常量
STATUS_PENDING = 0  # 待处理
STATUS_RUNNING = 1  # 执行中
STATUS_SUCCESS = 2  # 成功
STATUS_FAILED = 3   # 失败

# 不需要查询结果的状态（失败状态需要查询 error_message，因此不在此集合中）
STATUS_WITHOUT_RESULT = {STATUS_PENDING, STATUS_RUNNING}


def _build_response(data: Dict[str, Any]) -> Dict[str, Any]:
    """构建统一响应格式"""
    response = {
        "code": 0,
        "msg": "ok",
        "data": data
    }
    return response


def _parse_result_payload(payload: Optional[str]) -> Any:
    """解析结果负载（JSON字符串转对象）"""
    if not payload:
        return None
    
    try:
        return json.loads(payload)
    except (json.JSONDecodeError, TypeError):
        return payload


def _parse_web_url(web_url: Optional[str]) -> Optional[Dict[str, str]]:
    """解析结果表中的 web_url 字段，拆分出 pointUrl 和 weightUrl"""
    if not web_url:
        return None

    segments = [segment.strip() for segment in web_url.split(",") if segment.strip()]
    if not segments:
        return None

    point_url: Optional[str] = None
    weight_url: Optional[str] = None

    for segment in segments:
        if "weight_visualization" in segment:
            weight_url = segment
        else:
            point_url = segment

    if not point_url and not weight_url:
        return None

    return {
        "pointUrl": point_url or "",
        "weightUrl": weight_url or ""
    }


def _fetch_decode_result(task_id: str) -> Optional[Dict[str, Any]]:
    """获取解构任务结果"""
    sql = "SELECT result_payload, error_message, web_url FROM workflow_decode_task_result WHERE task_id = %s"
    result_record = mysql.fetchone(sql, (task_id,))
    
    if not result_record:
        return None
    
    return {
        "result": _parse_result_payload(result_record.get("result_payload")),
        "error_message": result_record.get("error_message"),
        "url": _parse_web_url(result_record.get("web_url"))
    }


def _build_result_data(
    task_id: str,
    status: int,
    result: Any = None,
    reason: Optional[str] = None,
    url: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
    """构建结果数据"""
    data: Dict[str, Any] = {
        "taskId": task_id,
        "status": status,
        "result": result,
        "reason": reason
    }

    # 对于解构任务，增加 url 字段（data.url.pointUrl / data.url.weightUrl）
    if url is not None:
        data["url"] = url

    return data


def _handle_success_status(task_id: str, capability: int) -> Dict[str, Any]:
    """处理成功状态（status=2）"""
    # 只有解构任务需要查询结果表
    if capability != CapabilityEnum.DECODE.value:
        return _build_response(_build_result_data(task_id, STATUS_SUCCESS))
    
    # 查询解构结果
    decode_result = _fetch_decode_result(task_id)
    if not decode_result:
        return _build_response(_build_result_data(task_id, STATUS_SUCCESS))
    
    result_data = _build_result_data(
        task_id=task_id,
        status=STATUS_SUCCESS,
        result=decode_result.get("result"),
        reason=decode_result.get("error_message"),
        url=decode_result.get("url")
    )
    
    return _build_response(
        result_data
    )


def get_decode_detail_by_task_id(task_id: str) -> Optional[Dict[str, Any]]:
    """获取任务详情"""
    # 查询任务基本信息
    sql = "SELECT task_id, status, capability FROM workflow_task WHERE task_id = %s"
    task = mysql.fetchone(sql, (task_id,))
    
    if not task:
        logger.info(f"task_id = {task_id} , 任务不存在")
        return None
    
    task_id_value = task.get("task_id")
    status = task.get("status")
    capability = task.get("capability")
    
    # 不需要查询结果的状态，直接返回
    if status in STATUS_WITHOUT_RESULT:
        result_data = _build_result_data(task_id_value, status)
        return _build_response(result_data)
    
    # 成功状态，需要查询结果
    if status == STATUS_SUCCESS:
        return _handle_success_status(task_id_value, capability)
    
    # 失败状态，需要返回 error_message 到 reason 字段，result 固定为 "[]"
    if status == STATUS_FAILED:
        error_message: Optional[str] = None
        # 仅对解构任务从结果表中查询失败原因
        if capability == CapabilityEnum.DECODE.value:
            decode_result = _fetch_decode_result(task_id_value)
            if decode_result:
                error_message = decode_result.get("error_message")
        
        result_data = _build_result_data(
            task_id=task_id_value,
            status=STATUS_FAILED,
            result="[]",
            reason=error_message or ""
        )
        return _build_response(
            result_data
        )
    
    # 其他未知状态，返回基础数据
    result_data = _build_result_data(task_id_value, status)
    return _build_response(result_data)