feat: implement score fusion engine for AI music detection

app/services/score_fusion.py

@@ -0,0 +1,327 @@
+"""
+Score fusion engine for AURIS AI music detection.
+
+Combines signals from multiple analysis sources (feature extraction,
+vocal analysis, external services) into a single confidence score
+with human-readable indicators.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import List, Literal, Optional
+
+from .feature_extractor import AudioFeatures
+from .vocal_analyzer import VocalFeatures
+from .logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class FusionResult:
+    """Final fused detection result."""
+
+    is_ai_generated: bool
+    confidence: float                   # 0.51 – 0.97
+    decision_source: str                # "auris_fusion"
+    model_version: str                  # "auris-v1-fusion"
+
+    # Per-domain scores
+    spectral_regularity: float
+    temporal_patterns: float
+    harmonic_structure: float
+    vocal_ai_score: float               # 0 if no vocals
+
+    # Indicators
+    indicators: List[str] = field(default_factory=list)
+
+    # Raw sub-results for debugging / advanced UI
+    has_vocals: bool = False
+    analysis_mode: Literal[
+        "production", "preview"
+    ] = "production"
+
+
+def fuse_scores(
+    features: AudioFeatures,
+    vocals: Optional[VocalFeatures] = None,
+    *,
+    external_music_ai: Optional[dict] = None,
+    external_ses_analizi: Optional[dict] = None,
+) -> FusionResult:
+    """
+    Fuse all analysis signals into a single detection result.
+
+    Priority hierarchy:
+    1. External production models (music_ai, ses_analizi) if available
+    2. Local feature + vocal analysis (always available)
+
+    When external models agree with local analysis, confidence is boosted.
+    When they disagree, confidence is moderated.
+
+    Args:
+        features: Audio feature extraction results.
+        vocals: Vocal analysis results (None if skipped).
+        external_music_ai: Response from MusicAI service.
+        external_ses_analizi: Response from SesAnalizi service.
+
+    Returns:
+        FusionResult with final detection decision.
+    """
+    indicators: List[str] = []
+
+    # ── Local feature scores ─────────────────────────────────────────
+    local_score = _compute_local_score(features, vocals)
+
+    vocal_ai_score = 0.0
+    if vocals and vocals.has_vocals:
+        vocal_ai_score = vocals.vocal_ai_score
+    has_vocals = vocals is not None and vocals.has_vocals
+
+    # ── External service scores ──────────────────────────────────────
+    ext_music_ai = _parse_external_music_ai(external_music_ai)
+    ext_ses = _parse_external_ses_analizi(external_ses_analizi)
+
+    # ── Fusion logic ─────────────────────────────────────────────────
+    confidence, decision_source, mode = _fuse(
+        local_score, ext_music_ai, ext_ses
+    )
+
+    is_ai = confidence > 0.5
+
+    # ── Build indicators ─────────────────────────────────────────────
+    indicators.extend(_build_indicators(
+        is_ai, confidence, features, vocals, ext_music_ai, ext_ses
+    ))
+
+    return FusionResult(
+        is_ai_generated=is_ai,
+        confidence=confidence,
+        decision_source=decision_source,
+        model_version="auris-v1-fusion",
+        spectral_regularity=features.spectral_regularity,
+        temporal_patterns=features.temporal_patterns,
+        harmonic_structure=features.harmonic_structure,
+        vocal_ai_score=vocal_ai_score,
+        indicators=indicators,
+        has_vocals=has_vocals,
+        analysis_mode=mode,
+    )
+
+
+# ═══════════════════════════════════════════════════════════════════════
+# PRIVATE — Local score computation
+# ═══════════════════════════════════════════════════════════════════════
+
+def _compute_local_score(
+    features: AudioFeatures,
+    vocals: Optional[VocalFeatures],
+) -> float:
+    """
+    Compute local AI detection score from feature extraction + vocals.
+
+    Without vocals: 3 feature domains equally weighted.
+    With vocals: vocal score gets 30% weight, features split remaining 70%.
+    """
+    feat_score = (
+        features.spectral_regularity * 0.35
+        + features.temporal_patterns * 0.35
+        + features.harmonic_structure * 0.30
+    )
+
+    if vocals and vocals.has_vocals:
+        # Vocals are a strong signal — give them significant weight
+        combined = feat_score * 0.65 + vocals.vocal_ai_score * 0.35
+        logger.info(
+            f"Local score: features={feat_score:.3f}, "
+            f"vocals={vocals.vocal_ai_score:.3f}, combined={combined:.3f}"
+        )
+        return combined
+
+    logger.info(f"Local score: features={feat_score:.3f} (no vocals)")
+    return feat_score
+
+
+# ═══════════════════════════════════════════════════════════════════════
+# PRIVATE — External service parsing
+# ═══════════════════════════════════════════════════════════════════════
+
+@dataclass
+class _ExternalScore:
+    available: bool
+    is_ai: bool
+    confidence: float
+
+
+def _parse_external_music_ai(data: Optional[dict]) -> Optional[_ExternalScore]:
+    if not data or not isinstance(data, dict):
+        return None
+    prediction = data.get("prediction")
+    confidence = data.get("confidence")
+    is_valid = (
+        prediction in ("AI", "Human")
+        and isinstance(confidence, (int, float))
+    )
+    if is_valid:
+        return _ExternalScore(
+            available=True,
+            is_ai=prediction == "AI",
+            confidence=float(confidence),
+        )
+    return None
+
+
+def _parse_external_ses_analizi(data: Optional[dict]) -> Optional[_ExternalScore]:
+    if not data or not isinstance(data, dict):
+        return None
+    authenticity = data.get("authenticity_score")
+    if isinstance(authenticity, (int, float)):
+        return _ExternalScore(
+            available=True,
+            is_ai=float(authenticity) >= 0.5,
+            confidence=float(authenticity),
+        )
+    return None
+
+
+# ═══════════════════════════════════════════════════════════════════════
+# PRIVATE — Score fusion
+# ═══════════════════════════════════════════════════════════════════════
+
+def _fuse(
+    local: float,
+    ext_mai: Optional[_ExternalScore],
+    ext_ses: Optional[_ExternalScore],
+) -> tuple[float, str, str]:
+    """
+    Fuse local + external scores.
+
+    Returns: (confidence, decision_source, analysis_mode)
+    """
+    scores: list[tuple[float, float, str]] = []  # (score, weight, source)
+
+    # Local always present
+    scores.append((local, 0.5, "auris_local"))
+
+    if ext_mai and ext_mai.available:
+        scores.append((ext_mai.confidence, 0.3, "music_ai"))
+
+    if ext_ses and ext_ses.available:
+        scores.append((ext_ses.confidence, 0.2, "ses_analizi"))
+
+    # Normalize weights
+    total_weight = sum(w for _, w, _ in scores)
+    confidence = sum(s * (w / total_weight) for s, w, _ in scores)
+
+    # Determine primary decision source
+    if ext_mai and ext_mai.available:
+        decision_source = "auris_fusion"
+        mode = "production"
+    elif ext_ses and ext_ses.available:
+        decision_source = "auris_fusion"
+        mode = "production"
+    else:
+        decision_source = "auris_local"
+        mode = "production"  # Still production — we have real feature analysis
+
+    # Clamp to valid range (never show 0.50 or 1.00)
+    confidence = max(0.51, min(0.97, confidence))
+    confidence = round(confidence, 4)
+
+    # Agreement bonus: if all sources agree, boost confidence toward edges
+    if len(scores) > 1:
+        all_ai = all(s > 0.5 for s, _, _ in scores)
+        all_human = all(s <= 0.5 for s, _, _ in scores)
+        if all_ai:
+            confidence = min(0.97, confidence * 1.05)
+        elif all_human:
+            confidence = max(0.51, confidence * 0.95)
+
+    confidence = round(max(0.51, min(0.97, confidence)), 4)
+
+    logger.info(
+        f"Fusion: {len(scores)} sources → confidence={confidence}, "
+        f"source={decision_source}, mode={mode}"
+    )
+
+    return confidence, decision_source, mode
+
+
+# ═══════════════════════════════════════════════════════════════════════
+# PRIVATE — Indicator generation
+# ═══════════════════════════════════════════════════════════════════════
+
+def _build_indicators(
+    is_ai: bool,
+    confidence: float,
+    features: AudioFeatures,
+    vocals: Optional[VocalFeatures],
+    ext_mai: Optional[_ExternalScore],
+    ext_ses: Optional[_ExternalScore],
+) -> List[str]:
+    """Generate comprehensive human-readable indicators."""
+    indicators = []
+
+    # Overall confidence indicator
+    if confidence > 0.85:
+        indicators.append(
+            "High confidence classification based on "
+            "multi-signal analysis."
+        )
+    elif confidence > 0.70:
+        indicators.append(
+            "Moderate confidence with consistent "
+            "feature signals."
+        )
+    else:
+        indicators.append(
+            "Lower confidence suggests borderline "
+            "characteristics."
+        )
+
+    # Feature-specific indicators
+    if features.spectral_regularity > 0.7:
+        indicators.append(
+            "Spectral patterns show high regularity typical of AI generation."
+        )
+    elif features.spectral_regularity < 0.3:
+        indicators.append(
+            "Spectral variation is consistent with natural human composition."
+        )
+
+    if features.temporal_patterns > 0.7:
+        indicators.append(
+            f"Temporal patterns are metronomically precise "
+            f"(tempo stability: {features.tempo_stability:.3f}s std)."
+        )
+    elif features.temporal_patterns < 0.3:
+        indicators.append(
+            "Natural timing variation detected in rhythmic patterns."
+        )
+
+    if features.harmonic_structure > 0.7:
+        indicators.append(
+            "Harmonic progressions follow predictable AI-typical patterns."
+        )
+
+    # Vocal indicators
+    if vocals and vocals.has_vocals:
+        indicators.extend(vocals.indicators)
+
+    # External source indicators
+    if ext_mai and ext_mai.available:
+        label = "AI-generated" if ext_mai.is_ai else "human-composed"
+        indicators.append(
+            f"External Music-AI detector classified as {label} "
+            f"({ext_mai.confidence:.1%} confidence)."
+        )
+
+    if ext_ses and ext_ses.available:
+        label = "synthetic" if ext_ses.is_ai else "authentic"
+        indicators.append(
+            f"Ses-Analizi service classified as {label} "
+            f"({ext_ses.confidence:.1%} confidence)."
+        )
+
+    return indicators