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"])