osint-dashboard/news/summerizer/summarizer.py

#!/usr/bin/env python3
"""News summarizer — Nous map-reduce of scraped articles into brief/ticker/map.

Reads articles scraped within the last SUMMARY_WINDOW_MINUTES from the shared `articles` table,
maps them with Nous (per-article English fact blocks), reduces to one JSON
object (summary_en + ticker + map_items), and stores the brief in
`article_summaries` plus flagged rows in `news_items`. Tables live in the
EXISTING osint-db (alembic 003_news + 005_news_items, idempotent).

Everything is env-driven (12-factor). Secrets/config are resolved at the start
of each summarize_news() — env wins, else api_keys / app_settings:

  DB_HOST / DB_NAME / DB_USER / DB_PASSWORD / DB_PORT   PostgreSQL (osint-db)
  NOUS_API_KEY             Nous Portal key (else api_keys.name='NOUS_API_KEY')
  NOUS_BASE_URL            default https://inference-api.nousresearch.com/v1
  SUMMARY_MODEL            default Hermes-4.3-36B (else app_settings)
  BATCH_SIZE               articles per map-phase batch (default 50)
  SUMMARY_WINDOW_MINUTES   look-back window (default 15; SUMMARY_WINDOW_HOURS wins if set)
  NEWS_RECAP               "1" for the 23:00 daily recap (24h window, recap prompt)
  NEWS_SUMMARIZE_FORCE     "1" to ignore the interval/recap idempotency skip
  TZ                       IANA tz for recap-day bounds (default America/New_York)
  INCLUDE_FUTURES          legacy; ignored — prompts never inject futures data
"""

from __future__ import annotations

import logging
import os
from datetime import datetime
from zoneinfo import ZoneInfo

import psycopg2

from intel import parse_reduce_json, select_map, select_ticker
from nous_client import chat

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")),
}

DEFAULT_NOUS_BASE_URL = "https://inference-api.nousresearch.com/v1"
DEFAULT_SUMMARY_MODEL = "Hermes-4.3-36B"
BATCH_SIZE = int(os.getenv("BATCH_SIZE", "50"))


def _summary_window_minutes() -> int:
    hours = (os.getenv("SUMMARY_WINDOW_HOURS") or "").strip()
    if hours:
        return max(1, int(hours) * 60)
    mins = (os.getenv("SUMMARY_WINDOW_MINUTES") or "").strip()
    if mins:
        return max(1, int(mins))
    return 15


def _summarize_interval_seconds() -> int:
    return max(1, int(os.getenv("NEWS_SUMMARIZE_INTERVAL_S", "900")))


def is_recap_run() -> bool:
    return os.getenv("NEWS_RECAP", "0").lower() in ("1", "true", "yes")


def effective_window_minutes(*, recap: bool | None = None) -> int:
    if recap is None:
        recap = is_recap_run()
    if recap:
        return 24 * 60
    return _summary_window_minutes()


SUMMARY_WINDOW_MINUTES = _summary_window_minutes()
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.

Focus on breaking important news (geopolitical, military/conflict, security, disasters, major political developments). Ignore futures prices, commodity tape, ticker chatter, and routine market moves unless they themselves are the breaking event. If the batch has no critical/high stories, still extract minor incidents and crime reports.

Write every field in English. Translate if the article is not English.

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: country/city/region or Unknown. If you can estimate coordinates, emit them as numbers; otherwise omit.
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 a breaking event with geopolitical, security, military, or disaster significance, say so in one short sentence. Otherwise write: "No notable OSINT signal."
6. Importance: critical (breaking geopolitical/military/disaster with immediate impact), high, medium, low, none.

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: ...
- Importance: ...
- Lat: ...
- Lon: ...

Article 2:
...

Articles in this batch:
{batch_text}
"""

SUMMARY_PROMPT_DEFAULT = """\
You are writing an English operator HUD brief from the article facts in DATA below. Use ONLY that data. Do not invent events, names, dates, places, or implications.

Always write a real summary_en that recaps the most important stories present in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news.

Lead with critical and high breaking events. If DATA has no critical/high stories, fill the brief with minor incidents and crime reports rather than writing an empty or unfinished brief. Never truncate mid-sentence; finish every sentence. If you run out of room, drop the lowest-priority item instead of cutting a line short.

ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank.

map_items may be empty if no located critical/high event is explicit in the data.

Demand a single JSON object (no markdown fences) with this exact shape:

{
  "summary_en": "English markdown brief of the provided stories",
  "ticker": [{"headline": "", "importance": "critical", "url": "", "location_name": ""}],
  "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}]
}

ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low.
map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
summary_en: English markdown executive brief for an operator HUD (4–8 complete bullets or short paragraphs). Cover the actual stories in DATA. Complete — never an unfinished sentence.

DATA:
{final_input}
"""

RECAP_PROMPT_DEFAULT = """\
You are writing a daily recap of the last 24 hours of news for an OSINT operator HUD, using ONLY the article facts in DATA below. Do not invent events, names, dates, places, or implications.

