feat: add FST client for AI music detection using HuggingFace API

app/services/fst_client.py

@@ -0,0 +1,266 @@
+"""
+FST (Fusion Segment Transformer) external API client (Layer 3).
+
+Calls the HuggingFace Space ``mippia/AI-Music-Detection-FST``
+Gradio API for high-accuracy AI music detection.
+
+FST uses MERT + beat-aware segmentation and reports 99.99%
+accuracy on benchmark datasets.  We treat it as a strong
+external signal in the score-fusion pipeline.
+
+Gracefully returns unavailable result on timeout or error.
+"""
+
+from __future__ import annotations
+
+import io
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional, Union
+
+from .logging_config import get_logger
+
+logger = get_logger(__name__)
+
+# HuggingFace Space endpoint
+FST_SPACE_ID = "mippia/AI-Music-Detection-FST"
+FST_API_URL = f"https://{FST_SPACE_ID.replace('/', '-')}.hf.space"
+
+# Timeouts
+FST_CONNECT_TIMEOUT = 10.0   # seconds
+FST_PREDICT_TIMEOUT = 120.0  # seconds (model inference can be slow)
+
+
+@dataclass
+class FSTResult:
+    """Result from FST external service."""
+
+    available: bool
+    is_ai: bool = False
+    confidence: float = 0.5
+    label: str = "unknown"
+    raw_scores: Optional[dict] = None
+    error: Optional[str] = None
+
+
+class FSTClientService:
+    """
+    Client for FST AI Music Detection HuggingFace Space.
+
+    Uses the Gradio Client API to submit audio and receive
+    predictions. Falls back gracefully if the space is
+    sleeping, overloaded, or unreachable.
+    """
+
+    def __init__(self) -> None:
+        self._client = None
+        self._available: Optional[bool] = None
+
+    def _ensure_client(self) -> bool:
+        """Lazy-initialize Gradio client."""
+        if self._available is not None:
+            return self._available
+
+        try:
+            from gradio_client import Client
+            self._client = Client(
+                FST_SPACE_ID,
+                hf_token=None,  # Public space
+            )
+            self._available = True
+            logger.info(f"FST client connected: {FST_SPACE_ID}")
+            return True
+        except ImportError:
+            logger.warning(
+                "gradio_client not installed — FST layer disabled"
+            )
+            self._available = False
+            return False
+        except Exception as e:
+            logger.warning(f"FST client init failed: {e}")
+            self._available = False
+            return False
+
+    async def predict(
+        self,
+        source: Union[Path, bytes, io.BytesIO],
+    ) -> FSTResult:
+        """
+        Submit audio to FST Space for AI detection.
+
+        Args:
+            source: Audio file path, raw bytes, or BytesIO.
+
+        Returns:
+            FSTResult with detection outcome.
+        """
+        if not self._ensure_client():
+            return FSTResult(
+                available=False,
+                error="fst_client_unavailable",
+            )
+
+        try:
+            audio_path = self._to_file_path(source)
+            return await self._call_api(audio_path)
+        except Exception as e:
+            logger.warning(f"FST prediction failed: {e}")
+            return FSTResult(
+                available=False,
+                error=str(e),
+            )
+
+    async def _call_api(self, audio_path: Path) -> FSTResult:
+        """Call the FST Gradio API."""
+        import asyncio
+
+        try:
+            # Run synchronous Gradio client in executor
+            loop = asyncio.get_event_loop()
+            result = await asyncio.wait_for(
+                loop.run_in_executor(
+                    None,
+                    self._sync_predict,
+                    audio_path,
+                ),
+                timeout=FST_PREDICT_TIMEOUT,
+            )
+            return result
+
+        except asyncio.TimeoutError:
+            logger.warning(
+                f"FST prediction timed out after "
+                f"{FST_PREDICT_TIMEOUT}s"
+            )
+            return FSTResult(
+                available=False,
+                error="fst_timeout",
+            )
+
+    def _sync_predict(self, audio_path: Path) -> FSTResult:
+        """Synchronous Gradio predict call."""
+        try:
+            result = self._client.predict(
+                str(audio_path),
+                api_name="/predict",
+            )
+
+            # Parse Gradio response
+            # FST typically returns label + confidence dict
+            return self._parse_response(result)
+
+        except Exception as e:
+            logger.warning(f"FST sync predict error: {e}")
+            return FSTResult(
+                available=False,
+                error=str(e),
+            )
+
+    def _parse_response(self, response) -> FSTResult:
+        """
+        Parse FST Gradio API response.
+
+        FST response format varies — handle multiple formats:
+        1. Dict with 'label' and 'confidences'
+        2. String label with confidence
+        3. Raw dict with scores
+        """
+        try:
+            if isinstance(response, dict):
+                return self._parse_dict_response(response)
+            elif isinstance(response, str):
+                return self._parse_string_response(response)
+            elif isinstance(response, (list, tuple)):
+                # First element is usually the classification
+                if len(response) > 0:
+                    return self._parse_response(response[0])
+            else:
+                logger.warning(
+                    f"Unexpected FST response type: "
+                    f"{type(response)}"
+                )
+                return FSTResult(
+                    available=True,
+                    label="parse_error",
+                    error=f"unexpected_type: {type(response).__name__}",
+                )
+        except Exception as e:
+            logger.warning(f"FST response parse error: {e}")
+            return FSTResult(
+                available=False,
+                error=f"parse_error: {e}",
+            )
+
+    def _parse_dict_response(self, data: dict) -> FSTResult:
+        """Parse dict-style response."""
+        # Format: {"label": "AI", "confidences": [{"label": "AI", "confidence": 0.99}, ...]}
+        label = data.get("label", "unknown")
+        confidences = data.get("confidences", [])
+
+        is_ai = "ai" in label.lower() or "fake" in label.lower()
+
+        confidence = 0.5
+        if confidences and isinstance(confidences, list):
+            for item in confidences:
+                if isinstance(item, dict):
+                    item_label = item.get("label", "")
+                    if "ai" in item_label.lower() or "fake" in item_label.lower():
+                        confidence = float(
+                            item.get("confidence", 0.5)
+                        )
+                        break
+
+        # If no AI confidence found, use first confidence
+        if confidence == 0.5 and confidences:
+            first = confidences[0]
+            if isinstance(first, dict):
+                confidence = float(
+                    first.get("confidence", 0.5)
+                )
+                if not is_ai:
+                    confidence = 1.0 - confidence
+
+        return FSTResult(
+            available=True,
+            is_ai=is_ai,
+            confidence=round(
+                max(0.01, min(0.99, confidence)), 4
+            ),
+            label=label,
+            raw_scores=data,
+        )
+
+    def _parse_string_response(self, text: str) -> FSTResult:
+        """Parse string-style response."""
+        lower = text.lower().strip()
+        is_ai = any(
+            kw in lower
+            for kw in ("ai", "fake", "generated", "synthetic")
+        )
+        # Conservative confidence for string-only responses
+        confidence = 0.75 if is_ai else 0.25
+
+        return FSTResult(
+            available=True,
+            is_ai=is_ai,
+            confidence=confidence,
+            label=text.strip(),
+        )
+
+    @staticmethod
+    def _to_file_path(
+        source: Union[Path, bytes, io.BytesIO],
+    ) -> Path:
+        """Convert source to a file path for Gradio upload."""
+        if isinstance(source, Path):
+            return source
+        if isinstance(source, bytes):
+            source = io.BytesIO(source)
+        tmp = tempfile.NamedTemporaryFile(
+            suffix=".wav", delete=False,
+        )
+        tmp.write(source.read())
+        tmp.flush()
+        tmp.close()
+        return Path(tmp.name)