api/analyze.py

import os
import asyncio
import logging
from typing import Optional
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel

from services.auth_service import get_current_user
from services.database_service import get_video, update_video_status, get_confirmed_track_id, set_confirmed_track_id
from services.storage_service import (
    get_local_video_path,
    get_local_highlight_path,
)
from services.progress_service import get_latest_progress, set_progress_sync
from services.gpu_gateway_service import (
    check_gpu_health,
    wakeup_gpu,
    submit_to_gpu,
    poll_gpu_progress,
    download_gpu_results,
    get_gpu_status,
    is_gpu_configured,
)
from orchestrator.analysis_orchestrator import AnalysisOrchestrator
from orchestrator.rendering_orchestrator import RenderingOrchestrator

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api", tags=["analyze"])

_analysis_tasks = {}


@router.post("/analyze/{video_id}")
async def api_analyze(video_id: str, token: str = Query(None)):
    if not token:
        raise HTTPException(status_code=401, detail="缺少认证 token")
    try:
        user = await get_current_user(token)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

    video = await get_video(video_id)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    if video["user_id"] != user["user_id"]:
        raise HTTPException(status_code=403, detail="无权操作")
    if video["status"] not in ["uploading", "uploaded"]:
        raise HTTPException(status_code=400, detail=f"视频状态不允许分析: {video['status']}")

    await update_video_status(video_id, "queued")

    person_query = video.get("person_query", "")
    task = asyncio.create_task(_run_analysis(video_id, user["user_id"], video.get("mode", "balanced"), person_query))
    _analysis_tasks[video_id] = task

    return {"video_id": video_id, "status": "queued", "message": "分析任务已提交"}


class PersonQueryRequest(BaseModel):
    person_query: str


class ConfirmPersonRequest(BaseModel):
    track_id: int


@router.post("/analyze/{video_id}/person-candidates")
async def api_person_candidates(video_id: str, req: PersonQueryRequest, token: str = Query(None)):
    """检测人物候选列表，返回 top-3 候选人物及缩略图"""
    if not token:
        raise HTTPException(status_code=401, detail="缺少认证 token")
    try:
        user = await get_current_user(token)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

    video = await get_video(video_id)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    if video["user_id"] != user["user_id"]:
        raise HTTPException(status_code=403, detail="无权操作")

    video_path = _get_video_file_path(video, user["user_id"])
    if not video_path:
        raise HTTPException(status_code=400, detail="视频文件未找到")

    analysis = AnalysisOrchestrator()
    candidates = await analysis.detect_person_candidates(video_path, video_id, req.person_query)

    return {"video_id": video_id, "candidates": candidates}


@router.post("/analyze/{video_id}/confirm-person")
async def api_confirm_person(video_id: str, req: ConfirmPersonRequest, token: str = Query(None)):
    """确认人物轨迹 ID，并使用人物筛选重新分析"""
    if not token:
        raise HTTPException(status_code=401, detail="缺少认证 token")
    try:
        user = await get_current_user(token)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

    video = await get_video(video_id)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    if video["user_id"] != user["user_id"]:
        raise HTTPException(status_code=403, detail="无权操作")

    await set_confirmed_track_id(video_id, req.track_id)

    # 重新分析（带人物筛选）
    person_query = video.get("person_query", "")
    task = asyncio.create_task(
        _run_analysis_with_person(video_id, user["user_id"], video.get("mode", "balanced"), person_query, req.track_id)
    )
    _analysis_tasks[video_id] = task

    return {"video_id": video_id, "status": "queued", "message": "已确认人物，正在重新分析"}


