feat: add acoustic dedupe pipeline and docker runtime
This commit is contained in:
@@ -0,0 +1,38 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
MAX_SECONDS = 120
|
||||
|
||||
|
||||
def chromaprint(path: Path) -> str | None:
|
||||
try:
|
||||
p = subprocess.run(
|
||||
["fpcalc", "-length", str(MAX_SECONDS), "-json", str(path)],
|
||||
capture_output=True, text=True, timeout=180, check=True,
|
||||
)
|
||||
return json.loads(p.stdout).get("fingerprint")
|
||||
except (OSError, subprocess.SubprocessError, ValueError, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
def similarity(a: str, b: str) -> float:
|
||||
"""Conservative normalized positional similarity for equal-length prefixes."""
|
||||
if not a or not b:
|
||||
return 0.0
|
||||
n = min(len(a), len(b))
|
||||
if not n:
|
||||
return 0.0
|
||||
# Chromaprint JSON fingerprints are comma-separated integer features.
|
||||
try:
|
||||
aa, bb = [int(x) for x in a.split(",")], [int(x) for x in b.split(",")]
|
||||
except ValueError:
|
||||
return 0.0
|
||||
n = min(len(aa), len(bb))
|
||||
if not n:
|
||||
return 0.0
|
||||
# Feature-wise equality is intentionally conservative. A future backend
|
||||
# can replace this with a library-specific fingerprint distance.
|
||||
return sum(x == y for x, y in zip(aa[:n], bb[:n])) / n
|
||||
Reference in New Issue
Block a user