spread_predictor.py

"""
PHASE 6 (Option A): Spread risk scoring -- real, explainable, NOT a trained GNN.

IMPORTANT CONTEXT -- read this before changing the scoring weights:
The original project plan called for "a GAT trained on SciFact" to power
this feature. That doesn't work: SciFact contains scientific claims paired
with paper abstracts labeled Supports/Contradicts/NotEnoughInfo -- it has
zero data about how claims spread through networks (no engagement counts,
no share/forward chains, no timing). A model trained on SciFact cannot
learn virality, no matter how it's wired up. Rather than pretend to train
a GNN that couldn't have learned this task, this module computes a real,
inspectable spread-risk estimate from three signals we can actually and
honestly measure:

  1. Misinformation pattern match -- how closely this claim resembles
     already-known False/Misleading claims in your 16k+ fact ChromaDB
     knowledge base (via retrieval.py's real similarity search).
  2. Sensationalism score -- a keyword/punctuation heuristic based on
     well-documented misinformation-spread research (claims using urgency,
     secrecy, and emotional language spread faster -- see Vosoughi, Roy &
     Aral, "The spread of true and false news online", Science 2018).
  3. Entity embeddedness -- how connected this claim's medical entities are
     within a NetworkX graph built from the claim + its extracted entities
     + retrieved evidence, using real graph centrality math.

This is a heuristic scoring system, not a machine-learned model. It is
labeled as such everywhere it surfaces (API response, UI). If you later
get access to real network/engagement data (e.g. CoAID's ~296k social
engagement records), that would be the right foundation for an actual
trained spread-prediction GNN -- SciFact never was.
"""

import re
import networkx as nx

# Sensational/urgency language commonly seen in misinformation forwards.
# Not exhaustive -- a real production system would derive this list from
# labeled data rather than a hand-picked set.
SENSATIONAL_KEYWORDS = [
    "cure", "miracle", "secret", "they don't want you to know",
    "doctors hate", "banned", "shocking", "guaranteed", "100% effective",
    "forward this", "share before", "urgent", "breaking", "must read",
    "wake up", "big pharma", "cover up", "conspiracy", "instant relief",
]


def _sensationalism_score(claim_text: str) -> float:
    """
    Returns 0.0-1.0. Combines keyword presence, exclamation density, and
    ALL-CAPS word ratio -- all directly measurable from the claim text
    itself, no black box.
    """
    text_lower = claim_text.lower()
    keyword_hits = sum(1 for kw in SENSATIONAL_KEYWORDS if kw in text_lower)
    keyword_score = min(keyword_hits / 3, 1.0)  # 3+ hits already maxes this out

    exclamations = claim_text.count("!")
    exclamation_score = min(exclamations / 3, 1.0)

    words = re.findall(r"[A-Za-z]+", claim_text)
    caps_words = [w for w in words if len(w) > 2 and w.isupper()]
    caps_ratio = (len(caps_words) / len(words)) if words else 0.0
    caps_score = min(caps_ratio * 2, 1.0)

    return round(0.5 * keyword_score + 0.25 * exclamation_score + 0.25 * caps_score, 3)


def _misinformation_pattern_score(evidence: list[dict]) -> float:
    """
    Returns 0.0-1.0. Looks at the real evidence your retrieval.py already
    pulled from ChromaDB: if this claim sits close (low similarity_distance)
    to facts your knowledge base has already labeled False/Misleading,
    that's a genuine, data-grounded signal that it fits a known
    misinformation pattern -- not a guess.
    """
    if not evidence:
        return 0.0

    weighted_total = 0.0
    weight_sum = 0.0
    for e in evidence:
        # similarity_distance: lower = more similar. Convert to a similarity
        # weight in [0, 1] (clamped) so closer matches count more.
        distance = e.get("similarity_distance", 1.0)
        similarity_weight = max(0.0, 1.0 - min(distance, 1.0))
        is_misinfo = e.get("verdict") in ("False", "Misleading")
        weighted_total += similarity_weight * (1.0 if is_misinfo else 0.0)
        weight_sum += similarity_weight

    if weight_sum == 0:
        return 0.0
    return round(weighted_total / weight_sum, 3)