Always write a real summary_en daily recap of the most important stories in DATA. Rank geopolitics, military/conflict, security, disasters, and major political developments first. Ignore futures prices, commodity tape, ticker chatter, and routine market data — do not treat price ticks as news.

Lead with critical and high breaking events. If DATA has no critical/high stories, fill the recap with minor incidents and crime reports rather than writing an empty or unfinished recap. Never truncate mid-sentence; finish every sentence.

ticker: prefer critical and high. If nothing is critical or high, fill ticker with medium then low incidents and crime so the HUD is not blank.

Demand a single JSON object (no markdown fences) with this exact shape:

{
  "summary_en": "English markdown daily recap of the provided stories",
  "ticker": [{"headline": "", "importance": "critical", "url": "", "location_name": ""}],
  "map_items": [{"headline": "", "importance": "critical", "location_name": "", "lat": 0, "lon": 0, "location_confidence": "city", "category": "military/conflict", "url": ""}]
}

ticker: max 12, ≤140 chars, no markdown. Rank critical > high > medium > low.
map_items: only where a real-world location is explicit in the data. Estimate lat/lon. If location is Unknown or not in the data, omit the item. Never invent a place. Max 20.
summary_en: English markdown daily recap of the last 24 hours. Complete sentences. Cover the actual stories in DATA.

DATA:
{final_input}
"""


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

def _kv(conn, table, name) -> str:
    cur = conn.cursor()
    cur.execute(f"SELECT value FROM {table} WHERE name = %s", (name,))
    row = cur.fetchone()
    return (row[0] or "").strip() if row else ""


def resolve_api_key() -> str:
    env = os.getenv("NOUS_API_KEY", "").strip()
    if env:
        return env
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        try:
            return _kv(conn, "api_keys", "NOUS_API_KEY")
        finally:
            conn.close()
    except Exception:  # noqa: BLE001
        return ""


def resolve_model() -> str:
    env = os.getenv("SUMMARY_MODEL", "").strip()
    if env:
        return env
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        try:
            value = _kv(conn, "app_settings", "SUMMARY_MODEL")
            return value or DEFAULT_SUMMARY_MODEL
        finally:
            conn.close()
    except Exception:  # noqa: BLE001
        return DEFAULT_SUMMARY_MODEL


def resolve_base_url() -> str:
    return os.getenv("NOUS_BASE_URL", DEFAULT_NOUS_BASE_URL).strip() or DEFAULT_NOUS_BASE_URL


def call_llm(prompt: str, *, api_key: str, model: str, base_url: str, json_mode: bool = False) -> str:
    """Send a prompt to Nous chat completions and return the text (\"\" on failure)."""
    if not api_key:
        logger.warning("NOUS_API_KEY not set — skipping LLM call")
        return ""
    return chat(prompt, api_key=api_key, model=model, base_url=base_url, json_mode=json_mode)


# ── 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 + 005_news_items 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()
        );
        ALTER TABLE article_summaries ADD COLUMN IF NOT EXISTS model TEXT;
        ALTER TABLE article_summaries ADD COLUMN IF NOT EXISTS kind TEXT;
        CREATE TABLE IF NOT EXISTS news_items (
            id              SERIAL PRIMARY KEY,
            summary_id      INTEGER REFERENCES article_summaries(id) ON DELETE CASCADE,
            kind            TEXT NOT NULL,
            headline        TEXT NOT NULL,
            importance      TEXT NOT NULL,
            location_name   TEXT,
            lat             DOUBLE PRECISION,
            lon             DOUBLE PRECISION,
            location_confidence TEXT,
            category        TEXT,
            url             TEXT,
            created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
        );
        CREATE INDEX IF NOT EXISTS ix_news_items_kind_created
            ON news_items (kind, created_at DESC);
        CREATE INDEX IF NOT EXISTS ix_news_items_map_bbox
            ON news_items (lon, lat)
            WHERE kind = 'map' AND lat IS NOT NULL AND lon IS NOT NULL;
    """
    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(window_minutes: int | None = None) -> list[dict]:
    """Fetch articles from the look-back window (content > 100 chars)."""
    mins = window_minutes if window_minutes is not None else effective_window_minutes()
    query = """
        SELECT title, content, url, domain
        FROM articles
        WHERE timestamp > NOW() - make_interval(mins => %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, (mins,))
        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 _already_summarized_this_interval() -> bool:
    """True when article_summaries already has a row in the last interval."""
    if os.getenv("NEWS_SUMMARIZE_FORCE", "") == "1":
        return False
    query = (
        "SELECT 1 FROM article_summaries "
        "WHERE batch_timestamp >= NOW() - make_interval(secs => %s)"
    )
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        cur.execute(query, (_summarize_interval_seconds(),))
        row = cur.fetchone()
        cur.close()
        conn.close()
        return row is not None
    except Exception as exc:  # noqa: BLE001
        logger.error("Error checking interval idempotency: %s", exc)
        return False


