sirius/osint-dashboard
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: summarize hourly news via Nous portal into brief/ticker/map

Browse source
This commit is contained in:
Sirius DevOps 2026-08-27 23:17:45 -04:00
parent 1616413e65
commit 8f3699f133
2 changed files with 218 additions and 77 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
news/summerizer/run_news_summarizer.py
View file

@ -10,7 +10,7 @@ The loop is serial, so a slow LLM pass never overlaps the next run.
Env (all optional, 12-factor):
NEWS_SUMMARIZE_MINUTE minute of the hour to fire (default 5)
NEWS_SUMMARIZE_RUN_ON_START "1" to summarize once immediately on boot (default 1)
GEMINI_API_KEY required to do real work; unset = idle
NOUS_API_KEY optional in env; Keys UI / api_keys also works
"""
from __future__ import annotations
@ -46,10 +46,9 @@ def run_summarize() -> None:
def main() -> None:
if not os.getenv("GEMINI_API_KEY", "").strip():
if not os.getenv("NOUS_API_KEY", "").strip():
logger.warning(
"GEMINI_API_KEY not set — summarizer will idle (set it in .env and "
"recreate the service to enable)"
"NOUS_API_KEY unset in env — will read api_keys on each run; idle if both empty"
)
logger.info(
"news summarizer loop starting (minute=%s, run_on_start=%s)",

288
news/summerizer/summarizer.py
View file

@ -1,19 +1,25 @@
#!/usr/bin/env python3
"""News summarizer — LLM (Gemini) map-reduce summarization of scraped articles.
"""News summarizer — Nous map-reduce of scraped articles into brief/ticker/map.
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).
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:
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)
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_HOURS look-back window in hours (default 1)
OSINT_USER_AGENT default osint-dashboard-news-summarizer
MAP_PROMPT override map-phase prompt (uses {batch_text})
SUMMARY_PROMPT override reduce-phase prompt (uses {final_input})
NEWS_SUMMARIZE_FORCE "1" to ignore the current-UTC-hour idempotency skip
INCLUDE_FUTURES "1" to prepend live futures prices (default 0)
The futures/markets coupling from the original pipeline is gated behind
@ -30,6 +36,9 @@ from datetime import datetime
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")
@ -42,8 +51,8 @@ DB_CONFIG = {
"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()
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"))
SUMMARY_WINDOW_HOURS = int(os.getenv("SUMMARY_WINDOW_HOURS", "1"))
INCLUDE_FUTURES = os.getenv("INCLUDE_FUTURES", "0").lower() in ("1", "true", "yes")
@ -62,12 +71,15 @@ FUTURES_TICKERS = {
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.
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: name the country / city / region mentioned if determinable from the text, else "Unknown".
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 an event with geopolitical, security, military, economic, 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]".
@ -80,6 +92,9 @@ Article 1:
- Entities: ...
- Category: ...
- OSINT signal: ...
- Importance: ...
- Lat: ...
- Lon: ...
Article 2:
...
@ -89,9 +104,21 @@ Articles in this batch:
"""
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.
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, set summary_en to exactly: "No qualifying impactful or unusual events in the recent hourly news data." and use empty ticker and map_items arrays. 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.
All text in English.
Demand a single JSON object (no markdown fences) with this exact shape:
{
"summary_en": "English markdown brief or the no-qualifying-events sentence",
"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: only critical and high, max 12, ≤140 chars, no markdown.
map_items: only critical and high 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 brief for an operator HUD.
DATA:
{final_input}
@ -100,56 +127,52 @@ DATA:
# ── LLM helpers ────────────────────────────────────────────────────────────
_client = None
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 _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
def resolve_api_key() -> str:
env = os.getenv("NOUS_API_KEY", "").strip()
if env:
return env
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)
conn = psycopg2.connect(**DB_CONFIG)
try:
return _kv(conn, "api_keys", "NOUS_API_KEY")
finally:
conn.close()
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 ""
def resolve_model() -> str:
env = os.getenv("SUMMARY_MODEL", "").strip()
if env:
return env
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)
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) ────────────────────────────────────────────────
@ -206,10 +229,11 @@ def build_futures_context() -> str:
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.
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 (
@ -225,6 +249,26 @@ def ensure_tables() -> None:
summary_text TEXT NOT NULL,
batch_timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
ALTER TABLE article_summaries ADD COLUMN IF NOT EXISTS model 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)
@ -262,24 +306,90 @@ def get_recent_news() -> list[dict]:
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:
def _already_summarized_this_hour() -> bool:
"""True when article_summaries already has a row for the current UTC hour."""
if os.getenv("NEWS_SUMMARIZE_FORCE", "") == "1":
return False
query = (
"SELECT 1 FROM article_summaries "
"WHERE batch_timestamp >= date_trunc('hour', NOW() AT TIME ZONE 'utc')"
)
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
cur.execute(query)
row = cur.fetchone()
cur.close()
conn.close()
return row is not None
except Exception as exc: # noqa: BLE001
logger.error("Error checking hourly idempotency: %s", exc)
return False
def save_batch(summary_en: str, model: str, ticker: list, map_items: list) -> 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) VALUES (%s)",
(summary_text.strip(),),
"INSERT INTO article_summaries (summary_text, model) VALUES (%s, %s) RETURNING id",
(text, model),
)
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 to database successfully.")
logger.info(
"Master summary saved id=%s model=%s ticker=%d map=%d",
summary_id, model, len(ticker_rows), len(map_rows),
)
cur.close()
conn.close()
except Exception as exc: # noqa: BLE001
logger.error("Error saving summary to DB: %s", exc)
logger.error("Error saving batch to DB: %s", exc)
# ── Orchestration ──────────────────────────────────────────────────────────
@ -307,8 +417,22 @@ def build_master_prompt(final_input: str) -> str:
def summarize_news() -> None:
"""Map-reduce summarize recent articles and store the master summary."""
"""Map-reduce summarize recent articles and store brief + ticker + map."""
ensure_tables()
if _already_summarized_this_hour():
logger.info(
"Skipping summarize: article_summaries already has a row this UTC hour "
"(set NEWS_SUMMARIZE_FORCE=1 to override)"
)
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()
if not articles:
logger.info("No new articles found in the last %sh.", SUMMARY_WINDOW_HOURS)
@ -316,14 +440,23 @@ def summarize_news() -> None:
logger.info(
"Processing %d articles with %s (batch_size=%d, futures=%s)...",
len(articles), MODEL_NAME, BATCH_SIZE, INCLUDE_FUTURES,
len(articles), model, 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))
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)
@ -333,9 +466,18 @@ def summarize_news() -> None:
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)
master_raw = call_llm(
build_master_prompt(final_input),
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"])
if __name__ == "__main__":