File size: 2,564 Bytes
23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | """
Duplicate detection forensics provider.
Delegates pHash / dHash / SHA-256 / Hamming distance to cores.vision —
no duplicated hashing logic. Maintains an in-memory registry of seen
hashes for duplicate detection across jobs.
"""
from __future__ import annotations
import numpy as np
from config.settings import Settings, settings as _default_settings
from cores.vision import phash, dhash, sha256_bytes, hamming_distance
from pipeline.feature_extraction import PipelineOutput
from providers.base import BaseProvider, ProviderCapability
class DuplicateDetectorProvider(BaseProvider):
name = "duplicate_detector"
capability = ProviderCapability.FORENSICS
DUPLICATE_THRESHOLD = 5 # bits of 64
def __init__(self, settings: Settings | None = None) -> None:
super().__init__(settings=settings or _default_settings)
# In-memory hash registry: phash -> source label (sha256[:12])
self._seen: dict[str, str] = {}
def is_available(self) -> bool:
return True
def _run(self, pipeline_output: PipelineOutput) -> tuple[dict, dict]:
img: np.ndarray = pipeline_output.image
p = phash(img)
d = dhash(img)
sha = sha256_bytes(pipeline_output.original_bytes) if pipeline_output.original_bytes else None
# Check for duplicates against in-memory set
duplicate_of = None
is_duplicate = False
similarity = 1.0
for stored_hash, label in self._seen.items():
dist = hamming_distance(p, stored_hash)
similarity = 1.0 - (dist / 64.0)
if dist <= self.DUPLICATE_THRESHOLD:
duplicate_of = label
is_duplicate = True
break
# Register this image's hash
if sha:
self._seen[p] = sha[:12]
raw = {
"phash": p,
"dhash": d,
"sha256": sha,
"is_duplicate": is_duplicate,
"duplicate_of": duplicate_of,
"similarity_score": round(similarity, 4),
"registered_hashes": len(self._seen),
}
normalized = {
"integrity_score": None,
"is_duplicate": is_duplicate,
"duplicate_of": duplicate_of,
"similarity_score": round(similarity, 4),
"manipulation_indicators": [],
"details": {
"phash": p,
"dhash": d,
"sha256": sha,
"registered_hashes": len(self._seen),
},
}
return raw, normalized
|