Blame for stance_classifier.py

"""
stance_classifier.py
====================
Three-layer stance classification for source ↔ claim pairs.

Layer 1 — Semantic Entailment Heuristics (always runs, zero cost)
  Detects indirect contradictions and supports that keyword lists miss.
  Example: "Earth is round" contradicts "Earth is flat" via antonym matching.

Layer 2 — Expanded Keyword Matching (always runs, zero cost)
  Larger keyword lists, threshold=1 for strong signals (false, debunked, etc.)
  Falls back on Layer 1 result if ambiguous.

Layer 3 — facebook/bart-large-mnli NLI (completely free, no API key needed)
  Natural Language Inference: premise=source_text, hypothesis=claim.
  ENTAILMENT  → SUPPORTS
  CONTRADICTION → CONTRADICTS 
  NEUTRAL     → NEUTRAL

  Primary:  HF Inference API  (set HF_TOKEN in .env for higher rate limits;
            works without token on public models — just slower)
  Fallback: local `transformers` pipeline (pip install transformers torch)

  Cached in-process — same (claim, source) pair never classified twice.

Priority: NLI result > L1 result > L2 result > neutral
"""

from __future__ import annotations

import hashlib
import logging
import os
import re
from typing import Dict, Optional, Tuple

logger = logging.getLogger("stance_classifier")

# ─────────────────────────────────────────────────────────────────────────────
# Layer 1 — Semantic Entailment Heuristics
# ─────────────────────────────────────────────────────────────────────────────

# Pairs of (claim_pattern → contradicting_signal) for common flat-earth-style claims
# Each entry: (regex that matches the CLAIM, regex that in the SOURCE indicates contradiction)
_ENTAILMENT_CONTRA_RULES: list[tuple[str, str]] = [
    # Shape / science denial
    (r"\bflat\s+earth\b|\bearth\s+is\s+flat\b",
     r"\bglobe\b|\bspherical\b|\bround\b|\borbits\b|\bnasa\b|\bscientists\b"),
    # Vaccine safety
    (r"\bvaccine.{0,20}(cause|causes|cause[sd]|causing).{0,20}(autism|dna|death|cancer|infertil)",
     r"\bno\s+evidence\b|\bdebunked\b|\bsafe\b|\bclinical\s+trial\b|\bfda\s+approved\b"),
    # 5G / radiation
    (r"\b5g.{0,30}(spread|cause|transmit|covid|virus|cancer|radiation)",
     r"\bno\s+link\b|\bno\s+evidence\b|\bdebunked\b|\bscientifically\b|\bimpossible\b"),
    # Climate denial
    (r"\bclimate\s+change\s+is\s+(fake|hoax|not\s+real)",
     r"\bscientific\s+consensus\b|\bdata\s+shows\b|\bglobal\s+warming\b|\b97\s*%\b"),
    # Election / fraud
    (r"\belection.{0,20}(stolen|rigged|fraud|fake)",
     r"\bno\s+evidence\b|\bcourts?\s+reject\b|\baudit\b.*\bno\s+fraud\b|\bofficial\s+results?\b"),
    # General "X never happened"
    (r"\b(never\s+happened|didn.t\s+happen|is\s+a\s+hoax|is\s+(fake|fabricated))",
     r"\bconfirmed\b|\bverified\b|\bofficial\s+records?\b|\bdocumented\b"),
]

# claim_pattern → source_pattern for SUPPORTING signal
_ENTAILMENT_SUPPORT_RULES: list[tuple[str, str]] = [
    # Vaccine efficacy
    (r"\bvaccine.{0,20}(effective|efficacy|prevent|protect)",
     r"\bclinical\s+trial\b|\befficacy\b|\bphase\s+[23]\b|\bfda\b|\bwho\b"),
    # Climate consensus
    (r"\bclimate\s+change\s+is\s+real\b|\bglobal\s+warming\b",
     r"\bscientific\s+consensus\b|\b97\s*%\b|\bipcc\b|\bnasa\b|\bnoaa\b"),
    # Official announcements
    (r"\b(government|ministry|official).{0,30}(announce|confirm|pass|enact)",
     r"\bofficial\b|\bgovernment\b|\bministry\b|\bgazette\b|\bpib\b|\bofficial\s+statement\b"),
]