@router.get("/analyze/{video_id}/person-thumbnails/{filename}")
async def api_person_thumbnail(video_id: str, filename: str, token: str = Query(None)):
    """获取人物候选缩略图"""
    if not token:
        raise HTTPException(status_code=401, detail="缺少认证 token")
    try:
        user = await get_current_user(token)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

    video = await get_video(video_id)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    if video["user_id"] != user["user_id"]:
        raise HTTPException(status_code=403, detail="无权操作")

    thumbnail_dir = os.path.join(
        os.path.dirname(os.path.dirname(__file__)), "data", "uploads", video_id, "person_candidates"
    )
    thumbnail_path = os.path.join(thumbnail_dir, filename)

    if not os.path.exists(thumbnail_path):
        raise HTTPException(status_code=404, detail="缩略图不存在")

    from fastapi.responses import FileResponse
    return FileResponse(thumbnail_path, media_type="image/jpeg")


@router.get("/gpu/status")
async def api_gpu_status():
    status = get_gpu_status()
    gpu_healthy = await check_gpu_health()
    status["gpu_available"] = gpu_healthy
    return status


@router.get("/debug/system")
async def api_debug_system():
    gpu_available = False
    gpu_info = {"cuda_available": False}
    try:
        import torch
        gpu_available = torch.cuda.is_available()
        gpu_info = {
            "cuda_available": gpu_available,
            "cuda_device_count": torch.cuda.device_count() if gpu_available else 0,
        }
        if gpu_available:
            gpu_info["cuda_device_name"] = torch.cuda.get_device_name(0)
            gpu_info["cuda_memory_allocated"] = f"{torch.cuda.memory_allocated(0) / 1024**2:.1f} MB"
            gpu_info["cuda_memory_reserved"] = f"{torch.cuda.memory_reserved(0) / 1024**2:.1f} MB"
    except ImportError:
        gpu_info["error"] = "torch not installed"

    active_tasks = []
    for vid, task in _analysis_tasks.items():
        active_tasks.append({
            "video_id": vid,
            "done": task.done(),
            "cancelled": task.cancelled(),
        })

    return {
        "gpu": gpu_info,
        "active_analysis_tasks": active_tasks,
        "analysis_task_count": len(_analysis_tasks),
    }


@router.post("/debug/test-progress/{video_id}")
async def api_test_progress(video_id: str):
    async def _simulate():
        stages = [
            ("coarse", 0.0, "开始粗筛分析..."),
            ("coarse", 0.1, "运动分析中... 10%"),
            ("coarse", 0.2, "运动分析完成"),
            ("coarse", 0.4, "姿态分析中... 40%"),
            ("coarse", 0.7, "姿态分析完成"),
            ("coarse", 0.9, "音频分析完成"),
            ("coarse", 1.0, "粗筛完成"),
            ("fine", 0.0, "开始精分析..."),
            ("fine", 0.5, "精分析中... 50%"),
            ("fine", 1.0, "精分析完成"),
        ]
        for stage, progress, message in stages:
            set_progress_sync(video_id, stage, progress, message)
            await asyncio.sleep(1)

    asyncio.create_task(_simulate())
    return {"status": "test started", "video_id": video_id, "message": "轮询 /api/analyze/{video_id}/progress 查看进度"}


