osint-dashboard/news/summerizer/summarizer.py

#!/usr/bin/env python3
"""News summarizer — LLM (Gemini) map-reduce summarization of scraped articles.

Reads articles scraped within the last hour from the shared `articles` table,
summarizes them with Gemini (map phase per batch, reduce phase into one master
summary), and stores the result in `article_summaries` — both tables live in
the EXISTING osint-db (created by alembic migration 003_news, idempotent).

Everything is env-driven (12-factor):
  DB_HOST / DB_NAME / DB_USER / DB_PASSWORD / DB_PORT   PostgreSQL (osint-db)
  GEMINI_API_KEY           Google AI Studio key (required to actually run)
  SUMMARY_MODEL            Gemini model id (default gemini-2.0-flash)
  BATCH_SIZE               articles per map-phase batch (default 50)
  SUMMARY_WINDOW_HOURS     look-back window in hours (default 1)
  MAP_PROMPT               override map-phase prompt (uses {batch_text})
  SUMMARY_PROMPT           override reduce-phase prompt (uses {final_input})
  INCLUDE_FUTURES          "1" to prepend live futures prices (default 0)

The futures/markets coupling from the original pipeline is gated behind
INCLUDE_FUTURES and OFF by default — it is irrelevant to the OSINT dashboard
and pulled yfinance into the image. Re-enable by installing yfinance and
setting INCLUDE_FUTURES=1.
"""

from __future__ import annotations

import logging
import os
from datetime import datetime

import psycopg2

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("news.summarizer")

# ── Configuration (12-factor, container-friendly defaults) ─────────────────
DB_CONFIG = {
    "host": os.getenv("DB_HOST", "db").strip(),
    "database": os.getenv("DB_NAME", "osint_data").strip(),
    "user": os.getenv("DB_USER", "osint").strip(),
    "password": os.getenv("DB_PASSWORD", "").strip(),
    "port": int(os.getenv("DB_PORT", "5432")),
}

GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "").strip()
MODEL_NAME = os.getenv("SUMMARY_MODEL", "gemini-2.0-flash").strip()
BATCH_SIZE = int(os.getenv("BATCH_SIZE", "50"))
SUMMARY_WINDOW_HOURS = int(os.getenv("SUMMARY_WINDOW_HOURS", "1"))
INCLUDE_FUTURES = os.getenv("INCLUDE_FUTURES", "0").lower() in ("1", "true", "yes")

# Only touched when INCLUDE_FUTURES=1 (legacy markets coupling, OSINT-off).
FUTURES_TICKERS = {
    "Equity Indices": ["ES=F", "NQ=F", "YM=F", "RTY=F"],
    "Energy": ["CL=F", "NG=F", "HO=F", "RB=F"],
    "Metals": ["GC=F", "SI=F", "HG=F"],
    "Agriculture": ["ZC=F", "ZS=F", "ZW=F", "ZL=F", "KE=F"],
    "Currencies": ["6E=F", "6J=F", "6B=F"],
}

# ── OSINT-neutral default prompts (env-overridable via MAP_PROMPT/SUMMARY_PROMPT) ──

MAP_PROMPT_DEFAULT = """\
You are a precise, factual OSINT news processor. Your ONLY source of information is the articles provided below. Do NOT add external knowledge, assumptions, training data, or invented facts.

For EACH article in the batch:
1. Extract 2-4 key factual bullet points (who, what, when, where, numbers, quotes — stay very close to the text).
2. Location: name the country / city / region mentioned if determinable from the text, else "Unknown".
3. Entities: list the key people, organizations, or governments mentioned (comma-separated, only names present in the text), else "None".
4. Category: pick one — politics, military/conflict, economy, technology, environment/disaster, health, crime, society, sport, other.
5. OSINT signal: if the article describes an event with geopolitical, security, military, economic, or disaster significance, say so in one short sentence. Otherwise write: "No notable OSINT signal."

If several articles cover the same story, add one short batch-level note at the end: "Batch theme: [one sentence]".

Output format — strictly one block per article:

Article 1:
- Fact bullet 1
- Fact bullet 2
- Location: ...
- Entities: ...
- Category: ...
- OSINT signal: ...

Article 2:
...

Articles in this batch:
{batch_text}
"""