# Antonym pairs: if the claim contains word A and source contains word B (or vice versa)
# in the same context window, it's likely a contradiction
_ANTONYM_PAIRS: list[tuple[str, str]] = [
    ("flat", "spherical"), ("flat", "globe"), ("flat", "round"),
    ("cure", "no cure"), ("safe", "dangerous"), ("safe", "unsafe"),
    ("effective", "ineffective"), ("effective", "no evidence"),
    ("real", "fake"), ("real", "hoax"), ("true", "false"),
    ("confirmed", "unconfirmed"), ("proven", "disproven"),
    ("increases", "decreases"), ("causes", "does not cause"),
    ("linked", "no link"), ("linked", "unrelated"),
    ("harmful", "harmless"), ("harmful", "safe"),
]


class SemanticHeuristicLayer:
    """
    Pattern-matching entailment: catches indirect stances that keyword
    lists miss because the source paraphrases rather than explicitly labels.
    """

    def classify(self, claim: str, source_text: str) -> Optional[str]:
        """
        Returns 'supporting' | 'contradicting' | None.
        None means this layer has no signal — defer to next layer.
        """
        claim_low  = claim.lower()
        source_low = source_text.lower()

        # Check contradiction entailment rules
        for claim_pat, source_pat in _ENTAILMENT_CONTRA_RULES:
            if re.search(claim_pat, claim_low) and re.search(source_pat, source_low):
                logger.debug("SemanticHeuristic CONTRA rule matched: claim_pat=%s", claim_pat[:40])
                return "contradicting"

        # Check support entailment rules
        for claim_pat, source_pat in _ENTAILMENT_SUPPORT_RULES:
            if re.search(claim_pat, claim_low) and re.search(source_pat, source_low):
                logger.debug("SemanticHeuristic SUPPORT rule matched: claim_pat=%s", claim_pat[:40])
                return "supporting"

        # Antonym detection: extract key noun/adj from claim, look for antonym in source
        for word_a, word_b in _ANTONYM_PAIRS:
            if word_a in claim_low and word_b in source_low:
                return "contradicting"
            if word_b in claim_low and word_a in source_low:
                return "contradicting"

        return None   # no signal


# ─────────────────────────────────────────────────────────────────────────────
# Layer 2 — Expanded Keyword Matching
# ─────────────────────────────────────────────────────────────────────────────

# Strong single-hit contradicting signals (threshold=1)
_STRONG_CONTRA = {
    "false", "debunked", "disproven", "misinformation", "misleading",
    "no evidence", "lacks evidence", "unfounded", "refuted", "myth",
    "conspiracy theory", "fact-check", "factcheck", "incorrect",
    "not true", "inaccurate", "unproven", "pseudoscience", "hoax",
    "wrong", "fake", "fabricated", "baseless", "discredited",
}

# Weak contradicting signals (threshold=2 — need two hits)
_WEAK_CONTRA = {
    "disputed", "controversial", "unverified", "claims", "alleged",
    "questioned", "skeptics", "critics", "challenged", "contradicts",
    "refutes", "rejects", "warns against", "cautioned",
}

# Strong single-hit supporting signals (threshold=1)
_STRONG_SUPPORT = {
    "confirms", "confirmed", "verified", "validated", "proven",
    "evidence shows", "data shows", "research shows", "study finds",
    "scientists confirm", "officially announced", "official statement",
    "press release", "gazette notification", "government confirms",
    "approved", "endorsed", "supported by evidence",
}

# Weak supporting signals (threshold=2)
_WEAK_SUPPORT = {
    "suggests", "indicates", "shows", "demonstrates", "according to",
    "reports", "states", "announced", "published", "found that",
}


