Complete workflow for detecting duplicate and near-duplicate images using MD5 hashes and perceptual hashing (dHash/pHash). Use when implementing duplicate detection features.
This skill provides the complete workflow for finding exact and near-duplicate images across platforms.
Fastest method - catches identical files even if renamed.
import hashlib
from pathlib import Path
def compute_md5(file_path: Path) -> str:
"""Compute MD5 hash of file."""
hash_md5 = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
def find_exact_duplicates(image_paths: list[Path]) -> dict[str, list[Path]]:
"""Group images by MD5 hash to find exact duplicates."""
hash_map = {}
for path in image_paths:
try:
md5 = compute_md5(path)
if md5 not in hash_map:
hash_map[md5] = []
hash_map[md5].append(path)
except Exception as e:
print(f"Error hashing {path}: {e}")
# Return only groups with duplicates
return {h: paths for h, paths in hash_map.items() if len(paths) > 1}
Use imagehash library for robust similarity detection.
from PIL import Image
import imagehash
def compute_perceptual_hashes(file_path: Path) -> dict:
"""Compute multiple perceptual hashes for robust comparison."""
try:
img = Image.open(file_path)
return {
'dhash': imagehash.dhash(img), # Difference hash (fast)
'phash': imagehash.phash(img), # Perceptual hash (robust)
'average_hash': imagehash.average_hash(img), # Average hash (simple)
'whash': imagehash.whash(img), # Wavelet hash (best quality)
}
except Exception as e:
print(f"Error computing hash for {file_path}: {e}")
return None
def compare_hashes(hash1, hash2, threshold=10) -> tuple[bool, int]:
"""
Compare two perceptual hashes.
Returns:
(is_similar, hamming_distance)
Threshold guidelines:
0: Identical
1-5: Very similar (likely duplicates)
6-10: Similar (possibly duplicates)
11-20: Somewhat similar
>20: Different images
"""
hamming_distance = hash1 - hash2
return hamming_distance <= threshold, hamming_distance
def find_near_duplicates(image_paths: list[Path], threshold=10) -> list[tuple]:
"""Find near-duplicate images using perceptual hashing."""
from collections import defaultdict
# Compute hashes for all images
hashes = {}
for path in image_paths:
hash_result = compute_perceptual_hashes(path)
if hash_result:
hashes[path] = hash_result['dhash'] # Use dhash by default
# Compare all pairs (can be optimized with spatial indexing)
duplicates = []
paths_list = list(hashes.keys())
for i in range(len(paths_list)):
for j in range(i + 1, len(paths_list)):
path1, path2 = paths_list[i], paths_list[j]
is_similar, distance = compare_hashes(
hashes[path1],
hashes[path2],
threshold
)
if is_similar:
duplicates.append({
'image1': path1,
'image2': path2,
'distance': distance,
'confidence': 1.0 - (distance / threshold) # 0-1 score
})
return duplicates
Quick filter before expensive hash computation.
from PIL import Image
def get_image_metadata(file_path: Path) -> dict:
"""Extract metadata for quick comparison."""
stat = file_path.stat()
metadata = {
'size': stat.st_size,
'modified': stat.st_mtime,
'name': file_path.name,
}
# Try to get image dimensions
try:
with Image.open(file_path) as img:
metadata['dimensions'] = img.size
metadata['format'] = img.format
except:
pass
return metadata
def filter_likely_duplicates(image_paths: list[Path]) -> list[list[Path]]:
"""Group images with identical size and dimensions (likely duplicates)."""
from collections import defaultdict
groups = defaultdict(list)
for path in image_paths:
meta = get_image_metadata(path)
# Group by size and dimensions
key = (meta['size'], meta.get('dimensions'))
groups[key].append(path)
# Return only groups with 2+ images
return [paths for paths in groups.values() if len(paths) > 1]
class DuplicateDetector:
def __init__(self, threshold=10):
self.threshold = threshold
self.exact_duplicates = {}
self.near_duplicates = []
def scan(self, image_paths: list[Path]):
"""Run complete duplicate detection workflow."""
print(f"Scanning {len(image_paths)} images...")
# Step 1: Quick metadata filtering
print("Step 1: Filtering by metadata...")
likely_groups = filter_likely_duplicates(image_paths)
candidates = [img for group in likely_groups for img in group]
print(f"Found {len(candidates)} candidates in {len(likely_groups)} groups")
# Step 2: Exact duplicates (MD5)
print("Step 2: Finding exact duplicates...")
self.exact_duplicates = find_exact_duplicates(candidates)
print(f"Found {len(self.exact_duplicates)} exact duplicate groups")
# Step 3: Near duplicates (perceptual hashing)
# Only scan images not already found as exact duplicates
remaining = [
p for p in candidates
if not any(p in group for group in self.exact_duplicates.values())
]
print(f"Step 3: Finding near duplicates ({len(remaining)} images)...")
self.near_duplicates = find_near_duplicates(remaining, self.threshold)
print(f"Found {len(self.near_duplicates)} near-duplicate pairs")
return {
'exact': self.exact_duplicates,
'near': self.near_duplicates
}
def generate_report(self, output_path: Path):
"""Generate detailed JSON report."""
import json
report = {
'scan_timestamp': datetime.now().isoformat(),
'total_exact_groups': len(self.exact_duplicates),
'total_near_pairs': len(self.near_duplicates),
'threshold': self.threshold,
'exact_duplicates': {
h: [str(p) for p in paths]
for h, paths in self.exact_duplicates.items()
},
'near_duplicates': [
{
'image1': str(d['image1']),
'image2': str(d['image2']),
'distance': d['distance'],
'confidence': d['confidence']
}
for d in self.near_duplicates
]
}
with open(output_path, 'w') as f:
json.dump(report, f, indent=2)
from concurrent.futures import ThreadPoolExecutor
import multiprocessing
def compute_hashes_parallel(image_paths: list[Path], max_workers=None):
"""Compute hashes in parallel for better performance."""
if max_workers is None:
max_workers = multiprocessing.cpu_count()
with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = executor.map(compute_perceptual_hashes, image_paths)
return {path: hash_result for path, hash_result in zip(image_paths, results) if hash_result}
from tqdm import tqdm
def scan_with_progress(image_paths: list[Path]):
"""Show progress bar during scanning."""
hashes = {}
for path in tqdm(image_paths, desc="Computing hashes"):
hash_result = compute_perceptual_hashes(path)
if hash_result:
hashes[path] = hash_result
return hashes
def get_google_drive_images(service, mime_type_filter='image/'):
"""List all images in Google Drive with MD5 checksums."""
images = []
page_token = None
while True:
query = f"mimeType contains '{mime_type_filter}'"
results = service.files().list(
q=query,
spaces='drive',
fields='nextPageToken, files(id, name, size, md5Checksum, mimeType)',
pageToken=page_token
).execute()
images.extend(results.get('files', []))
page_token = results.get('nextPageToken')
if not page_token:
break
return images
def find_drive_duplicates_by_md5(service):
"""Find exact duplicates in Google Drive using API-provided MD5."""
from collections import defaultdict
images = get_google_drive_images(service)
hash_map = defaultdict(list)
for img in images:
if 'md5Checksum' in img:
hash_map[img['md5Checksum']].append(img)
return {h: files for h, files in hash_map.items() if len(files) > 1}
def create_test_duplicates(source_image: Path, output_dir: Path):
"""Create test images for validation."""
from PIL import Image
img = Image.open(source_image)
# Exact duplicate (same file)
shutil.copy(source_image, output_dir / "exact_copy.jpg")
# Near duplicate (slightly resized)
resized = img.resize((img.width - 10, img.height - 10))
resized.save(output_dir / "resized.jpg")
# Near duplicate (slightly compressed)
img.save(output_dir / "compressed.jpg", quality=85)
# Not a duplicate (rotated)
rotated = img.rotate(90)
rotated.save(output_dir / "rotated.jpg")
| Distance | Interpretation | Action |
|---|---|---|
| 0 | Identical | Exact duplicate |
| 1-5 | Nearly identical | High confidence duplicate |
| 6-10 | Very similar | Likely duplicate (review) |
| 11-15 | Similar | Possible duplicate (manual check) |
| 16-20 | Somewhat similar | Probably different |
| >20 | Different | Not duplicates |
imagehash library: https://github.com/JohannesBuchner/imagehash.github/copilot-instructions.md for full project context