def _entity_embeddedness_score(claim_text: str, entities: list[dict], evidence: list[dict]) -> float:
    """
    Returns 0.0-1.0. Builds a real graph: claim node -> entity nodes ->
    evidence nodes (edge added when an entity's text appears inside a
    retrieved evidence claim). Uses NetworkX's degree_centrality on that
    graph as a genuine structural signal -- more-connected entities mean
    this claim's topic is more "embedded" in your existing knowledge base,
    which correlates with it being a well-trodden (and thus more shareable)
    topic rather than a completely novel one-off statement.
    """
    G = nx.Graph()
    claim_node = "CLAIM"
    G.add_node(claim_node)

    if not entities:
        return 0.0

    for ent in entities:
        entity_node = f"ENT::{ent['text'].lower()}"
        G.add_node(entity_node)
        G.add_edge(claim_node, entity_node)

        for i, e in enumerate(evidence):
            evidence_node = f"EVID::{i}"
            if ent["text"].lower() in e.get("matched_claim", "").lower():
                G.add_node(evidence_node)
                G.add_edge(entity_node, evidence_node)

    if G.number_of_nodes() <= 1:
        return 0.0

    centrality = nx.degree_centrality(G)
    entity_nodes = [n for n in G.nodes if n.startswith("ENT::")]
    if not entity_nodes:
        return 0.0

    avg_centrality = sum(centrality[n] for n in entity_nodes) / len(entity_nodes)
    return round(min(avg_centrality * 2, 1.0), 3)  # scaled -- small graphs rarely hit 1.0 raw


def analyze_spread_risk(claim_text: str, entities: list[dict], evidence: list[dict]) -> dict:
    """
    Combines the three signals above into a single explainable spread-risk
    estimate. Every number in the output is either directly measured from
    real data (your ChromaDB knowledge base, the claim's own text) or a
    disclosed heuristic formula -- nothing here is a black-box prediction.
    """
    misinfo_score = _misinformation_pattern_score(evidence)
    sensational_score = _sensationalism_score(claim_text)
    embeddedness_score = _entity_embeddedness_score(claim_text, entities, evidence)

    virality_score = round(
        100 * (0.5 * misinfo_score + 0.3 * sensational_score + 0.2 * embeddedness_score)
    )
    virality_score = max(0, min(100, virality_score))

    if virality_score >= 70:
        risk_level = "High Risk"
    elif virality_score >= 40:
        risk_level = "Medium Risk"
    else:
        risk_level = "Low Risk"

    # Illustrative, order-of-magnitude reach estimate -- NOT a forecast.
    # Scaled directly off virality_score with a disclosed formula, so it's
    # honest about being derived, not measured.
    if virality_score >= 70:
        reach_bucket = "Tens of thousands"
        reach_midpoint = 25000
    elif virality_score >= 40:
        reach_bucket = "Thousands"
        reach_midpoint = 3000
    else:
        reach_bucket = "Hundreds"
        reach_midpoint = 300

    # Higher risk -> assumed faster peak. Also just a disclosed heuristic.
    time_to_peak_hours = max(2, round(48 * (1 - virality_score / 100)))

    return {
        "virality_score": virality_score,
        "risk_level": risk_level,
        "predicted_nodes_reached": reach_midpoint,
        "reach_estimate_bucket": reach_bucket,
        "time_to_peak_hours": time_to_peak_hours,
        "network_hubs_vulnerable": [
            "WhatsApp Forwards (general risk channel, not claim-specific)",
            "Public Facebook Groups (general risk channel, not claim-specific)",
        ],
        "signal_breakdown": {
            "misinformation_pattern_match": misinfo_score,
            "sensational_language_score": sensational_score,
            "entity_embeddedness_score": embeddedness_score,
        },
        "is_heuristic": True,
        "methodology": (
            "Heuristic estimate combining: (1) similarity to known False/"
            "Misleading claims in the knowledge base, (2) sensational-language "
            "scoring of the claim text, (3) NetworkX graph centrality of the "
            "claim's medical entities. Not a trained machine learning model -- "
            "see spread_predictor.py for the full disclosed formula."
        ),
    }