@router.post("/debug/e2e-test")
async def api_e2e_test():
    import tempfile

    from services.auth_service import guest_login
    from services.database_service import create_video

    auth_result = await guest_login("e2e_tester")
    token = auth_result["token"]
    user_id = auth_result["user_id"]

    tmp_dir = tempfile.mkdtemp()
    video_path = os.path.join(tmp_dir, "test.mp4")

    try:
        import cv2
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        out = cv2.VideoWriter(video_path, fourcc, 30, (320, 240))
        import numpy as np
        for i in range(90):
            frame = np.zeros((240, 320, 3), dtype=np.uint8)
            cx = 160 + int(100 * np.sin(i * 0.1))
            cy = 120 + int(60 * np.cos(i * 0.15))
            cv2.circle(frame, (cx, cy), 30, (0, 140, 255), -1)
            out.write(frame)
        out.release()
    except ImportError:
        import subprocess
        result = subprocess.run(
            ["ffmpeg", "-y", "-f", "lavfi", "-i", "testsrc=duration=3:size=320x240:rate=30",
             "-c:v", "libx264", "-preset", "ultrafast", video_path],
            capture_output=True, text=True
        )
        if result.returncode != 0:
            return {"error": f"ffmpeg failed: {result.stderr[-300:]}"}

    if not os.path.exists(video_path):
        return {"error": "Failed to create test video"}

    video_id = await create_video(
        user_id=user_id,
        original_filename="test.mp4",
        mode="fast",
        file_size_bytes=os.path.getsize(video_path),
    )

    await update_video_status(video_id, "uploaded")

    task = asyncio.create_task(_run_analysis_with_path(video_id, user_id, "fast", video_path))
    _analysis_tasks[video_id] = task

    return {
        "video_id": video_id,
        "token": token,
        "message": "测试分析已启动，轮询 /api/analyze/{video_id}/progress?token={token} 查看进度",
    }


async def _run_analysis_with_path(video_id: str, user_id: str, mode: str, video_path: str, person_query: str = ""):
    try:
        analysis = AnalysisOrchestrator()
        highlights = await analysis.analyze(video_path, video_id, mode, person_query=person_query)

        if highlights:
            rendering = RenderingOrchestrator()
            await rendering.render(video_path, video_id, highlights)
            set_progress_sync(video_id, "completed", 1.0, "分析完成！")
            await update_video_status(video_id, "completed")
        else:
            set_progress_sync(video_id, "completed", 1.0, "分析完成，未发现高光片段")
            await update_video_status(video_id, "completed")

    except Exception as e:
        logger.error(f"E2E test analysis failed for {video_id}: {e}", exc_info=True)
        await update_video_status(video_id, "error")
    finally:
        _analysis_tasks.pop(video_id, None)


@router.get("/analyze/{video_id}/progress")
async def api_get_progress(video_id: str, token: str = Query(None)):
    if not token:
        raise HTTPException(status_code=401, detail="缺少认证 token")
    try:
        user = await get_current_user(token)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

    video = await get_video(video_id)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    if video["user_id"] != user["user_id"]:
        raise HTTPException(status_code=403, detail="无权操作")

    progress = get_latest_progress(video_id)
    if progress:
        return progress

    status = video.get("status", "unknown")
    if status == "completed":
        return {"stage": "completed", "progress": 1.0, "message": "分析完成", "highlights_found": 0, "eta_seconds": 0}
    return {"stage": status, "progress": 0, "message": f"视频状态: {status}", "highlights_found": 0, "eta_seconds": 0}


async def _try_gpu_analysis(video_id: str, user_id: str, mode: str, video_path: str, person_query: str = "") -> bool:
    if not is_gpu_configured():
        return False

    gpu_healthy = await check_gpu_health()
    if not gpu_healthy:
        logger.info(f"[Analysis] GPU backend offline for {video_id}, trying wakeup")
        asyncio.create_task(wakeup_gpu())
        return False

    logger.info(f"[Analysis] Using GPU backend for {video_id}")
    set_progress_sync(video_id, "coarse", 0.0, "使用 GPU 加速 (T4)...")

    gpu_video_id = await submit_to_gpu(video_path, video_id, mode, user_id, person_query=person_query)
    if not gpu_video_id:
        logger.warning(f"[Analysis] GPU submit failed for {video_id}, falling back to CPU")
        return False

    max_polls = 720
    for i in range(max_polls):
        await asyncio.sleep(5)

        progress = await poll_gpu_progress(gpu_video_id)
        if progress:
            stage = progress.get("stage", "processing")
            prog = progress.get("progress", 0)
            msg = progress.get("message", "处理中...")
            set_progress_sync(video_id, stage, prog, f"[GPU] {msg}")

            if stage == "completed":
                logger.info(f"[Analysis] GPU analysis completed for {video_id}, downloading results")
                highlights_dir = os.path.join(
                    os.path.dirname(os.path.dirname(__file__)), "data", "uploads", "highlights", video_id
                )
                results = await download_gpu_results(gpu_video_id, highlights_dir)
                if results:
                    await update_video_status(video_id, "completed")
                    set_progress_sync(video_id, "completed", 1.0, "GPU 加速分析完成")
                    return True
                else:
                    logger.warning(f"[Analysis] GPU results download failed for {video_id}")
                    return False

            if stage == "error":
                logger.warning(f"[Analysis] GPU analysis error for {video_id}: {msg}")
                return False

    logger.warning(f"[Analysis] GPU analysis timed out for {video_id}")
    return False