class KeywordLayer:
    """
    Expanded keyword matching with differentiated strong/weak thresholds.
    Threshold for strong signals = 1 (any single hit is decisive).
    Threshold for weak signals   = 2 (need corroboration).
    """

    def classify(self, text: str) -> Tuple[str, float]:
        """
        Returns (stance, confidence).
        stance     = 'supporting' | 'contradicting' | 'neutral'
        confidence = 0.0–1.0 (how certain this layer is)
        """
        low = text.lower()

        strong_contra = sum(1 for kw in _STRONG_CONTRA if kw in low)
        weak_contra   = sum(1 for kw in _WEAK_CONTRA   if kw in low)
        strong_supp   = sum(1 for kw in _STRONG_SUPPORT if kw in low)
        weak_supp     = sum(1 for kw in _WEAK_SUPPORT   if kw in low)

        contra_score = strong_contra * 2 + weak_contra
        supp_score   = strong_supp * 2 + weak_supp

        # Strong single signal is decisive
        if strong_contra >= 1 and strong_contra >= strong_supp:
            conf = min(1.0, 0.6 + 0.1 * strong_contra)
            return "contradicting", conf

        if strong_supp >= 1 and strong_supp >= strong_contra:
            conf = min(1.0, 0.6 + 0.1 * strong_supp)
            return "supporting", conf

        # Weak signals need 2+
        if contra_score >= 2 and contra_score > supp_score:
            return "contradicting", 0.45

        if supp_score >= 2 and supp_score > contra_score:
            return "supporting", 0.45

        return "neutral", 0.20


# ─────────────────────────────────────────────────────────────────────────────
# Layer 3 — facebook/bart-large-mnli  (NLI, completely free)
# ─────────────────────────────────────────────────────────────────────────────

_HF_API_URL   = "https://router.huggingface.co/hf-inference/models/facebook/bart-large-mnli/v1/zero-shot-classification"
_NLI_ENABLED  = os.getenv("STANCE_NLI_ENABLED", "true").lower() != "false"

# Label mapping: HuggingFace returns "ENTAILMENT", "CONTRADICTION", "NEUTRAL"
_NLI_LABEL_MAP = {
    "entailment":   "supporting",
    "contradiction": "contradicting",
    "neutral":       "neutral",
}


def classify_stance(claim: str, evidence: str) -> dict:
    """
    Public NLI helper used by tests / external callers.

    premise   = evidence  (the source text)
    hypothesis = claim    (what we are verifying)

    Returns:
        {"stance": "SUPPORTS" | "CONTRADICTS" | "NEUTRAL", "confidence": float}
    """
    result = _bart_nli(claim=claim, text=evidence[:512], title="")
    if result is None:
        return {"stance": "NEUTRAL", "confidence": 0.20}
    stance, confidence = result
    return {
        "stance": {
            "supporting": "SUPPORTS",
            "contradicting": "CONTRADICTS",
            "neutral": "NEUTRAL",
        }.get(stance, "NEUTRAL"),
        "confidence": confidence,
    }


def _bart_nli(claim: str, title: str, text: str) -> Optional[Tuple[str, float]]:
    """
    Core NLI call.  Returns (stance_internal, confidence) or None on failure.
    stance_internal = "supporting" | "contradicting" | "neutral"

    Strategy
    --------
    1. Try HF Inference API  (zero-install, rate-limited but free)
    2. Try local `transformers` pipeline  (no rate limit, needs torch)
    3. Return None → caller falls back to keyword result
    """
    combined = f"{title} {text}".strip()[:512]
    if not combined:
        return None

    # ── attempt 1: HF Inference API ──────────────────────────────────────────
    result = _hf_api_nli(claim=claim, premise=combined)
    if result is not None:
        return result

    # ── attempt 2: local transformers ────────────────────────────────────────
    result = _local_nli(claim=claim, premise=combined)
    return result   # None if both failed


# ── HF Inference API ─────────────────────────────────────────────────────────