SUMMARY_PROMPT_DEFAULT = """\
CRITICAL INSTRUCTION - REPEAT 3 TIMES: YOU MUST USE ONLY THE DATA PROVIDED BELOW. DO NOT INVENT, RECALL, OR ADD ANY EVENTS, NAMES, DATES, IMPLICATIONS, PROJECTS, OR DETAILS NOT EXPLICITLY PRESENT IN THE DATA. IF THE DATA HAS NO MAJOR GEOPOLITICAL/TECH/MILITARY/ECONOMIC/IMPACTFUL EVENTS OR UNUSUAL STORIES, OUTPUT ONLY: "No qualifying impactful or unusual events in the recent hourly news data." AND STOP. NO EXTERNAL KNOWLEDGE FROM TRAINING.

Write a concise executive summary of the most impactful items as a short markdown list, one line per story, using only the data.

DATA:
{final_input}
"""


# ── LLM helpers ────────────────────────────────────────────────────────────

_client = None


def _get_client():
    """Lazily build the Gemini client (avoids import/init when key unset)."""
    global _client
    if _client is None:
        from google import genai

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


def _extract_text(resp) -> str:
    """Defensively pull text out of the google-genai GenerateContentResponse.

    The modern SDK returns the response directly (``resp.text``); some older
    wrappers exposed it as ``resp.response``. Handle both plus a candidates
    fallback so a provider/SDK change degrades to "" instead of crashing.
    """
    if not resp:
        return ""
    if hasattr(resp, "text") and resp.text:
        return resp.text
    inner = getattr(resp, "response", None)
    if inner is not None and hasattr(inner, "text") and inner.text:
        return inner.text
    try:
        parts = []
        for cand in getattr(resp, "candidates", None) or []:
            content = getattr(cand, "content", None)
            for part in getattr(content, "parts", None) or []:
                if getattr(part, "text", None):
                    parts.append(part.text)
        return "\n".join(parts)
    except Exception:  # noqa: BLE001
        return str(resp)


def call_llm(prompt: str) -> str:
    """Send a prompt to Gemini and return the text ("" on any failure)."""
    if not GEMINI_API_KEY:
        logger.warning("GEMINI_API_KEY not set — skipping LLM call")
        return ""
    try:
        resp = _get_client().models.generate_content(model=MODEL_NAME, contents=prompt)
        return _extract_text(resp)
    except Exception as exc:  # noqa: BLE001
        logger.error("Gemini API error: %s", exc)
        return ""


# ── Futures (legacy, gated) ────────────────────────────────────────────────

def fetch_current_futures_prices() -> dict:
    """Live futures prices. Only meaningful when INCLUDE_FUTURES=1."""
    if not INCLUDE_FUTURES:
        return {}
    try:
        import yfinance as yf  # noqa: PLC0415
    except ImportError:
        logger.warning(
            "INCLUDE_FUTURES=1 but yfinance is not installed — install it to enable futures prices"
        )
        return {}

    prices: dict = {}
    for category, tickers in FUTURES_TICKERS.items():
        for ticker in tickers:
            try:
                data = yf.Ticker(ticker).history(period="1d", interval="1m")
                if not data.empty:
                    last_price = data["Close"].iloc[-1]
                    prices[ticker] = {
                        "price": round(last_price, 2),
                        "change_pct": round(
                            (last_price - data["Open"].iloc[0]) / data["Open"].iloc[0] * 100, 2
                        ) if len(data) > 1 else 0,
                        "timestamp": datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC"),
                        "category": category,
                    }
                else:
                    prices[ticker] = {"price": None, "error": "No data"}
            except Exception as exc:  # noqa: BLE001
                prices[ticker] = {"price": None, "error": str(exc)}
    return prices


def build_futures_context() -> str:
    ctx = f"CURRENT FUTURES PRICES (as of {datetime.now().strftime('%Y-%m-%d %H:%M UTC')}):\n"
    for ticker, info in fetch_current_futures_prices().items():
        if info.get("price") is not None:
            ctx += (
                f"- {ticker} ({info['category']}): ${info['price']:.2f}  "
                f"({info['change_pct']:+.2f}% today)\n"
            )
        else:
            ctx += f"- {ticker}: unavailable ({info.get('error', 'unknown error')})\n"
    return ctx


# ── DB helpers ─────────────────────────────────────────────────────────────

