src/fusion/fuser.py

from __future__ import annotations

import numpy as np

from src.types import DetectionResponse, EngineResult

ENGINE_WEIGHTS = {
    "fingerprint": 0.45,
    "coherence": 0.35,
    "sstgnn": 0.20,
}

ENGINE_WEIGHTS_VIDEO = {
    "fingerprint": 0.30,
    "coherence": 0.50,
    "sstgnn": 0.20,
}

ATTRIBUTION_PRIORITY = {
    "fingerprint": 1,
    "sstgnn": 2,
    "coherence": 3,
}


def _normalize_generator(value: str | None) -> str:
    if not value:
        return "real"
    return str(value).strip().lower().replace(" ", "_")


def fuse(results: list[EngineResult], is_video: bool = False) -> tuple[str, float, str]:
    """Return (verdict, confidence_for_verdict, attributed_generator)."""

    weights = ENGINE_WEIGHTS_VIDEO if is_video else ENGINE_WEIGHTS
    active = [result for result in results if result.verdict != "UNKNOWN"]

    if not active:
        return "UNKNOWN", 0.5, "unknown_gan"

    wf = sum(
        result.confidence * weights.get(result.engine, 0.1)
        for result in active
        if result.verdict == "FAKE"
    )
    wr = sum(
        (1.0 - result.confidence) * weights.get(result.engine, 0.1)
        for result in active
        if result.verdict == "REAL"
    )

    denom = wf + wr + 1e-9
    fake_prob = float(np.clip(wf / denom, 0.0, 1.0))
    verdict = "FAKE" if fake_prob > 0.5 else "REAL"
    confidence = fake_prob if verdict == "FAKE" else (1.0 - fake_prob)

    generator = "real"
    if verdict == "FAKE":
        for result in sorted(active, key=lambda r: ATTRIBUTION_PRIORITY.get(r.engine, 9)):
            candidate = _normalize_generator(result.attributed_generator)
            if candidate and candidate != "real":
                generator = candidate
                break
        if generator == "real":
            generator = "unknown_gan"

    return verdict, confidence, generator


class Fuser:
    """Compatibility wrapper returning `DetectionResponse` objects."""

    def fuse(
        self,
        results: list[EngineResult],
        media_type: str = "image",
        total_ms: float = 0.0,
    ) -> DetectionResponse:
        if not results:
            return DetectionResponse(
                verdict="REAL",
                confidence=0.5,
                attributed_generator="unknown_gan",
                explanation="No engine results available.",
                processing_time_ms=round(total_ms, 2),
                engine_breakdown=[],
            )

        verdict, confidence, generator = fuse(results, is_video=(media_type == "video"))

        if verdict == "UNKNOWN":
            explanation = "No active engine outputs were available."
        else:
            summary = ", ".join(
                f"{result.engine}:{result.verdict}({result.confidence:.2f})"
                for result in results
            )
            explanation = f"Fused {media_type} analysis from engines: {summary}."

        return DetectionResponse(
            verdict=verdict,
            confidence=confidence,
            attributed_generator=generator,
            explanation=explanation,
            processing_time_ms=round(total_ms, 2),
            engine_breakdown=results,
        )