"""FpcalcFingerprinter — Chromaprint fingerprint via the ``fpcalc`` binary. ``fpcalc -json `` emits ``{"duration": float, "fingerprint": str}``. The binary ships in the Docker image (``libchromaprint-tools``). Any failure (binary missing, bad file, timeout) degrades to ``None`` — the pipeline then falls back to tag-only metadata (plan §6.2: one external dependency must never crash it). """ import asyncio import json import shutil from pathlib import Path from app.core.logging import get_logger from app.domain.entities.metadata import Fingerprint log = get_logger(__name__) _TIMEOUT_SECONDS = 30 class FpcalcFingerprinter: """Implements :class:`app.domain.ports.AudioFingerprinter`.""" def __init__(self, binary: str = "fpcalc") -> None: self._binary = binary def is_available(self) -> bool: return shutil.which(self._binary) is not None async def calculate(self, path: Path) -> Fingerprint | None: if not self.is_available(): return None try: proc = await asyncio.create_subprocess_exec( self._binary, "-json", str(path), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) async with asyncio.timeout(_TIMEOUT_SECONDS): stdout, _stderr = await proc.communicate() except (TimeoutError, OSError): log.warning("fpcalc_failed", path=str(path)) return None if proc.returncode != 0: log.warning("fpcalc_nonzero", path=str(path), returncode=proc.returncode) return None try: data = json.loads(stdout) fingerprint = str(data["fingerprint"]) duration = round(float(data["duration"])) except (json.JSONDecodeError, KeyError, ValueError): log.warning("fpcalc_bad_output", path=str(path)) return None if not fingerprint or duration <= 0: return None return Fingerprint(fingerprint=fingerprint, duration_seconds=duration)