123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
"""Duplikate unter den TMDB-Backdrops finden.
|
|
|
|
TMDB kennzeichnet nicht, welches Bild ein Szenenfoto und welches ein Crop
|
|
desselben Frames ist. Also vergleichen wir die Bilder selbst.
|
|
|
|
Zwei Metriken, weil eine allein nicht reicht:
|
|
- dHash über eine 9x8-Graustufenminiatur erkennt gleiche Bilder zuverlässig,
|
|
scheitert aber an Crops. Gemessen: ein echter Crop lag bei Distanz 14, zwei
|
|
völlig verschiedene Bilder bei 13.
|
|
- Das Farbhistogramm trennt genau diese Fälle. Der Crop lag bei 0.12, die
|
|
verschiedenen Bilder bei über 0.27.
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
import logging
|
|
|
|
import httpx
|
|
from PIL import Image
|
|
|
|
from .config import IMAGE_BASE
|
|
|
|
log = logging.getLogger("imagehash")
|
|
|
|
HASH_SIZE = 8
|
|
# Bis hierher entscheidet der dHash allein
|
|
STRICT_DISTANCE = 10
|
|
# Darüber hinaus nur zusammen mit sehr ähnlicher Farbverteilung
|
|
LOOSE_DISTANCE = 18
|
|
MAX_HIST_DISTANCE = 0.15
|
|
|
|
HIST_BINS = 32
|
|
THUMB_SIZE = "w300"
|
|
CONCURRENCY = 12
|
|
|
|
|
|
def dhash(img):
|
|
"""64-Bit-Fingerabdruck: je Pixelpaar ein Bit, ob links heller ist."""
|
|
thumb = img.convert("L").resize((HASH_SIZE + 1, HASH_SIZE), Image.LANCZOS)
|
|
pixels = list(thumb.getdata())
|
|
|
|
bits = 0
|
|
for row in range(HASH_SIZE):
|
|
offset = row * (HASH_SIZE + 1)
|
|
for col in range(HASH_SIZE):
|
|
bits <<= 1
|
|
if pixels[offset + col] > pixels[offset + col + 1]:
|
|
bits |= 1
|
|
return bits
|
|
|
|
|
|
def histogram(img):
|
|
"""96 Bytes Farbverteilung, unabhängig von Bildausschnitt und Größe."""
|
|
small = img.convert("RGB").resize((64, 64), Image.LANCZOS)
|
|
raw = small.histogram()
|
|
total = 64 * 64
|
|
step = 256 // HIST_BINS
|
|
|
|
packed = bytearray()
|
|
for channel in range(3):
|
|
base = channel * 256
|
|
for start in range(0, 256, step):
|
|
share = sum(raw[base + start : base + start + step]) / total
|
|
packed.append(min(255, round(share * 255)))
|
|
return bytes(packed)
|
|
|
|
|
|
def fingerprint(data):
|
|
with Image.open(io.BytesIO(data)) as img:
|
|
return dhash(img), histogram(img)
|
|
|
|
|
|
def distance(a, b):
|
|
return bin(a ^ b).count("1")
|
|
|
|
|
|
def hist_distance(a, b):
|
|
"""0 = gleiche Farbverteilung, 1 = keine Überschneidung."""
|
|
overlap = sum(min(x, y) for x, y in zip(a, b)) / 255
|
|
return max(0.0, 1 - overlap / 3)
|
|
|
|
|
|
def is_duplicate(a, b):
|
|
gap = distance(a[0], b[0])
|
|
if gap <= STRICT_DISTANCE:
|
|
return True
|
|
return gap <= LOOSE_DISTANCE and hist_distance(a[1], b[1]) <= MAX_HIST_DISTANCE
|
|
|
|
|
|
async def fetch_fingerprint(http, path, semaphore):
|
|
async with semaphore:
|
|
try:
|
|
resp = await http.get(f"{IMAGE_BASE}/{THUMB_SIZE}{path}")
|
|
resp.raise_for_status()
|
|
return path, fingerprint(resp.content)
|
|
except (httpx.HTTPError, OSError):
|
|
log.warning("Bild nicht lesbar: %s", path)
|
|
return path, None
|
|
|
|
|
|
async def fingerprint_many(paths):
|
|
semaphore = asyncio.Semaphore(CONCURRENCY)
|
|
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as http:
|
|
pairs = await asyncio.gather(
|
|
*(fetch_fingerprint(http, p, semaphore) for p in paths)
|
|
)
|
|
return {path: value for path, value in pairs if value is not None}
|
|
|
|
|
|
def dedupe(paths, prints):
|
|
"""Reihenfolge bleibt, jedes weitere Bild muss sich von allen bisherigen
|
|
unterscheiden. Bilder ohne Fingerabdruck fallen raus."""
|
|
kept, kept_prints = [], []
|
|
for path in paths:
|
|
current = prints.get(path)
|
|
if current is None:
|
|
continue
|
|
if any(is_duplicate(current, other) for other in kept_prints):
|
|
continue
|
|
kept.append(path)
|
|
kept_prints.append(current)
|
|
return kept
|