def _hf_api_nli(claim: str, premise: str) -> Optional[Tuple[str, float]]:
    """
    Calls the HF router zero-shot classification endpoint.
    Requires HF_TOKEN — set it in .env for free access.
    Without a token, skips to local transformers.
    Returns (stance, confidence) or None.
    """
    try:
        import requests as _req
    except ImportError:
        return None

    hf_token = os.getenv("HF_TOKEN", "")
    if not hf_token or hf_token.startswith("your_"):
        logger.debug("No HF_TOKEN set — skipping HF router API, using local transformers")
        return None

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {hf_token}",
    }

    # router.huggingface.co zero-shot-classification format:
    # inputs = the text to classify (premise)
    # parameters.candidate_labels = the hypothesis options
    # multi_label=False → softmax across labels (NLI-style)
    nli_payload = {
        "inputs": premise,
        "parameters": {
            "candidate_labels": ["supports the claim", "contradicts the claim", "unrelated to the claim"],
            "hypothesis_template": "{}.",
            "multi_label": False,
        },
    }

    try:
        resp = _req.post(_HF_API_URL, headers=headers, json=nli_payload, timeout=15)
        if resp.status_code in (503, 502):
            # Model loading — skip silently, fall through to local
            logger.debug("HF API: model loading (%s), falling back to local", resp.status_code)
            return None
        if not resp.ok:
            logger.debug("HF API error %s: %s", resp.status_code, resp.text[:120])
            return None

        data = resp.json()
        return _parse_hf_router_response(data)

    except Exception as exc:
        logger.debug("HF API NLI call failed: %s", exc)
        return None


def _parse_hf_router_response(data) -> Optional[Tuple[str, float]]:
    """
    Parse the HuggingFace router zero-shot-classification response.

    Expected shape:
      {
        "sequence": "...",
        "labels": ["supports the claim", "contradicts the claim", "unrelated to the claim"],
        "scores": [0.85, 0.10, 0.05]
      }
    """
    if not data or not isinstance(data, dict):
        return None

    labels = data.get("labels", [])
    scores = data.get("scores", [])
    if not labels or not scores or len(labels) != len(scores):
        logger.debug("HF router NLI: unexpected response shape: %s", str(data)[:120])
        return None

    # Map back to internal stance labels
    label_to_stance = {
        "supports the claim":     "supporting",
        "contradicts the claim":  "contradicting",
        "unrelated to the claim": "neutral",
    }

    # Pick the highest-scoring label
    best_idx = scores.index(max(scores))
    best_label = labels[best_idx].lower()
    best_score = float(scores[best_idx])

    stance = label_to_stance.get(best_label, "neutral")
    logger.debug("HF router NLI: '%s' → %s (%.2f)", best_label, stance, best_score)
    return stance, best_score


def _parse_hf_nli_response(data) -> Optional[Tuple[str, float]]:
    """
    Legacy parser kept for reference — not called by current code.

    Two possible shapes from old api-inference.huggingface.co:
      A) List of {"label": "ENTAILMENT", "score": 0.97} dicts
      B) Nested list from zero-shot endpoint
    """
    if not data:
        return None

    # Shape A: direct label list
    if isinstance(data, list) and isinstance(data[0], dict) and "label" in data[0]:
        best = max(data, key=lambda x: x.get("score", 0))
        label = best["label"].lower()
        score = float(best.get("score", 0.5))
        stance = _NLI_LABEL_MAP.get(label, "neutral")
        logger.debug("HF NLI API: %s → %s (%.2f)", label, stance, score)
        return stance, score

    # Shape B: nested list [[{label, score}, ...]]
    if isinstance(data, list) and isinstance(data[0], list):
        flat = data[0]
        best = max(flat, key=lambda x: x.get("score", 0))
        label = best["label"].lower()
        score = float(best.get("score", 0.5))
        stance = _NLI_LABEL_MAP.get(label, "neutral")
        logger.debug("HF NLI API (nested): %s → %s (%.2f)", label, stance, score)
        return stance, score

    logger.debug("HF NLI: unexpected response shape: %s", str(data)[:120])
    return None


