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