Files
atm-curs-dl/audio_fingerprint.py
Claude Agent f0b8f28fa9 Detectare duplicate: Q&A 18.08.2026 este acelasi video cu 11.08.2026
audio_fingerprint.py - amprenta pe anvelopa de energie, fara transcriere.
md5 nu prinde duplicatele (urcari encodate separat, durate la 0.08s distanta).
Validat: 1.000 pe perechea duplicat, 0.038 pe sesiuni diferite.
check_duplicates.py - acelasi lucru pe transcrieri, cu --selftest.
Scanat tot modulul Q&A: un singur duplicat.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MtSTyTmt6AbCL9j5ajEDm1
2026-09-12 17:40:18 +00:00

85 lines
2.9 KiB
Python

"""
Detectează sesiuni duplicate (același videoclip urcat sub două date) direct
din audio, fără transcriere.
De ce nu md5: cele două urcări sunt encodate separat, deci bytes diferiți și
durate care diferă cu zecimi de secundă. De ce nu whisper: ar însemna să
transcrii tot ca să afli că n-avea rost.
Amprenta = energia (RMS) pe ferestre de o secundă, normalizată. Două urcări
ale aceleiași înregistrări au aceeași curbă de energie, chiar dacă sunt
decalate cu o fracțiune de secundă sau encodate diferit.
"""
import subprocess
import sys
from itertools import combinations
from pathlib import Path
import numpy as np
SR = 1000 # decodăm la 1 kHz: energia vorbirii se vede, e de 16x mai ieftin
WIN = SR # o fereastră = o secundă
MAX_SHIFT = 15 # secunde de decalaj tolerate între cele două urcări
def envelope(path: Path) -> np.ndarray:
raw = subprocess.run(
["ffmpeg", "-v", "error", "-i", str(path), "-f", "s16le", "-ar", str(SR), "-ac", "1", "-"],
capture_output=True, check=True,
).stdout
x = np.frombuffer(raw, dtype=np.int16).astype(np.float32)
n = len(x) // WIN
rms = np.sqrt((x[: n * WIN].reshape(n, WIN) ** 2).mean(axis=1))
return rms / (rms.std() or 1.0)
def similarity(a: np.ndarray, b: np.ndarray) -> float:
"""Cea mai bună corelație normalizată peste toate decalajele plauzibile."""
if min(len(a), len(b)) < 60:
return 0.0
best = 0.0
for shift in range(-MAX_SHIFT, MAX_SHIFT + 1):
x, y = (a[shift:], b) if shift >= 0 else (a, b[-shift:])
n = min(len(x), len(y))
if n < 60:
continue
x, y = x[:n] - x[:n].mean(), y[:n] - y[:n].mean()
denom = np.linalg.norm(x) * np.linalg.norm(y)
if denom:
best = max(best, float(x @ y / denom))
return best
def main():
args = [a for a in sys.argv[1:] if not a.startswith("--")]
expanded = set()
for a in args:
pa = Path(a)
expanded |= {pa} if pa.is_file() else set(Path(pa.anchor or ".").glob(str(pa.relative_to(pa.anchor or "."))))
files = sorted(expanded or Path("audio").glob("*/*.mp3"))
if len(files) < 2:
print("Nimic de comparat.")
return 0
print(f"Amprentez {len(files)} fișiere...")
fps = {f: envelope(f) for f in files}
found = False
for x, y in combinations(files, 2):
# durate foarte diferite => sigur nu e aceeași înregistrare
if abs(len(fps[x]) - len(fps[y])) > MAX_SHIFT:
continue
s = similarity(fps[x], fps[y])
if s > 0.90:
found = True
print(f"DUPLICAT ({s:.3f}): {x.stem} == {y.stem}")
elif s > 0.70:
print(f"suspect ({s:.3f}): {x.stem} vs {y.stem} — verifică manual")
if not found:
print("Niciun duplicat.")
return 0
if __name__ == "__main__":
sys.exit(main())