def _already_recapped_today() -> bool:
    """True when a daily_recap row already exists for the local calendar day."""
    if os.getenv("NEWS_SUMMARIZE_FORCE", "") == "1":
        return False
    tz_name = (os.getenv("TZ") or "America/New_York").strip() or "America/New_York"
    start = datetime.now(ZoneInfo(tz_name)).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    query = (
        "SELECT 1 FROM article_summaries "
        "WHERE kind = 'daily_recap' AND batch_timestamp >= %s"
    )
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        cur.execute(query, (start,))
        row = cur.fetchone()
        cur.close()
        conn.close()
        return row is not None
    except Exception as exc:  # noqa: BLE001
        logger.error("Error checking recap idempotency: %s", exc)
        return False


def save_batch(
    summary_en: str, model: str, ticker: list, map_items: list, *, kind: str = "interval"
) -> None:
    """Insert the master brief plus flagged ticker/map rows."""
    ticker_rows = select_ticker(ticker or [])
    map_rows = select_map(map_items or [])
    text = (summary_en or "").strip()
    if len(text) < 10 and not ticker_rows and not map_rows:
        logger.info("Summary too short or empty. Skipping save.")
        return
    insert_item = """
        INSERT INTO news_items (
            summary_id, kind, headline, importance, location_name,
            lat, lon, location_confidence, category, url
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO article_summaries (summary_text, model, kind) VALUES (%s, %s, %s) RETURNING id",
            (text, model, kind),
        )
        summary_id = cur.fetchone()[0]
        for row in ticker_rows:
            cur.execute(
                insert_item,
                (
                    summary_id,
                    "ticker",
                    row.get("headline"),
                    row.get("importance"),
                    row.get("location_name"),
                    None,
                    None,
                    None,
                    None,
                    row.get("url"),
                ),
            )
        for row in map_rows:
            cur.execute(
                insert_item,
                (
                    summary_id,
                    "map",
                    row.get("headline"),
                    row.get("importance"),
                    row.get("location_name"),
                    row.get("lat"),
                    row.get("lon"),
                    row.get("location_confidence"),
                    row.get("category"),
                    row.get("url"),
                ),
            )
        conn.commit()
        logger.info(
            "Master summary saved id=%s model=%s kind=%s ticker=%d map=%d",
            summary_id, model, kind, len(ticker_rows), len(map_rows),
        )
        cur.close()
        conn.close()
    except Exception as exc:  # noqa: BLE001
        logger.error("Error saving batch 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)
    try:
        return template.format(batch_text=batch_text)
    except KeyError:
        return template


def build_master_prompt(final_input: str, recap: bool = False) -> str:
    if recap:
        template = os.getenv("RECAP_PROMPT", RECAP_PROMPT_DEFAULT)
    else:
        template = os.getenv("SUMMARY_PROMPT", SUMMARY_PROMPT_DEFAULT)
    if "{final_input}" in template:
        return template.replace("{final_input}", final_input)
    return template


def summarize_news() -> None:
    """Map-reduce summarize recent articles and store brief + ticker + map."""
    recap = is_recap_run()
    window = effective_window_minutes(recap=recap)
    ensure_tables()
    if recap:
        if _already_recapped_today():
            logger.info(
                "Skipping recap: article_summaries already has daily_recap today "
                "(set NEWS_SUMMARIZE_FORCE=1 to override)"
            )
            return
    elif _already_summarized_this_interval():
        logger.info(
            "Skipping summarize: article_summaries already has a row in the last %ss "
            "(set NEWS_SUMMARIZE_FORCE=1 to override)",
            _summarize_interval_seconds(),
        )
        return

    api_key = resolve_api_key()
    model = resolve_model()
    base_url = resolve_base_url()
    if not api_key:
        logger.warning("NOUS_API_KEY unset in env and api_keys — idle this run")
        return

    articles = get_recent_news(window)
    if not articles:
        logger.info("No new articles found in the last %s min.", window)
        return

    logger.info(
        "Processing %d articles with %s (batch_size=%d, recap=%s, window_min=%s)...",
        len(articles), model, BATCH_SIZE, recap, window,
    )

    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),
            api_key=api_key,
            model=model,
            base_url=base_url,
            json_mode=False,
        )
        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_raw = call_llm(
        build_master_prompt(final_input, recap=recap),
        api_key=api_key,
        model=model,
        base_url=base_url,
        json_mode=True,
    )
    if not master_raw:
        logger.warning("Reduce phase returned empty — nothing to persist.")
        return
    parsed = parse_reduce_json(master_raw)
    save_batch(
        parsed["summary_en"],
        model,
        parsed["ticker"],
        parsed["map_items"],
        kind="daily_recap" if recap else "interval",
    )


if __name__ == "__main__":
    summarize_news()