110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
"""Characterize hashing + duplicate ledger (donor: _sha1_file, _phash_image,
|
|
ensure_hashes, cluster_duplicates, mark_duplicates, reconcile_moved).
|
|
|
|
Complements the pre-existing root-level self-check test_dedup.py.
|
|
"""
|
|
import hashlib
|
|
|
|
import photo_analyzer as pa
|
|
from conftest import write_fixture
|
|
|
|
# Captured goldens (Pillow/scipy in current env; a change here means the
|
|
# decoder/algorithm changed — exactly what characterization should catch).
|
|
GOLDEN_PHASH = {
|
|
"fx-blocks-01": "851e7ae08fa4347b",
|
|
"fx-blocks-02": "84333bcce3338ce9",
|
|
"fx-blocks-03": "841f32f069cfcd38",
|
|
"fx-blocks-04": "864c3c7073b761b9",
|
|
}
|
|
|
|
|
|
def test_phash_goldens(tmp_path):
|
|
for fid, expected in GOLDEN_PHASH.items():
|
|
p = write_fixture(fid, tmp_path / f"{fid}.jpg")
|
|
assert pa._phash_image(p) == expected, fid
|
|
|
|
|
|
def test_phash_survives_resize_and_recompress(tmp_path, img):
|
|
from PIL import Image
|
|
small = tmp_path / "small.jpg"
|
|
with Image.open(img) as im:
|
|
im.resize((400, 300)).save(small, quality=70)
|
|
a = int(pa._phash_image(img), 16)
|
|
b = int(pa._phash_image(small), 16)
|
|
assert bin(a ^ b).count("1") <= pa.PHASH_THRESHOLD
|
|
|
|
|
|
def test_phash_format_16_hex_chars(img):
|
|
ph = pa._phash_image(img)
|
|
assert len(ph) == 16
|
|
int(ph, 16)
|
|
|
|
|
|
def test_sha1_matches_hashlib(img):
|
|
assert pa._sha1_file(img) == hashlib.sha1(img.read_bytes()).hexdigest()
|
|
|
|
|
|
def test_sha1_unreadable_returns_none(tmp_path):
|
|
assert pa._sha1_file(tmp_path / "missing.jpg") is None
|
|
|
|
|
|
def test_ensure_hashes_skips_hashed_and_is_resume_safe(db, tmp_path):
|
|
p1 = write_fixture("fx-blocks-01", tmp_path / "a.jpg")
|
|
p2 = write_fixture("fx-blocks-02", tmp_path / "b.jpg")
|
|
for p in (p1, p2):
|
|
pa.upsert_pending(db, str(p))
|
|
assert pa.ensure_hashes(db, [p1, p2]) == 2
|
|
# second run: nothing to do
|
|
assert pa.ensure_hashes(db, [p1, p2]) == 0
|
|
row = db.execute("SELECT phash, file_sha1 FROM photos WHERE path=?",
|
|
(str(p1),)).fetchone()
|
|
assert row["phash"] == GOLDEN_PHASH["fx-blocks-01"]
|
|
assert row["file_sha1"] == pa._sha1_file(p1)
|
|
|
|
|
|
def test_cluster_and_mark_duplicates_largest_is_canonical(db, tmp_path):
|
|
from PIL import Image
|
|
big = write_fixture("fx-blocks-01", tmp_path / "big.jpg", quality=95)
|
|
small = tmp_path / "small.jpg"
|
|
with Image.open(big) as im:
|
|
im.resize((400, 300)).save(small, quality=60)
|
|
other = write_fixture("fx-blocks-02", tmp_path / "other.jpg")
|
|
for p in (big, small, other):
|
|
pa.upsert_pending(db, str(p))
|
|
pa.ensure_hashes(db, [big, small, other])
|
|
|
|
clusters = pa.cluster_duplicates(db, pa.PHASH_THRESHOLD)
|
|
assert len(clusters) == 1
|
|
assert clusters[0][0]["path"] == str(big), "canonical = largest file"
|
|
assert {c["path"] for c in clusters[0]} == {str(big), str(small)}
|
|
|
|
marked, n = pa.mark_duplicates(db, pa.PHASH_THRESHOLD)
|
|
assert (marked, n) == (1, 1)
|
|
row = db.execute("SELECT status, dup_of FROM photos WHERE path=?",
|
|
(str(small),)).fetchone()
|
|
assert row["status"] == "duplicate"
|
|
assert row["dup_of"] == str(big)
|
|
# duplicates never re-enter the pending queue
|
|
assert str(small) not in pa.get_pending(db, reanalyze=False)
|
|
assert str(small) not in pa.get_pending(db, reanalyze=True)
|
|
|
|
|
|
def test_reconcile_moved_preserves_row_by_sha1(db, tmp_path):
|
|
lib = tmp_path / "lib"
|
|
old = write_fixture("fx-blocks-03", lib / "album" / "pic.jpg")
|
|
pa.upsert_pending(db, str(old))
|
|
pa.ensure_hashes(db, [old])
|
|
db.execute("UPDATE photos SET status='analyzed', description='kept' WHERE path=?",
|
|
(str(old),))
|
|
db.commit()
|
|
|
|
new = lib / "renamed" / "pic_new.jpg"
|
|
new.parent.mkdir(parents=True)
|
|
old.rename(new)
|
|
|
|
assert pa.reconcile_moved(db, lib) == 1
|
|
row = db.execute("SELECT path, status, description FROM photos").fetchone()
|
|
assert row["path"] == str(new)
|
|
assert row["status"] == "analyzed"
|
|
assert row["description"] == "kept", "analysis survives the move"
|