osint-dashboard/app/sources.py

"""Data source ingestors — fetch from external APIs and push to NATS."""

from __future__ import annotations

import json
import logging
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime

import httpx
import feedparser
import nats

from config import NATS_URL, OSINT_USER_AGENT
from upstream_cache import rss_cache

logger = logging.getLogger("osint.sources")


def _parse_rfc822(date_str: object) -> str | None:
    """Parse RFC-822 date string from feedparser entries."""
    if not isinstance(date_str, str):
        return None
    try:
        return parsedate_to_datetime(date_str).astimezone(timezone.utc).isoformat()
    except (ValueError, TypeError):
        return None


NATS_URLS = NATS_URL
_nc = None


async def _jetstream():
    """Reuse one NATS connection across publishes (no connect/close per event)."""
    global _nc
    if _nc is None or _nc.is_closed:
        _nc = await nats.connect(NATS_URLS)
    return _nc.jetstream()


async def publish_event(subject: str, event: dict):
    """Publish an event to NATS JetStream."""
    js = await _jetstream()
    await js.publish(subject, json.dumps(event).encode())
    logger.debug("Published event to %s", subject)


def event_dedup_key(msg: dict) -> str | None:
    """Natural key for generic events. URL when present; else None (always insert)."""
    url = msg.get("url")
    if not isinstance(url, str):
        return None
    url = url.strip()
    return url or None


async def existing_event_urls(urls: list[str]) -> set[str]:
    """URLs already claimed in event_dedup. Empty input -> empty set."""
    if not urls:
        return set()
    from sqlalchemy import select

    from database import async_session
    from models import event_dedup as event_dedup_table

    async with async_session() as session:
        result = await session.execute(
            select(event_dedup_table.c.url).where(event_dedup_table.c.url.in_(urls))
        )
        return {row[0] for row in result}


async def _publish_unknown(subject: str, events: list[dict]) -> int:
    """Publish only events whose URL is not already in event_dedup."""
    keys = [event_dedup_key(e) for e in events]
    known = await existing_event_urls([k for k in keys if k])
    published = 0
    for event, key in zip(events, keys):
        if key and key in known:
            continue
        await publish_event(subject, event)
        published += 1
    return published


def _ua_headers() -> dict[str, str]:
    return {"User-Agent": OSINT_USER_AGENT}


# ─── RSS Feed Ingestor ──────────────────────────────────────────────────

async def ingest_rss_feed(feed_url: str, source_id: str | None = None):
    """Fetch and parse an RSS feed, publish items to NATS."""
    text = rss_cache.get(feed_url)
    if text is None:
        async with httpx.AsyncClient(timeout=30, headers=_ua_headers()) as client:
            resp = await client.get(feed_url)
            resp.raise_for_status()
            text = resp.text
        rss_cache[feed_url] = text
    feed = feedparser.parse(text)

    count = 0
    for entry in feed.entries[:100]:  # max 100 per run
        event = {
            "source_type": "rss",
            "title": entry.get("title"),
            "body": entry.get("summary") or entry.get("description"),
            "url": entry.get("link"),
            "source_timestamp": _parse_rfc822(entry.get("published"))
                or datetime.now(timezone.utc).isoformat(),
            "tags": [t.get("term") for t in entry.get("tags", []) if t.get("term")],
            "raw": {
                "feed_url": feed_url,
                "feed_title": feed.feed.get("title"),
                "author": entry.get("author"),
                "categories": [c.get("term") for c in entry.get("categories", [])],
            },
        }
        await publish_event("events.rss", event)
        count += 1

    logger.info("Ingested %d items from RSS feed %s", count, feed_url)
    return count


# ─── GDELT 2.0 DOC API ──────────────────────────────────────────────────

GDELT_API = "https://api.gdeltproject.org/api/v2/doc/doc"
GDELT_DEFAULT_QUERY = '(unrest OR protest OR outage OR cyber OR "power outage")'


def gdelt_params(query: str = "", max_articles: int = 50) -> dict[str, str]:
    """DOC 2.0 query string (not the retired gdeltv2 ``search`` param)."""
    q = (query or "").strip() or GDELT_DEFAULT_QUERY
    return {
        "query": q,
        "mode": "ArtList",
        "format": "json",
        "maxrecords": str(int(max_articles)),
        "timespan": "1d",
    }