async def _run_analysis(video_id: str, user_id: str, mode: str, person_query: str = ""):
    try:
        logger.info(f"[Analysis] Starting _run_analysis for video_id={video_id}, user_id={user_id}, mode={mode}, person_query='{person_query}'")

        local_path = get_local_video_path(user_id, video_id)
        if not local_path:
            logger.error(f"[Analysis] Video file not found locally for {video_id}")
            await update_video_status(video_id, "error")
            set_progress_sync(video_id, "error", 0, "视频文件未找到")
            return
        video_path = local_path

        logger.info(f"[Analysis] Video path: {video_path}, starting analysis...")

        gpu_success = await _try_gpu_analysis(video_id, user_id, mode, video_path, person_query)
        if gpu_success:
            return

        logger.info(f"[Analysis] Using CPU backend for {video_id}")
        set_progress_sync(video_id, "coarse", 0.0, "开始粗筛分析 (CPU)...")

        analysis = AnalysisOrchestrator()
        highlights = await analysis.analyze(video_path, video_id, mode, person_query=person_query)

        if highlights:
            rendering = RenderingOrchestrator()
            await rendering.render(video_path, video_id, highlights)
            set_progress_sync(video_id, "completed", 1.0, "分析完成！")
            await update_video_status(video_id, "completed")
        else:
            set_progress_sync(video_id, "completed", 1.0, "分析完成，未发现高光片段")
            await update_video_status(video_id, "completed")

    except Exception as e:
        logger.error(f"[Analysis] Failed for {video_id}: {e}", exc_info=True)
        await update_video_status(video_id, "error")
        set_progress_sync(video_id, "error", 0, f"分析失败: {str(e)}")
    finally:
        _analysis_tasks.pop(video_id, None)


async def _run_analysis_with_person(video_id: str, user_id: str, mode: str, person_query: str, confirmed_track_id: int):
    """使用已确认的人物轨迹 ID 重新分析"""
    try:
        local_path = get_local_video_path(user_id, video_id)
        if not local_path:
            await update_video_status(video_id, "error")
            set_progress_sync(video_id, "error", 0, "视频文件未找到")
            return
        video_path = local_path

        analysis = AnalysisOrchestrator()
        highlights = await analysis.analyze(video_path, video_id, mode, person_query=person_query)

        if highlights:
            rendering = RenderingOrchestrator()
            await rendering.render(video_path, video_id, highlights)
            set_progress_sync(video_id, "completed", 1.0, "人物筛选分析完成！")
            await update_video_status(video_id, "completed")
        else:
            set_progress_sync(video_id, "completed", 1.0, "分析完成，未发现高光片段")
            await update_video_status(video_id, "completed")

    except Exception as e:
        logger.error(f"[Analysis] Person-filtered analysis failed for {video_id}: {e}", exc_info=True)
        await update_video_status(video_id, "error")
        set_progress_sync(video_id, "error", 0, f"分析失败: {str(e)}")
    finally:
        _analysis_tasks.pop(video_id, None)


def _get_video_file_path(video: dict, user_id: str) -> Optional[str]:
    """获取视频文件路径（始终使用本地路径）"""
    return get_local_video_path(user_id, video["video_id"])