"""General analysis route for CrownCode. Handles YouTube URLs and file uploads through a unified pipeline. Response format matches platform/hooks/analysisTypes.ts Pipeline: 1. Source acquisition (YouTube download or file upload) 2. Feature extraction, vocal analysis, CLAP and wav2vec2 in parallel 3. FST comparison model (external Space) 4. Score fusion, then the calibrated XAI verdict (LightGBM + votes + SHAP) 5. Meta-classifier second opinion, response formatting 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. """ 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 from fastapi import ( APIRouter, File, UploadFile, Form, Request, Depends, HTTPException, status, ) from fastapi.responses import JSONResponse from .analyze_schemas import ( AnalysisFeatures, AnalysisResult, AnalyzeResponse, AudioInfo, ConfidenceBandModel, FeatureContributionModel, MetaClassifierExplanation, MetaFeatureImportance, ModelVoteModel, Timeline, VocalAnalysis, XAIExplanation, ) 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 ( MAX_TRACK_SEC, decode_clip, extract_features, load_track, AudioFeatures, ) from ..services.timeline import scan_track from ..services.vocal_analyzer import ( analyze_vocals, VocalFeatures, ) from ..services.score_fusion import fuse_scores, FusionResult from ..services.clap_detector import CLAPDetectorService from ..services.fst_client import FSTClientService from ..services.wav2vec2_detector import Wav2Vec2DetectorService from ..services.inference_xai import get_xai_service from ..services.meta_classifier import MetaClassifierService, MODEL_VERSION_TRAINED from ..services.url_parser import parse_youtube_url from ..services.youtube_downloader import YouTubeDownloadError, YouTubeDownloader from ..services.client_ip import client_ip as _client_ip from ..services.logging_config import get_logger router = APIRouter() logger = get_logger(__name__) MAX_UPLOAD_BYTES = 30 * 1024 * 1024 MIN_UPLOAD_BYTES = 1024 AUDIO_EXTENSIONS = {".mp3", ".wav", ".flac", ".m4a", ".mp4", ".aac", ".ogg", ".opus", ".webm"} _EXT_MIME = { ".mp3": "audio/mpeg", ".wav": "audio/wav", ".flac": "audio/flac", ".m4a": "audio/mp4", ".mp4": "audio/mp4", ".aac": "audio/aac", ".ogg": "audio/ogg", ".opus": "audio/ogg", ".webm": "audio/webm", } URL_SOURCES = ("youtube", "tiktok", "instagram", "soundcloud", "twitter") # A link needs the part the timeline scan reads; the clip layers use its first 60 s. YOUTUBE_CLIP_SEC = MAX_TRACK_SEC YOUTUBE_MAX_BYTES = 100 * 1024 * 1024 # Bump when the pipeline's output changes, so results cached from the old one aren't served. PIPELINE_VERSION = "3" # Rate limiter for public heavy endpoints _analyze_rate_store: dict[str, list[float]] = defaultdict(list) _ANALYZE_RATE_WINDOW = 60 # seconds _ANALYZE_RATE_MAX = 20 # requests per minute per IP _ANALYZE_MAX_IPS = 10_000 async def _analyze_rate_limit(request: Request) -> None: """Rate limit for analysis endpoints.""" client_ip = _client_ip(request) now = time.time() cutoff = now - _ANALYZE_RATE_WINDOW if len(_analyze_rate_store) > _ANALYZE_MAX_IPS: stale = [ ip for ip, ts in _analyze_rate_store.items() if not ts or all(t <= cutoff for t in ts) ] for ip in stale: del _analyze_rate_store[ip] hits = [t for t in _analyze_rate_store[client_ip] if t > cutoff] if len(hits) >= _ANALYZE_RATE_MAX: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail={ "code": "rate_limit_exceeded", "message": "Too many requests. Please wait.", } ) hits.append(now) _analyze_rate_store[client_ip] = hits clap_service = CLAPDetectorService() fst_service = FSTClientService() wav2vec2_service = Wav2Vec2DetectorService() xai_service = get_xai_service() meta_classifier_service = MetaClassifierService() # Thread pool for CPU-bound audio analysis (feature extraction + vocal) _analysis_pool = ThreadPoolExecutor(max_workers=4) # Downloads wait on the network; keep them off the analysis workers. _download_pool = ThreadPoolExecutor(max_workers=2) # ── helpers ─────────────────────────────────────────────────────────── def _unique(values: List[str]) -> List[str]: return list(dict.fromkeys(values)) def _probe_audio(content: bytes) -> Tuple[Optional[int], Optional[int], Optional[float]]: """Original sample rate, channels and full duration, when soundfile can read the container (WAV/FLAC/OGG/MP3). The analysis itself loads mono at 22.05 kHz and only the first 60 s, so it can't tell these.""" try: import soundfile as sf info = sf.info(io.BytesIO(content)) return int(info.samplerate), int(info.channels), float(info.duration) except Exception: return None, None, None async def _scan_timeline( request_id: str, content: bytes, total_duration: Optional[float], progress: Progress, loop: "asyncio.AbstractEventLoop", ) -> Optional[Dict[str, Any]]: """Read the whole track window by window. Never fails the analysis: on any problem the result simply has no timeline.""" progress.set("timeline", STEP_RUNNING) try: y, sr = await loop.run_in_executor(_analysis_pool, load_track, content) total = total_duration or _probe_audio(content)[2] or y.size / sr payload = await loop.run_in_executor( _analysis_pool, lambda: scan_track( y, sr, xai_service, total_sec=total, on_progress=lambda done, n: progress.advance("timeline", done, n), cancelled=lambda: progress.cancelled, ), ) progress.set("timeline", STEP_DONE) return payload except ValueError as e: logger.info(f"[{request_id}] Timeline skipped: {e}") progress.set("timeline", STEP_SKIPPED) except Exception as e: # noqa: BLE001 - the verdict stands without it logger.warning(f"[{request_id}] Timeline failed: {e}", exc_info=True) progress.set("timeline", STEP_FAILED) return None def _track(future: "asyncio.Future[Any]", progress: Progress, step: str, ok: Optional[Callable[[Any], bool]] = None) -> None: """Report a parallel step as done/skipped/failed the moment it settles.""" def _settled(f: "asyncio.Future[Any]") -> None: if f.cancelled() or f.exception() is not None: progress.set(step, STEP_FAILED) elif ok is not None and not ok(f.result()): progress.set(step, STEP_SKIPPED) else: progress.set(step, STEP_DONE) future.add_done_callback(_settled) async def _read_upload(file: UploadFile) -> Tuple[Optional[bytes], Optional[str], str]: """Validate an upload. Returns (content, error_code, content_type).""" filename = file.filename or "audio" ext = Path(filename).suffix.lower() content_type = (file.content_type or "").lower() # Browsers label some audio containers as video (webm, mp4); trust the extension then. if not (content_type.startswith("audio/") or ext in AUDIO_EXTENSIONS): return None, "invalid_file_type", content_type if not content_type.startswith("audio/"): content_type = _EXT_MIME.get(ext, "audio/unknown") content = await file.read(MAX_UPLOAD_BYTES + 1) if len(content) > MAX_UPLOAD_BYTES: return None, "file_too_large", content_type if len(content) < MIN_UPLOAD_BYTES: return None, "file_too_small", content_type return content, None, content_type # ── the pipeline ────────────────────────────────────────────────────── async def _run_pipeline( request_id: str, content: bytes, *, filename: str, content_type: str, 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. 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 timeline_payload: Optional[Dict[str, Any]] = None 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() # The rest of the track, window by window (same slot, same models). scan = xai_service.available and feat_future.exception() is None timeline_payload = await _scan_timeline(request_id, content, total_duration, progress, loop) if scan else None if scan and timeline_payload is None: warnings.append("timeline_unavailable") progress.check() progress.check() except BaseException: if fst_task is not None: fst_task.cancel() raise # Feature extraction is required for everything below. try: 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 Exception as e: logger.error(f"[{request_id}] Feature extraction error: {e}", exc_info=True) fst_task.cancel() progress.skip_pending() return AnalyzeResponse(errors=["audio_decode_failed"], warnings=_unique(warnings)) vocals: Optional[VocalFeatures] = None try: vocals = vocal_future.result() if vocals.has_vocals: logger.info( f"[{request_id}] Vocals: " f"ai={vocals.vocal_ai_score:.3f} " f"pitch={vocals.pitch_std_cents:.1f}c" ) else: logger.info(f"[{request_id}] No vocals detected") except Exception as e: logger.warning(f"[{request_id}] Vocal analysis failed: {e}") warnings.append("vocal_analysis_unavailable") clap_result = None try: clap_result = clap_future.result() if clap_result.available: logger.info( f"[{request_id}] CLAP ({clap_result.classifier_used}): " f"ai={clap_result.is_ai}, conf={clap_result.confidence:.3f}" ) else: logger.info(f"[{request_id}] CLAP unavailable: {clap_result.error}") warnings.append("clap_analysis_unavailable") except Exception as e: logger.warning(f"[{request_id}] CLAP failed: {e}") warnings.append("clap_analysis_unavailable") wav2vec2_result = None try: wav2vec2_result = wav2vec2_future.result() if wav2vec2_result.available: logger.info(f"[{request_id}] wav2vec2: p_ai={wav2vec2_result.p_ai:.3f}") else: logger.info(f"[{request_id}] wav2vec2 unavailable: {wav2vec2_result.error}") warnings.append("wav2vec2_unavailable") except Exception as e: logger.warning(f"[{request_id}] wav2vec2 failed: {e}") warnings.append("wav2vec2_unavailable") # ── FST detection (external Space), started with the CPU stage ── fst_result = None try: fst_result = await fst_task if fst_result.available: progress.set("fst", STEP_DONE) logger.info( f"[{request_id}] FST: ai={fst_result.is_ai}, " f"conf={fst_result.confidence:.3f}, label={fst_result.label}" ) else: progress.set("fst", STEP_SKIPPED) logger.info(f"[{request_id}] FST unavailable: {fst_result.error}") warnings.append("fst_analysis_unavailable") except Exception as e: 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( features, vocals, clap_result=clap_result, fst_result=fst_result, wav2vec2_result=wav2vec2_result, ) logger.info( f"[{request_id}] Fusion: ai={fusion.is_ai_generated}, " f"confidence={fusion.confidence}, source={fusion.decision_source}" ) # ── XAI inference (calibrated verdict + votes + SHAP) ── xai_payload: Optional[XAIExplanation] = None progress.set("xai", STEP_RUNNING) try: # 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( probability=xai_dict["probability"], threshold=xai_dict["threshold"], baseProbability=xai_dict["baseProbability"], confidenceBand=ConfidenceBandModel(**xai_dict["confidenceBand"]), modelVotes=[ModelVoteModel(**v) for v in xai_dict["modelVotes"]], bestModel=xai_dict["bestModel"], topContributions=[ FeatureContributionModel(**c) for c in xai_dict["topContributions"] ], allFeatures={ k: FeatureContributionModel(**v) for k, v in xai_dict["allFeatures"].items() }, featureCount=xai_dict["featureCount"], ) progress.set("xai", STEP_DONE) logger.info( f"[{request_id}] XAI: p={xai_result.probability:.3f}, " f"band={xai_result.confidence_band.tier}, " f"top={xai_result.top_contributions[0].name if xai_result.top_contributions else 'n/a'}" ) else: progress.set("xai", STEP_SKIPPED) warnings.append("xai_unavailable") except Exception as e: progress.set("xai", STEP_FAILED) logger.warning(f"[{request_id}] XAI failed: {e}") warnings.append("xai_unavailable") # ── Meta-classifier: a second opinion, never overrides the verdict ── meta_payload: Optional[MetaClassifierExplanation] = None progress.set("meta", STEP_RUNNING) try: 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( isAIGenerated=meta_result.is_ai_generated, confidence=meta_result.confidence, modelVersion=meta_result.model_version, towerScores={k: round(float(v), 3) for k, v in meta_result.tower_scores.items()}, indicators=meta_result.indicators, topFeatures=[ MetaFeatureImportance(**f) for f in meta_result.top_features ], ) progress.set("meta", STEP_DONE) logger.info( f"[{request_id}] Meta-classifier: ai={meta_result.is_ai_generated}, " f"confidence={meta_result.confidence}" ) else: progress.set("meta", STEP_SKIPPED) warnings.append("meta_classifier_unavailable") except Exception as e: progress.set("meta", STEP_FAILED) logger.warning(f"[{request_id}] Meta-classifier failed: {e}") warnings.append("meta_classifier_unavailable") # ── Tower scores (P(AI) per independent layer) ── tower_scores: Dict[str, float] = { "local_features": round( (features.spectral_regularity + features.temporal_patterns + features.harmonic_structure) / 3.0, 3, ), } if wav2vec2_result and wav2vec2_result.available: tower_scores["wav2vec2"] = round(wav2vec2_result.p_ai, 3) if vocals and vocals.has_vocals: tower_scores["vocals"] = round(vocals.vocal_ai_score, 3) if clap_result and clap_result.available: tower_scores["clap"] = round(clap_result.confidence, 3) if fst_result and fst_result.available: tower_scores["fst"] = round(fst_result.confidence, 3) if xai_payload is not None: tower_scores["xai_ensemble"] = round(xai_payload.probability, 3) # Which implementation each layer actually used this time. layers: Dict[str, Dict[str, Any]] = { "wav2vec2": { "available": bool(wav2vec2_result and wav2vec2_result.available), "error": getattr(wav2vec2_result, "error", None), }, "clap": { "available": bool(clap_result and clap_result.available), "mode": getattr(clap_result, "classifier_used", None), "error": getattr(clap_result, "error", None), }, "fst": { "available": bool(fst_result and fst_result.available), "label": getattr(fst_result, "label", None), "error": getattr(fst_result, "error", None), }, "xai": { "available": xai_payload is not None, "ensembleLoaded": len(xai_service.ensemble_models), }, "meta": {"available": meta_payload is not None}, } vocal_response = None if vocals and vocals.has_vocals: vocal_response = VocalAnalysis( hasVocals=True, vocalConfidence=round(vocals.vocal_confidence, 3), vocalAiScore=round(vocals.vocal_ai_score, 3), pitchStabilityScore=round(vocals.pitch_stability_score, 3), vibratoRegularityScore=round(vocals.vibrato_regularity_score, 3), formantConsistencyScore=round(vocals.formant_consistency_score, 3), breathPatternScore=round(vocals.breath_pattern_score, 3), vocalTextureScore=round(vocals.vocal_texture_score, 3), pitchMeanHz=round(vocals.pitch_mean_hz, 1), pitchStdCents=round(vocals.pitch_std_cents, 1), vibratoRateHz=round(vocals.vibrato_rate_hz, 2), vibratoExtentCents=round(vocals.vibrato_extent_cents, 1), indicators=vocals.indicators, ) # ── Audio facts: from the container when readable, else what was analysed ── orig_sr, orig_channels, probed_duration = _probe_audio(content) duration = total_duration or probed_duration or features.duration_sec bitrate = 0 if probed_duration and probed_duration > 0: bitrate = max(1, int((file_size * 8) / (probed_duration * 1000))) # Prefer the XAI verdict — it's the calibrated classifier. final_is_ai = fusion.is_ai_generated final_confidence = fusion.confidence final_decision = fusion.decision_source if xai_payload is not None: final_is_ai = xai_payload.probability >= xai_payload.threshold final_confidence = round(max(0.03, min(0.97, xai_payload.probability)), 4) final_decision = f"auris_xai_{xai_payload.bestModel.lower()}" result = AnalysisResult( isAIGenerated=final_is_ai, confidence=final_confidence, processingTime=round(time.monotonic() - start_time, 3), modelVersion=fusion.model_version, decisionSource=final_decision, analysisMode=fusion.analysis_mode, source=source, features=AnalysisFeatures( spectralRegularity=fusion.spectral_regularity, temporalPatterns=fusion.temporal_patterns, harmonicStructure=fusion.harmonic_structure, artificialIndicators=fusion.indicators, ), audioInfo=AudioInfo( duration=round(duration, 2), analysedSec=round(features.duration_sec, 2), sampleRate=orig_sr or features.sample_rate, bitrate=bitrate, format=Path(filename).suffix.lstrip(".").lower() or content_type.replace("audio/", ""), channels=orig_channels or 0, ), vocalAnalysis=vocal_response, xai=xai_payload, towerScores=tower_scores, metaClassifier=meta_payload, timeline=Timeline(**timeline_payload) if timeline_payload else None, layers=layers, ) return AnalyzeResponse(result=result, warnings=_unique(warnings), errors=[]) # ── sources ─────────────────────────────────────────────────────────── def _download_youtube_sync( normalized_url: str, video_id: str, start_sec: Optional[int], ) -> Tuple[bytes, str, Optional[str], Optional[float], List[str]]: """Download the opening of a video (from ?t= if given) to memory.""" start = float(start_sec or 0) with tempfile.TemporaryDirectory() as tmp: downloader = YouTubeDownloader( Path(tmp), clip=(start, start + YOUTUBE_CLIP_SEC), max_filesize=YOUTUBE_MAX_BYTES, ) result = downloader.download(normalized_url, video_id) content = result.file_path.read_bytes() return content, result.file_path.suffix.lower(), result.title, result.duration_sec, result.warnings async def _analyze_url( 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: parsed = parse_youtube_url(url) except ValueError as e: logger.warning(f"[{request_id}] Invalid URL: {e}") progress.set("download", STEP_FAILED) return AnalyzeResponse(errors=["invalid_youtube_url"]) 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) except YouTubeDownloadError as exc: progress.set("download", STEP_FAILED) logger.warning(f"[{request_id}] YouTube download failed ({exc.error_code}): {exc}") errors = ["youtube_analysis_failed"] if exc.error_code == "youtube_authentication_required": errors.insert(0, "youtube_authentication_required") return AnalyzeResponse(errors=errors, warnings=_unique(exc.warnings)) except Exception as exc: progress.set("download", STEP_FAILED) 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"]) source: Dict[str, Any] = { "kind": "youtube", "url": url, "normalizedUrl": parsed.normalized_url, "videoId": parsed.video_id, "startTimeSec": parsed.start_time_sec, } if title: source["title"] = title return await _run_pipeline( request_id, content, filename=f"{parsed.video_id}{ext}", content_type=_EXT_MIME.get(ext, "audio/wav"), source=source, start_time=start_time, progress=progress, ticket=ticket, warnings=dl_warnings, total_duration=duration, ) 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( request_id: str, source_type: str, url: Optional[str], file: Optional[UploadFile], ) -> Tuple[Optional[AnalyzeResponse], Optional[Plan]]: """Validate a request. Returns an immediate error response, or the 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 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: return AnalyzeResponse(errors=["missing_file"]), None content, error, content_type = await _read_upload(file) if error or content is None: return AnalyzeResponse(errors=[error or "invalid_file_type"]), None filename = file.filename or "audio" source = { "kind": "file", "fileName": filename, "fileSizeBytes": len(content), "mimeType": content_type, } 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, 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( "/api/analyze", response_model=AnalyzeResponse, dependencies=[Depends(_analyze_rate_limit)], ) async def analyze( request: Request, sourceType: str = Form(...), url: Optional[str] = Form(None), file: Optional[UploadFile] = File(None), ) -> AnalyzeResponse: """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: 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( "/api/analyze/jobs", dependencies=[Depends(_analyze_rate_limit)], ) async def start_analysis_job( sourceType: str = Form(...), url: Optional[str] = Form(None), file: Optional[UploadFile] = File(None) ) -> JSONResponse: """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: response = immediate or AnalyzeResponse(errors=["internal_error"]) return JSONResponse(status_code=200, content={ "jobId": None, "sourceType": sourceType, "status": "error", "steps": [], "elapsedSec": 0, "response": response.model_dump(), }) async def run(job: Job) -> Dict[str, Any]: try: 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() 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()) 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 @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)