Build a three-stage video dedup worker: SHA-256, perceptual hash, then FFmpeg's MPEG-7 signature

sudo apt-get install -y ffmpeg     # 7.x or 8.x; the signature filter has shipped since 2017
python3 -m venv .venv && source .venv/bin/activate
pip install videohash2            # maintained fork of videohash; pin the current release in requirements.txt
ffmpeg -hide_banner -filters | grep signature

 ... signature          N->V       Calculate the MPEG-7 video signature

# scripts/make_fixtures.sh
set -e
mkdir -p fixtures && cd fixtures

# 20-second source
ffmpeg -y -f lavfi -i "testsrc2=duration=20:size=640x360:rate=25,format=yuv420p" \
       -f lavfi -i "sine=frequency=440:duration=20" \
       -c:v libx264 -crf 20 -c:a aac -shortest source.mp4

# A: re-encode at lower quality and resolution
ffmpeg -y -i source.mp4 -vf scale=320:180 -c:v libx264 -crf 32 -c:a aac dup_reencode.mp4

# B: watermark in the corner
ffmpeg -y -i source.mp4 -vf "drawbox=x=10:y=10:w=80:h=30:color=white@0.8:t=fill" \
       -c:v libx264 -crf 23 -c:a copy dup_watermark.mp4

# C: 6-second excerpt starting at 0:07 (containment case)
ffmpeg -y -ss 7 -i source.mp4 -t 6 -c:v libx264 -crf 23 -c:a aac excerpt.mp4

# D: unrelated video
ffmpeg -y -f lavfi -i "mandelbrot=size=640x360:rate=25" -t 20 \
       -c:v libx264 -crf 23 unrelated.mp4

# dedup/stage1.py
import hashlib
from pathlib import Path

def sha256(path: Path, chunk: int = 1 << 20) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        while blk := f.read(chunk):
            h.update(blk)
    return h.hexdigest()

# dedup/stage2.py
from pathlib import Path
from videohash2 import VideoHash

def phash(path: Path) -> int:
    vh = VideoHash(path=str(path))
    return int(vh.hash_hex, 16)          # 64-bit int

def hamming(a: int, b: int) -> int:
    return (a ^ b).bit_count()

>>> from pathlib import Path
>>> from dedup.stage2 import phash, hamming
>>> src = phash(Path("fixtures/source.mp4"))
>>> for name in ["dup_reencode", "dup_watermark", "excerpt", "unrelated"]:
...     print(name, hamming(src, phash(Path(f"fixtures/{name}.mp4"))))

# one-time per asset: write a binary signature next to it
ffmpeg -hide_banner -i fixtures/source.mp4 \
  -vf "signature=format=binary:filename=fixtures/source.sig" -f null -

ffmpeg -hide_banner -i fixtures/source.mp4 -i fixtures/excerpt.mp4 \
  -filter_complex "signature=nb_inputs=2:detectmode=full" -f null - 2>&1 | grep -i match

[Parsed_signature_0 @ 0x...] matching of video 0 at 7.000000 and 1 at 0.000000, 150 frames matching

# dedup/stage3.py
import re, subprocess
from pathlib import Path

MATCH = re.compile(r"matching of video 0 at ([\d.]+) and 1 at ([\d.]+), (\d+) frames matching")
WHOLE = re.compile(r"whole video matching")

def signature_compare(a: Path, b: Path) -> dict | None:
    cmd = ["ffmpeg", "-hide_banner", "-nostats", "-i", str(a), "-i", str(b),
           "-filter_complex", "signature=nb_inputs=2:detectmode=full:th_di=50",
           "-f", "null", "-"]
    out = subprocess.run(cmd, capture_output=True, text=True).stderr
    m = MATCH.search(out)
    if not m:
        return None
    return {
        "offset_a": float(m.group(1)),
        "offset_b": float(m.group(2)),
        "frames": int(m.group(3)),
        "whole": bool(WHOLE.search(out)),
    }

# dedup/verdict.py
def coverage(frames: int, frames_a: int, frames_b: int) -> float:
    return frames / min(frames_a, frames_b)

# dedup/worker.py
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from .stage1 import sha256
from .stage2 import phash, hamming
from .stage3 import signature_compare
from .verdict import coverage

PHASH_MAX_DISTANCE = 6
MIN_COVERAGE = 0.4

@dataclass
class Verdict:
    kind: str            # exact | near | contains | new
    match_id: int | None = None
    detail: dict | None = None

def check(path: Path, db: sqlite3.Connection, frames_of) -> Verdict:
    digest = sha256(path)
    row = db.execute("SELECT id FROM assets WHERE sha256=?", (digest,)).fetchone()
    if row:
        return Verdict("exact", row[0])

    h = phash(path)
    candidates = []
    for aid, other_hash, other_path in db.execute("SELECT id, phash, path FROM assets"):
        d = hamming(h, int(other_hash))
        if d <= PHASH_MAX_DISTANCE:
            return Verdict("near", aid, {"hamming": d})
        if d <= 20:                          # loose band: worth a stage-3 look
            candidates.append((aid, Path(other_path)))

    for aid, other in candidates:
        m = signature_compare(other, path)
        if m and coverage(m["frames"], frames_of(other), frames_of(path)) >= MIN_COVERAGE:
            return Verdict("contains", aid, m)

    db.execute("INSERT INTO assets(path, sha256, phash) VALUES (?,?,?)",
               (str(path), digest, str(h)))
    db.commit()
    return Verdict("new")

-- dedup/schema.sql
CREATE TABLE IF NOT EXISTS assets (
  id     INTEGER PRIMARY KEY,
  path   TEXT NOT NULL,
  sha256 TEXT NOT NULL UNIQUE,
  phash  TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_phash ON assets(phash);

python -m dedup.cli fixtures/source.mp4 fixtures/dup_reencode.mp4 \
                    fixtures/dup_watermark.mp4 fixtures/excerpt.mp4 fixtures/unrelated.mp4
# source.mp4        new
# dup_reencode.mp4  near      (hamming=<small>)
# dup_watermark.mp4 near      (hamming=<small>)
# excerpt.mp4       contains  (offset_a≈7.0, frames≈150, coverage≈1.00)
# unrelated.mp4     new
# exact distances and frame counts depend on your FFmpeg build; the verdicts should not