Add opt-in song identification (ShazamIO), offline-safe & cached

- oida/songid.py: SongIdProvider abstraction (ShazamIOProvider + description
  fallback), default OFF (OIDA_SONGID toggle), results shaped for an akousma's
  extensions.songid, cached by audio content hash, graceful offline/no-match so
  oída's descriptive path is never lost. enrich_akousma() lands results in a record.
- shazamio as an optional extra (pip install 'oida[songid]').
- tests: mocked provider (no network) — toggle, match shape, offline, cache, enrich.

Author: emeisazam

oida/songid.py

@@ -0,0 +1,158 @@
+"""Song identification for oída — opt-in, provider-abstracted, offline-safe.
+
+When oída's listening classifies a source as music, an optional "song id" toggle can try to
+name the track. Default is OFF. Uses ShazamIO (an *unofficial* API — no SLA; sends an audio
+fingerprint, not the raw recording) behind a provider interface, so open-data providers
+(AcoustID/MusicBrainz) can be added later without touching callers. On any failure or when
+disabled, it degrades gracefully to "no match" and oída's descriptive path is unaffected.
+
+Results are shaped for an akousma's ``extensions.songid`` block.
+"""
+from __future__ import annotations
+
+import os
+import time
+from dataclasses import dataclass, field
+from hashlib import sha256
+from pathlib import Path
+from typing import Any, Callable, Protocol
+
+
+def songid_enabled(explicit: bool | None = None) -> bool:
+    """Toggle state: explicit arg wins, else ``OIDA_SONGID`` env (default OFF)."""
+    if explicit is not None:
+        return explicit
+    return os.getenv("OIDA_SONGID", "0").strip().lower() in {"1", "true", "yes", "on"}
+
+
+@dataclass
+class SongIdResult:
+    provider: str
+    matched: bool = False
+    title: str | None = None
+    artist: str | None = None
+    album: str | None = None
+    isrc: str | None = None
+    track_id: str | None = None
+    confidence: float | None = None
+    note: str | None = None
+    checked_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
+
+    def to_extension(self) -> dict[str, Any]:
+        data = {
+            "provider": self.provider,
+            "matched": self.matched,
+            "checked_at": self.checked_at,
+        }
+        for k in ("title", "artist", "album", "isrc", "track_id", "confidence", "note"):
+            v = getattr(self, k)
+            if v is not None:
+                data[k] = v
+        return data
+
+
+class SongIdProvider(Protocol):
+    name: str
+
+    def identify(self, audio_path: Path) -> SongIdResult:
+        ...
+
+
+class DescriptionFallbackProvider:
+    """Never matches — signals callers to keep oída's descriptive path."""
+
+    name = "description"
+
+    def identify(self, audio_path: Path) -> SongIdResult:
+        return SongIdResult(provider=self.name, matched=False, note="song id disabled; describing instead")
+
+
+class ShazamIOProvider:
+    """ShazamIO-backed identification. Lazy-imports shazamio; any error → matched=False."""
+
+    name = "shazamio"
+
+    def identify(self, audio_path: Path) -> SongIdResult:
+        try:
+            import asyncio
+
+            from shazamio import Shazam  # optional dep: pip install 'oida[songid]'
+
+            async def _run() -> dict[str, Any]:
+                shazam = Shazam()
+                recognize = getattr(shazam, "recognize", None) or getattr(shazam, "recognize_song")
+                return await recognize(str(audio_path))
+
+            data = asyncio.run(_run()) or {}
+        except ModuleNotFoundError:
+            return SongIdResult(provider=self.name, matched=False, note="shazamio not installed")
+        except Exception as exc:  # network/format/API — stay graceful and offline-safe
+            return SongIdResult(provider=self.name, matched=False, note=f"offline or unrecognized: {exc}".strip())
+
+        track = (data.get("track") or {}) if isinstance(data, dict) else {}
+        if not track:
+            return SongIdResult(provider=self.name, matched=False, note="no match")
+        isrc = track.get("isrc")
+        if not isrc:
+            for section in track.get("sections", []) or []:
+                for meta in section.get("metadata", []) or []:
+                    if str(meta.get("title", "")).upper() == "ISRC":
+                        isrc = meta.get("text")
+        return SongIdResult(
+            provider=self.name,
+            matched=True,
+            title=track.get("title"),
+            artist=track.get("subtitle"),
+            isrc=isrc,
+            track_id=str(track.get("key")) if track.get("key") is not None else None,
+        )
+
+
+def _file_hash(audio_path: Path) -> str:
+    return sha256(Path(audio_path).read_bytes()).hexdigest()
+
+
+def identify_song(
+    audio_path: str | Path,
+    *,
+    enabled: bool | None = None,
+    provider: SongIdProvider | None = None,
+    cache: dict[str, SongIdResult] | None = None,
+) -> dict[str, Any]:
+    """Identify the track at ``audio_path`` and return an ``extensions.songid`` dict.
+
+    Off by default. Caches by audio content hash so the same fragment is never re-queried.
+    Never raises for identification failures — returns ``matched=False`` instead.
+    """
+    if not songid_enabled(enabled):
+        return SongIdResult(provider="description", matched=False, note="song id off").to_extension()
+
+    provider = provider or ShazamIOProvider()
+    path = Path(audio_path).expanduser()
+    try:
+        key = _file_hash(path)
+    except OSError as exc:
+        return SongIdResult(provider=provider.name, matched=False, note=f"unreadable audio: {exc}").to_extension()
+
+    if cache is not None and key in cache:
+        return cache[key].to_extension()
+
+    result = provider.identify(path)
+    if cache is not None:
+        cache[key] = result
+    return result.to_extension()
+
+
+def enrich_akousma(
+    record: dict[str, Any],
+    audio_path: str | Path,
+    *,
+    enabled: bool | None = None,
+    provider: SongIdProvider | None = None,
+    cache: dict[str, SongIdResult] | None = None,
+) -> dict[str, Any]:
+    """Attach a song-id result to ``record['extensions']['songid']`` and return the record."""
+    record.setdefault("extensions", {})["songid"] = identify_song(
+        audio_path, enabled=enabled, provider=provider, cache=cache
+    )
+    return record