# ── Local transformers pipeline ───────────────────────────────────────────────

_LOCAL_NLI_PIPE = None          # cached pipeline instance
_LOCAL_NLI_TRIED = False        # avoid retrying failed imports


def _local_nli(claim: str, premise: str) -> Optional[Tuple[str, float]]:
    """
    Local BART-large-MNLI via the `transformers` library.
    Returns (stance, confidence) or None.
    First call downloads ~1.6 GB model; subsequent calls are fast.
    """
    global _LOCAL_NLI_PIPE, _LOCAL_NLI_TRIED
    if _LOCAL_NLI_TRIED and _LOCAL_NLI_PIPE is None:
        return None    # already failed once — skip

    if _LOCAL_NLI_PIPE is None:
        _LOCAL_NLI_TRIED = True
        try:
            from transformers import pipeline as hf_pipeline
            logger.info("Loading facebook/bart-large-mnli locally (first call downloads ~1.6 GB)…")
            _LOCAL_NLI_PIPE = hf_pipeline(
                "zero-shot-classification",
                model="facebook/bart-large-mnli",
                device=-1,       # CPU; set device=0 for GPU
            )
            logger.info("BART-MNLI local pipeline ready.")
        except Exception as exc:
            logger.warning(
                "transformers import or model load failed: %s. "
                "Install with: pip install transformers torch", exc
            )
            _LOCAL_NLI_PIPE = None
            return None

    try:
        # zero-shot-classification with a three-way NLI hypothesis template
        # We use claim as the candidate_label so BART tests:
        #   premise ENTAILS claim / CONTRADICTS claim / NEUTRAL
        out = _LOCAL_NLI_PIPE(
            premise,
            candidate_labels=["true", "false", "unrelated"],
            hypothesis_template=f"This text {{}} the following: {claim[:200]}",
        )
        # out = {"labels": [...], "scores": [...]}
        label_map = {"true": "supporting", "false": "contradicting", "unrelated": "neutral"}
        top_label = out["labels"][0].lower()
        top_score = float(out["scores"][0])
        stance = label_map.get(top_label, "neutral")
        logger.debug("Local BART NLI: '%s' → %s (%.2f)", premise[:40], stance, top_score)
        return stance, top_score

    except Exception as exc:
        logger.warning("Local BART NLI inference failed: %s", exc)
        return None


# ── In-process NLI result cache ───────────────────────────────────────────────

_NLI_CACHE: Dict[str, Tuple[str, float]] = {}


class NLIStanceLayer:
    """
    Layer 3: facebook/bart-large-mnli NLI stance classifier.

    Completely free — no API key required for public HF models.
    Optionally set HF_TOKEN in .env for higher Inference API rate limits.

    Usage is identical to the old LLMStanceLayer interface so the
    orchestrator (StanceClassifier) needs no changes.
    """

    def __init__(self) -> None:
        self._enabled = _NLI_ENABLED
        if self._enabled:
            logger.info(
                "NLI stance layer enabled (facebook/bart-large-mnli). "
                "Primary: HF Inference API. Fallback: local transformers."
            )
        else:
            logger.info("NLI stance layer disabled via STANCE_NLI_ENABLED=false")

    def available(self) -> bool:
        return self._enabled

    def classify(self, claim: str, title: str, text: str) -> Optional[str]:
        """
        Returns 'supporting' | 'contradicting' | 'neutral' | None.
        None = both API and local inference failed; caller uses keyword result.
        """
        if not self._enabled:
            return None

        cache_key = hashlib.md5(
            f"{claim[:120]}||{title[:80]}||{text[:300]}".encode()
        ).hexdigest()
        if cache_key in _NLI_CACHE:
            cached_stance, _ = _NLI_CACHE[cache_key]
            return cached_stance

        result = _bart_nli(claim=claim, title=title, text=text)
        if result is None:
            return None

        stance, confidence = result
        _NLI_CACHE[cache_key] = (stance, confidence)
        logger.debug("NLI layer: '%s…' → %s (%.2f)", claim[:40], stance, confidence)
        return stance


