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 _fetch_decode_result(task_id: str) -> Optional[Dict[str, Any]]: """获取解构任务结果""" sql = "SELECT result_payload, error_message 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") } def _build_result_data(task_id: str, status: int, result: Any = None, reason: Optional[str] = None) -> Dict[str, Any]: """构建结果数据""" return { "taskId": task_id, "status": status, "result": result, "reason": reason } 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") ) 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)