src/explainability/explainer.py

from __future__ import annotations

import logging
import os
from typing import Any

from src.types import DetectionResponse, EngineResult

logger = logging.getLogger(__name__)

SYSTEM_INSTRUCTION = (
    "You are a deepfake forensics analyst writing reports for security professionals. "
    "Given detection engine outputs, write exactly 2-3 sentences in plain English "
    "explaining why the content is real or fake. "
    "Be specific and name the strongest signals. "
    "Use direct declarative sentences. "
    "Output only the explanation text."
)

DEFAULT_MODEL_CANDIDATES = (
    "gemini-3.1-flash-lite-preview",
    "gemini-2.5-flash",
)

_configured_candidates = [
    value.strip()
    for value in os.environ.get("GEMINI_MODEL_CANDIDATES", "").split(",")
    if value.strip()
]
MODEL_CANDIDATES = (
    tuple(_configured_candidates)
    if _configured_candidates
    else DEFAULT_MODEL_CANDIDATES
)

REQUEST_TIMEOUT_S = float(os.environ.get("GEMINI_REQUEST_TIMEOUT_S", "20"))
MAX_MODEL_ATTEMPTS = max(1, int(os.environ.get("GEMINI_MAX_MODEL_ATTEMPTS", "3")))
TEMPERATURE = float(os.environ.get("GEMINI_EXPLAIN_TEMPERATURE", "0.3"))
TOP_P = float(os.environ.get("GEMINI_EXPLAIN_TOP_P", "0.95"))
MAX_TOKENS = int(os.environ.get("GEMINI_EXPLAIN_MAX_TOKENS", "300"))

_client: Any | None = None


def _get_api_key() -> str:
    return os.environ.get("GEMINI_API_KEY", "").strip()


def _get_client():
    global _client
    if _client is not None:
        return _client

    api_key = _get_api_key()
    if not api_key:
        raise RuntimeError("GEMINI_API_KEY is not configured")

    try:
        from google import genai
    except Exception as exc:
        raise RuntimeError("google-genai package is not installed") from exc

    _client = genai.Client(api_key=api_key)
    return _client


def _generate(prompt: str) -> str:
    client = _get_client()
    last_error: Exception | None = None

    try:
        from google.genai import types
    except Exception as exc:
        raise RuntimeError("google-genai types module is unavailable") from exc

    for model_name in MODEL_CANDIDATES[:MAX_MODEL_ATTEMPTS]:
        try:
            response = client.models.generate_content(
                model=model_name,
                contents=prompt,
                config=types.GenerateContentConfig(
                    system_instruction=SYSTEM_INSTRUCTION,
                    temperature=TEMPERATURE,
                    top_p=TOP_P,
                    max_output_tokens=MAX_TOKENS,
                ),
            )
            content = getattr(response, "text", None)
            if content and content.strip():
                logger.info("Gemini explain model selected: %s", model_name)
                return content.strip()
        except Exception as exc:
            last_error = exc
            logger.debug("Gemini explain model %s failed: %s", model_name, exc)

    if last_error is not None:
        raise last_error
    raise RuntimeError("No Gemini model candidates succeeded")


def explain(
    verdict: str,
    confidence: float,
    engine_results: list[EngineResult],
    generator: str,
) -> str:
    breakdown = "\n".join(
        f"- {result.engine}: {result.verdict} ({result.confidence:.0%}) - {result.explanation}"
        for result in engine_results
    )

    prompt = (
        f"Verdict: {verdict} ({confidence:.0%} confidence)\n"
        f"Attributed generator: {generator}\n"
        f"Engine breakdown:\n{breakdown}\n\n"
        "Write the forensics explanation."
    )

    try:
        return _generate(prompt)
    except Exception as exc:
        logger.error("Gemini explain failed: %s", exc)
        top = engine_results[0] if engine_results else None
        primary = f"Primary signal came from the {top.engine} engine." if top else ""
        return (
            f"Content classified as {verdict} with {confidence:.0%} confidence. "
            f"Attributed generator: {generator}. "
            f"{primary}"
        ).strip()


class Explainer:
    """Compatibility wrapper for legacy callers expecting an object API."""

    def explain(self, response: DetectionResponse) -> str:
        return explain(
            verdict=response.verdict,
            confidence=response.confidence,
            engine_results=response.engine_breakdown,
            generator=response.attributed_generator,
        )