# ─────────────────────────────────────────────────────────────────────────────
# Orchestrator — combines all three layers
# ─────────────────────────────────────────────────────────────────────────────

class StanceClassifier:
    """
    Three-layer stance classifier for source ↔ claim pairs.

    Priority cascade:
      1. Semantic heuristics  (pattern matching, no cost)
      2. Keyword layer        (expanded lists, no cost)
      3. Claude LLM           (only when uncertain + high-reliability source)

    Usage (drop-in replacement for the old StanceClassifier)
    ---------------------------------------------------------
        classifier = StanceClassifier()
        stance = classifier.classify(text=source_text, claim=claim_text)
        # returns: "supporting" | "contradicting" | "neutral"

    Advanced (with confidence + layer attribution)
    ----------------------------------------------
        stance, confidence, layer = classifier.classify_detailed(
            title=source_title, text=source_text, claim=claim_text,
            source_reliability=0.85,
        )
    """

    # Minimum reliability a source needs to trigger LLM call when uncertain
    _LLM_MIN_RELIABILITY = float(os.getenv("STANCE_LLM_MIN_REL", "0.40"))

    def __init__(self) -> None:
        self._heuristic = SemanticHeuristicLayer()
        self._keywords  = KeywordLayer()
        self._llm       = NLIStanceLayer()

    # ------------------------------------------------------------------
    # Drop-in API (matches old StanceClassifier.classify signature)
    # ------------------------------------------------------------------

    def classify(self, text: str, claim: str) -> str:
        """
        Backward-compatible interface: returns stance string only.
        title defaults to first 80 chars of text.
        """
        title = text[:80]
        stance, _, _ = self.classify_detailed(
            title=title, text=text, claim=claim, source_reliability=0.50
        )
        return stance

    # ------------------------------------------------------------------
    # Full API
    # ------------------------------------------------------------------

    def classify_detailed(
        self,
        title: str,
        text: str,
        claim: str,
        source_reliability: float = 0.50,
    ) -> Tuple[str, float, str]:
        """
        Returns (stance, confidence, winning_layer).

        stance          = "supporting" | "contradicting" | "neutral"
        confidence      = 0.0–1.0
        winning_layer   = "heuristic" | "keyword" | "llm" | "default"
        """
        combined_text = (title + " " + text).strip()

        # ── Layer 1: Semantic heuristics ─────────────────────────────────
        h_stance = self._heuristic.classify(claim, combined_text)
        if h_stance is not None:
            return h_stance, 0.75, "heuristic"

        # ── Layer 2: Keywords ─────────────────────────────────────────────
        kw_stance, kw_conf = self._keywords.classify(combined_text)

        # If keyword layer is confident, use it
        if kw_conf >= 0.55:
            return kw_stance, kw_conf, "keyword"

        # ── Layer 3: NLI (BART-MNLI) — only for uncertain + credible sources ─
        if (
            self._llm.available()
            and source_reliability >= self._LLM_MIN_RELIABILITY
        ):
            llm_stance = self._llm.classify(claim, title, text)
            if llm_stance is not None:
                return llm_stance, 0.90, "nli"

        # ── Fallback: use whatever keyword layer found ────────────────────
        return kw_stance, kw_conf, "keyword"


# ─────────────────────────────────────────────────────────────────────────────
# Backwards-compat shim: old code did StanceClassifier(scorer)
# New code supports both StanceClassifier() and StanceClassifier(scorer)
# ─────────────────────────────────────────────────────────────────────────────

def build_stance_classifier(scorer=None) -> StanceClassifier:
    """Factory that accepts an optional DomainReliabilityScorer (ignored — kept for compat)."""
    return StanceClassifier()