import asyncio import shutil import time from contextlib import asynccontextmanager import httpx from fastapi import Depends, FastAPI, Header, HTTPException from app.audio import classify_audio from app.config import settings from app.image import analyze_images from app.media import download_media from app.profile_store import profile_store from app.schemas import AnalysisRequest, AnalysisResponse, AnalysisType from app.video import VisionModelError, analyze_video def authorize(authorization: str | None = Header(default=None)) -> None: if not settings.api_key: return if authorization != f"Bearer {settings.api_key}": raise HTTPException(status_code=401, detail="Invalid analysis service credential") @asynccontextmanager async def lifespan(_: FastAPI): settings.jobs_dir.mkdir(parents=True, exist_ok=True) settings.artifacts_dir.mkdir(parents=True, exist_ok=True) profile_store.reload() yield app = FastAPI(title="CMVR Media Analysis Service", version="1.0.0", lifespan=lifespan) @app.get("/health") async def health() -> dict: async def probe(url: str) -> str: try: async with httpx.AsyncClient(timeout=2) as client: response = await client.get(url) response.raise_for_status() return "UP" except Exception: return "DOWN" sglang_urls = { "fast": settings.sglang_fast_base_url, "accurate": settings.sglang_accurate_base_url, } configured_names = [name for name, url in sglang_urls.items() if url] checks = [ probe(f"{sglang_urls[name].rstrip('/')}/models") for name in configured_names ] statuses = await asyncio.gather(*checks) configured_statuses = dict(zip(configured_names, statuses)) sglang = { name: configured_statuses.get(name, "NOT_CONFIGURED") for name in sglang_urls } return { "status": "UP", "sglang": sglang, "profiles": profile_store.status(), } @app.post( "/api/v1/analysis/run", response_model=AnalysisResponse, dependencies=[Depends(authorize)], ) async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: started = time.monotonic() media_path = None media_paths = [] try: profile = profile_store.get(request.profileCode, request.analysisType.value) if request.analysisType == AnalysisType.AUDIO_CLASSIFICATION: media_path = await download_media( str(request.mediaUrl), ".audio", settings.max_audio_bytes ) result = classify_audio(profile, media_path) model = {"provider": "CMVR", "name": "mfcc-dtw-audio-fingerprint-v2"} elif request.analysisType == AnalysisType.VIDEO_ANALYSIS: media_path = await download_media( str(request.mediaUrl), ".video", settings.max_video_bytes ) detailed_option = request.options.get("detailedOutput", False) detailed_output = ( detailed_option if isinstance(detailed_option, bool) else str(detailed_option).strip().lower() == "true" ) result, duration, video_metadata = await analyze_video( profile, media_path, str(request.options.get("instruction", "")), str(request.options.get("analysisMode", "AUTO")), request.options.get("tuning"), str(request.options.get("decisionPolicy", "FAIL_CLOSED")), detailed_output, ) evidence = result.get("evidence") if not isinstance(evidence, dict): evidence = {} result["evidence"] = evidence evidence.update( { "sampledFrameCount": video_metadata["sampledFrameCount"], "sampleFps": video_metadata["effectiveSampleFps"], "samplingFrameLimit": video_metadata["samplingFrameLimit"], "configuredSamplingFrameLimit": video_metadata["configuredSamplingFrameLimit"], "samplingCapped": video_metadata["samplingCapped"], "numCtx": video_metadata["numCtx"], "maximumWidth": video_metadata["maximumWidth"], "durationSeconds": duration, } ) result["analysisMode"] = video_metadata["mode"] result["fallback"] = video_metadata["fallback"] result["fallbackReason"] = video_metadata["fallbackReason"] result["effectiveTuning"] = video_metadata["tuning"] result["detailedOutput"] = video_metadata["detailedOutput"] model = { "provider": video_metadata["provider"], "name": video_metadata["model"], "primaryProvider": video_metadata["primaryProvider"], "primaryModel": video_metadata["primaryModel"], "providerFallback": video_metadata["providerFallback"], "requestedMode": video_metadata["requestedMode"], "usedMode": video_metadata["mode"], "fallback": video_metadata["fallback"], } else: image_urls = [str(url) for url in request.mediaUrls] if not image_urls and request.mediaUrl is not None: image_urls = [str(request.mediaUrl)] for image_url in image_urls: media_paths.append(await download_media( image_url, ".image", settings.max_image_bytes )) reference_path = None reference_url = str(request.options.get("referenceImageUrl", "")).strip() if reference_url: reference_path = await download_media( reference_url, ".reference-image", settings.max_image_bytes ) media_paths.append(reference_path) prompt = str(request.options.get("prompt", "")).strip() if not prompt: target = str(request.options.get("targetDescription", "")).strip() instruction = str(request.options.get("instruction", "")).strip() prompt = "\n".join(part for part in (target, instruction) if part) if prompt: prompt += "\n只返回分析结果，不输出通过、未通过或置信度。" result, image_metadata = await analyze_images( profile, media_paths[:-1] if reference_path is not None else media_paths, reference_path, prompt, str(request.options.get("analysisMode", "AUTO")), request.options.get("tuning"), str(request.options.get("analysisMethod", "AUTO")), ) result["analysisMode"] = image_metadata["mode"] result["fallback"] = image_metadata["fallback"] result["imageCount"] = image_metadata["imageCount"] result["analysisMethod"] = image_metadata["analysisMethod"] model = { "provider": image_metadata["provider"], "name": image_metadata["model"], "primaryProvider": image_metadata["provider"], "primaryModel": image_metadata["model"], "providerFallback": False, "requestedMode": image_metadata["requestedMode"], "usedMode": image_metadata["mode"], "fallback": image_metadata["fallback"], } return AnalysisResponse( requestId=request.requestId, analysisType=request.analysisType, profileCode=request.profileCode, status="SUCCEEDED", result=result, model=model, timingMs=round((time.monotonic() - started) * 1000), ) except (KeyError, ValueError) as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except httpx.HTTPError as exc: raise HTTPException(status_code=502, detail=f"Remote service request failed: {exc}") from exc except VisionModelError as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc finally: if media_path is not None: media_path.unlink(missing_ok=True) for path in media_paths: path.unlink(missing_ok=True)