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
This commit is contained in:
84
audio_fingerprint.py
Normal file
84
audio_fingerprint.py
Normal file
@@ -0,0 +1,84 @@
|
||||
"""
|
||||
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())
|
||||
Reference in New Issue
Block a user