Spaces:

infoclassifier
/

InfoClassifier

Sleeping

InfoClassifier / data_collector.py
UTKARSH MISHRA
Initial commit: InfoClassifier project
e5e36db 7 months ago
48.3 kB
"""
data_collector.py
=================
Production data-collection pipeline for InformationIntegrity OpenEnv.
Takes any claim string(s), fetches real evidence from multiple search/news
backends, scores source reliability, classifies stance, and caches results
in Redis.  The output is a Scenario JSON ready for the environment.
Backends (all optional — falls back gracefully):
  1. SerpAPI          SERPAPI_KEY
  2. NewsAPI          NEWSAPI_KEY
  3. Brave Search     BRAVE_SEARCH_KEY
  4. Google Fact Check GOOGLE_FACTCHECK_KEY
  5. DuckDuckGo HTML  (no key, rate-limited fallback)
Environment variables:
  SERPAPI_KEY
  NEWSAPI_KEY
  BRAVE_SEARCH_KEY
  GOOGLE_FACTCHECK_KEY
  REDIS_URL              (default: redis://localhost:6379)
  CACHE_TTL_SECONDS      (default: 86400 — 24h)
  RELIABILITY_CONFIG     (default: domain_reliability.yaml)
  MAX_SOURCES_PER_CLAIM  (default: 8)
  REQUEST_TIMEOUT        (default: 10s)
Usage:
    python data_collector.py --claims "Claim one" "Claim two" \
                             --content "The original post text" \
                             --difficulty medium \
                             --out scenario.json
    from data_collector import DataCollector, CollectionRequest
    collector = DataCollector()
    scenario  = collector.collect(CollectionRequest(
        content="Post text",
        claims=["Claim A"],
        difficulty="medium",
    ))
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
import re
import time
import urllib.parse
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from urllib.parse import urlparse

import fnmatch
import requests
import yaml

# Load .env BEFORE any os.getenv() calls so API keys are available
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass  # python-dotenv not installed; rely on shell environment

try:
    from retrieval import QueryRewriter as _QueryRewriter
    _QUERY_REWRITER = _QueryRewriter()
except Exception:
    _QUERY_REWRITER = None  # type: ignore

# Import the enhanced stance classifier with fallback to original
try:
    from enhanced_stance_classifier import EnhancedStanceClassifier as _StanceClassifier, build_enhanced_stance_classifier
    _USE_ENHANCED_STANCE = True
    logger_tmp = logging.getLogger("data_collector")
    logger_tmp.info("Using enhanced stance classifier with improved context understanding")
except Exception as _e:
    logger_tmp = logging.getLogger("data_collector")
    logger_tmp.warning("Enhanced stance_classifier unavailable (%s) — falling back to original", _e)
    try:
        from stance_classifier import StanceClassifier as _StanceClassifier, build_stance_classifier
        _USE_ENHANCED_STANCE = False
        _USE_NEW_STANCE = True
    except Exception as _e2:
        logger_tmp.warning("Original stance_classifier also unavailable (%s) — using legacy keywords", _e2)
        _USE_ENHANCED_STANCE = False
        _USE_NEW_STANCE = False

try:
    import redis
    _REDIS_AVAILABLE = True
except ImportError:
    _REDIS_AVAILABLE = False

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s — %(message)s",
)
logger = logging.getLogger("data_collector")


# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

RELIABILITY_CONFIG_PATH = os.getenv("RELIABILITY_CONFIG", "domain_reliability.yaml")
REDIS_URL               = os.getenv("REDIS_URL", "redis://localhost:6379")
CACHE_TTL               = int(os.getenv("CACHE_TTL_SECONDS", 86400))
MAX_SOURCES             = int(os.getenv("MAX_SOURCES_PER_CLAIM", 8))
REQUEST_TIMEOUT         = int(os.getenv("REQUEST_TIMEOUT", 10))


# ---------------------------------------------------------------------------
# Domain Reliability Scorer
# ---------------------------------------------------------------------------

class DomainReliabilityScorer:
    """
    Loads domain_reliability.yaml and scores any URL.
    Fully enterprise-configurable without touching Python code.
    """

    def __init__(self, config_path: str = RELIABILITY_CONFIG_PATH) -> None:
        self._config = self._load(config_path)
        self._exact: Dict[str, tuple[float, str]] = {}   # domain → (score, type)
        self._patterns: List[tuple[str, float, str]] = [] # (glob, score, type)
        self._build_index()

    def _load(self, path: str) -> dict:
        if not os.path.exists(path):
            logger.warning("Reliability config not found at %s — using defaults", path)
            return {}
        with open(path, "r", encoding="utf-8") as fh:
            return yaml.safe_load(fh) or {}

    def _build_index(self) -> None:
        skip = {"version", "defaults", "stance_thresholds", "evidence_thresholds"}
        for section_name, section in self._config.items():
            if section_name in skip or not isinstance(section, dict):
                continue
            score = float(section.get("score", 0.35))
            stype = str(section.get("source_type", "unknown"))
            for domain in section.get("domains", []):
                self._exact[domain.lower().lstrip("www.")] = (score, stype)
            for pat in section.get("patterns", []):
                self._patterns.append((pat.lower(), score, stype))

    def score(self, url: str) -> tuple[float, str]:
        """Return (reliability_score, source_type) for a URL."""
        try:
            host = urlparse(url).hostname or ""
        except Exception:
            host = ""
        host = host.lower().lstrip("www.")

        # Exact match (including subdomains stripped to base)
        for candidate in self._candidates(host):
            if candidate in self._exact:
                return self._exact[candidate]

        # Glob patterns (e.g. *.blogspot.com)
        for pat, score, stype in self._patterns:
            if fnmatch.fnmatch(host, pat):
                return score, stype

        defaults = self._config.get("defaults", {})
        return (
            float(defaults.get("unknown_domain_score", 0.35)),
            str(defaults.get("unknown_source_type", "unknown")),
        )

    @staticmethod
    def _candidates(host: str) -> List[str]:
        """Generate host + progressively stripped parent domains."""
        parts = host.split(".")
        candidates = []
        for i in range(len(parts)):
            candidates.append(".".join(parts[i:]))
        return candidates

    @property
    def stance_config(self) -> dict:
        return self._config.get("stance_thresholds", {})

    @property
    def evidence_thresholds(self) -> dict:
        return self._config.get("evidence_thresholds", {})


# ---------------------------------------------------------------------------
# Stance Classifier
# ---------------------------------------------------------------------------

class StanceClassifier:
    """
    Heuristic stance classification using keyword lists from the YAML config.
    No ML dependency — deterministic and inspectable.
    """

    def __init__(self, scorer: DomainReliabilityScorer) -> None:
        cfg = scorer.stance_config
        self._contra_kw: List[str] = [k.lower() for k in cfg.get("contradicting_keywords", [])]
        self._supp_kw: List[str]   = [k.lower() for k in cfg.get("supporting_keywords", [])]
        self._contra_min: int      = int(cfg.get("contradicting_keyword_min", 2))
        self._supp_min: int        = int(cfg.get("supporting_keyword_min", 2))

    def classify(self, text: str, claim: str) -> str:
        """Return 'supporting' | 'contradicting' | 'neutral'."""
        lowered = text.lower()
        contra_hits = sum(1 for kw in self._contra_kw if kw in lowered)
        supp_hits   = sum(1 for kw in self._supp_kw   if kw in lowered)

        if contra_hits >= self._contra_min and contra_hits > supp_hits:
            return "contradicting"
        if supp_hits >= self._supp_min and supp_hits >= contra_hits:
            return "supporting"
        return "neutral"


# ---------------------------------------------------------------------------
# Data models for collection pipeline
# ---------------------------------------------------------------------------

@dataclass
class RawResult:
    title: str
    snippet: str
    url: str
    source: str       # which backend returned this


@dataclass
class CollectionRequest:
    content: str
    claims: List[str]
    difficulty: str = "medium"
    scenario_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    tags: List[str] = field(default_factory=list)
    topic: str = ""
    force_refresh: bool = False


# ---------------------------------------------------------------------------
# Search backend interface + implementations
# ---------------------------------------------------------------------------

class SearchBackend(ABC):
    name: str = "base"

    @abstractmethod
    def search(self, query: str, num_results: int = 8) -> List[RawResult]:
        ...

    def available(self) -> bool:
        return True


class SerpAPIBackend(SearchBackend):
    name = "serpapi"

    def __init__(self) -> None:
        self._key = os.getenv("SERPAPI_KEY", "")

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

    def search(self, query: str, num_results: int = 8) -> List[RawResult]:
        # Detect query intent to choose the right SerpAPI parameters
        is_factcheck  = re.search(r'\bfact.?check\b|debunked|hoax|false|misleading', query, re.I)
        is_news       = re.search(r'\bofficial\b|\bgovernment\b|\bannouncement\b', query, re.I)

        params: dict = {
            "q":        query,
            "api_key":  self._key,
            "num":      num_results,
            "hl":       "en",
            "gl":       "us",       # default country; overridden below if entity found
            "safe":     "off",
        }

        # For very recent claims (date entities present) restrict to last month
        if re.search(r'\b(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)\b|\b\d{4}\b',
                     query, re.I):
            params["tbs"] = "qdr:m"   # last month

        # Fact-check / debunking queries → force news tab
        if is_factcheck or is_news:
            params["tbm"] = "nws"

        # Adjust country context from entity
        country_gl = {
            "india": "in", "pakistan": "pk", "uk": "uk", "australia": "au",
            "canada": "ca", "germany": "de", "france": "fr", "japan": "jp",
            "china": "cn", "brazil": "br", "nigeria": "ng",
        }
        low_q = query.lower()
        for country, gl in country_gl.items():
            if country in low_q:
                params["gl"] = gl
                break

        resp = requests.get(
            "https://serpapi.com/search",
            params=params,
            timeout=REQUEST_TIMEOUT,
        )
        resp.raise_for_status()
        data = resp.json()
        results: List[RawResult] = []

        # Answer box — featured snippet is very high signal
        if ab := data.get("answer_box"):
            snippet = ab.get("answer") or ab.get("snippet") or ab.get("result", "")
            link    = ab.get("link", "")
            if snippet and link:
                results.append(RawResult(
                    title   = ab.get("title", "Featured Answer"),
                    snippet = str(snippet)[:600],
                    url     = link,
                    source  = f"{self.name}:answer_box",
                ))

        # Top stories (news carousel)
        for story in data.get("top_stories", [])[:4]:
            link = story.get("link", "")
            if link:
                results.append(RawResult(
                    title   = story.get("title", ""),
                    snippet = story.get("snippet", story.get("title", "")),
                    url     = link,
                    source  = f"{self.name}:news",
                ))

        # Organic results
        for item in data.get("organic_results", [])[:num_results]:
            link = item.get("link", "")
            if link:
                results.append(RawResult(
                    title   = item.get("title", ""),
                    snippet = item.get("snippet", ""),
                    url     = link,
                    source  = self.name,
                ))

        return results


class NewsAPIBackend(SearchBackend):
    name = "newsapi"

    def __init__(self) -> None:
        self._key = os.getenv("NEWSAPI_KEY", "")

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

    def search(self, query: str, num_results: int = 8) -> List[RawResult]:
        params = {
            "q": query,
            "apiKey": self._key,
            "language": "en",
            "sortBy": "relevancy",
            "pageSize": num_results,
        }
        resp = requests.get(
            "https://newsapi.org/v2/everything",
            params=params,
            timeout=REQUEST_TIMEOUT,
        )
        resp.raise_for_status()
        data = resp.json()
        results = []
        for article in data.get("articles", [])[:num_results]:
            content = article.get("content") or ""
            # NewsAPI truncates content with "[+N chars]" — strip that artifact
            content = re.sub(r"\s*\[\+\d+ chars\]$", "", content).strip()
            snippet = article.get("description") or content or ""
            results.append(RawResult(
                title=article.get("title", ""),
                snippet=snippet[:500],
                url=article.get("url", ""),
                source=self.name,
            ))
        return results


class BraveSearchBackend(SearchBackend):
    name = "brave"

    def __init__(self) -> None:
        self._key = os.getenv("BRAVE_SEARCH_KEY", "")

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

    def search(self, query: str, num_results: int = 8) -> List[RawResult]:
        headers = {
            "Accept": "application/json",
            "Accept-Encoding": "gzip",
            "X-Subscription-Token": self._key,
        }
        params = {"q": query, "count": num_results, "text_decorations": False}
        resp = requests.get(
            "https://api.search.brave.com/res/v1/web/search",
            headers=headers,
            params=params,
            timeout=REQUEST_TIMEOUT,
        )
        resp.raise_for_status()
        data = resp.json()
        results = []
        for item in data.get("web", {}).get("results", [])[:num_results]:
            results.append(RawResult(
                title=item.get("title", ""),
                snippet=item.get("description", ""),
                url=item.get("url", ""),
                source=self.name,
            ))
        return results


class GoogleFactCheckBackend(SearchBackend):
    """
    Google Fact Check Tools API.
    Returns structured fact-check claims with verdicts from IFCN signatories.
    """
    name = "google_factcheck"

    def __init__(self) -> None:
        self._key = os.getenv("GOOGLE_FACTCHECK_KEY", "")

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

    def search(self, query: str, num_results: int = 8) -> List[RawResult]:
        params = {
            "query": query,
            "key": self._key,
            "languageCode": "en",
            "pageSize": num_results,
        }
        resp = requests.get(
            "https://factchecktools.googleapis.com/v1alpha1/claims:search",
            params=params,
            timeout=REQUEST_TIMEOUT,
        )
        resp.raise_for_status()
        data = resp.json()
        results = []
        for item in data.get("claims", [])[:num_results]:
            reviews = item.get("claimReview", [])
            for review in reviews[:1]:
                publisher = review.get("publisher", {}).get("name", "")
                rating    = review.get("textualRating", "")
                title     = f"{publisher}: {rating}"
                snippet   = (
                    f"Claim: {item.get('text', '')} | "
                    f"Rating: {rating} | "
                    f"Review: {review.get('title', '')}"
                )
                url = review.get("url", "")
                results.append(RawResult(
                    title=title,
                    snippet=snippet,
                    url=url,
                    source=self.name,
                ))
        return results


class DuckDuckGoFallbackBackend(SearchBackend):
    """
    No-key fallback using DuckDuckGo.
    Rate-limit strategy
    -------------------
    DDG aggressively rate-limits when the same IP fires multiple queries
    in quick succession.  To work around this:
    1. Only 1 query is ever sent per search() call (controlled by the
       n_queries=1 limit set in _fetch_all_backends).
    2. Exponential back-off: 5s → 15s → 45s between retries.
    3. On the first RateLimitException, promote directly to the HTML
       fallback (no waiting) — the HTML endpoint has a separate rate limit.
    4. A class-level cooldown flag prevents hammering DDG when it's known
       to be rate-limiting; the flag resets after _COOLDOWN_SECS seconds.
    """
    name = "duckduckgo"
    _MAX_RETRIES   = 2
    _BASE_DELAY    = 5.0    # seconds — first retry wait
    _BACKOFF_MULT  = 3.0    # each retry waits BASE * MULT^attempt
    _COOLDOWN_SECS = 60.0   # after a rate-limit, skip DDG for this many seconds

    # Class-level cooldown state (shared across all calls in the process)
    _cooldown_until: float = 0.0

    def available(self) -> bool:
        return True

    def search(self, query: str, num_results: int = 8) -> List[RawResult]:
        import time as _time

        # Respect session-level cooldown
        if _time.time() < DuckDuckGoFallbackBackend._cooldown_until:
            remaining = DuckDuckGoFallbackBackend._cooldown_until - _time.time()
            logger.debug("DDG in cooldown for %.0fs — skipping to HTML fallback", remaining)
            return self._html_fallback(query, num_results)

        for attempt in range(self._MAX_RETRIES):
            try:
                from duckduckgo_search import DDGS
                from duckduckgo_search.exceptions import RatelimitException
                results = []
                with DDGS() as ddgs:
                    for r in ddgs.text(query, max_results=num_results):
                        results.append(RawResult(
                            title   = r.get("title", ""),
                            snippet = r.get("body", ""),
                            url     = r.get("href", ""),
                            source  = self.name,
                        ))
                if results:
                    return results
                logger.warning("DuckDuckGo returned 0 results (attempt %d)", attempt + 1)
                time.sleep(self._BASE_DELAY * (self._BACKOFF_MULT ** attempt))

            except ImportError:
                return self._html_fallback(query, num_results)

            except Exception as exc:
                exc_str = str(exc)
                is_ratelimit = (
                    "202" in exc_str
                    or "Ratelimit" in exc_str
                    or "ratelimit" in exc_str.lower()
                    or "rate" in exc_str.lower()
                )
                if is_ratelimit:
                    # Set cooldown and immediately try HTML — no point waiting
                    DuckDuckGoFallbackBackend._cooldown_until = (
                        time.time() + self._COOLDOWN_SECS
                    )
                    logger.warning(
                        "DDG rate-limited — switching to HTML fallback "
                        "(cooldown %ds)", self._COOLDOWN_SECS
                    )
                    return self._html_fallback(query, num_results)

                logger.warning("DuckDuckGo attempt %d failed: %s", attempt + 1, exc)
                time.sleep(self._BASE_DELAY * (self._BACKOFF_MULT ** attempt))

        # All DDGS attempts exhausted — try HTML
        return self._html_fallback(query, num_results)

    def _html_fallback(self, query: str, num_results: int) -> List[RawResult]:
        """Scrape DuckDuckGo HTML endpoint — separate rate limit from the API."""
        import random
        _USER_AGENTS = [
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15",
        ]
        encoded = urllib.parse.quote_plus(query)
        url     = f"https://html.duckduckgo.com/html/?q={encoded}"
        headers = {
            "User-Agent":      random.choice(_USER_AGENTS),
            "Accept-Language": "en-US,en;q=0.9",
            "Accept":          "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        }
        try:
            resp = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT)
            resp.raise_for_status()
            clean_tag = re.compile(r"<[^>]+>")
            titles    = re.findall(r'class="result__title">.*?<a[^>]+>(.*?)</a>', resp.text, re.S)
            snippets  = re.findall(r'class="result__snippet">(.*?)</a>', resp.text, re.S)
            urls_raw  = re.findall(r'class="result__url">(.*?)</span>', resp.text, re.S)
            results   = []
            for t, s, u in zip(titles, snippets, urls_raw):
                u = u.strip()
                results.append(RawResult(
                    title   = clean_tag.sub("", t).strip()[:200],
                    snippet = clean_tag.sub("", s).strip()[:500],
                    url     = ("https://" + u) if not u.startswith("http") else u,
                    source  = self.name,
                ))
                if len(results) >= num_results:
                    break
            return results
        except Exception as exc:
            logger.error("DuckDuckGo HTML fallback failed: %s", exc)
            return []


class WikipediaBackend(SearchBackend):
    """
    Wikipedia REST API — zero rate-limits, no key, always returns results.
    Used as the guaranteed last-resort fallback when all other backends fail.
    """
    name = "wikipedia"

    def available(self) -> bool:
        return True

    def search(self, query: str, num_results: int = 5) -> List[RawResult]:
        results = []
        # 1. Search for matching article titles
        try:
            search_resp = requests.get(
                "https://en.wikipedia.org/w/api.php",
                params={
                    "action": "query", "list": "search",
                    "srsearch": query, "srlimit": num_results,
                    "format": "json", "utf8": 1,
                },
                timeout=REQUEST_TIMEOUT,
                headers={"User-Agent": "InformationIntegrityEnv/1.0 (research)"},
            )
            search_resp.raise_for_status()
            hits = search_resp.json().get("query", {}).get("search", [])
        except Exception as exc:
            logger.warning("Wikipedia search failed: %s", exc)
            return []

        # 2. Fetch intro paragraph for each hit
        for hit in hits[:num_results]:
            title = hit.get("title", "")
            snippet = re.sub(r"<[^>]+>", "", hit.get("snippet", "")).strip()
            try:
                summ_resp = requests.get(
                    f"https://en.wikipedia.org/api/rest_v1/page/summary/{urllib.parse.quote(title)}",
                    timeout=REQUEST_TIMEOUT,
                    headers={"User-Agent": "InformationIntegrityEnv/1.0 (research)"},
                )
                if summ_resp.ok:
                    data    = summ_resp.json()
                    snippet = data.get("extract", snippet)[:600]
                    url     = data.get("content_urls", {}).get("desktop", {}).get("page",
                              f"https://en.wikipedia.org/wiki/{urllib.parse.quote(title)}")
                else:
                    url = f"https://en.wikipedia.org/wiki/{urllib.parse.quote(title)}"
            except Exception:
                url = f"https://en.wikipedia.org/wiki/{urllib.parse.quote(title)}"

            results.append(RawResult(
                title   = f"Wikipedia: {title}",
                snippet = snippet,
                url     = url,
                source  = self.name,
            ))

        return results


# ---------------------------------------------------------------------------
# Full Content Fetcher
# ---------------------------------------------------------------------------

class FullContentFetcher:
    """
    Fetches and extracts the main article body from a URL.
    Uses newspaper3k when available (best extraction quality), then falls back
    to a BeautifulSoup heuristic that strips boilerplate and joins <p> tags.
    Only called for sources whose reliability score >= FULL_CONTENT_MIN_RELIABILITY.
    Returns None on any failure so callers can fall back to the snippet.
    """

    MIN_RELIABILITY: float = float(os.getenv("FULL_CONTENT_MIN_RELIABILITY", "0.6"))
    MAX_CHARS: int = int(os.getenv("FULL_CONTENT_MAX_CHARS", "4000"))
    FETCH_TIMEOUT: int = int(os.getenv("FULL_CONTENT_TIMEOUT", "12"))

    _USER_AGENTS = [
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15",
    ]

    # Simple cache to avoid re-fetching the same URL within a single run
    _cache: Dict[str, Optional[str]] = {}

    def fetch(self, url: str) -> Optional[str]:
        """Return extracted article text (up to MAX_CHARS) or None on failure."""
        if url in self._cache:
            return self._cache[url]

        text = self._fetch_newspaper(url) or self._fetch_bs4(url)
        if text:
            text = text[:self.MAX_CHARS]
        self._cache[url] = text
        return text

    def _fetch_newspaper(self, url: str) -> Optional[str]:
        """Extract article using newspaper3k (best quality)."""
        try:
            from newspaper import Article  # type: ignore
            article = Article(url, request_timeout=self.FETCH_TIMEOUT)
            article.download()
            article.parse()
            body = (article.text or "").strip()
            return body if len(body) > 100 else None
        except Exception:
            return None

    def _fetch_bs4(self, url: str) -> Optional[str]:
        """Extract main content using BeautifulSoup as fallback."""
        import random
        try:
            from bs4 import BeautifulSoup  # type: ignore
        except ImportError:
            return None

        try:
            headers = {
                "User-Agent": random.choice(self._USER_AGENTS),
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "en-US,en;q=0.9",
            }
            resp = requests.get(url, headers=headers, timeout=self.FETCH_TIMEOUT, allow_redirects=True)
            if not resp.ok or "text/html" not in resp.headers.get("Content-Type", ""):
                return None

            soup = BeautifulSoup(resp.text, "html.parser")

            # Remove boilerplate elements
            for tag in soup(["script", "style", "nav", "header", "footer",
                              "aside", "form", "noscript", "iframe", "figure"]):
                tag.decompose()

            # Heuristic: find the container with the most <p> text
            candidates = soup.find_all(["article", "main", "div", "section"])
            best_block = None
            best_len = 0
            for block in candidates:
                text = " ".join(p.get_text(" ", strip=True) for p in block.find_all("p"))
                if len(text) > best_len:
                    best_len = len(text)
                    best_block = text

            if best_block and best_len > 150:
                # Normalise whitespace
                return re.sub(r"\s{2,}", " ", best_block).strip()

            # Last resort: just get all paragraph text
            all_p = " ".join(p.get_text(" ", strip=True) for p in soup.find_all("p"))
            cleaned = re.sub(r"\s{2,}", " ", all_p).strip()
            return cleaned if len(cleaned) > 150 else None

        except Exception as exc:
            logger.debug("BS4 content fetch failed for %s: %s", url[:60], exc)
            return None


# ---------------------------------------------------------------------------
# Redis Cache
# ---------------------------------------------------------------------------

class ScenarioCache:
    """
    Redis-backed cache for collected scenarios.
    Key = sha256(claim_fingerprint).
    Falls back to an in-process dict if Redis is unavailable.
    """

    def __init__(self, redis_url: str = REDIS_URL, ttl: int = CACHE_TTL) -> None:
        self._ttl = ttl
        self._mem: Dict[str, Any] = {}
        self._redis: Optional[Any] = None
        if _REDIS_AVAILABLE:
            try:
                self._redis = redis.from_url(redis_url, decode_responses=True)
                self._redis.ping()
                logger.info("Redis cache connected at %s", redis_url)
            except Exception as exc:
                logger.warning("Redis unavailable (%s) — falling back to in-process cache", exc)
                self._redis = None

    @staticmethod
    def _key(request: CollectionRequest) -> str:
        fingerprint = json.dumps({
            "claims": sorted(request.claims),
            "difficulty": request.difficulty,
        }, sort_keys=True)
        return "scenario:" + hashlib.sha256(fingerprint.encode()).hexdigest()

    def get(self, request: CollectionRequest) -> Optional[dict]:
        k = self._key(request)
        if self._redis:
            raw = self._redis.get(k)
            if raw:
                return json.loads(raw)
        return self._mem.get(k)

    def set(self, request: CollectionRequest, scenario: dict) -> None:
        k = self._key(request)
        if self._redis:
            self._redis.setex(k, self._ttl, json.dumps(scenario))
        self._mem[k] = scenario

    def invalidate(self, request: CollectionRequest) -> None:
        k = self._key(request)
        if self._redis:
            self._redis.delete(k)
        self._mem.pop(k, None)


# ---------------------------------------------------------------------------
# Main DataCollector
# ---------------------------------------------------------------------------

class DataCollector:
    """
    Orchestrates multi-backend search, reliability scoring, stance
    classification, deduplication, and caching into a Scenario dict.
    """

    def __init__(
        self,
        backends: Optional[List[SearchBackend]] = None,
        cache: Optional[ScenarioCache] = None,
        scorer: Optional[DomainReliabilityScorer] = None,
    ) -> None:
        self._scorer    = scorer or DomainReliabilityScorer()
        self._cache     = cache or ScenarioCache()
        self._backends  = backends or self._default_backends()

        # Use enhanced stance classifier with fallbacks
        if _USE_ENHANCED_STANCE:
            self._stancer = build_enhanced_stance_classifier(self._scorer)
        elif _USE_NEW_STANCE:
            self._stancer = build_stance_classifier(self._scorer)
        else:
            self._stancer = StanceClassifier(self._scorer)  # type: ignore

    @staticmethod
    def _default_backends() -> List[SearchBackend]:
        # Disable DuckDuckGo due to rate limiting and reliability issues
        candidates = [
            SerpAPIBackend(),
            NewsAPIBackend(),
            BraveSearchBackend(),
            GoogleFactCheckBackend(),
            # DuckDuckGoFallbackBackend(),  # Disabled - causes rate limiting issues
            WikipediaBackend(),          # guaranteed fallback — no rate-limits
        ]
        available = [b for b in candidates if b.available()]
        if not available:
            logger.warning("No search backends available — results will be empty")
        logger.info(
            "Active search backends: %s",
            [b.name for b in available],
        )
        return available

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def collect(self, request: CollectionRequest) -> dict:
        """
        Main entry point.  Returns a Scenario-compatible dict.
        Checks Redis cache first; fetches from APIs on cache miss.
        """
        if not request.force_refresh:
            cached = self._cache.get(request)
            if cached:
                logger.info("Cache hit for scenario %s", request.scenario_id)
                return cached

        scenario = self._build_scenario(request)
        self._cache.set(request, scenario)
        logger.info(
            "Collected scenario %s — %d claims, %d sources",
            scenario["id"],
            len(scenario["claims"]),
            len(scenario["sources"]),
        )
        return scenario

    def collect_batch(self, requests: List[CollectionRequest]) -> List[dict]:
        return [self.collect(r) for r in requests]

    # ------------------------------------------------------------------
    # Internal pipeline
    # ------------------------------------------------------------------

    def _build_scenario(self, req: CollectionRequest) -> dict:
        all_sources = []
        seen_urls: set[str] = set()
        fetcher = FullContentFetcher()

        for claim in req.claims:
            raw_results = self._fetch_all_backends(claim)
            for raw in raw_results:
                norm_url = self._normalise_url(raw.url)
                if not norm_url or norm_url in seen_urls:
                    continue
                seen_urls.add(norm_url)

                score, stype = self._scorer.score(norm_url)

                # For high-reliability sources, fetch the full article body so
                # stance classification has rich context rather than a 500-char snippet.
                full_text: Optional[str] = None
                if score >= FullContentFetcher.MIN_RELIABILITY:
                    full_text = fetcher.fetch(norm_url)
                    if full_text:
                        logger.info(
                            "Full content fetched for %s (%d chars, rel=%.2f)",
                            norm_url[:70], len(full_text), score,
                        )

                # The text used for classification is full article when available,
                # otherwise fall back to the snippet returned by the backend.
                text_for_classification = full_text or raw.snippet

                # Use enhanced classifier when available (returns detailed tuple),
                # fall back to legacy interface (returns string only)
                try:
                    if (_USE_ENHANCED_STANCE or _USE_NEW_STANCE) and hasattr(self._stancer, "classify_detailed"):
                        stance, _conf, _layer = self._stancer.classify_detailed(
                            title=raw.title,
                            text=text_for_classification,
                            claim=claim,
                            source_reliability=score,
                        )
                        logger.debug(
                            "Stance '%s' via %s (conf=%.2f): %s",
                            stance, _layer, _conf, raw.title[:60],
                        )
                    else:
                        stance = self._stancer.classify(text_for_classification + " " + raw.title, claim)
                        logger.debug("Stance '%s' via legacy: %s", stance, raw.title[:60])
                except Exception as stance_exc:
                    logger.warning(
                        "Stance classification failed for '%s': %s - defaulting to neutral",
                        raw.title[:60], stance_exc
                    )
                    stance = "neutral"
                src_id = f"src_{len(all_sources) + 1:03d}"

                all_sources.append({
                    "id":                  src_id,
                    "title":               raw.title[:200],
                    "text":                (full_text or raw.snippet)[:800],
                    "url":                 norm_url,
                    "stance":              stance,
                    "reliability":         round(score, 3),
                    "source_type":         stype,
                    "verification_weight": round(score, 3),
                    "fetched_by":          raw.source,
                    "has_full_content":    full_text is not None,
                })

                if len(all_sources) >= MAX_SOURCES * len(req.claims):
                    break

        claims_payload = [
            {
                "id":                      f"claim_{i+1:03d}",
                "text":                    c,
                "ground_truth_status":     self._infer_ground_truth_status(c, all_sources),
                "ground_truth_confidence": self._infer_confidence(all_sources),
            }
            for i, c in enumerate(req.claims)
        ]

        ground_truth = self._infer_decision(claims_payload, all_sources)

        return {
            "id":               req.scenario_id,
            "difficulty":       req.difficulty,
            "topic":            req.topic or self._guess_topic(req.content),
            "tags":             req.tags,
            "content":          req.content,
            "claims":           claims_payload,
            "sources":          all_sources,
            "ground_truth":     ground_truth,
            "explanation_hint": self._build_explanation_hint(
                req.claims, all_sources, ground_truth
            ),
            "_meta": {
                "collected_at":    time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                "backends_used":   list({s["fetched_by"] for s in all_sources}),
                "total_sources":   len(all_sources),
            },
        }

    def _fetch_all_backends(self, query: str) -> List[RawResult]:
        """Try backends in priority order; aggregate unique results."""
        results: List[RawResult] = []
        seen_urls: set[str] = set()

        # Use QueryRewriter for semantically rich query variants
        if _QUERY_REWRITER is not None:
            enriched_queries = _QUERY_REWRITER.expand(query)
        else:
            keywords = query.strip().rstrip("?.")
            enriched_queries = [
                keywords,
                f'fact check "{keywords}"',
                f'"{keywords}" evidence',
                f'is it true that {keywords}',
            ]

        for backend in self._backends:
            # SerpAPI gets all query variants (primary backend, paid, high rate-limit)
            # Other backends get 2 queries for better coverage
            if backend.name == "serpapi":
                n_queries = len(enriched_queries)
            else:
                n_queries = 2
            try:
                for q in enriched_queries[:n_queries]:
                    for r in backend.search(q, num_results=MAX_SOURCES):
                        if r.url and r.url not in seen_urls and (r.snippet or r.title):
                            seen_urls.add(r.url)
                            results.append(r)
            except Exception as exc:
                logger.warning("Backend %s failed for query '%s': %s", backend.name, query, exc)

        # ── Keyword gate: drop results with <2 overlapping claim keywords ──────
        gated = [r for r in results if self._keyword_overlap(query, r) >= self._min_overlap(query)]
        # Safety net: if gate drops everything, keep original results (no empty pool)
        if not gated and results:
            logger.debug("Keyword gate dropped all results for '%s…' — keeping unfiltered", query[:60])
            return results[:MAX_SOURCES * 3]

        return gated[:MAX_SOURCES * 3]

    @staticmethod
    def _keyword_overlap(claim: str, raw: RawResult) -> int:
        """Count shared non-trivial keywords between the claim and a result."""
        _stop = {
            "the","a","an","is","are","was","were","in","on","at","to","from",
            "of","and","or","but","that","this","it","be","will","has","have",
            "its","by","for","not","with","also","into","than","more","about",
        }
        claim_kws  = {w for w in re.findall(r'\b[a-z]{3,}\b', claim.lower())  if w not in _stop}
        result_kws = {w for w in re.findall(r'\b[a-z]{3,}\b',
                      (raw.title + " " + raw.snippet).lower()) if w not in _stop}
        return len(claim_kws & result_kws)

    @staticmethod
    def _min_overlap(claim: str) -> int:
        """
        Adaptive minimum overlap threshold based on claim length.
        Short claims (≤4 meaningful words) → require 1 match.
        Longer claims → require 2 matches (avoids generic Wikipedia hits).
        """
        _stop = {"the","a","an","is","are","was","were","in","on","at","to",
                 "from","of","and","or","but","that","this","it","be","will"}
        n_kws = len({w for w in re.findall(r'\b[a-z]{3,}\b', claim.lower()) if w not in _stop})
        return 1 if n_kws <= 4 else 2

    @staticmethod
    def _normalise_url(url: str) -> str:
        url = url.strip()
        if not url.startswith(("http://", "https://")):
            url = "https://" + url
        try:
            parsed = urlparse(url)
            return parsed.geturl()
        except Exception:
            return url

    def _infer_ground_truth_status(self, claim: str, sources: List[dict]) -> str:
        """
        Heuristic: look at stance distribution weighted by reliability.
        Returns 'true' | 'false' | 'partially_true' | 'unknown'
        """
        contra_score = sum(
            s["reliability"] for s in sources if s["stance"] == "contradicting"
        )
        support_score = sum(
            s["reliability"] for s in sources if s["stance"] == "supporting"
        )
        total = contra_score + support_score
        if total == 0:
            return "unknown"
        ratio = support_score / total
        if ratio < 0.2:
            return "false"
        if ratio > 0.7:
            return "true"
        return "partially_true"

    def _infer_confidence(self, sources: List[dict]) -> float:
        if not sources:
            return 0.0
        max_rel = max(s["reliability"] for s in sources)
        return round(min(max_rel, 1.0), 3)

    def _infer_decision(self, claims: List[dict], sources: List[dict]) -> str:
        """
        Derive a moderation recommendation from weighted evidence.
          remove          → weighted false evidence dominant, high reliability
          flag            → mixed or uncertain evidence
          allow           → predominantly true, high-reliability support
        """
        false_weight = sum(
            s["reliability"]
            for s in sources
            if s["stance"] == "contradicting"
        )
        support_weight = sum(
            s["reliability"]
            for s in sources
            if s["stance"] == "supporting"
        )
        neutral_weight = sum(
            s["reliability"]
            for s in sources
            if s["stance"] == "neutral"
        )
        total = false_weight + support_weight + neutral_weight or 1.0

        false_pct   = false_weight / total
        support_pct = support_weight / total

        # Strong contradicting signal → remove
        if false_pct > 0.55 and false_weight > 1.5:
            return "remove"
        # Weak or mixed signal → flag
        if false_pct > 0.30 or (abs(false_pct - support_pct) < 0.20):
            return "flag"
        # Predominantly supported and reliable → allow
        if support_pct > 0.55 and support_weight > 1.0:
            return "allow"
        return "flag"

    @staticmethod
    def _guess_topic(content: str) -> str:
        TOPICS = [
            ("health",      ["vaccine", "covid", "cancer", "virus", "drug", "health", "disease"]),
            ("politics",    ["election", "president", "government", "vote", "congress", "parliament"]),
            ("technology",  ["5g", "ai", "chip", "microchip", "algorithm", "robot", "data"]),
            ("climate",     ["climate", "warming", "carbon", "emission", "fossil", "renewable"]),
            ("economics",   ["inflation", "gdp", "unemployment", "economy", "recession", "market"]),
            ("science",     ["study", "research", "scientist", "experiment", "proof", "evidence"]),
        ]
        low = content.lower()
        for topic, keywords in TOPICS:
            if any(kw in low for kw in keywords):
                return topic
        return "general"

    @staticmethod
    def _build_explanation_hint(claims: List[str], sources: List[dict], decision: str) -> str:
        contra = [s for s in sources if s["stance"] == "contradicting"]
        supp   = [s for s in sources if s["stance"] == "supporting"]
        lines  = [
            f"Decision: {decision}.",
            f"Claims examined: {len(claims)}.",
            f"Contradicting sources: {len(contra)}, Supporting sources: {len(supp)}.",
        ]
        if contra:
            top = max(contra, key=lambda s: s["reliability"])
            lines.append(f"Strongest contradicting: {top['title'][:80]} (reliability {top['reliability']}).")
        if supp:
            top = max(supp, key=lambda s: s["reliability"])
            lines.append(f"Strongest supporting: {top['title'][:80]} (reliability {top['reliability']}).")
        return " ".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def _cli() -> None:
    import argparse

    parser = argparse.ArgumentParser(
        description="Collect real-world evidence for a set of claims."
    )
    parser.add_argument(
        "--claims", nargs="+", required=True,
        help="One or more claim strings to investigate"
    )
    parser.add_argument(
        "--content", default="",
        help="The original piece of content containing the claims"
    )
    parser.add_argument(
        "--difficulty", default="medium",
        choices=["easy", "medium", "hard"],
    )
    parser.add_argument(
        "--topic", default="",
        help="Optional topic label"
    )
    parser.add_argument(
        "--tags", nargs="*", default=[],
        help="Optional tags"
    )
    parser.add_argument(
        "--out", default="scenario.json",
        help="Output file path"
    )
    parser.add_argument(
        "--force-refresh", action="store_true",
        help="Ignore Redis cache and re-fetch"
    )
    args = parser.parse_args()

    collector = DataCollector()
    request   = CollectionRequest(
        content=args.content or " ".join(args.claims),
        claims=args.claims,
        difficulty=args.difficulty,
        topic=args.topic,
        tags=args.tags,
        force_refresh=args.force_refresh,
    )
    scenario = collector.collect(request)

    with open(args.out, "w", encoding="utf-8") as fh:
        json.dump(scenario, fh, indent=2, ensure_ascii=False)

    logger.info("Scenario written to %s", args.out)
    print(json.dumps(scenario, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    _cli()