confidence/engine.py

"""
Confidence engine — decomposes every score into weighted sub-scores
with a human-readable explanation.

Sub-scores:
  - source_reliability   (provider's track record)
  - cross_provider_consensus (how many providers agree)
  - detection_confidence (raw provider confidence)
  - latency_penalty      (slower = lower confidence)
"""

from __future__ import annotations

from typing import Any, List, Protocol

from models.reports import (
    ConfidenceScore,
    ConflictReport,
    FaceDetection,
    FaceMatch,
    ForensicsResult,
    ImageAnalysisResult,
)

WEIGHTS = {
    "source_reliability": 0.25,
    "cross_provider_consensus": 0.30,
    "detection_confidence": 0.35,
    "latency_penalty": 0.10,
}

# Default reliability per provider (can be overridden by metrics history)
DEFAULT_RELIABILITY = {
    "haar": 0.65,
    "dnn": 0.85,
    "mtcnn": 0.92,
    "retinaface": 0.95,
    "face_recognition": 0.90,
    "deepface": 0.88,
    "insightface": 0.93,
    "beautifulsoup": 0.75,
    "selenium": 0.70,
    "bing": 0.85,
    "duckduckgo": 0.65,
    "google_lens": 0.55,
    "serpapi": 0.90,
    "yandex": 0.55,
    "tineye": 0.88,
    # analysis / metadata / forensics
    "image_quality": 0.85,
    "image_properties": 0.85,
    "visual_features": 0.80,
    "exif": 0.95,
    "xmp": 0.90,
    "image_integrity": 0.90,
    "duplicate_detector": 0.85,
    "manipulation_analyzer": 0.75,
}


# Duck-typed protocols (confidence must not import from normalization)
class _BoxLike(Protocol):
    detector: str
    confidence: float


class _MatchLike(Protocol):
    query_face_index: int
    best_match: str | None
    distance: float
    recognizer: str


class _AnalysisLike(Protocol):
    provider: str
    quality_score: float | None


class _MetadataLike(Protocol):
    provider: str
    format: str | None


class _ForensicsLike(Protocol):
    provider: str
    integrity_score: float | None