def _parse_gdelt_seendate(value: object) -> str:
    if isinstance(value, str) and len(value) >= 15:
        try:
            return datetime.strptime(value[:15], "%Y%m%dT%H%M%S").replace(
                tzinfo=timezone.utc
            ).isoformat()
        except ValueError:
            pass
    return datetime.now(timezone.utc).isoformat()


def parse_gdelt_articles(data: dict) -> list[dict]:
    events = []
    for article in data.get("articles") or []:
        if not isinstance(article, dict):
            continue
        url = article.get("url")
        if not url:
            continue
        events.append({
            "source_type": "gdel-t2",
            "title": article.get("title"),
            "body": article.get("domain") or article.get("language"),
            "url": url,
            "location_name": article.get("sourcecountry"),
            "source_timestamp": _parse_gdelt_seendate(article.get("seendate")),
            "tags": [t for t in (article.get("language"), article.get("sourcecountry")) if t],
            "raw": article,
        })
    return events


async def ingest_gdelt(query: str = "", max_articles: int = 50):
    """Fetch articles from the GDELT DOC 2.0 API."""
    params = gdelt_params(query=query, max_articles=max_articles)
    data: dict = {"articles": []}
    async with httpx.AsyncClient(
        timeout=60, headers=_ua_headers(), follow_redirects=True,
    ) as client:
        try:
            resp = await client.get(GDELT_API, params=params)
            resp.raise_for_status()
            data = resp.json()
        except (httpx.TransportError, httpx.HTTPStatusError) as exc:
            # gdeltproject.org certs have expired in the wild; HTTP fallback.
            logger.warning("GDELT HTTPS failed (%s); retrying HTTP", exc)
            http_url = GDELT_API.replace("https://", "http://", 1)
            resp = await client.get(http_url, params=params)
            resp.raise_for_status()
            data = resp.json()

    events = parse_gdelt_articles(data if isinstance(data, dict) else {})
    for event in events:
        await publish_event("events.gdelt", event)
    logger.info("Ingested %d articles from GDELT", len(events))
    return len(events)


def _parse_gdelt_tone(tone: str) -> float | None:
    """Parse GDELT tone string to a -1..1 sentiment score."""
    try:
        tone_float = float(tone)
        return max(-1.0, min(1.0, tone_float / 4249.0))  # GDELT tone range
    except (ValueError, TypeError):
        return None


# ─── Earthquake Ingestor (USGS) ─────────────────────────────────────────

USGS_API = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson"


def parse_usgs_feature(feature: dict) -> dict:
    """Map one USGS GeoJSON feature, keeping the stable event id."""
    props = feature.get("properties") or {}
    geometry = (feature.get("geometry") or {}).get("coordinates") or []
    usgs_id = feature.get("id")
    url = props.get("url") or (
        f"https://earthquake.usgs.gov/earthquakes/eventpage/{usgs_id}" if usgs_id else None
    )
    raw = dict(props)
    raw["usgs_id"] = usgs_id
    return {
        "source_type": "earthquake",
        "title": props.get("title"),
        "body": props.get("description"),
        "url": url,
        "location_lat": geometry[1] if len(geometry) > 1 else None,
        "location_lon": geometry[0] if len(geometry) > 0 else None,
        "location_name": props.get("place"),
        "sentiment_label": "neutral",
        "tags": [f"magnitude:{props.get('mag')}"] if props.get("mag") else [],
        "source_timestamp": (
            datetime.utcfromtimestamp(props.get("time", 0) / 1000)
            .replace(tzinfo=timezone.utc)
            .isoformat()
        ),
        "raw": raw,
    }


async def ingest_earthquakes():
    """Fetch recent earthquakes from USGS."""
    async with httpx.AsyncClient(timeout=30, headers=_ua_headers()) as client:
        resp = await client.get(USGS_API)
        resp.raise_for_status()
        data = resp.json()

    count = 0
    for feature in data.get("features", []):
        event = parse_usgs_feature(feature)
        await publish_event("events.earthquake", event)
        count += 1

    logger.info("Ingested %d earthquake events", count)
    return count


# ─── NASA EONET v3 ──────────────────────────────────────────────────────

EONET_API = "https://eonet.gsfc.nasa.gov/api/v3/events"


