video_decode/tasks/detail.py

from utils.sync_mysql_help import mysql
from utils.params import CapabilityEnum, SceneEnum
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]) -> Any:
    """解析结果表中的 web_url 字段（JSON 字符串 -> 对象）"""
    if not web_url:
        return None

    try:
        return json.loads(web_url)
    except (json.JSONDecodeError, TypeError):
        # 如果不是合法 JSON，则原样返回，避免接口直接报错
        return web_url


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 _fetch_script_decode_result(task_id: str) -> Optional[Dict[str, Any]]:
    """获取创作解构任务结果"""
    sql = "SELECT result_payload, error_message, web_url FROM workflow_script_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 _fetch_pattern_result(task_id: str) -> Optional[Dict[str, Any]]:
    """获取聚类任务结果"""
    sql = "SELECT result_payload, error_message, web_url FROM workflow_pattern_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: Any = None
) -> Dict[str, Any]:
    """构建结果数据"""
    data: Dict[str, Any] = {
        "taskId": task_id,
        "status": status,
        "result": result,
        "reason": reason
    }

    # 对于有可视化页面的任务，增加 url 字段
    if url is not None:
        data["url"] = url

    return data


def _handle_success_status(task_id: str, capability: int, scene: Optional[int] = None) -> Dict[str, Any]:
    """处理成功状态（status=2）"""
    # 解构任务
    if capability == CapabilityEnum.DECODE.value:
        # 按业务场景区分：0 选题 -> 选题解构表；1 创作 -> 创作解构表
        if scene is not None and scene == SceneEnum.CREATION.value:
            decode_result = _fetch_script_decode_result(task_id)
        else:
            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)

    # 聚类任务
    if capability == CapabilityEnum.PATTERN.value:
        pattern_result = _fetch_pattern_result(task_id)
        if not pattern_result:
            return _build_response(_build_result_data(task_id, STATUS_SUCCESS))

        result_data = _build_result_data(
            task_id=task_id,
            status=STATUS_SUCCESS,
            result=pattern_result.get("result"),
            reason=pattern_result.get("error_message"),
            url=pattern_result.get("url")
        )
        return _build_response(result_data)

    # 其他能力：只返回基础任务信息
    return _build_response(_build_result_data(task_id, STATUS_SUCCESS))


def get_decode_detail_by_task_id(task_id: str) -> Optional[Dict[str, Any]]:
    """获取任务详情"""
    # 查询任务基本信息
    sql = "SELECT task_id, status, capability, scene 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")
    scene = task.get("scene")
    
    # 不需要查询结果的状态，直接返回
    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, scene)
    
    # 失败状态，需要返回 error_message 到 reason 字段，result 固定为 "[]"
    if status == STATUS_FAILED:
        error_message: Optional[str] = None
        # 解构任务失败原因
        if capability == CapabilityEnum.DECODE.value:
            # 按业务场景区分：0 选题 -> 选题解构表；1 创作 -> 创作解构表
            if scene is not None and scene == SceneEnum.CREATION.value:
                decode_result = _fetch_script_decode_result(task_id_value)
            else:
                decode_result = _fetch_decode_result(task_id_value)

            if decode_result:
                error_message = decode_result.get("error_message")
        # 聚类任务失败原因
        elif capability == CapabilityEnum.PATTERN.value:
            pattern_result = _fetch_pattern_result(task_id_value)
            if pattern_result:
                error_message = pattern_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)