class ConfidenceEngine:
    """Computes weighted confidence scores."""

    def __init__(self, reliability_overrides: dict | None = None) -> None:
        self._reliability = {**DEFAULT_RELIABILITY, **(reliability_overrides or {})}

    # ------------------------------------------------------------------ #
    # Detection confidence
    # ------------------------------------------------------------------ #
    def score_detection(
        self,
        nbox: _BoxLike,
        all_results: dict[str, Any],
    ) -> ConfidenceScore:
        provider = nbox.detector
        source_rel = self._reliability.get(provider, 0.7)

        num_detectors = sum(
            1 for r in all_results.values()
            if r.success and r.capability.value == "detection"
        )
        consensus = min(1.0, num_detectors / 3.0)

        det_conf = nbox.confidence

        provider_result = all_results.get(provider)
        latency = provider_result.elapsed_ms if provider_result else 500.0
        latency_score = max(0.5, 1.0 - (latency / 4000.0))

        components = {
            "source_reliability": round(source_rel, 4),
            "cross_provider_consensus": round(consensus, 4),
            "detection_confidence": round(det_conf, 4),
            "latency_penalty": round(latency_score, 4),
        }
        overall = sum(components[k] * WEIGHTS[k] for k in WEIGHTS)
        explanation = (
            f"{provider} detected a face with raw confidence {det_conf:.2f}. "
            f"{num_detectors} detector(s) ran in this job (consensus={consensus:.2f}). "
            f"Provider reliability={source_rel:.2f}, latency={latency:.0f}ms (score={latency_score:.2f})."
        )
        return ConfidenceScore(
            overall=round(overall, 4),
            components=components,
            explanation=explanation,
            method="weighted_average",
        )

    # ------------------------------------------------------------------ #
    # Recognition match confidence
    # ------------------------------------------------------------------ #
    def score_match(self, nm: _MatchLike) -> ConfidenceScore:
        provider = nm.recognizer
        source_rel = self._reliability.get(provider, 0.7)
        match_conf = max(0.0, 1.0 - nm.distance)
        components = {
            "source_reliability": round(source_rel, 4),
            "cross_provider_consensus": 0.5,
            "detection_confidence": round(match_conf, 4),
            "latency_penalty": 0.9,
        }
        overall = sum(components[k] * WEIGHTS[k] for k in WEIGHTS)
        explanation = (
            f"{provider} matched face #{nm.query_face_index} "
            f"to '{nm.best_match}' with distance {nm.distance:.3f} "
            f"(match confidence={match_conf:.2f})."
        )
        return ConfidenceScore(
            overall=round(overall, 4),
            components=components,
            explanation=explanation,
            method="weighted_average",
        )

    # ------------------------------------------------------------------ #
    # Image analysis confidence
    # ------------------------------------------------------------------ #
    def score_image_analysis(self, nia: _AnalysisLike) -> ConfidenceScore:
        provider = nia.provider
        source_rel = self._reliability.get(provider, 0.8)
        # If the provider returned a quality_score use it, else neutral 0.7
        quality = nia.quality_score if nia.quality_score is not None else 0.7
        components = {
            "source_reliability": round(source_rel, 4),
            "cross_provider_consensus": 0.7,
            "detection_confidence": round(quality, 4),
            "latency_penalty": 0.9,
        }
        overall = sum(components[k] * WEIGHTS[k] for k in WEIGHTS)
        explanation = (
            f"{provider} analyzed image with quality score {quality:.2f}. "
            f"Provider reliability={source_rel:.2f}."
        )
        return ConfidenceScore(
            overall=round(overall, 4),
            components=components,
            explanation=explanation,
            method="weighted_average",
        )

    # ------------------------------------------------------------------ #
    # Metadata confidence
    # ------------------------------------------------------------------ #
    def score_metadata(self, nm: _MetadataLike) -> ConfidenceScore:
        provider = nm.provider
        source_rel = self._reliability.get(provider, 0.9)
        # Higher confidence if we actually extracted a format
        extraction_conf = 0.95 if nm.format else 0.5
        components = {
            "source_reliability": round(source_rel, 4),
            "cross_provider_consensus": 0.8,
            "detection_confidence": round(extraction_conf, 4),
            "latency_penalty": 0.95,
        }
        overall = sum(components[k] * WEIGHTS[k] for k in WEIGHTS)
        explanation = (
            f"{provider} extracted metadata (format={nm.format or 'unknown'}). "
            f"Provider reliability={source_rel:.2f}."
        )
        return ConfidenceScore(
            overall=round(overall, 4),
            components=components,
            explanation=explanation,
            method="weighted_average",
        )

    # ------------------------------------------------------------------ #
    # Forensics confidence
    # ------------------------------------------------------------------ #
    def score_forensics(self, nf: _ForensicsLike) -> ConfidenceScore:
        provider = nf.provider
        source_rel = self._reliability.get(provider, 0.8)
        # Use integrity_score if present, else fall back to neutral
        integrity = nf.integrity_score if nf.integrity_score is not None else 0.7
        components = {
            "source_reliability": round(source_rel, 4),
            "cross_provider_consensus": 0.6,
            "detection_confidence": round(integrity, 4),
            "latency_penalty": 0.85,
        }
        overall = sum(components[k] * WEIGHTS[k] for k in WEIGHTS)
        explanation = (
            f"{provider} forensics analysis integrity_score={integrity:.2f}. "
            f"Provider reliability={source_rel:.2f}."
        )
        return ConfidenceScore(
            overall=round(overall, 4),
            components=components,
            explanation=explanation,
            method="weighted_average",
        )

    # ------------------------------------------------------------------ #
    # Overall report confidence
    # ------------------------------------------------------------------ #
    def score_overall(
        self,
        report_detections: List[FaceDetection] = None,
        matches: List[FaceMatch] = None,
        conflicts: List[ConflictReport] = None,
        image_analyses: List[ImageAnalysisResult] = None,
        forensics: List[ForensicsResult] = None,
    ) -> ConfidenceScore:
        report_detections = report_detections or []
        matches = matches or []
        conflicts = conflicts or []
        image_analyses = image_analyses or []
        forensics = forensics or []

        # Collect all sub-scores available
        subscores: list[float] = []
        if report_detections:
            subscores.append(sum(d.confidence.overall for d in report_detections) / len(report_detections))
        if matches:
            subscores.append(sum(m.confidence.overall for m in matches) / len(matches))
        if image_analyses:
            subscores.append(sum(a.confidence.overall for a in image_analyses if a.confidence) / len(image_analyses))
        if forensics:
            subscores.append(sum(f.confidence.overall for f in forensics if f.confidence) / len(forensics))

        if not subscores:
            return ConfidenceScore(
                overall=0.0,
                components={},
                explanation="No results produced.",
            )

        base_avg = sum(subscores) / len(subscores)
        conflict_penalty = max(0.0, 1.0 - (0.15 * len(conflicts)))
        overall = base_avg * 0.8 + conflict_penalty * 0.2
        components = {
            "result_avg": round(base_avg, 4),
            "conflict_penalty": round(conflict_penalty, 4),
            "num_detections": float(len(report_detections)),
            "num_matches": float(len(matches)),
            "num_image_analyses": float(len(image_analyses)),
            "num_forensics": float(len(forensics)),
        }
        explanation = (
            f"Report confidence from {len(report_detections)} detection(s), "
            f"{len(matches)} match(es), {len(image_analyses)} image analyses, "
            f"{len(forensics)} forensics results, {len(conflicts)} conflict(s)."
        )
        return ConfidenceScore(
            overall=round(overall, 4),
            components=components,
            explanation=explanation,
            method="weighted_average",
        )