def parse_eonet_events(payload: dict) -> list[dict]:
    events = []
    for item in payload.get("events") or []:
        if not isinstance(item, dict):
            continue
        eid = item.get("id")
        geoms = item.get("geometry") or []
        point = None
        for g in geoms:
            if isinstance(g, dict) and g.get("type") == "Point":
                point = g
        if point is None:
            continue
        coords = point.get("coordinates") or []
        if len(coords) < 2:
            continue
        lon, lat = float(coords[0]), float(coords[1])
        cats = item.get("categories") or []
        tags = []
        for c in cats:
            if isinstance(c, dict) and c.get("id"):
                tags.append(str(c["id"]))
        url = item.get("link") or (f"https://eonet.gsfc.nasa.gov/api/v3/events/{eid}" if eid else None)
        ts = point.get("date") or datetime.now(timezone.utc).isoformat()
        events.append({
            "source_type": "disaster",
            "title": item.get("title"),
            "body": ", ".join(tags) if tags else None,
            "url": url,
            "location_lat": lat,
            "location_lon": lon,
            "location_name": item.get("title"),
            "tags": tags,
            "source_timestamp": ts,
            "raw": {**item, "eonet_id": eid},
        })
    return events


async def ingest_eonet():
    """Volcanoes, storms, floods, drought — gaps USGS/FIRMS don't cover."""
    async with httpx.AsyncClient(timeout=30, headers=_ua_headers()) as client:
        resp = await client.get(EONET_API, params={"status": "open", "limit": 100})
        resp.raise_for_status()
        data = resp.json()
    events = parse_eonet_events(data if isinstance(data, dict) else {})
    published = await _publish_unknown("events.disaster", events)
    logger.info("Ingested %d EONET events (%d already known)", published, len(events) - published)
    return published


# ─── CISA KEV ───────────────────────────────────────────────────────────

CISA_KEV_API = (
    "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
)


def parse_cisa_kev(payload: dict) -> list[dict]:
    events = []
    for row in payload.get("vulnerabilities") or []:
        if not isinstance(row, dict):
            continue
        cve = row.get("cveID")
        if not cve:
            continue
        title = row.get("vulnerabilityName") or cve
        vendor = row.get("vendorProject") or ""
        product = row.get("product") or ""
        events.append({
            "source_type": "disaster",
            "title": f"{cve}: {title}",
            "body": row.get("shortDescription") or f"{vendor} {product}".strip(),
            "url": f"https://nvd.nist.gov/vuln/detail/{cve}",
            "location_lat": None,
            "location_lon": None,
            "location_name": None,
            "tags": ["cisa-kev", cve, "ransomware" if row.get("knownRansomwareCampaignUse") == "Known" else None],
            "source_timestamp": row.get("dateAdded") or datetime.now(timezone.utc).isoformat(),
            "raw": {**row, "cveID": cve},
        })
        events[-1]["tags"] = [t for t in events[-1]["tags"] if t]
    return events


async def ingest_cisa_kev():
    """Exploited-in-the-wild CVEs. No fake map coords — ticker/events only."""
    async with httpx.AsyncClient(timeout=30, headers=_ua_headers(), follow_redirects=True) as client:
        resp = await client.get(CISA_KEV_API)
        resp.raise_for_status()
        data = resp.json()
    events = parse_cisa_kev(data if isinstance(data, dict) else {})
    published = await _publish_unknown("events.disaster", events)
    logger.info("Ingested %d CISA KEV rows (%d already known)", published, len(events) - published)
    return published


# ─── Social Signals (Twitter/X-like placeholder) ────────────────────────

async def ingest_social_signals(query: str = "", max_items: int = 50):
    """Placeholder for social media signal ingestion.

    In production, this would connect to Twitter API, Reddit, NewsAPI, etc.
    For now, it publishes a heartbeat to signal the pipeline is active.
    """
    event = {
        "source_type": "social",
        "title": f"Social signal scan: {query}",
        "body": f"Scanned for '{query}' — placeholder connector",
        "source_timestamp": datetime.now(timezone.utc).isoformat(),
        "tags": [query] if query else [],
        "raw": {"query": query, "max_items": max_items, "connector": "placeholder"},
    }
    await publish_event("events.social", event)
    logger.info("Social signal scan complete for '%s'", query)
    return 1