def ensure_tables() -> None:
    """Idempotently create the news tables if missing.

    Normally created by alembic 003_news when the app container starts, but
    this summarizer may boot before the app has run migrations (compose only
    guarantees `db` is up, not that alembic has run). Mirrors the scraper
    pipeline's own CREATE TABLE IF NOT EXISTS so either start order is safe.
    """
    ddl = """
        CREATE TABLE IF NOT EXISTS articles (
            id SERIAL PRIMARY KEY,
            title TEXT,
            url TEXT UNIQUE,
            content TEXT,
            domain TEXT,
            timestamp TIMESTAMPTZ
        );
        CREATE TABLE IF NOT EXISTS article_summaries (
            id SERIAL PRIMARY KEY,
            summary_text TEXT NOT NULL,
            batch_timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
        );
    """
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        cur.execute(ddl)
        conn.commit()
        cur.close()
        conn.close()
    except Exception as exc:  # noqa: BLE001
        logger.error("Error ensuring news tables: %s", exc)


def get_recent_news() -> list[dict]:
    """Fetch articles from the last SUMMARY_WINDOW_HOURS (content > 100 chars)."""
    query = """
        SELECT title, content, url, domain
        FROM articles
        WHERE timestamp > NOW() - make_interval(hours => %s)
          AND content IS NOT NULL AND length(content) > 100
        ORDER BY timestamp DESC;
    """
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        cur.execute(query, (SUMMARY_WINDOW_HOURS,))
        rows = cur.fetchall()
        cur.close()
        conn.close()
        return [
            {"title": r[0], "content": r[1], "url": r[2], "domain": r[3]}
            for r in rows
        ]
    except Exception as exc:  # noqa: BLE001
        logger.error("Database error reading articles: %s", exc)
        return []


def save_summary_to_db(summary_text: str) -> None:
    """Insert one master summary row (table created by alembic 003_news)."""
    if not summary_text or len(summary_text.strip()) < 10:
        logger.info("Summary too short or empty. Skipping save.")
        return
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO article_summaries (summary_text) VALUES (%s)",
            (summary_text.strip(),),
        )
        conn.commit()
        logger.info("Master summary saved to database successfully.")
        cur.close()
        conn.close()
    except Exception as exc:  # noqa: BLE001
        logger.error("Error saving summary to DB: %s", exc)


# ── Orchestration ──────────────────────────────────────────────────────────

def build_map_prompt(batch: list[dict]) -> str:
    batch_text = "\n\n".join(
        f"Title: {a['title']}\nSource: {a['domain']}\nURL: {a['url']}\nContent: {a['content'][:1500]}"
        for a in batch
    )
    template = os.getenv("MAP_PROMPT", MAP_PROMPT_DEFAULT)
    prefix = build_futures_context() + "\n" if INCLUDE_FUTURES else ""
    try:
        return prefix + template.format(batch_text=batch_text)
    except KeyError:
        return prefix + template


def build_master_prompt(final_input: str) -> str:
    template = os.getenv("SUMMARY_PROMPT", SUMMARY_PROMPT_DEFAULT)
    prefix = build_futures_context() + "\n" if INCLUDE_FUTURES else ""
    try:
        return prefix + template.format(final_input=final_input)
    except KeyError:
        return prefix + template


def summarize_news() -> None:
    """Map-reduce summarize recent articles and store the master summary."""
    ensure_tables()
    articles = get_recent_news()
    if not articles:
        logger.info("No new articles found in the last %sh.", SUMMARY_WINDOW_HOURS)
        return

    logger.info(
        "Processing %d articles with %s (batch_size=%d, futures=%s)...",
        len(articles), MODEL_NAME, BATCH_SIZE, INCLUDE_FUTURES,
    )

    partial_summaries: list[str] = []
    for i in range(0, len(articles), BATCH_SIZE):
        batch = articles[i : i + BATCH_SIZE]
        logger.info("map batch %d/%d (%d articles)", i // BATCH_SIZE + 1, -(-len(articles) // BATCH_SIZE), len(batch))
        summary = call_llm(build_map_prompt(batch))
        if summary:
            partial_summaries.append(summary)

    final_input = "\n\n".join(partial_summaries)
    if not final_input.strip():
        logger.warning("No partial summaries produced — nothing to reduce.")
        return

    logger.info("reduce phase over %d partial summaries", len(partial_summaries))
    master_summary = call_llm(build_master_prompt(final_input))
    if master_summary:
        save_summary_to_db(master_summary)


if __name__ == "__main__":
    summarize_news()