providers/forensics/duplicate_detector.py

"""
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