confidence/conflicts.py

"""
Conflict detector — finds cross-provider disagreements.

Currently detects:
  - face_count_mismatch   (detectors disagree on number of faces)
  - match_disagreement    (recognizers disagree on best match for same face)
  - quality_disagreement  (image-quality providers disagree substantially)
  - integrity_disagreement (forensics providers disagree on integrity)
  - format_mismatch       (metadata providers disagree on format)
"""

from __future__ import annotations

from typing import Dict, List, Protocol

from models.reports import ConflictReport
from providers.base import ProviderResult


# 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
    recognizer: str


class ConflictDetector:
    """Detects cross-provider disagreements."""

    def detect(
        self,
        results: Dict[str, ProviderResult],
        boxes: List[_BoxLike],
        matches: List[_MatchLike],
    ) -> List[ConflictReport]:
        conflicts: List[ConflictReport] = []

        conflicts.extend(self._detect_face_count_mismatch(results))
        conflicts.extend(self._detect_match_disagreement(matches))
        conflicts.extend(self._detect_quality_disagreement(results))
        conflicts.extend(self._detect_integrity_disagreement(results))
        conflicts.extend(self._detect_format_mismatch(results))

        return conflicts

    # ------------------------------------------------------------------ #
    # Individual conflict detectors
    # ------------------------------------------------------------------ #
    def _detect_face_count_mismatch(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
        detector_counts: Dict[str, int] = {}
        for r in results.values():
            if r.success and r.capability.value == "detection":
                detector_counts[r.provider] = r.normalized.get("num_faces", 0)
        if len(detector_counts) < 2:
            return []
        counts = list(detector_counts.values())
        if max(counts) == min(counts):
            return []
        return [ConflictReport(
            kind="face_count_mismatch",
            providers=list(detector_counts.keys()),
            description=(
                f"Detectors disagree on face count: {detector_counts}"
            ),
            severity="warning",
        )]

    def _detect_match_disagreement(self, matches: List[_MatchLike]) -> List[ConflictReport]:
        by_face: Dict[int, List[_MatchLike]] = {}
        for m in matches:
            by_face.setdefault(m.query_face_index, []).append(m)
        out: List[ConflictReport] = []
        for face_idx, face_matches in by_face.items():
            if len(face_matches) < 2:
                continue
            best_matches = {m.best_match for m in face_matches if m.best_match}
            if len(best_matches) > 1:
                out.append(ConflictReport(
                    kind="match_disagreement",
                    providers=[m.recognizer for m in face_matches],
                    description=(
                        f"Recognizers disagree on best match for face #{face_idx}: "
                        f"{[(m.recognizer, m.best_match) for m in face_matches]}"
                    ),
                    severity="warning",
                ))
        return out

    def _detect_quality_disagreement(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
        """Flag if image-quality providers disagree on quality by > 0.3."""
        quality_scores: Dict[str, float] = {}
        for r in results.values():
            if r.success and r.capability.value == "image_analysis":
                qs = r.normalized.get("quality_score")
                if qs is not None:
                    quality_scores[r.provider] = float(qs)
        if len(quality_scores) < 2:
            return []
        scores = list(quality_scores.values())
        if max(scores) - min(scores) > 0.3:
            return [ConflictReport(
                kind="quality_disagreement",
                providers=list(quality_scores.keys()),
                description=f"Image-quality providers disagree by >0.3: {quality_scores}",
                severity="info",
            )]
        return []

    def _detect_integrity_disagreement(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
        """Flag if forensics providers disagree on integrity by > 0.3."""
        integrity_scores: Dict[str, float] = {}
        for r in results.values():
            if r.success and r.capability.value == "forensics":
                ii = r.normalized.get("integrity_score")
                if ii is not None:
                    integrity_scores[r.provider] = float(ii)
        if len(integrity_scores) < 2:
            return []
        scores = list(integrity_scores.values())
        if max(scores) - min(scores) > 0.3:
            return [ConflictReport(
                kind="integrity_disagreement",
                providers=list(integrity_scores.keys()),
                description=f"Forensics providers disagree on integrity: {integrity_scores}",
                severity="warning",
            )]
        return []

    def _detect_format_mismatch(self, results: Dict[str, ProviderResult]) -> List[ConflictReport]:
        """Flag if metadata providers disagree on image format."""
        formats: Dict[str, str] = {}
        for r in results.values():
            if r.success and r.capability.value == "metadata":
                fmt = r.normalized.get("format")
                if fmt:
                    formats[r.provider] = fmt
        if len(formats) < 2:
            return []
        unique_formats = set(formats.values())
        if len(unique_formats) > 1:
            return [ConflictReport(
                kind="format_mismatch",
                providers=list(formats.keys()),
                description=f"Metadata providers disagree on format: {formats}",
                severity="info",
            )]
        return []