pyproject.toml

@@ -30,6 +30,9 @@ dev = [
   "httpx>=0.27",
   "pytest>=8.2"
 ]
+songid = [
+  "shazamio>=0.4"
+]
 
 [project.scripts]
 oida = "oida.cli:main"

tests/test_songid.py

@@ -0,0 +1,77 @@
+"""Song identification: opt-in, offline-safe, cached (mocked provider — no network)."""
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from oida import songid
+
+
+class FakeProvider:
+    def __init__(self, result: songid.SongIdResult):
+        self.name = "fake"
+        self.result = result
+        self.calls = 0
+
+    def identify(self, audio_path: Path) -> songid.SongIdResult:
+        self.calls += 1
+        return self.result
+
+
+class TestSongId(unittest.TestCase):
+    def setUp(self):
+        self.f = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
+        self.f.write(b"RIFF....fake-audio-bytes")
+        self.f.close()
+        self.path = self.f.name
+        os.environ.pop("OIDA_SONGID", None)
+
+    def tearDown(self):
+        os.unlink(self.path)
+        os.environ.pop("OIDA_SONGID", None)
+
+    def test_off_by_default(self):
+        ext = songid.identify_song(self.path)
+        self.assertFalse(ext["matched"])
+        self.assertEqual(ext["provider"], "description")
+
+    def test_enabled_match_shape(self):
+        fake = FakeProvider(songid.SongIdResult(
+            provider="fake", matched=True, title="Windowlicker", artist="Aphex Twin", isrc="GBAAA0000001"
+        ))
+        ext = songid.identify_song(self.path, enabled=True, provider=fake)
+        self.assertTrue(ext["matched"])
+        self.assertEqual(ext["title"], "Windowlicker")
+        self.assertEqual(ext["artist"], "Aphex Twin")
+        self.assertEqual(ext["isrc"], "GBAAA0000001")
+        self.assertIn("checked_at", ext)
+
+    def test_offline_degrades_to_no_match(self):
+        fake = FakeProvider(songid.SongIdResult(provider="fake", matched=False, note="offline"))
+        ext = songid.identify_song(self.path, enabled=True, provider=fake)
+        self.assertFalse(ext["matched"])
+        self.assertEqual(ext["note"], "offline")
+
+    def test_cache_by_audio_hash(self):
+        fake = FakeProvider(songid.SongIdResult(provider="fake", matched=True, title="T"))
+        cache: dict = {}
+        songid.identify_song(self.path, enabled=True, provider=fake, cache=cache)
+        songid.identify_song(self.path, enabled=True, provider=fake, cache=cache)
+        self.assertEqual(fake.calls, 1)  # second call served from cache
+
+    def test_enrich_akousma_places_result_in_extensions(self):
+        fake = FakeProvider(songid.SongIdResult(provider="fake", matched=True, title="T", artist="A"))
+        record = {"extensions": {}}
+        songid.enrich_akousma(record, self.path, enabled=True, provider=fake)
+        self.assertTrue(record["extensions"]["songid"]["matched"])
+        self.assertEqual(record["extensions"]["songid"]["title"], "T")
+
+    def test_env_toggle(self):
+        fake = FakeProvider(songid.SongIdResult(provider="fake", matched=True, title="T"))
+        self.assertFalse(songid.identify_song(self.path, provider=fake)["matched"])  # env unset -> off
+        os.environ["OIDA_SONGID"] = "1"
+        self.assertTrue(songid.identify_song(self.path, provider=fake)["matched"])   # env on
+
+
+if __name__ == "__main__":
+    unittest.main()