feat: health check ve audio analysis API routes eklendi

app/routes/analyze.py

@@ -14,6 +14,7 @@ Endpoints:
   POST /api/analyze                 one request, waits for the result
   POST /api/analyze/jobs            starts a background job, returns its id
   GET  /api/analyze/jobs/{job_id}   per-step progress, then the result
+  DELETE /api/analyze/jobs/{job_id} cancel
 
 If the audio can't be decoded or a link can't be downloaded, the
 response has an error code and no result.
@@ -22,12 +23,14 @@ response has an error code and no result.
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import io
 import tempfile
 import time
 import uuid
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
 
@@ -52,15 +55,22 @@ from .analyze_schemas import (
 )
 from ..services.analysis_jobs import (
     FILE_STEPS,
+    MAX_PENDING,
     URL_STEPS,
     STEP_DONE,
     STEP_FAILED,
     STEP_RUNNING,
     STEP_SKIPPED,
+    Job,
+    JobCancelled,
     Progress,
+    QueueFull,
+    cpu_gate,
     job_store,
+    result_cache,
 )
 from ..services.feature_extractor import (
+    decode_clip,
     extract_features,
     AudioFeatures,
 )
@@ -95,8 +105,8 @@ URL_SOURCES = ("youtube", "tiktok", "instagram", "soundcloud", "twitter")
 YOUTUBE_CLIP_SEC = 120.0
 YOUTUBE_MAX_BYTES = 100 * 1024 * 1024
 
-# cpu-basic has 2 vCPUs; more concurrent jobs only queue in the pool.
-MAX_RUNNING_JOBS = 4
+# Bump when the pipeline's output changes, so results cached from the old one aren't served.
+PIPELINE_VERSION = "2"
 
 # Rate limiter for public heavy endpoints
 _analyze_rate_store: dict[str, list[float]] = defaultdict(list)
@@ -153,7 +163,7 @@ xai_service = get_xai_service()
 meta_classifier_service = MetaClassifierService()
 
 # Thread pool for CPU-bound audio analysis (feature extraction + vocal)
-_analysis_pool = ThreadPoolExecutor(max_workers=3)
+_analysis_pool = ThreadPoolExecutor(max_workers=4)
 # Downloads wait on the network; keep them off the analysis workers.
 _download_pool = ThreadPoolExecutor(max_workers=2)
 
@@ -219,51 +229,77 @@ async def _run_pipeline(
     source: Dict[str, Any],
     start_time: float,
     progress: Progress,
+    ticket: str,
     warnings: Optional[List[str]] = None,
     total_duration: Optional[float] = None,
 ) -> AnalyzeResponse:
-    """Run every layer on one piece of audio and build the response."""
+    """Run every layer on one piece of audio and build the response.
+
+    The CPU-heavy part waits its turn in `cpu_gate` (ticket = job id); the
+    external FST call starts alongside it and never holds the CPU slot.
+    """
     warnings = list(warnings or [])
     file_size = len(content)
     logger.info(f"[{request_id}] Audio: {filename} ({file_size / 1024 / 1024:.2f}MB)")
     loop = asyncio.get_running_loop()
+    fst_task: Optional["asyncio.Task[Any]"] = None
 
