Spaces:
Build error
Build error
| """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]] | |
| 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 ───────────────────────────────────────────────────────── | |
| 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 | |
| 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 | |
| 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() | |
| 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) | |