video_decode/main.py

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from utils.params import DecodeContentParam, PatternContentParam
from dotenv import load_dotenv, find_dotenv
from typing import List, Dict, Any, Optional

from tasks.decode import begin_decode_task
from tasks.detail import get_decode_detail_by_task_id
from tasks.pattern import begin_pattern_task

from loguru import logger
import sys

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

# 响应消息映射
RESPONSE_MSG_MAP = {
    0: "success",
    1002: "视频不存在",
    2001: "解构/聚类任务创建失败",
    -1: "failed",
    404: "任务不存在"
}

load_dotenv(find_dotenv(), override=False)
app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """统一处理 HTTPException，保证返回结构与其它接口一致"""
    msg = RESPONSE_MSG_MAP.get(exc.status_code, "failed")
    content = {
        "code": exc.status_code,
        "msg": msg,
        "data": None,
    }
    # 将异常 detail 写入 reason，便于排查问题
    if exc.detail:
        content["reason"] = str(exc.detail)
    return JSONResponse(status_code=200, content=content)

def _build_api_response(
    code: int,
    data: Any = None,
    reason: Optional[str] = None
) -> JSONResponse:
    """构建统一的API响应"""
    msg = RESPONSE_MSG_MAP.get(code, "failed")
    content = {
        "code": code,
        "msg": msg,
        "data": data
    }
    
    # 失败时添加 reason 字段
    if code != 0 and reason:
        content["reason"] = reason
    
    return JSONResponse(status_code=200, content=content)


@app.post("/api/v1/content/tasks/decode")
def decode_content(param: DecodeContentParam):
    """创建解构任务"""
    res = begin_decode_task(param)
    code = res.get("code", -1)
    task_id = res.get("task_id")
    reason = res.get("reason", "")
    
    return _build_api_response(
        code=code,
        data={"task_id": task_id} if task_id else None,
        reason=reason
    )

@app.get("/api/v1/content/tasks/{taskId}")
def get_task_detail(taskId: str):
    """获取任务详情"""
    result = get_decode_detail_by_task_id(taskId)
    
    # 任务不存在
    if result is None:
        return _build_api_response(code=404, data=None)
    
    # 直接返回结果（已经包含 code、msg、data、reason）
    return JSONResponse(status_code=200, content=result)

@app.post("/api/v1/content/tasks/pattern")
def pattern_content(param: PatternContentParam):
    """创建聚类任务"""
    res = begin_pattern_task(param)
    code = res.get("code", -1)
    task_id = res.get("task_id")
    reason = res.get("reason", "")
    
    return _build_api_response(
        code=code,
        data={"task_id": task_id} if task_id else None,
        reason=reason
    )