-    # ── Feature extraction + vocal analysis + CLAP + wav2vec2 in parallel ──
-    progress.running("features", "vocals", "clap", "wav2vec2")
-    feat_future = loop.run_in_executor(_analysis_pool, lambda: extract_features(io.BytesIO(content)))
-    vocal_future = loop.run_in_executor(_analysis_pool, lambda: analyze_vocals(io.BytesIO(content)))
-    clap_future = loop.run_in_executor(_analysis_pool, lambda: clap_service.predict(io.BytesIO(content)))
-    wav2vec2_future = loop.run_in_executor(_analysis_pool, lambda: wav2vec2_service.predict(io.BytesIO(content)))
-    _track(feat_future, progress, "features")
-    _track(vocal_future, progress, "vocals")
-    _track(clap_future, progress, "clap", ok=lambda r: bool(r.available))
-    _track(wav2vec2_future, progress, "wav2vec2", ok=lambda r: bool(r.available))
+    try:
+        async with cpu_gate.slot(ticket, progress):
+            # Network, not CPU: runs while the models work.
+            progress.set("fst", STEP_RUNNING)
+            fst_task = asyncio.create_task(fst_service.predict(content, suffix=Path(filename).suffix.lower() or ".wav"))
+
+            # Decode once; every layer reads the same 60 s clip.
+            progress.running("features", "vocals", "clap", "wav2vec2")
+            try:
+                clip = await loop.run_in_executor(_analysis_pool, decode_clip, content)
+            except ValueError as e:
+                logger.warning(f"[{request_id}] Audio rejected: {e}")
+                message = str(e).lower()
+                code = "audio_too_short" if "short" in message else "audio_silent" if "silent" in message else "audio_decode_failed"
+                progress.skip_pending()
+                return AnalyzeResponse(errors=[code], warnings=_unique(warnings))
+            except Exception as e:
+                logger.error(f"[{request_id}] Decode error: {e}", exc_info=True)
+                progress.skip_pending()
+                return AnalyzeResponse(errors=["audio_decode_failed"], warnings=_unique(warnings))
+            progress.check()
+
+            feat_future = loop.run_in_executor(_analysis_pool, lambda: extract_features(io.BytesIO(clip)))
+            vocal_future = loop.run_in_executor(_analysis_pool, lambda: analyze_vocals(io.BytesIO(clip)))
+            clap_future = loop.run_in_executor(_analysis_pool, lambda: clap_service.predict(io.BytesIO(clip)))
+            wav2vec2_future = loop.run_in_executor(_analysis_pool, lambda: wav2vec2_service.predict(io.BytesIO(clip)))
+            _track(feat_future, progress, "features")
+            _track(vocal_future, progress, "vocals")
+            _track(clap_future, progress, "clap", ok=lambda r: bool(r.available))
+            _track(wav2vec2_future, progress, "wav2vec2", ok=lambda r: bool(r.available))
+            # Hold the slot until every thread is done, even if one fails.
+            await asyncio.gather(feat_future, vocal_future, clap_future, wav2vec2_future, return_exceptions=True)
+        progress.check()
+    except BaseException:
+        if fst_task is not None:
+            fst_task.cancel()
+        raise
 
     # Feature extraction is required for everything below.
     try:
-        features: AudioFeatures = await feat_future
+        features: AudioFeatures = feat_future.result()
         logger.info(
             f"[{request_id}] Features: "
             f"S={features.spectral_regularity:.3f} "
             f"T={features.temporal_patterns:.3f} "
             f"H={features.harmonic_structure:.3f}"
         )
-    except ValueError as e:
-        logger.warning(f"[{request_id}] Feature extraction rejected the audio: {e}")
-        for step in ("fst", "xai", "meta"):
-            progress.set(step, STEP_SKIPPED)
-        message = str(e).lower()
-        code = "audio_too_short" if "short" in message else "audio_silent" if "silent" in message else "audio_decode_failed"
-        return AnalyzeResponse(errors=[code], warnings=_unique(warnings))
     except Exception as e:
         logger.error(f"[{request_id}] Feature extraction error: {e}", exc_info=True)
-        for step in ("fst", "xai", "meta"):
-            progress.set(step, STEP_SKIPPED)
+        fst_task.cancel()
+        progress.skip_pending()
         return AnalyzeResponse(errors=["audio_decode_failed"], warnings=_unique(warnings))
 
     vocals: Optional[VocalFeatures] = None
     try:
