123 lines
5.5 KiB
Python
123 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Self-check for the phash/dedup/reconcile additions. Runnable, no framework:
|
|
python test_dedup.py
|
|
Creates synthetic images in a temp dir + a temp DB, asserts behaviour, cleans up.
|
|
"""
|
|
import io
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from PIL import Image
|
|
|
|
import photo_analyzer as pa
|
|
|
|
|
|
def _make_photo(path: Path, seed: int, size=(800, 600)):
|
|
"""Deterministic, structured (not pure-noise) image so phash is stable."""
|
|
rng = np.random.default_rng(seed)
|
|
# low-frequency structure: a few coloured blocks + gradient — survives resize
|
|
base = np.zeros((size[1], size[0], 3), dtype=np.uint8)
|
|
for _ in range(6):
|
|
x0, y0 = rng.integers(0, size[0] - 100), rng.integers(0, size[1] - 100)
|
|
col = rng.integers(0, 256, 3)
|
|
base[y0:y0 + 150, x0:x0 + 150] = col
|
|
grad = np.linspace(0, 120, size[0], dtype=np.uint8)
|
|
base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
|
|
Image.fromarray(base).save(path, quality=95)
|
|
|
|
|
|
def main():
|
|
tmp = Path(tempfile.mkdtemp(prefix="dedup_test_"))
|
|
db_path = tmp / "test.db"
|
|
try:
|
|
# ── synthetic library ────────────────────────────────────────────────
|
|
orig = tmp / "A.jpg"
|
|
_make_photo(orig, seed=1)
|
|
|
|
# exact byte copy (backup-of-backup)
|
|
exact = tmp / "A_backup.jpg"
|
|
exact.write_bytes(orig.read_bytes())
|
|
|
|
# near-dupe: same picture, resized + recompressed (different phone)
|
|
near = tmp / "A_phone.jpg"
|
|
with Image.open(orig) as im:
|
|
im.resize((400, 300)).save(near, quality=60)
|
|
|
|
# a genuinely different photo
|
|
other = tmp / "B.jpg"
|
|
_make_photo(other, seed=999)
|
|
|
|
# ── phash: dupes close, different far ────────────────────────────────
|
|
def dist(p1, p2):
|
|
h1, h2 = int(pa._phash_image(p1), 16), int(pa._phash_image(p2), 16)
|
|
return bin(h1 ^ h2).count("1")
|
|
|
|
d_exact = dist(orig, exact)
|
|
d_near = dist(orig, near)
|
|
d_other = dist(orig, other)
|
|
print(f"phash Hamming: exact={d_exact} near={d_near} other={d_other}")
|
|
assert d_exact == 0, f"exact copy should have phash distance 0, got {d_exact}"
|
|
assert d_near <= pa.PHASH_THRESHOLD, f"near-dupe {d_near} > threshold {pa.PHASH_THRESHOLD}"
|
|
assert d_other > pa.PHASH_THRESHOLD, f"different photo {d_other} <= threshold — false match"
|
|
|
|
# ── sha1: identical bytes match, different bytes don't ───────────────
|
|
assert pa._sha1_file(orig) == pa._sha1_file(exact), "byte-identical files must share sha1"
|
|
assert pa._sha1_file(orig) != pa._sha1_file(near), "recompressed file must differ in sha1"
|
|
|
|
# ── DB: register, hash, cluster, mark ────────────────────────────────
|
|
conn = pa.get_db(str(db_path))
|
|
photos = [orig, exact, near, other]
|
|
for p in photos:
|
|
pa.upsert_pending(conn, str(p))
|
|
n = pa.ensure_hashes(conn, photos)
|
|
assert n == 4, f"expected 4 hashed, got {n}"
|
|
|
|
# all four should carry phash + sha1 now
|
|
rows = conn.execute("SELECT path, phash, file_sha1 FROM photos").fetchall()
|
|
assert all(r["phash"] and r["file_sha1"] for r in rows), "some hashes missing"
|
|
|
|
clusters = pa.cluster_duplicates(conn, pa.PHASH_THRESHOLD)
|
|
# exactly one cluster: {orig, exact, near}; 'other' stands alone
|
|
assert len(clusters) == 1, f"expected 1 cluster, got {len(clusters)}"
|
|
cluster_paths = {it["path"] for it in clusters[0]}
|
|
assert cluster_paths == {str(orig), str(exact), str(near)}, cluster_paths
|
|
# canonical = largest file (orig, saved at q95 full-res)
|
|
assert clusters[0][0]["path"] == str(orig), "canonical should be the largest file"
|
|
|
|
marked, ncl = pa.mark_duplicates(conn, pa.PHASH_THRESHOLD)
|
|
assert (marked, ncl) == (2, 1), f"expected (2 marked, 1 cluster), got ({marked}, {ncl})"
|
|
|
|
dupes = conn.execute(
|
|
"SELECT path, dup_of FROM photos WHERE status='duplicate'"
|
|
).fetchall()
|
|
assert {r["path"] for r in dupes} == {str(exact), str(near)}
|
|
assert all(r["dup_of"] == str(orig) for r in dupes), "dup_of must point at canonical"
|
|
|
|
# get_pending must exclude duplicates
|
|
pending = pa.get_pending(conn, reanalyze=True)
|
|
assert str(exact) not in pending and str(near) not in pending, "dupes leaked into pending"
|
|
assert str(orig) in pending and str(other) in pending
|
|
|
|
# ── reconcile_moved: rename a file, path follows by sha1 ─────────────
|
|
moved_dst = tmp / "B_renamed.jpg"
|
|
os.rename(other, moved_dst) # 'other' now missing at old path
|
|
n_moved = pa.reconcile_moved(conn, tmp)
|
|
assert n_moved == 1, f"expected 1 reconciled, got {n_moved}"
|
|
row = conn.execute("SELECT path FROM photos WHERE path=?", (str(moved_dst),)).fetchone()
|
|
assert row is not None, "moved file's row was not repathed"
|
|
gone = conn.execute("SELECT 1 FROM photos WHERE path=?", (str(other),)).fetchone()
|
|
assert gone is None, "old path should no longer exist in DB"
|
|
|
|
conn.close()
|
|
print("ALL DEDUP/RECONCILE CHECKS PASSED")
|
|
finally:
|
|
import shutil
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|