-        vocals = await vocal_future
+        vocals = vocal_future.result()
         if vocals.has_vocals:
             logger.info(
                 f"[{request_id}] Vocals: "
@@ -278,7 +314,7 @@ async def _run_pipeline(
 
     clap_result = None
     try:
-        clap_result = await clap_future
+        clap_result = clap_future.result()
         if clap_result.available:
             logger.info(
                 f"[{request_id}] CLAP ({clap_result.classifier_used}): "
@@ -293,7 +329,7 @@ async def _run_pipeline(
 
     wav2vec2_result = None
     try:
-        wav2vec2_result = await wav2vec2_future
+        wav2vec2_result = wav2vec2_future.result()
         if wav2vec2_result.available:
             logger.info(f"[{request_id}] wav2vec2: p_ai={wav2vec2_result.p_ai:.3f}")
         else:
@@ -303,11 +339,10 @@ async def _run_pipeline(
         logger.warning(f"[{request_id}] wav2vec2 failed: {e}")
         warnings.append("wav2vec2_unavailable")
 
-    # ── FST detection (external Space) ──
+    # ── FST detection (external Space), started with the CPU stage ──
     fst_result = None
-    progress.set("fst", STEP_RUNNING)
     try:
-        fst_result = await fst_service.predict(io.BytesIO(content))
+        fst_result = await fst_task
         if fst_result.available:
             progress.set("fst", STEP_DONE)
             logger.info(
@@ -322,6 +357,7 @@ async def _run_pipeline(
         progress.set("fst", STEP_FAILED)
         logger.warning(f"[{request_id}] FST failed: {e}")
         warnings.append("fst_analysis_unavailable")
+    progress.check()
 
     # ── Score fusion ──
     fusion: FusionResult = fuse_scores(
@@ -339,7 +375,11 @@ async def _run_pipeline(
     xai_payload: Optional[XAIExplanation] = None
     progress.set("xai", STEP_RUNNING)
     try:
-        xai_result = xai_service.predict(features, vocals) if xai_service.available else None
+        # 11 models + SHAP: short, but CPU; keep it off the event loop.
+        xai_result = (
+            await loop.run_in_executor(_analysis_pool, xai_service.predict, features, vocals)
+            if xai_service.available else None
+        )
         if xai_result is not None:
             xai_dict = xai_service.to_dict(xai_result)
             xai_payload = XAIExplanation(
@@ -377,9 +417,12 @@ async def _run_pipeline(
     meta_payload: Optional[MetaClassifierExplanation] = None
     progress.set("meta", STEP_RUNNING)
     try:
-        meta_result = meta_classifier_service.predict(
-            features, vocals=vocals,
-            wav2vec2=wav2vec2_result, clap=clap_result, fst=fst_result,
+        meta_result = await loop.run_in_executor(
+            _analysis_pool,
+            lambda: meta_classifier_service.predict(
+                features, vocals=vocals,
+                wav2vec2=wav2vec2_result, clap=clap_result, fst=fst_result,
+            ),
         )
         if meta_result.model_version == MODEL_VERSION_TRAINED:
             meta_payload = MetaClassifierExplanation(
@@ -531,7 +574,7 @@ def _download_youtube_sync(
 
 
 async def _analyze_url(
-    request_id: str, url: str, start_time: float, progress: Progress,
+    request_id: str, url: str, start_time: float, progress: Progress, ticket: str,
 ) -> AnalyzeResponse:
     """Download a YouTube link's audio and run the same pipeline as a file."""
     try:
@@ -541,13 +584,15 @@ async def _analyze_url(
         progress.set("download", STEP_FAILED)
         return AnalyzeResponse(errors=["invalid_youtube_url"])
 
-    progress.set("download", STEP_RUNNING)
+    def download() -> Tuple[bytes, str, Optional[str], Optional[float], List[str]]:
+        # Marked running once a download thread is free, not while waiting for one.
+        progress.phase = "running"
+        progress.set("download", STEP_RUNNING)
+        return _download_youtube_sync(parsed.normalized_url, parsed.video_id, parsed.start_time_sec)
+
     loop = asyncio.get_running_loop()
     try:
-        content, ext, title, duration, dl_warnings = await loop.run_in_executor(
-            _download_pool, _download_youtube_sync,
-            parsed.normalized_url, parsed.video_id, parsed.start_time_sec,
-        )
+        content, ext, title, duration, dl_warnings = await loop.run_in_executor(_download_pool, download)
     except YouTubeDownloadError as exc:
         progress.set("download", STEP_FAILED)
         logger.warning(f"[{request_id}] YouTube download failed ({exc.error_code}): {exc}")
@@ -560,6 +605,7 @@ async def _analyze_url(
         logger.error(f"[{request_id}] YouTube download error: {exc}", exc_info=True)
         return AnalyzeResponse(errors=["youtube_analysis_failed"])
     progress.set("download", STEP_DONE)
+    progress.check()
 
     if len(content) < MIN_UPLOAD_BYTES:
         return AnalyzeResponse(errors=["youtube_analysis_failed"])
@@ -581,12 +627,21 @@ async def _analyze_url(
         source=source,
         start_time=start_time,
         progress=progress,
+        ticket=ticket,
         warnings=dl_warnings,
         total_duration=duration,
     )
 
 
-Work = Callable[[Progress], Awaitable[AnalyzeResponse]]
+Work = Callable[[Progress, str], Awaitable[AnalyzeResponse]]
+
+
+@dataclass
+class Plan:
+    steps: List[str]
+    work: Work
+    # Same file or same link gives the same key: shared while in flight, cached after.
+    key: Optional[str]
 
 
 async def _plan(
@@ -594,17 +649,21 @@ async def _plan(
     source_type: str,
     url: Optional[str],
     file: Optional[UploadFile],
-) -> Tuple[Optional[AnalyzeResponse], Optional[Tuple[List[str], Work]]]:
+) -> Tuple[Optional[AnalyzeResponse], Optional[Plan]]:
     """Validate a request. Returns an immediate error response, or the
-    steps plus the work to run (uploads are read here, while the request
-    is still open)."""
+    plan to run (uploads are read here, while the request is still open)."""
     start_time = time.monotonic()
     logger.info(f"[{request_id}] Analysis: sourceType={source_type}")
 
     if source_type in URL_SOURCES:
         if not url:
             return AnalyzeResponse(errors=["missing_url"]), None
-        return None, (URL_STEPS, lambda p: _analyze_url(request_id, url, start_time, p))
+        try:
+            parsed = parse_youtube_url(url)
+        except ValueError:
+            return AnalyzeResponse(errors=["invalid_youtube_url"]), None
+        key = f"yt:{parsed.video_id}:{parsed.start_time_sec or 0}:{PIPELINE_VERSION}"
+        return None, Plan(URL_STEPS, lambda p, t: _analyze_url(request_id, url, start_time, p, t), key)
 
     if source_type == "file":
         if not file:
@@ -619,17 +678,26 @@ async def _plan(
             "fileSizeBytes": len(content),
             "mimeType": content_type,
         }
-        return None, (FILE_STEPS, lambda p: _run_pipeline(
+        key = f"file:{hashlib.sha256(content).hexdigest()}:{PIPELINE_VERSION}"
+        return None, Plan(FILE_STEPS, lambda p, t: _run_pipeline(
             request_id, content,
             filename=filename, content_type=content_type,
-            source=source, start_time=start_time, progress=p,
-        ))
+            source=source, start_time=start_time, progress=p, ticket=t,
+        ), key)
 
     if source_type in ("spotify", "apple"):
         return AnalyzeResponse(errors=["unsupported_source"]), None
     return AnalyzeResponse(errors=["invalid_source_type"]), None
 
 
+def _busy() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        detail={"code": "server_busy", "message": "Too many analyses waiting. Try again shortly."},
+        headers={"Retry-After": "30"},
+    )
+
+
 # ── endpoints ─────────────────────────────────────────────────────────
 
 @router.post(
@@ -638,21 +706,33 @@ async def _plan(
     dependencies=[Depends(_analyze_rate_limit)],
 )
 async def analyze(
+    request: Request,
     sourceType: str = Form(...),
     url: Optional[str] = Form(None),
-    file: Optional[UploadFile] = File(None)
+    file: Optional[UploadFile] = File(None),
 ) -> AnalyzeResponse:
-    """Unified analysis endpoint (waits for the result)."""
+    """Unified analysis endpoint (waits for the result, same queue as jobs)."""
     request_id = str(uuid.uuid4())[:8]
+    immediate, plan = await _plan(request_id, sourceType, url, file)
+    if immediate is not None or plan is None:
+        return immediate or AnalyzeResponse(errors=["internal_error"])
+    cached = result_cache.get(plan.key)
+    if cached is not None:
+        return AnalyzeResponse(**cached)
+    if cpu_gate.waiting + job_store.pending_count() >= MAX_PENDING:
+        raise _busy()
+
+    progress = Progress(plan.steps)
+    progress.gone = request.is_disconnected
     try:
-        immediate, plan = await _plan(request_id, sourceType, url, file)
-        if immediate is not None or plan is None:
-            return immediate or AnalyzeResponse(errors=["internal_error"])
-        steps, work = plan
-        return await work(Progress(steps))
+        response = await plan.work(progress, f"sync-{request_id}")
+    except JobCancelled as stop:
+        return AnalyzeResponse(errors=[stop.code])
     except Exception as e:
         logger.error(f"[{request_id}] Error: {e}", exc_info=True)
         return AnalyzeResponse(errors=["internal_error"])
+    result_cache.put(plan.key, response.model_dump())
+    return response
 
 
 @router.post(
@@ -664,7 +744,7 @@ async def start_analysis_job(
     url: Optional[str] = Form(None),
     file: Optional[UploadFile] = File(None)
 ) -> JSONResponse:
-    """Start an analysis in the background; poll GET /api/analyze/jobs/{jobId}."""
+    """Queue an analysis; poll GET /api/analyze/jobs/{jobId}."""
     request_id = str(uuid.uuid4())[:8]
     immediate, plan = await _plan(request_id, sourceType, url, file)
     if immediate is not None or plan is None:
@@ -674,33 +754,80 @@ async def start_analysis_job(
             "steps": [], "elapsedSec": 0, "response": response.model_dump(),
         })
 
-    if job_store.running_count() >= MAX_RUNNING_JOBS:
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail={"code": "server_busy", "message": "Too many analyses running. Try again shortly."},
-        )
-
-    steps, work = plan
-
-    async def run(progress: Progress) -> Dict[str, Any]:
+    async def run(job: Job) -> Dict[str, Any]:
         try:
-            return (await work(progress)).model_dump()
+            return (await plan.work(job.progress, job.id)).model_dump()
+        except JobCancelled:
+            raise
         except Exception as e:
             logger.error(f"[{request_id}] Job error: {e}", exc_info=True)
             return AnalyzeResponse(errors=["internal_error"]).model_dump()
 
-    job = job_store.start(sourceType, steps, run)
-    logger.info(f"[{request_id}] Job {job.id} started")
+    try:
+        job = job_store.start(sourceType, plan.steps, run, key=plan.key)
+    except QueueFull:
+        raise _busy() from None
+    logger.info(f"[{request_id}] Job {job.id} {'from cache' if job.cached else 'queued'}")
     return JSONResponse(status_code=202, content=job.to_dict())
 
 
-@router.get("/api/analyze/jobs/{job_id}")
-async def get_analysis_job(job_id: str) -> Dict[str, Any]:
-    """Per-step progress of a job; `response` is filled once it settles."""
-    job = job_store.get(job_id)
+def _job_or_404(job: Optional[Job]) -> Job:
     if job is None:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail={"code": "job_not_found", "message": "Unknown or expired job."},
         )
-    return job.to_dict()
+    return job
+
+
+@router.get("/api/analyze/jobs/{job_id}")
+async def get_analysis_job(job_id: str) -> Dict[str, Any]:
+    """Per-step progress of a job; `response` is filled once it settles.
+    Polling is also what keeps a queued job alive."""
+    return _job_or_404(job_store.get(job_id)).to_dict()
+
+
+@router.delete("/api/analyze/jobs/{job_id}")
+async def cancel_analysis_job(job_id: str) -> Dict[str, Any]:
+    """Cancel a job: dropped if still queued, stopped at the next step if running."""
+    return _job_or_404(job_store.cancel(job_id)).to_dict()
+
+
+# ── warm-up ───────────────────────────────────────────────────────────
+
+warm_state: Dict[str, Any] = {"done": False, "seconds": None}
+
+
+async def warm_up() -> None:
+    """Load every model and let librosa compile its numba code on a short
+    synthetic signal, so the first visitor after a restart doesn't pay for it.
+    Holds the CPU slot while it runs; visitors arriving meanwhile queue."""
+    import numpy as np
+    import soundfile as sf
+
+    started = time.monotonic()
+    sr = 22050
+    t = np.arange(sr * 8) / sr
+    rng = np.random.default_rng(7)
+    y = (0.3 * np.sin(2 * np.pi * 220 * t) * (1 + 0.3 * np.sin(2 * np.pi * 2 * t))
+         + 0.15 * np.sin(2 * np.pi * 330 * t) + 0.02 * rng.standard_normal(t.size)).astype(np.float32)
+    buf = io.BytesIO()
+    sf.write(buf, y, sr, format="WAV", subtype="PCM_16")
+    loop = asyncio.get_running_loop()
+    try:
+        async with cpu_gate.slot("warm-up"):
+            clip = await loop.run_in_executor(_analysis_pool, decode_clip, buf.getvalue())
+            features, vocals, *_ = await asyncio.gather(
+                loop.run_in_executor(_analysis_pool, lambda: extract_features(io.BytesIO(clip))),
+                loop.run_in_executor(_analysis_pool, lambda: analyze_vocals(io.BytesIO(clip))),
+                loop.run_in_executor(_analysis_pool, lambda: clap_service.predict(io.BytesIO(clip))),
+                loop.run_in_executor(_analysis_pool, lambda: wav2vec2_service.predict(io.BytesIO(clip))),
+            )
+            if xai_service.available:
+                await loop.run_in_executor(_analysis_pool, xai_service.predict, features, vocals)
+            await loop.run_in_executor(_analysis_pool, lambda: meta_classifier_service.predict(features, vocals=vocals))
+        warm_state["done"] = True
+        warm_state["seconds"] = round(time.monotonic() - started, 1)
+        logger.info(f"Warm-up finished in {warm_state['seconds']}s")
+    except Exception as e:  # noqa: BLE001 — a failed warm-up must not take the API down
+        logger.warning(f"Warm-up failed: {e}", exc_info=True)