Compare commits

...

14 commits

Author SHA1 Message Date
f1d70d3f4c Merge pull request 'feat: Nous Portal news summarizer, ticker flags, map pins' (#8) from feat/nous-news-summarizer into master
All checks were successful
build-and-deploy / build-push-deploy (push) Successful in 3m27s
Reviewed-on: #8
2026-08-28 17:43:40 -04:00
Sirius DevOps
d2bf55183c fix: let Settings model reach summarizer; cap map pins at 20 2026-08-28 17:06:55 -04:00
Sirius DevOps
450de45bfc docs: nous portal news summarizer 2026-08-28 17:06:55 -04:00
Sirius DevOps
496d13b8b0 feat: critical news pins on the osint map 2026-08-28 17:06:55 -04:00
Sirius DevOps
3b4dd49253 feat: news ticker shows flagged hourly items 2026-08-28 17:06:55 -04:00
Sirius DevOps
610666ef6a feat: settings model selector for news summarizer 2026-08-28 17:06:55 -04:00
Sirius DevOps
d93ad88956 chore: point news-summarizer compose at Nous portal 2026-08-28 17:06:55 -04:00
Sirius DevOps
8f3699f133 feat: summarize hourly news via Nous portal into brief/ticker/map 2026-08-28 17:06:55 -04:00
Sirius DevOps
1616413e65 feat: persist summarizer model in app_settings 2026-08-28 17:06:55 -04:00
Sirius DevOps
f50268ef81 feat: register NOUS_API_KEY in dashboard keystore 2026-08-28 17:06:55 -04:00
Sirius DevOps
5b23578302 feat: news ticker and map API contracts 2026-08-28 17:06:55 -04:00
Sirius DevOps
cbc573b830 feat: add news_items table and summary model column 2026-08-28 17:06:55 -04:00
Sirius DevOps
4b00eed566 feat: nous portal chat client for news summarizer 2026-08-28 17:06:55 -04:00
Sirius DevOps
96590bc16d test: add news summarizer intel parser tests + impl 2026-08-28 17:06:55 -04:00
19 changed files with 1666 additions and 169 deletions

View file

@ -60,7 +60,7 @@ FIRMS_INTERVAL=900
INGEST_FIRES=1
# ── API keys (managed from the dashboard UI) ──────────────────────────────
# Keys such as GEMINI_API_KEY and TELEGRAM_TOKEN are stored in the Postgres
# Keys such as NOUS_API_KEY and TELEGRAM_TOKEN are stored in the Postgres
# `api_keys` table and managed from the dashboard's "Keys" tab
# (GET/POST/DELETE /api/keys/{name}) — see app/keystore.py. The FIRMS ingestor
# currently reads FIRMS_MAP_KEY from .env (above); wiring the Keys-UI store as
@ -68,13 +68,18 @@ INGEST_FIRES=1
# ── News pipeline (scraper + summarizer, profile `ingest`) ────────────────
# Hourly: the scraper crawls 257 RSS sources at minute :00 and the summarizer
# runs the Gemini map-reduce at minute :05, both writing to the shared osint-db
# runs the Nous map-reduce at minute :05, both writing to the shared osint-db
# (tables `articles` + `article_summaries`, created by alembic 003_news).
# Consume via GET /api/news and GET /api/news/summaries.
# GEMINI_API_KEY is REQUIRED for summarization; unset = summarizer idles.
GEMINI_API_KEY=
# NOUS_API_KEY is also (preferably) set in the Keys UI; env is an override.
# Unset in both env and api_keys = summarizer logs and idles (never crashes).
NOUS_API_KEY=
NOUS_BASE_URL=https://inference-api.nousresearch.com/v1
# Optional LLM knobs
SUMMARY_MODEL=gemini-2.0-flash
# SUMMARY_MODEL is an optional override. Leave unset so Settings
# (app_settings.SUMMARY_MODEL) can reach the summarizer. Code default
# Hermes-4.3-36B remains after a Postgres miss. Env wins when set.
# SUMMARY_MODEL=
NEWS_BATCH_SIZE=50
SUMMARY_WINDOW_HOURS=1
# Futures/markets coupling from the upstream pipeline is OFF by default
@ -87,6 +92,8 @@ NEWS_SUMMARIZE_MINUTE=5
# scheduled minute.
NEWS_SCRAPE_RUN_ON_START=1
NEWS_SUMMARIZE_RUN_ON_START=1
# "1" ignores the current-UTC-hour idempotency skip (double-pins on recreate).
NEWS_SUMMARIZE_FORCE=0
NEWS_LOG_LEVEL=INFO
# Reserved for the (out-of-scope) Telegram delivery bot.
TELEGRAM_TOKEN=

View file

@ -0,0 +1,64 @@
"""news_items table + article_summaries.model
Revision ID: 005_news_items
Revises: 004_camera_enum
Create Date: 2026-08-28
"""
from alembic import op
# revision identifiers, used by Alembic.
revision = "005_news_items"
down_revision = "004_camera_enum"
branch_labels = None
depends_on = None
def upgrade() -> None:
# Idempotent DDL: ingest (summarizer ensure_tables) may create the same
# shapes first depending on container startup order. IF NOT EXISTS makes
# both orders safe — whichever runs first wins, the other no-ops.
# One statement per op.execute: asyncpg rejects multi-command prepared
# statements (same style as 003_news).
op.execute(
"""
ALTER TABLE article_summaries
ADD COLUMN IF NOT EXISTS model TEXT
"""
)
op.execute(
"""
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()
)
"""
)
op.execute(
"""
CREATE INDEX IF NOT EXISTS ix_news_items_kind_created
ON news_items (kind, created_at DESC)
"""
)
op.execute(
"""
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
"""
)
def downgrade() -> None:
op.execute("DROP TABLE IF EXISTS news_items")
op.execute("ALTER TABLE article_summaries DROP COLUMN IF EXISTS model")

View file

@ -57,10 +57,10 @@ KEY_REGISTRY: dict[str, dict] = {
"pattern": r"^[0-9a-fA-F]{32}$",
"example": "32-char hex string (e.g. 5f3c…9a02)",
},
"GEMINI_API_KEY": {
"description": "Google Gemini API key — LLM event analysis / summarization.",
"pattern": r"^AIza[0-9A-Za-z_-]{35}$",
"example": "AIza… (Google API key, 39 chars)",
"NOUS_API_KEY": {
"description": "Nous Portal API key — hourly news summarizer (inference-api.nousresearch.com).",
"pattern": r"^.{16,}$",
"example": "key from https://portal.nousresearch.com (API keys page)",
},
"TELEGRAM_TOKEN": {
"description": "Telegram bot token — push alert notifications to a channel.",
@ -214,7 +214,7 @@ async def get_api_key(name: str) -> str | None:
"""Read a stored key value — used by ingest services, never by the API.
Returns the raw value (or None when unset) so producers can pass it to
external APIs (FIRMS, Gemini, Telegram, ). Reads live from Postgres, so a
external APIs (FIRMS, Nous, Telegram, ). Reads live from Postgres, so a
key set via the dashboard is picked up on the next poll no restart.
"""
await ensure_api_keys_table()

View file

@ -31,14 +31,16 @@ from sqlalchemy.ext.asyncio import AsyncSession
from database import async_session, init_extensions
from models import (
alerts, documents, entities, entity_events, events, feed_sources, fires,
articles, article_summaries,
articles, article_summaries, news_items,
)
from schemas import (
AlertCreate, AlertOut, AlertSeverity, AlertType, AlertUpdate,
DashboardSummary, EntityCreate, EntityKind, EntityOut,
EventCreate, EventOut, FireOut, NewsArticleOut, NewsSummaryOut,
EventCreate, EventOut, FireOut, NewsArticleOut, NewsMapItemOut,
NewsSummaryOut, NewsTickerItemOut,
FeedSourceCreate, FeedSourceOut,
KeyOut, KeyValueIn,
NewsModelsOut, SettingsIn, SettingsOut,
SearchResult, SentimentSummary, SourceType,
SearchQuery, TimelinePoint, VesselBboxUpdate,
GeofenceCreate, GeofenceUpdate,
@ -47,6 +49,7 @@ from ingestor import ingest_event, fetch_and_process
from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
from fire_sources import ingest_fires
from keystore import KeyFormatError, delete_key, list_keys, set_key
from settings_store import SettingsError, get_app_settings, list_models, set_summary_model
from live_layers import (
fetch_aircraft, fetch_fire_incidents, fetch_fire_perimeters,
fetch_radar_meta, fetch_storms, fetch_trains, fetch_vessels,
@ -602,7 +605,7 @@ async def list_documents(
async def list_api_keys():
"""List known API keys with set/missing status — masked, never raw.
Registered keys (FIRMS_MAP_KEY, GEMINI_API_KEY, TELEGRAM_TOKEN)
Registered keys (FIRMS_MAP_KEY, NOUS_API_KEY, TELEGRAM_TOKEN)
are always included. Any extra stored keys are appended.
"""
return await list_keys()
@ -633,6 +636,21 @@ async def remove_api_key(name: str):
return {"ok": True}
@app.get("/api/settings", response_model=SettingsOut)
async def get_settings():
"""Summarizer model + read-only Nous base URL."""
return await get_app_settings()
@app.put("/api/settings", response_model=SettingsOut)
async def put_settings(payload: SettingsIn):
"""Persist SUMMARY_MODEL. ``nous_base_url`` is ignored even if sent."""
try:
return await set_summary_model(payload.summary_model)
except SettingsError as exc:
raise HTTPException(status_code=422, detail=str(exc))
# ── Ingestion Triggers ───────────────────────────────────────────────────
@app.post("/api/ingest/rss")
@ -1128,12 +1146,113 @@ async def list_news_summaries(
return [
NewsSummaryOut(
id=r["id"], summary_text=r["summary_text"],
batch_timestamp=r["batch_timestamp"],
batch_timestamp=r["batch_timestamp"], model=r["model"],
)
for r in rows
]
_FLAGGED = ("critical", "high")
@app.get("/api/news/ticker", response_model=list[NewsTickerItemOut])
async def list_news_ticker(
since: datetime | None = Query(
None,
description="Only ticker items created at/after this UTC instant.",
),
limit: int = Query(20, ge=1, le=50),
):
"""Flagged ticker rows (critical/high), newest first. No LLM required."""
async with async_session() as session:
stmt = (
select(news_items)
.where(
news_items.c.kind == "ticker",
news_items.c.importance.in_(_FLAGGED),
)
.order_by(news_items.c.created_at.desc())
)
if since:
stmt = stmt.where(news_items.c.created_at >= since)
stmt = stmt.limit(limit)
rows = (await session.execute(stmt)).mappings().all()
return [
NewsTickerItemOut(
id=r["id"], headline=r["headline"], importance=r["importance"],
location_name=r["location_name"], url=r["url"],
created_at=r["created_at"],
)
for r in rows
]
@app.get("/api/news/map", response_model=list[NewsMapItemOut])
async def list_news_map(
bbox: str | None = Query(
None,
description="Comma-separated 'minlon,minlat,maxlon,maxlat' to bound the "
"result set by item coordinates. Omit for all flagged pins.",
),
since: datetime | None = Query(
None,
description="Only map items created at/after this UTC instant. "
"Defaults to the last 24 hours.",
),
limit: int = Query(200, ge=1, le=500),
):
"""Flagged map pins (critical/high with coords). No zoom skip — world view."""
if since is None:
since = datetime.now(timezone.utc) - timedelta(hours=24)
async with async_session() as session:
stmt = (
select(news_items)
.where(
news_items.c.kind == "map",
news_items.c.lat.isnot(None),
news_items.c.lon.isnot(None),
news_items.c.importance.in_(_FLAGGED),
news_items.c.created_at >= since,
)
.order_by(news_items.c.created_at.desc())
)
if bbox:
parts = [p.strip() for p in bbox.split(",")]
if len(parts) != 4:
raise HTTPException(
422, "bbox must be 'minlon,minlat,maxlon,maxlat' (4 comma-separated values)"
)
try:
minlon, minlat, maxlon, maxlat = (float(p) for p in parts)
except ValueError:
raise HTTPException(
422, "bbox values must be floats: 'minlon,minlat,maxlon,maxlat'"
)
stmt = stmt.where(
and_(
news_items.c.lon >= minlon, news_items.c.lon <= maxlon,
news_items.c.lat >= minlat, news_items.c.lat <= maxlat,
)
)
stmt = stmt.limit(limit)
rows = (await session.execute(stmt)).mappings().all()
return [
NewsMapItemOut(
id=r["id"], headline=r["headline"], importance=r["importance"],
location_name=r["location_name"], lat=r["lat"], lon=r["lon"],
location_confidence=r["location_confidence"], category=r["category"],
url=r["url"], created_at=r["created_at"],
)
for r in rows
]
@app.get("/api/news/models", response_model=NewsModelsOut)
async def list_news_models():
"""Nous model catalog for the summarizer selector. Never 502s."""
return await list_models()
# ── Frontend ──────────────────────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse)

View file

@ -185,8 +185,8 @@ Index("ix_fires_bbox", fires.c.longitude, fires.c.latitude)
# ── News pipeline (scraper + summarizer) ──────────────────────────────────
# Written by the vendored news-scraper (Scrapy) / news-summarizer (Gemini)
# services; schema must match the idempotent alembic migration 003_news.
# Written by the vendored news-scraper (Scrapy) / news-summarizer services;
# schema must match the idempotent alembic migrations 003_news + 005_news_items.
articles = Table(
"articles",
@ -209,6 +209,26 @@ article_summaries = Table(
Column("summary_text", Text, nullable=False),
Column("batch_timestamp", DateTime(timezone=True),
server_default=func.now(), nullable=False),
Column("model", Text), # LLM id used for this batch; nullable for old rows
)
Index("ix_article_summaries_batch_timestamp", article_summaries.c.batch_timestamp)
news_items = Table(
"news_items",
metadata,
Column("id", Integer, primary_key=True, autoincrement=True),
Column("summary_id", Integer),
Column("kind", Text, nullable=False),
Column("headline", Text, nullable=False),
Column("importance", Text, nullable=False),
Column("location_name", Text),
Column("lat", Float),
Column("lon", Float),
Column("location_confidence", Text),
Column("category", Text),
Column("url", Text),
Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
)
Index("ix_news_items_kind_created", news_items.c.kind, news_items.c.created_at)

View file

@ -7,7 +7,7 @@ from enum import Enum
from typing import Optional
from uuid import UUID
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
# ─── Enums ───────────────────────────────────────────────────────────────
@ -273,6 +273,69 @@ class NewsSummaryOut(BaseModel):
id: int
summary_text: str
batch_timestamp: datetime
model: Optional[str] = None
class NewsTickerItemOut(BaseModel):
"""One flagged ticker row as exposed by GET /api/news/ticker."""
id: int
headline: str
importance: str
location_name: Optional[str] = None
url: Optional[str] = None
created_at: datetime
class NewsMapItemOut(BaseModel):
"""One flagged map pin as exposed by GET /api/news/map."""
id: int
headline: str
importance: str
location_name: Optional[str] = None
lat: float
lon: float
location_confidence: Optional[str] = None
category: Optional[str] = None
url: Optional[str] = None
created_at: datetime
class NewsModelId(BaseModel):
"""One model id as exposed by GET /api/news/models."""
id: str
class NewsModelsOut(BaseModel):
"""Catalog for the summarizer model selector."""
source: str
models: list[NewsModelId]
class SettingsIn(BaseModel):
"""Body for PUT /api/settings. ``nous_base_url`` is not writable."""
summary_model: str = Field(..., min_length=1, max_length=128)
@field_validator("summary_model")
@classmethod
def summary_model_not_blank(cls, v: str) -> str:
stripped = v.strip()
if not stripped:
raise ValueError("summary_model must be 1128 chars, not whitespace-only")
if len(stripped) > 128:
raise ValueError("summary_model must be 1128 chars, not whitespace-only")
return stripped
class SettingsOut(BaseModel):
"""Current summarizer settings. ``nous_base_url`` is read-only."""
summary_model: str
nous_base_url: str
# ─── Aggregations ────────────────────────────────────────────────────────

219
app/settings_store.py Normal file
View file

@ -0,0 +1,219 @@
"""OSINT Dashboard — non-secret app settings (keyv-style Postgres table).
Model choice lives here so the summarizer container can read it from Postgres.
Only whitelisted names are stored this is not a generic dump.
Storage: the table is created lazily with ``CREATE TABLE IF NOT EXISTS`` on
first use in each process (same bootstrap pattern as ``api_keys``).
"""
from __future__ import annotations
import asyncio
import os
import time
from datetime import datetime, timezone
import httpx
from sqlalchemy import Column, DateTime, String, Table, Text, func, select, text
import keystore
from database import async_session, engine, metadata
DEFAULT_NOUS_BASE_URL = "https://inference-api.nousresearch.com/v1"
DEFAULT_SUMMARY_MODEL = "Hermes-4.3-36B"
SETTING_SUMMARY_MODEL = "SUMMARY_MODEL"
ALLOWED_SETTINGS = frozenset({SETTING_SUMMARY_MODEL})
MODELS_CACHE_TTL_S = 600.0
MODELS_TIMEOUT_S = 8.0
DEFAULT_MODELS_USER_AGENT = "osint-dashboard-news-summarizer"
FALLBACK_MODELS = [
"Hermes-4.3-36B",
"Hermes-4-70B",
"google/gemini-2.5-flash",
"anthropic/claude-haiku-4.5",
"openai/gpt-4.1-mini",
"x-ai/grok-4",
]
app_settings = Table(
"app_settings",
metadata,
Column("name", String(128), primary_key=True),
Column("value", Text, nullable=False),
Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
)
_CREATE_TABLE_SQL = text(
"""
CREATE TABLE IF NOT EXISTS app_settings (
name VARCHAR(128) PRIMARY KEY,
value TEXT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
"""
)
_ensure_lock = asyncio.Lock()
_ensured = False
_models_cache: tuple[float, dict] | None = None
class SettingsError(ValueError):
"""Raised when a setting name or value fails validation."""
async def ensure_app_settings_table() -> None:
"""Create the app_settings table if it doesn't exist (idempotent, per process)."""
global _ensured
if _ensured:
return
async with _ensure_lock:
if _ensured:
return
async with engine.begin() as conn:
await conn.execute(_CREATE_TABLE_SQL)
_ensured = True
def nous_base_url() -> str:
"""Read-only Nous inference base URL (env, never writable from the UI)."""
raw = (os.getenv("NOUS_BASE_URL") or "").strip().rstrip("/")
return raw or DEFAULT_NOUS_BASE_URL
def _validate_summary_model(value: str) -> str:
stripped = (value or "").strip()
if not stripped or len(stripped) > 128:
raise SettingsError("summary_model must be 1128 chars, not whitespace-only")
return stripped
async def get_summary_model() -> str:
"""Stored SUMMARY_MODEL, else env, else Hermes-4.3-36B."""
await ensure_app_settings_table()
async with async_session() as session:
row = (
await session.execute(
select(app_settings).where(app_settings.c.name == SETTING_SUMMARY_MODEL)
)
).mappings().one_or_none()
if row and row["value"]:
return row["value"]
return os.getenv("SUMMARY_MODEL", DEFAULT_SUMMARY_MODEL)
async def set_summary_model(value: str) -> dict:
"""Upsert SUMMARY_MODEL and return the public settings payload."""
value = _validate_summary_model(value)
now = datetime.now(timezone.utc)
await ensure_app_settings_table()
async with async_session() as session:
existing = (
await session.execute(
select(app_settings).where(app_settings.c.name == SETTING_SUMMARY_MODEL)
)
).mappings().one_or_none()
if existing:
await session.execute(
app_settings.update()
.where(app_settings.c.name == SETTING_SUMMARY_MODEL)
.values(value=value, updated_at=now)
)
else:
await session.execute(
app_settings.insert().values(
name=SETTING_SUMMARY_MODEL, value=value, updated_at=now
)
)
await session.commit()
return await get_app_settings()
async def get_app_settings() -> dict:
return {
"summary_model": await get_summary_model(),
"nous_base_url": nous_base_url(),
}
def _fallback_payload() -> dict:
return {
"source": "fallback",
"models": [{"id": mid} for mid in FALLBACK_MODELS],
}
async def _nous_api_key() -> str | None:
"""Keystore first, then env. Any lookup failure is treated as missing."""
try:
stored = await keystore.get_api_key("NOUS_API_KEY")
except Exception:
stored = None
if stored and str(stored).strip():
return str(stored).strip()
env = (os.getenv("NOUS_API_KEY") or "").strip()
return env or None
def _models_user_agent() -> str:
return os.getenv("OSINT_USER_AGENT") or DEFAULT_MODELS_USER_AGENT
def _parse_models_payload(body: object) -> list[dict[str, str]]:
if isinstance(body, dict):
raw = body.get("data", body.get("models", []))
elif isinstance(body, list):
raw = body
else:
raw = []
out: list[dict[str, str]] = []
for item in raw or []:
if isinstance(item, str) and item.strip():
out.append({"id": item.strip()})
elif isinstance(item, dict):
mid = item.get("id") or item.get("name")
if mid:
out.append({"id": str(mid)})
return out
async def _http_get(url: str, *, headers: dict[str, str], timeout: float) -> httpx.Response:
async with httpx.AsyncClient(timeout=timeout, headers=headers) as client:
return await client.get(url)
async def list_models() -> dict:
"""Live ``GET {base}/models`` when a key is present; otherwise curated fallback.
Never raises to the caller for missing key or upstream failure.
"""
global _models_cache
key = await _nous_api_key()
if not key:
return _fallback_payload()
now = time.monotonic()
hit = _models_cache
if hit and now - hit[0] < MODELS_CACHE_TTL_S:
return hit[1]
url = f"{nous_base_url()}/models"
headers = {
"Authorization": f"Bearer {key}",
"User-Agent": _models_user_agent(),
"Accept": "application/json",
}
try:
resp = await _http_get(url, headers=headers, timeout=MODELS_TIMEOUT_S)
resp.raise_for_status()
models = _parse_models_payload(resp.json())
if not models:
return _fallback_payload()
payload = {"source": "live", "models": models}
_models_cache = (now, payload)
return payload
except Exception:
return _fallback_payload()

View file

@ -264,6 +264,7 @@
.hud-chip .c.fires { background: #fb923c; box-shadow: 0 0 6px #fb923c; }
.hud-chip .c.cams { background: #4ade80; box-shadow: 0 0 6px #4ade80; }
.hud-chip .c.blips { background: var(--magenta); box-shadow: 0 0 6px var(--magenta); }
.hud-chip .c.news { background: var(--magenta); box-shadow: 0 0 6px var(--magenta); border-radius: 1px; transform: rotate(45deg); }
.hud-chip.off { opacity: 0.4; }
#dvr-bar {
position: absolute; left: 50%; bottom: 12px; transform: translateX(-50%);
@ -319,6 +320,7 @@
.lp-dot.fires { background: #fb923c; box-shadow: 0 0 7px #fb923c; }
.lp-dot.cams { background: #4ade80; box-shadow: 0 0 7px #4ade80; }
.lp-dot.blips { background: var(--magenta); box-shadow: 0 0 7px var(--magenta); }
.lp-dot.news { background: var(--magenta); box-shadow: 0 0 7px var(--magenta); border-radius: 1px; transform: rotate(45deg); }
.lp-dot.weather { background: var(--amber); box-shadow: 0 0 7px var(--amber); }
.lp-dot.flights { background: #7dd3fc; box-shadow: 0 0 7px #7dd3fc; }
.lp-dot.vessels { background: #2dd4bf; box-shadow: 0 0 7px #2dd4bf; }
@ -539,11 +541,14 @@
.tick-item .dom { font-family: 'Share Tech Mono', monospace; font-size: 0.64rem; color: var(--cyan); background: rgba(53,224,255,0.08); border: 1px solid rgba(53,224,255,0.22); border-radius: 3px; padding: 0.05rem 0.4rem; }
.tick-item .tt { font-family: 'Share Tech Mono', monospace; font-size: 0.66rem; color: var(--muted); }
.tick-item .sep { color: var(--line-hi); }
button.tick-item { cursor: pointer; background: transparent; border: 0; font: inherit; color: inherit; }
.tick-item.brief {
cursor: pointer; background: linear-gradient(90deg, rgba(255,46,151,0.12), transparent 70%);
border-left: 2px solid var(--magenta); font-weight: 600;
}
.tick-item.brief .b-tag { font-family: 'Share Tech Mono', monospace; font-size: 0.6rem; color: #1c0311; background: var(--magenta); border-radius: 2px; padding: 0.08rem 0.4rem; letter-spacing: 0.1em; }
.tick-item .b-tag { font-family: 'Share Tech Mono', monospace; font-size: 0.6rem; color: #1c0311; background: var(--magenta); border-radius: 2px; padding: 0.08rem 0.4rem; letter-spacing: 0.1em; }
.tick-item .b-tag.critical { background: var(--red); }
.tick-item .b-tag.high { background: var(--amber); }
.tick-item.brief:hover { color: var(--magenta); }
.tick-item.standby { color: var(--muted); opacity: 0.75; }
.tick-item.standby .price { color: var(--muted); }
@ -721,6 +726,13 @@
</div>
<div class="lp-note">Geolocated events from the ingest pipeline. Color = source type.</div>
</div>
<div class="lp-layer">
<div class="lp-row">
<label class="lp-name"><input type="checkbox" id="lp-news-on" checked onchange="toggleNewsPins()"> <span class="lp-dot news"></span> Critical News</label>
<span class="lp-count" id="lp-news-count">0</span>
</div>
<div class="lp-note">LLM-estimated locations from the hourly brief. Pins only for critical/high.</div>
</div>
<div class="lp-layer">
<div class="lp-row">
@ -801,6 +813,7 @@
<div class="hud-chip" id="hud-fires"><span class="c fires"></span>FIRMS <b id="hud-fires-n"></b></div>
<div class="hud-chip" id="hud-cams"><span class="c cams"></span>CAMS <b id="hud-cams-n"></b></div>
<div class="hud-chip" id="hud-blips"><span class="c blips"></span>BLIPS <b id="hud-blips-n"></b></div>
<div class="hud-chip" id="hud-news"><span class="c news"></span>NEWS <b id="hud-news-n"></b></div>
</div>
<div id="dvr-bar" class="live">
<button class="btn" id="dvr-live" type="button" onclick="dvrGoLive()">LIVE</button>
@ -827,7 +840,7 @@
<div class="ns-kicker">Executive Summary <span class="ns-badge" id="ns-badge">WAITING</span></div>
<div class="ns-time" id="ns-time"></div>
<div class="ns-body" id="ns-body">
<div class="ns-empty">No executive summary yet — the Gemini summarizer is idle until <code>GEMINI_API_KEY</code> is set on the Pi.</div>
<div class="ns-empty">No executive summary yet — the Nous summarizer is idle until <code>NOUS_API_KEY</code> is set on the Pi and a model is chosen in Settings.</div>
</div>
</div>
@ -955,7 +968,7 @@
<section class="view" id="view-keys" aria-label="API keys">
<div class="subview">
<h2><span class="tick"></span> API Keys</h2>
<p class="sub">Keys used by ingest services (FIRMS, Gemini, Telegram, …). Stored in Postgres, never shown again — only status + last 4 chars.</p>
<p class="sub">Keys used by ingest services (FIRMS, Nous Portal, Telegram, AIS, …). Stored in Postgres, never shown again — only status + last 4 chars.</p>
<div class="key-grid" id="keys-grid"></div>
</div>
</section>
@ -979,6 +992,30 @@
</div>
<div class="set-saved" id="mkt-saved"></div>
</div>
<div class="panel">
<h3>News Summarizer</h3>
<p class="set-note" style="margin-bottom:0.9rem">Takes effect on the next hourly run (:05). No container restart.</p>
<div class="set-row">
<label>Provider</label>
<span>Nous Portal (inference-api.nousresearch.com)</span>
</div>
<div class="set-row">
<label>API key</label>
<span>set in API Keys as NOUS_API_KEY (never shown here)</span>
</div>
<div class="set-row">
<label for="set-summary-model">Model</label>
<select id="set-summary-model"></select>
</div>
<div class="set-row">
<label>Current</label>
<span id="set-summary-model-now"></span>
</div>
<div class="set-actions">
<button class="btn" onclick="saveSummaryModel()">Save model</button>
</div>
<div class="set-saved" id="sum-saved"></div>
</div>
<div class="panel">
<h3>Map Defaults</h3>
<p class="set-note" style="margin-bottom:0.9rem">Applied to the live map on load.</p>
@ -1025,7 +1062,7 @@
<div class="sys-row"><span class="k">Active sources</span><span class="v" id="sys-sources"></span></div>
<div class="sys-row"><span class="k">Open alerts</span><span class="v" id="sys-alerts"></span></div>
<div class="sys-row"><span class="k">Tracked entities</span><span class="v" id="sys-entities"></span></div>
<div class="sys-row"><span class="k">News cycle</span><span class="v">15 min</span></div>
<div class="sys-row"><span class="k">News cycle</span><span class="v">hourly :05</span></div>
<div class="sys-row"><span class="k">Market feed</span><span class="v" id="sys-market">STANDBY</span></div>
</div>
<div class="set-actions">
@ -1051,7 +1088,7 @@
<div class="tick-wrap">
<div class="tick-track" id="nt-track"></div>
</div>
<div class="tick-tag" id="nt-tag">15-MIN CYCLE</div>
<div class="tick-tag" id="nt-tag">HOURLY</div>
</div>
</footer>
@ -1177,11 +1214,12 @@ function showView(name) {
if (firesOn) loadFires();
if (camsOn) loadCams();
if (blipsOn) loadBlips();
if (newsOn) loadNewsPins();
}
}
if (name === 'news') loadNews(false);
if (name === 'events') { loadSummary(); loadEvents(); }
if (name === 'settings') { loadSummary(); renderSysInfo(); }
if (name === 'settings') { loadSummary(); renderSysInfo(); loadSettings(); }
window.scrollTo(0, 0);
}
@ -1299,13 +1337,16 @@ async function loadNews(force) {
if (upd) upd.textContent = 'updating…';
let articles = [];
let summaries = [];
let tickerItems = [];
try {
const [rArticles, rSumm] = await Promise.all([
const [rArticles, rSumm, rTick] = await Promise.all([
fetch(`${API}/api/news?limit=100`),
fetch(`${API}/api/news/summaries?limit=1`)
fetch(`${API}/api/news/summaries?limit=1`),
fetch(`${API}/api/news/ticker?limit=20`),
]);
if (rArticles.ok) articles = await rArticles.json();
if (rSumm.ok) summaries = await rSumm.json();
if (rTick.ok) tickerItems = await rTick.json();
} catch (e) {
const list = document.getElementById('news-list');
const body = document.getElementById('ns-body');
@ -1317,7 +1358,7 @@ async function loadNews(force) {
}
renderNewsSummary(summaries);
renderNewsList(articles);
renderNewsTicker(articles, summaries);
renderNewsTicker(articles, summaries, tickerItems);
if (upd) upd.textContent = 'updated ' + new Date().toLocaleTimeString();
if (!newsInterval) {
newsInterval = setInterval(() => loadNews(false), NEWS_REFRESH_MS);
@ -1329,7 +1370,7 @@ function renderNewsSummary(summaries) {
const badgeEl = document.getElementById('ns-badge');
if (!bodyEl) return;
if (!summaries || !summaries.length) {
bodyEl.innerHTML = '<div class="ns-empty">No executive summary yet — the Gemini summarizer is idle until <code>GEMINI_API_KEY</code> is set on the Pi.</div>';
bodyEl.innerHTML = '<div class="ns-empty">No executive summary yet — the Nous summarizer is idle until <code>NOUS_API_KEY</code> is set on the Pi and a model is chosen in Settings.</div>';
timeEl.textContent = '—';
badgeEl.textContent = 'WAITING';
return;
@ -1357,25 +1398,45 @@ function renderNewsList(articles) {
'</a>';
}).join('');
}
function renderNewsTicker(articles, summaries) {
function renderNewsTicker(articles, summaries, tickerItems) {
const track = document.getElementById('nt-track');
if (!track) return;
let items = '';
// LLM executive summary flash first (click → News view)
const ticks = Array.isArray(tickerItems) ? tickerItems : [];
if (ticks.length) {
ticks.forEach(t => {
const imp = String(t.importance || '').toLowerCase();
const tag = imp === 'critical' ? 'CRITICAL' : 'HIGH';
const loc = (t.location_name || '').trim();
const inner =
`<span class="b-tag ${imp === 'critical' ? 'critical' : 'high'}">${tag}</span>` +
`<span>${newsEsc((t.headline || '').trim())}</span>` +
(loc ? `<span class="dom">${newsEsc(loc)}</span>` : '') +
`<span class="tt">${newsTimeAgo(t.created_at)}</span><span class="sep"></span>`;
const url = (t.url || '').trim();
if (url) {
items += `<a class="tick-item" href="${newsEsc(url)}" target="_blank" rel="noopener">${inner}</a>`;
} else {
items += `<button class="tick-item" onclick="showView('news')" title="Open news brief">${inner}</button>`;
}
});
} else {
// Quiet hour: brief chip + article titles so the dock is never blank.
const summ = summaries && summaries.length ? summaries[0] : null;
if (summ && summ.summary_text) {
const brief = summ.summary_text.replace(/\s+/g, ' ').trim();
const briefSrc = summ && (summ.summary_en || summ.summary_text);
if (briefSrc) {
const brief = String(briefSrc).replace(/\s+/g, ' ').trim();
items += `<button class="tick-item brief" onclick="showView('news')" title="Open full brief">` +
`<span class="b-tag">LLM BRIEF</span><span>${newsEsc(brief.slice(0, 200))}${brief.length > 200 ? '…' : ''}</span>` +
`<span class="b-tag">LLM BRIEF</span><span>${newsEsc(brief.slice(0, 120))}${brief.length > 120 ? '…' : ''}</span>` +
`<span class="tt">${newsTimeAgo(summ.batch_timestamp)}</span><span class="sep"></span></button>`;
}
// Top headlines
(articles || []).slice(0, 30).forEach(a => {
(articles || []).slice(0, 10).forEach(a => {
items += `<a class="tick-item" href="${newsEsc(a.url || '#')}" target="_blank" rel="noopener">` +
`<span class="dom">${newsEsc(a.domain || '?')}</span>` +
`<span>${newsEsc((a.title || '').trim())}</span>` +
`<span class="tt">${newsTimeAgo(a.timestamp)}</span><span class="sep"></span></a>`;
});
}
if (!items) {
items = `<span class="tick-item standby">No headlines yet — scraper idle</span>`;
}
@ -1486,7 +1547,7 @@ function sentimentBadge(label) {
/* ═══════════════ API KEYS ═══════════════ */
const KEY_PATTERNS = {
'FIRMS_MAP_KEY': /^[0-9a-fA-F]{32}$/,
'GEMINI_API_KEY': /^AIza[0-9A-Za-z_-]{35}$/,
'NOUS_API_KEY': /^.{16,}$/,
'TELEGRAM_TOKEN': /^\d{8,10}:[0-9A-Za-z_-]{35}$/,
};
async function loadKeys() {
@ -1657,12 +1718,62 @@ function initSettings() {
}
} catch (e) {}
}
async function loadSettings() {
const sel = document.getElementById('set-summary-model');
const now = document.getElementById('set-summary-model-now');
try {
const [setR, modR] = await Promise.all([
fetch(`${API}/api/settings`),
fetch(`${API}/api/news/models`),
]);
const settings = setR.ok ? await setR.json() : {};
const catalog = modR.ok ? await modR.json() : {models: []};
const saved = (settings.summary_model || '').trim();
const ids = (catalog.models || []).map(m => m && m.id).filter(Boolean);
if (saved && ids.indexOf(saved) === -1) ids.unshift(saved);
if (sel) {
sel.innerHTML = ids.map(id => `<option value="${esc(id)}">${esc(id)}</option>`).join('');
if (saved) sel.value = saved;
}
if (now) now.textContent = saved || '—';
} catch (e) {
console.error('Settings load failed', e);
}
}
async function saveSummaryModel() {
const sel = document.getElementById('set-summary-model');
const flash = document.getElementById('sum-saved');
const now = document.getElementById('set-summary-model-now');
const summary_model = ((sel && sel.value) || '').trim();
if (!summary_model) {
if (flash) flash.textContent = 'Choose a model to save.';
return;
}
try {
const r = await fetch(`${API}/api/settings`, {
method: 'PUT',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({summary_model}),
});
const d = await r.json().catch(() => ({}));
if (!r.ok) {
if (flash) flash.textContent = (typeof d.detail === 'string' ? d.detail : 'Save failed');
return;
}
if (now) now.textContent = d.summary_model || summary_model;
if (sel && d.summary_model) sel.value = d.summary_model;
if (flash) flash.textContent = 'Model saved.';
} catch (e) {
if (flash) flash.textContent = 'Network error.';
}
}
/* ═══════════════ MAP (NASA GIBS basemap + FIRMS + cameras + blips) ═══════
The satellite map core is production-proven — preserved verbatim. */
let map = null, mapLayer = null, mapInitStarted = false;
let firesHeat = null, camsGroup = null, firesOn = false, camsOn = false;
let blipsGroup = null, blipsOn = false, blipsSince = '24h';
let newsGroup = null, newsOn = false, newsReq = 0;
let firesOpacity = 0.65, firesColor = 'brightness', firesSince = '24h';
let camsOpacity = 1.0, baseOpacity = 1.0;
let camReq = 0, fireReq = 0, blipReq = 0; // stale-response guards (rapid zoom races)
@ -1855,6 +1966,7 @@ async function initMap() {
if (firesOn) loadFires();
if (camsOn) loadCams();
if (blipsOn) loadBlips();
if (newsOn) loadNewsPins();
refreshLiveOverlays();
sendLiveViewport();
}, 500);
@ -1866,6 +1978,7 @@ async function initMap() {
firesOn = document.getElementById('lp-fires-on').checked;
camsOn = document.getElementById('lp-cams-on').checked;
blipsOn = document.getElementById('lp-blips-on').checked;
newsOn = document.getElementById('lp-news-on').checked;
firesSince = document.getElementById('lp-fires-since').value;
blipsSince = document.getElementById('lp-blips-since').value;
radarOn = document.getElementById('lp-radar-on').checked;
@ -1880,6 +1993,7 @@ async function initMap() {
if (firesOn) loadFires();
if (camsOn) loadCams();
if (blipsOn) loadBlips();
if (newsOn) loadNewsPins();
refreshLiveOverlays();
connectLiveWs();
} catch(e) {
@ -1902,8 +2016,9 @@ function syncHud() {
set('fires', firesOn ? (hudFiresCount) : '—', firesOn);
set('cams', camsOn ? (hudCamsCount) : '—', camsOn);
set('blips', blipsOn ? (hudBlipsCount) : '—', blipsOn);
set('news', newsOn ? (hudNewsCount) : '—', newsOn);
}
let hudFiresCount = null, hudCamsCount = null, hudBlipsCount = null;
let hudFiresCount = null, hudCamsCount = null, hudBlipsCount = null, hudNewsCount = null;
function setMapAttribution(text) {
if (!map) return;
@ -2318,6 +2433,57 @@ async function loadBlips() {
}
}
/* ── CRITICAL NEWS (LLM-estimated locations, critical/high only) ── */
async function toggleNewsPins() {
newsOn = document.getElementById('lp-news-on').checked;
if (newsOn) {
await loadNewsPins();
return;
}
newsReq++;
if (newsGroup && map) { map.removeLayer(newsGroup); newsGroup = null; }
hudNewsCount = null;
const countEl = document.getElementById('lp-news-count');
if (countEl) countEl.textContent = '0';
syncHud();
}
async function loadNewsPins() {
if (!map) return;
const req = ++newsReq;
try {
const url = `${API}/api/news/map?bbox=${currentBBox()}&limit=200`;
const r = await overlayFetch(url);
const data = await r.json();
if (req !== newsReq) return; // superseded by a newer pan/zoom
const items = Array.isArray(data) ? data : [];
if (newsGroup) map.removeLayer(newsGroup);
const col = '#ff2e97';
const pins = items.filter(it => Number.isFinite(Number(it.lat)) && Number.isFinite(Number(it.lon)));
newsGroup = L.layerGroup(pins.map(it => {
const icon = L.divIcon({
className: '',
html: `<span style="display:inline-block;width:10px;height:10px;background:${col};box-shadow:0 0 8px ${col},0 0 2px #0f172a;border:1px solid #030609;transform:rotate(45deg);"></span>`,
iconSize: [14, 14], iconAnchor: [7, 7],
});
return L.marker([it.lat, it.lon], { icon })
.bindPopup(`<div class="blip-pop">` +
`<div class="blip-src">◈ ${esc(it.importance || '')}${it.category ? ' · ' + esc(it.category) : ''}</div>` +
`<b>${esc(it.headline || '')}</b>` +
`<div class="blip-time">${esc(it.location_name || '')}</div>` +
(it.url ? `<a href="${esc(it.url)}" target="_blank" rel="noopener">source ↗</a>` : '') +
`</div>`);
}));
newsGroup.addTo(map);
document.getElementById('lp-news-count').textContent = pins.length.toLocaleString();
hudNewsCount = pins.length;
syncHud();
} catch(e) {
if (isAbort(e)) return;
document.getElementById('map-hint').textContent = `News pins load failed: ${e.message || e}`;
console.error('News pins load failed', e);
}
}
/* ── Live overlays (radar / alerts / WFIGS / aircraft / trains / AIS / NHC) ── */
function refreshLiveOverlays() {
if (!map) return;

View file

@ -224,14 +224,16 @@ services:
DB_HOST: db
DB_PORT: ${DB_PORT:-5432}
DB_NAME: ${DB_NAME:-osint_data}
# Required to do real work; unset → the loop logs and idles.
GEMINI_API_KEY: ${GEMINI_API_KEY:-}
SUMMARY_MODEL: ${SUMMARY_MODEL:-gemini-2.0-flash}
NOUS_API_KEY: ${NOUS_API_KEY:-}
NOUS_BASE_URL: ${NOUS_BASE_URL:-https://inference-api.nousresearch.com/v1}
SUMMARY_MODEL: ${SUMMARY_MODEL:-}
OSINT_USER_AGENT: ${OSINT_USER_AGENT:-osint-dashboard-news-summarizer}
BATCH_SIZE: ${NEWS_BATCH_SIZE:-50}
SUMMARY_WINDOW_HOURS: ${SUMMARY_WINDOW_HOURS:-1}
INCLUDE_FUTURES: ${INCLUDE_FUTURES:-0}
NEWS_SUMMARIZE_MINUTE: ${NEWS_SUMMARIZE_MINUTE:-5}
NEWS_SUMMARIZE_RUN_ON_START: ${NEWS_SUMMARIZE_RUN_ON_START:-1}
NEWS_SUMMARIZE_FORCE: ${NEWS_SUMMARIZE_FORCE:-0}
command: ["python", "run_news_summarizer.py"]
volumes:

View file

@ -1,10 +1,11 @@
# News pipeline — scraper + summarizer
# News pipeline — scraper + Nous Portal summarizer
The OSINT dashboard ingests ~257 global news RSS sources hourly and produces
LLM master summaries. Both services were vendored from the upstream
`~/Projects/newsPipeline` project and re-integrated here to replace the old
k8s CronJob choreography with in-compose scheduling against the EXISTING
osint-db — **no second Postgres**.
an English LLM brief plus flagged ticker/map rows. Both services were vendored
from the upstream `~/Projects/newsPipeline` project and re-integrated here to
replace the old k8s CronJob choreography with in-compose scheduling against
the EXISTING osint-db — **no second Postgres**. The LLM is **Nous Portal**
(`inference-api.nousresearch.com`) — not Gemini.
## Architecture
@ -15,10 +16,11 @@ osint-db — **no second Postgres**.
news-scraper (Scrapy, hourly :00) ──► articles table (osint-db)
│ │
│ ▼
news-summarizer (Gemini map-reduce, hourly :05) ──► article_summaries table
news-summarizer (Nous Portal map-reduce, :05) ──► article_summaries + news_items
GET /api/news · GET /api/news/summaries
GET /api/news · /api/news/summaries · /api/news/ticker · /api/news/map
GET /api/news/models · GET/PUT /api/settings
```
| Component | Image | Container | Scheduling |
@ -29,6 +31,11 @@ news-summarizer (Gemini map-reduce, hourly :05) ──► article_summaries tabl
Both services live under the `ingest` compose profile (same as the ingester
and camera-scraper): `docker compose --profile ingest up -d`.
The summarizer is a batch sidecar, **not** a live overlay. Do **not** reuse
`GET /api/alerts` (dashboard entity/keyword alerts). Do **not** stuff news
into `overlay_catalog()``/api/map/layers` `overlays` stays live upstream
feeds (`GET /api/news` exact key set is unchanged on purpose).
## Data flow
1. **Scraper**`news/scraper/run_news_scraper.py` runs
@ -39,29 +46,52 @@ and camera-scraper): `docker compose --profile ingest up -d`.
(`ON CONFLICT (url) DO NOTHING`).
2. **Summarizer**`news/summerizer/run_news_summarizer.py` runs
`summarizer.py` at :05 past each hour. It reads articles from the last
`SUMMARY_WINDOW_HOURS`, map-reduces them through Gemini
(`SUMMARY_MODEL`, default `gemini-2.0-flash`), and inserts one master
summary into `article_summaries`.
`SUMMARY_WINDOW_HOURS`, map-reduces them through Nous Portal
(`SUMMARY_MODEL` / Settings, default `Hermes-4.3-36B`), writes the English
brief to `article_summaries` (column `model` is the LLM id), and flagged
ticker/map rows to `news_items`.
Scheduling is done with small in-compose wall-clock loops (not host cron): each
loop runs once on boot (`*_RUN_ON_START=1`, seeds data fast) then sleeps until
the next scheduled minute. The loop is serial, so a run that overruns its slot
simply shifts to the next boundary — two crawls/summaries never overlap.
Hour-truncation idempotency: if `article_summaries` already has a row for the
current UTC hour, the summarizer **skips** (prevents double-pins on
`RUN_ON_START` recreate). Set `NEWS_SUMMARIZE_FORCE=1` to ignore that skip.
The `articles` and `article_summaries` tables are created by the idempotent
alembic migration `003_news` (also created by the scraper's own
`CREATE TABLE IF NOT EXISTS`, so container startup order doesn't matter).
`CREATE TABLE IF NOT EXISTS`). `news_items` is alembic `005_news_items`.
Container startup order doesn't matter.
## Keys and Settings
- **`NOUS_API_KEY`** — paste in the dashboard **Keys** UI (`api_keys` /
`keystore.KEY_REGISTRY`). Env / `.env` is an **override** (env wins, same
as FIRMS). Never returned by any API; never emitted into `index.html`;
never proxied from the browser.
- **Idle without a key** — if env is unset **and** the keystore row is empty,
the summarizer logs and idles (never crashes). News intel APIs return `[]`.
- **Model** — non-secret. Settings UI model selector `PUT /api/settings`
`{ "summary_model": "…" }` stores `SUMMARY_MODEL` in `app_settings` (1128
chars). `GET /api/settings` echoes `{summary_model, nous_base_url}`.
`nous_base_url` is read-only. Default `Hermes-4.3-36B`. Live catalog is
best-effort `GET /api/news/models`.
## Endpoints
### GET /api/news — recent articles
Key set **unchanged** (no `lat`/`lon` on articles; geo lives on `/api/news/map`).
| Query param | Meaning | Default |
|---|---|---|
| `domain` | filter by source domain (e.g. `www.reuters.com`) | none |
| `since` | only articles captured at/after this UTC instant (ISO-8601) | none |
| `limit` | max rows | `50` (max `500`) |
| `offset` | pagination offset | `0` |
| `include_content` | include full article body | `false` |
```json
[
@ -69,14 +99,14 @@ alembic migration `003_news` (also created by the scraper's own
"id": 1,
"title": "…",
"url": "https://…",
"content": "full extracted article text…",
"content": null,
"domain": "www.reuters.com",
"timestamp": "2026-08-24T18:10:00Z"
}
]
```
### GET /api/news/summaries — master LLM summaries
### GET /api/news/summaries — master LLM briefs
| Query param | Meaning | Default |
|---|---|---|
@ -88,52 +118,178 @@ alembic migration `003_news` (also created by the scraper's own
[
{
"id": 1,
"summary_text": "master LLM summary (markdown)…",
"batch_timestamp": "2026-08-24T18:10:00Z"
"summary_text": "English markdown brief…",
"batch_timestamp": "2026-08-24T18:10:00Z",
"model": "Hermes-4.3-36B"
}
]
```
`model` is additive. Empty DB → `[]` (no crash).
### GET /api/news/ticker — flagged HUD headlines
Critical/high `news_items` with `kind=ticker` only. Do **not** reuse
`GET /api/alerts`. Bottom HUD `#nt-track` scrolls these rows, not a dump of
the whole brief.
| Query param | Meaning | Default |
|---|---|---|
| `since` | only items created at/after this UTC instant | none |
| `limit` | max rows | `20` (max `50`) |
```json
[
{
"id": 1,
"headline": "…",
"importance": "critical",
"location_name": "Kyiv",
"url": "https://…",
"created_at": "2026-08-24T18:10:00Z"
}
]
```
### GET /api/news/map — geolocated critical/high pins
Only rows with valid `lat`/`lon`. Optional bbox. **No zoom skip** — world
view is the point. Layer-panel toggle uses this dedicated path (same as
event blips), not `overlay_catalog`.
| Query param | Meaning | Default |
|---|---|---|
| `bbox` | `minlon,minlat,maxlon,maxlat` | all flagged pins |
| `since` | only items created at/after this UTC instant | last 24 hours |
| `limit` | max rows | `200` (max `500`) |
Malformed bbox → `422`.
```json
[
{
"id": 1,
"headline": "…",
"importance": "high",
"location_name": "Kyiv",
"lat": 50.45,
"lon": 30.52,
"location_confidence": "city",
"category": "military/conflict",
"url": "https://…",
"created_at": "2026-08-24T18:10:00Z"
}
]
```
Pins are LLM-estimated and clamped (`lat∈[-90,90]`, `lon∈[-180,180]`). No
Nominatim. No writes into `events`.
### GET /api/news/models — Settings dropdown catalog
Never 502s. `{ "source": "live"|"fallback", "models": [{"id": "…"}] }`.
### GET /api/settings · PUT /api/settings
```json
{ "summary_model": "Hermes-4.3-36B", "nous_base_url": "https://inference-api.nousresearch.com/v1" }
```
PUT body is `{ "summary_model": "<1128 char id>" }`. `nous_base_url` is
ignored even if sent.
## Reduce JSON contract
Reduce phase (`response_format: json_object`, English only) must be a single
object. Parser (`intel.parse_reduce_json`) strips `<think>…</think>` and
markdown json fences, then brace-slices:
```json
{
"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": ""
}
]
}
```
Persist ticker/map only for `importance` in `critical`/`high`. Map rows also
need valid coords; Unknown / invented places are dropped. Caps: 12 ticker
(≤140 chars, no markdown), 20 map. Empty ticker is allowed. `summary_en`
lands in `article_summaries.summary_text`.
## Configuration (all via env / `.env`)
| Var | Default | Notes |
|---|---|---|
| `GEMINI_API_KEY` | *(blank)* | **Required for summaries.** Unset = summarizer logs and idles (never crashes). |
| `SUMMARY_MODEL` | `gemini-2.0-flash` | Gemini model id. |
| `NEWS_BATCH_SIZE` | `50` | Articles per map-phase batch. |
| `NOUS_API_KEY` | *(blank)* | **Required for summaries.** Prefer Keys UI; env overrides. Unset in **both** env and `api_keys` = summarizer logs and idles (never crashes); APIs return `[]`. |
| `NOUS_BASE_URL` | `https://inference-api.nousresearch.com/v1` | Read-only in Settings. |
| `SUMMARY_MODEL` | `Hermes-4.3-36B` | Compose default. Operator-facing choice is Settings → `app_settings.SUMMARY_MODEL`. |
| `NEWS_BATCH_SIZE` | `50` | Articles per map-phase batch (compose maps to container `BATCH_SIZE`). |
| `SUMMARY_WINDOW_HOURS` | `1` | How far back the summarizer looks for new articles. |
| `INCLUDE_FUTURES` | `0` | Legacy futures-prices coupling (upstream pipeline). OFF for OSINT; set `1` + install `yfinance` to enable. |
| `NEWS_SCRAPE_MINUTE` | `0` | Wall-clock minute the scraper fires. |
| `NEWS_SUMMARIZE_MINUTE` | `5` | Wall-clock minute the summarizer fires. |
| `NEWS_SCRAPE_RUN_ON_START` | `1` | Run one scrape immediately on container start. |
| `NEWS_SUMMARIZE_RUN_ON_START` | `1` | Run one summarize immediately on container start. |
| `NEWS_SUMMARIZE_FORCE` | `0` | `1` ignores the current-UTC-hour idempotency skip (double-pins on recreate). |
| `NEWS_LOG_LEVEL` | `INFO` | Scrapy log level. |
| `OSINT_USER_AGENT` | `osint-dashboard-news-summarizer` | Sent on every outbound Nous call. |
| `TELEGRAM_TOKEN` / `TELEGRAM_CHAT_ID` | *(blank)* | Reserved for the (out-of-scope) Telegram delivery bot. |
DB_* for both services is mapped to the shared osint-db credentials
(`DB_HOST=db`, same `DB_USER/DB_PASSWORD/DB_NAME` as the rest of the stack).
Nous chat: `POST {NOUS_BASE_URL}/chat/completions` via `news/summerizer/nous_client.py`
(`httpx`, no `openai` SDK). Auth is a Bearer token from `NOUS_API_KEY`.
No Hermes-4 reasoning system prompt. Reduce uses `json_mode=True`.
## Prompts
Both prompts are env-overridable — the default `MAP_PROMPT` is OSINT-neutral
(facts, locations, entities, category, OSINT signal per article) and the default
`SUMMARY_PROMPT` produces a concise executive summary of the most impactful
items (with a "no qualifying events" escape hatch). Upstream's futures/markets
prompt language is gated behind `INCLUDE_FUTURES=1`.
(facts, locations, entities, category, OSINT signal per article; English) and
the default `SUMMARY_PROMPT` demands the reduce JSON above (with a
"no qualifying events" escape hatch). Upstream's futures/markets prompt
language is gated behind `INCLUDE_FUTURES=1`.
## Tests
`tests/test_api_news.py` — DB-backed API contract tests (auto-skip without a
reachable test database, same as the FIRMS tests):
```bash
DB_HOST=... DB_PORT=... DB_USER=osint DB_PASSWORD=... DB_NAME=osint_data \
pytest tests/test_api_news.py -v
PYTHONPATH=news/summerizer pytest news/summerizer/tests -v
# intel + nous_client tests PASS (no network)
PYTHONPATH=app pytest tests/test_api_news.py \
tests/test_api_settings.py tests/test_api_live_layers.py -v
# DB-marked tests skip without Postgres; live_layers must still PASS
# /api/map/layers overlays key set UNCHANGED
```
## Live verification
End-to-end (real crawl → DB → API) is verified after deploy on the Pi: check
`docker compose --profile ingest logs -f news-scraper news-summarizer`, then
`curl -s localhost:8000/api/news | head`. Summaries additionally require
`GEMINI_API_KEY` to be set in `.env` on the Pi.
After deploy / compose rebuild of `news-summarizer` on the Pi:
1. Keys UI: save `NOUS_API_KEY` → status `****last4`.
2. Settings: pick a model → Save → `GET /api/settings` echoes it.
3. `docker compose --profile ingest logs -f news-summarizer` — next run (or
`NEWS_SUMMARIZE_RUN_ON_START=1` recreate) logs `Processing N articles with <model>`.
4. `curl -s localhost:8000/api/news/summaries?limit=1` — English `summary_text`, `model` set.
5. `curl -s localhost:8000/api/news/ticker` — flagged headlines only.
6. `curl -s localhost:8000/api/news/map` — only rows with lat/lon.
7. HUD: NEWS ticker scrolls flagged items; map overlay pins popup with location.
8. Unset key + empty keystore → summarizer logs idle, APIs return `[]`, no crash.
**Operator action after merge:** paste a Nous Portal API key in API Keys; pick
a model in Settings if the default `Hermes-4.3-36B` is not wanted; rebuild
`osint-news-summarizer` on the Pi (`pi-app-deploy` / compose).

93
news/summerizer/intel.py Normal file
View file

@ -0,0 +1,93 @@
"""Pure parser for the news-summarizer reduce JSON / geo / importance contract."""
from __future__ import annotations
import json
import re
_EMPTY = {"summary_en": "", "ticker": [], "map_items": []}
_KEEP = frozenset({"critical", "high"})
_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
_FENCE_RE = re.compile(r"```(?:json)?", re.IGNORECASE)
TICKER_HEADLINE_MAX = 140
MAP_HEADLINE_MAX = 160
TICKER_CAP = 12
MAP_CAP = 20
def parse_reduce_json(raw: str) -> dict:
try:
text = _THINK_RE.sub("", raw or "")
text = _FENCE_RE.sub("", text)
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1 or end < start:
return dict(_EMPTY)
data = json.loads(text[start : end + 1])
if not isinstance(data, dict):
return dict(_EMPTY)
summary = data.get("summary_en", "")
ticker = data.get("ticker", [])
map_items = data.get("map_items", [])
return {
"summary_en": summary if isinstance(summary, str) else "",
"ticker": ticker if isinstance(ticker, list) else [],
"map_items": map_items if isinstance(map_items, list) else [],
}
except Exception:
return dict(_EMPTY)
def clamp_coords(lat, lon) -> tuple[float, float] | None:
try:
lat_f = float(lat)
lon_f = float(lon)
except (TypeError, ValueError):
return None
if not (-90 <= lat_f <= 90 and -180 <= lon_f <= 180):
return None
return (lat_f, lon_f)
def _trimmed_headline(row: dict, limit: int) -> str:
headline = row.get("headline") or ""
if not isinstance(headline, str):
headline = str(headline)
return headline.strip()[:limit]
def select_ticker(rows: list) -> list:
out = []
for row in rows:
if row.get("importance") not in _KEEP:
continue
headline = _trimmed_headline(row, TICKER_HEADLINE_MAX)
if not headline:
continue
item = dict(row)
item["headline"] = headline
out.append(item)
if len(out) >= TICKER_CAP:
break
return out
def select_map(items: list) -> list:
out = []
for row in items:
if row.get("importance") not in _KEEP:
continue
headline = _trimmed_headline(row, MAP_HEADLINE_MAX)
if not headline:
continue
coords = clamp_coords(row.get("lat"), row.get("lon"))
if coords is None:
continue
item = dict(row)
item["headline"] = headline
item["lat"], item["lon"] = coords
out.append(item)
if len(out) >= MAP_CAP:
break
return out

View file

@ -0,0 +1,36 @@
"""HTTP client for the Nous inference chat completions API."""
from __future__ import annotations
import os
import httpx
_DEFAULT_UA = "osint-dashboard-news-summarizer"
_DEFAULT_BASE = "https://inference-api.nousresearch.com/v1"
def chat(prompt, *, api_key, model, base_url, json_mode=False) -> str:
resolved = (base_url or os.environ.get("NOUS_BASE_URL", _DEFAULT_BASE)).rstrip("/")
url = f"{resolved}/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"User-Agent": os.environ.get("OSINT_USER_AGENT") or _DEFAULT_UA,
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2,
"max_tokens": 4096,
}
if json_mode:
payload["response_format"] = {"type": "json_object"}
try:
with httpx.Client(timeout=60.0) as client:
resp = client.post(url, headers=headers, json=payload)
if resp.status_code == 401 or resp.status_code >= 500:
return ""
data = resp.json()
return data["choices"][0]["message"]["content"]
except Exception:
return ""

View file

@ -1,6 +1,2 @@
# Database adapter for PostgreSQL (shared osint-db).
psycopg2-binary==2.9.11
# Gemini LLM SDK (google-genai). yfinance is NOT a dependency: futures prices
# are gated behind INCLUDE_FUTURES=1 and lazy-imported (install it to enable).
google-genai
httpx==0.28.1

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

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__":

View file

@ -0,0 +1,51 @@
from intel import parse_reduce_json, clamp_coords, select_ticker, select_map
FENCED = """```json
{"summary_en": "Brief.", "ticker": [
{"headline": "Blast in Kyiv", "importance": "critical", "url": "https://ex", "location_name": "Kyiv"}
], "map_items": [
{"headline": "Blast in Kyiv", "importance": "critical", "location_name": "Kyiv, Ukraine",
"lat": 50.45, "lon": 30.52, "location_confidence": "city", "category": "military/conflict", "url": "https://ex"}
]}
```"""
def test_parse_strips_fence_and_think_tags():
raw = "<think>nope</think>\n" + FENCED
out = parse_reduce_json(raw)
assert out["summary_en"] == "Brief."
assert len(out["ticker"]) == 1
def test_parse_empty_and_garbage_returns_empty_struct():
assert parse_reduce_json("")["summary_en"] == ""
assert parse_reduce_json("not json")["ticker"] == []
def test_clamp_coords_drops_out_of_range_and_unknown():
assert clamp_coords(50.45, 30.52) == (50.45, 30.52)
assert clamp_coords(95.0, 10.0) is None
assert clamp_coords(None, 10.0) is None
assert clamp_coords("50.45", "30.52") == (50.45, 30.52)
def test_select_ticker_keeps_critical_high_caps_12():
rows = [{"headline": f"h{i}", "importance": "critical"} for i in range(15)]
rows.append({"headline": "skip", "importance": "low"})
out = select_ticker(rows)
assert len(out) == 12
assert all(r["importance"] in ("critical", "high") for r in out)
def test_select_map_requires_valid_coords_and_flag():
items = [
{"headline": "A", "importance": "critical", "lat": 50.45, "lon": 30.52, "location_name": "Kyiv"},
{"headline": "B", "importance": "critical", "lat": None, "lon": None, "location_name": "Unknown"},
{"headline": "C", "importance": "low", "lat": 1.0, "lon": 2.0, "location_name": "x"},
]
out = select_map(items)
assert [r["headline"] for r in out] == ["A"]
def test_select_map_caps_20():
items = [
{"headline": f"h{i}", "importance": "critical", "lat": 1.0, "lon": 2.0}
for i in range(25)
]
out = select_map(items)
assert len(out) == 20
assert all(r["importance"] in ("critical", "high") for r in out)

View file

@ -0,0 +1,97 @@
from unittest.mock import MagicMock
import httpx
from nous_client import chat
DEFAULT_UA = "osint-dashboard-news-summarizer"
BASE = "https://inference-api.nousresearch.com/v1"
def _ok_response(content="hello"):
resp = MagicMock()
resp.status_code = 200
resp.json.return_value = {"choices": [{"message": {"content": content}}]}
return resp
def _install_fake(monkeypatch, post_impl):
captured = {}
class FakeClient:
def __init__(self, timeout=None, **kwargs):
captured["timeout"] = timeout
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def post(self, url, *, headers=None, json=None, **kwargs):
captured["url"] = url
captured["headers"] = headers
captured["json"] = json
return post_impl(url, headers, json)
monkeypatch.setattr(httpx, "Client", FakeClient)
return captured
def test_posts_chat_completions_with_auth_body_and_returns_content(monkeypatch):
captured = _install_fake(monkeypatch, lambda *a: _ok_response("the-content"))
out = chat(
"summarize this",
api_key="secret-key",
model="hermes-3",
base_url=BASE,
)
assert out == "the-content"
assert captured["url"] == f"{BASE}/chat/completions"
assert captured["headers"]["Authorization"] == "Bearer secret-key"
assert captured["headers"]["User-Agent"] == DEFAULT_UA
assert captured["json"]["model"] == "hermes-3"
assert captured["json"]["messages"] == [{"role": "user", "content": "summarize this"}]
assert captured["json"]["temperature"] == 0.2
assert captured["json"]["max_tokens"] == 4096
assert "response_format" not in captured["json"]
def test_user_agent_equals_osint_user_agent_env(monkeypatch):
monkeypatch.setenv("OSINT_USER_AGENT", "custom-ua/2.0")
captured = _install_fake(monkeypatch, lambda *a: _ok_response("ok"))
chat("p", api_key="k", model="m", base_url=BASE)
assert captured["headers"]["User-Agent"] == "custom-ua/2.0"
def test_json_mode_sets_response_format(monkeypatch):
captured = _install_fake(monkeypatch, lambda *a: _ok_response("{}"))
chat("p", api_key="k", model="m", base_url=BASE, json_mode=True)
assert captured["json"]["response_format"] == {"type": "json_object"}
def test_401_returns_empty_string(monkeypatch):
def post_impl(*a):
resp = MagicMock()
resp.status_code = 401
return resp
_install_fake(monkeypatch, post_impl)
assert chat("p", api_key="bad", model="m", base_url=BASE) == ""
def test_5xx_returns_empty_string(monkeypatch):
def post_impl(*a):
resp = MagicMock()
resp.status_code = 503
return resp
_install_fake(monkeypatch, post_impl)
assert chat("p", api_key="k", model="m", base_url=BASE) == ""
def test_timeout_returns_empty_string(monkeypatch):
def post_impl(*a):
raise httpx.TimeoutException("timed out")
_install_fake(monkeypatch, post_impl)
assert chat("p", api_key="k", model="m", base_url=BASE) == ""

View file

@ -1,9 +1,9 @@
"""Integration tests for the news pipeline API (GET /api/news + summaries).
"""Integration tests for the news pipeline API (GET /api/news + summaries + intel).
DB-backed: marked `requires_db` and auto-skip when the test database is
unreachable (see tests/conftest.py). Seeding writes directly to the shared
`articles` / `article_summaries` tables, exactly as the scraper + summarizer
services would.
`articles` / `article_summaries` / `news_items` tables, exactly as the scraper
+ summarizer services would.
"""
from __future__ import annotations
@ -37,7 +37,9 @@ def _truncate() -> None:
async def run():
conn = await asyncpg.connect(**_conn_kwargs())
try:
await conn.execute("TRUNCATE articles, article_summaries")
await conn.execute(
"TRUNCATE articles, article_summaries, news_items CASCADE"
)
finally:
await conn.close()
@ -66,14 +68,45 @@ def _seed_article(title: str, url: str, domain: str, ts: str, content: str = "bo
asyncio.run(run())
def _seed_summary(text: str, ts: str) -> None:
def _seed_summary(text: str, ts: str, model: str | None = None) -> int:
async def run() -> int:
conn = await asyncpg.connect(**_conn_kwargs())
try:
row = await conn.fetchrow(
"INSERT INTO article_summaries (summary_text, batch_timestamp, model) "
"VALUES ($1, $2, $3) RETURNING id",
text, datetime.fromisoformat(ts), model,
)
return int(row["id"])
finally:
await conn.close()
return asyncio.run(run())
def _seed_news_item(
summary_id: int,
kind: str,
headline: str,
importance: str,
*,
location_name: str | None = None,
lat: float | None = None,
lon: float | None = None,
location_confidence: str | None = None,
category: str | None = None,
url: str | None = None,
) -> None:
async def run():
conn = await asyncpg.connect(**_conn_kwargs())
try:
await conn.execute(
"INSERT INTO article_summaries (summary_text, batch_timestamp) "
"VALUES ($1, $2)",
text, datetime.fromisoformat(ts),
"INSERT INTO news_items "
"(summary_id, kind, headline, importance, location_name, "
" lat, lon, location_confidence, category, url) "
"VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
summary_id, kind, headline, importance, location_name,
lat, lon, location_confidence, category, url,
)
finally:
await conn.close()
@ -131,7 +164,7 @@ def test_api_news_summaries_contract(clean_news):
assert isinstance(body, list)
assert len(body) == 1
s = body[0]
assert set(s.keys()) == {"id", "summary_text", "batch_timestamp"}
assert set(s.keys()) == {"id", "summary_text", "batch_timestamp", "model"}
assert s["summary_text"] == "master summary markdown…"
assert s["batch_timestamp"].startswith("2026-08-24T18:05")
@ -140,3 +173,65 @@ def test_api_news_summaries_contract(clean_news):
def test_api_news_empty(clean_news):
assert _get("/api/news").json() == []
assert _get("/api/news/summaries").json() == []
assert _get("/api/news/ticker").json() == []
assert _get("/api/news/map").json() == []
TICKER_KEYS = {"id", "headline", "importance", "location_name", "url", "created_at"}
MAP_KEYS = {
"id", "headline", "importance", "location_name", "lat", "lon",
"location_confidence", "category", "url", "created_at",
}
def _seed_flagged_items() -> None:
sid = _seed_summary("batch brief", "2026-08-27T18:05:00+00:00", "Hermes-4.3-36B")
_seed_news_item(
sid, "ticker", "Critical ticker", "critical",
location_name="Kyiv", url="https://example.com/ticker",
)
_seed_news_item(
sid, "ticker", "Low ticker", "low",
location_name="Somewhere", url="https://example.com/low",
)
_seed_news_item(
sid, "map", "Critical map", "critical",
location_name="Taipei", lat=25.03, lon=121.56,
location_confidence="high", category="conflict",
url="https://example.com/map",
)
@requires_db
def test_api_news_ticker_returns_only_flagged(clean_news):
_seed_flagged_items()
resp = _get("/api/news/ticker")
assert resp.status_code == 200
body = resp.json()
assert isinstance(body, list)
assert len(body) == 1
item = body[0]
assert set(item.keys()) == TICKER_KEYS
assert item["headline"] == "Critical ticker"
assert item["importance"] == "critical"
assert item["location_name"] == "Kyiv"
assert item["url"] == "https://example.com/ticker"
@requires_db
def test_api_news_map_returns_only_flagged_with_coords(clean_news):
_seed_flagged_items()
resp = _get("/api/news/map")
assert resp.status_code == 200
body = resp.json()
assert isinstance(body, list)
assert len(body) == 1
item = body[0]
assert set(item.keys()) == MAP_KEYS
assert item["headline"] == "Critical map"
assert item["importance"] == "critical"
assert item["lat"] == 25.03
assert item["lon"] == 121.56
assert item["location_confidence"] == "high"
assert item["category"] == "conflict"
assert _get("/api/news/map?bbox=1,2,3").status_code == 422

172
tests/test_api_settings.py Normal file
View file

@ -0,0 +1,172 @@
"""API tests for GET/PUT /api/settings and GET /api/news/models."""
from __future__ import annotations
import asyncio
import os
import asyncpg
import httpx
import pytest
from conftest import requires_db
from main import app
BASE = "http://test"
def _conn_kwargs() -> dict:
return {
"host": os.environ["DB_HOST"],
"port": int(os.environ["DB_PORT"]),
"user": os.environ["DB_USER"],
"password": os.environ["DB_PASSWORD"],
"database": os.environ["DB_NAME"],
}
def _truncate_settings() -> None:
async def run():
conn = await asyncpg.connect(**_conn_kwargs())
try:
await conn.execute("DROP TABLE IF EXISTS app_settings")
finally:
await conn.close()
asyncio.run(run())
@pytest.fixture()
def clean_settings():
import settings_store
settings_store._ensured = False
_truncate_settings()
yield
settings_store._ensured = False
_truncate_settings()
def _request(method: str, path: str, json: dict | None = None) -> httpx.Response:
async def _run() -> httpx.Response:
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
return await client.request(method, path, json=json)
return asyncio.run(_run())
def _get(path: str) -> httpx.Response:
return _request("GET", path)
def _put(path: str, json: dict) -> httpx.Response:
return _request("PUT", path, json=json)
def test_get_news_models_without_key_returns_fallback(monkeypatch):
monkeypatch.delenv("NOUS_API_KEY", raising=False)
monkeypatch.setattr("keystore.get_api_key", _missing_key)
resp = _get("/api/news/models")
assert resp.status_code == 200
body = resp.json()
assert body["source"] == "fallback"
ids = [m["id"] for m in body["models"]]
assert "Hermes-4.3-36B" in ids
from settings_store import FALLBACK_MODELS
assert ids == FALLBACK_MODELS
async def _missing_key(name: str):
return None
async def _present_key(name: str):
return "test-nous-api-key-1234"
def test_get_news_models_live_from_upstream(monkeypatch):
monkeypatch.setattr("keystore.get_api_key", _present_key)
monkeypatch.setenv("NOUS_BASE_URL", "https://inference-api.nousresearch.com/v1")
monkeypatch.delenv("OSINT_USER_AGENT", raising=False)
captured: dict = {}
class FakeResponse:
status_code = 200
def raise_for_status(self):
return None
def json(self):
return {"data": [{"id": "live-model-a"}, {"id": "Hermes-4.3-36B"}]}
async def fake_http_get(url, *, headers, timeout):
captured["url"] = url
captured["headers"] = headers
captured["timeout"] = timeout
return FakeResponse()
import settings_store
monkeypatch.setattr(settings_store, "_http_get", fake_http_get)
settings_store._models_cache = None
resp = _get("/api/news/models")
assert resp.status_code == 200
body = resp.json()
assert body["source"] == "live"
ids = [m["id"] for m in body["models"]]
assert ids == ["live-model-a", "Hermes-4.3-36B"]
assert captured["url"] == "https://inference-api.nousresearch.com/v1/models"
assert captured["headers"]["User-Agent"] == "osint-dashboard-news-summarizer"
assert captured["headers"]["Authorization"] == "Bearer test-nous-api-key-1234"
timeout = captured["timeout"]
assert timeout == 8 or getattr(timeout, "read", timeout) == 8 or float(timeout) == 8.0
def test_get_news_models_upstream_failure_returns_fallback(monkeypatch):
monkeypatch.setattr("keystore.get_api_key", _present_key)
async def boom_http_get(url, *, headers, timeout):
raise httpx.ConnectError("upstream down")
import settings_store
monkeypatch.setattr(settings_store, "_http_get", boom_http_get)
settings_store._models_cache = None
resp = _get("/api/news/models")
assert resp.status_code == 200
body = resp.json()
assert body["source"] == "fallback"
ids = [m["id"] for m in body["models"]]
assert "Hermes-4.3-36B" in ids
def test_put_settings_empty_returns_422():
resp = _put("/api/settings", {"summary_model": ""})
assert resp.status_code == 422
def test_put_settings_whitespace_only_returns_422():
resp = _put("/api/settings", {"summary_model": " "})
assert resp.status_code == 422
@requires_db
def test_put_settings_round_trip(clean_settings, monkeypatch):
monkeypatch.delenv("SUMMARY_MODEL", raising=False)
monkeypatch.delenv("NOUS_BASE_URL", raising=False)
put_resp = _put("/api/settings", {"summary_model": "google/gemini-2.5-flash"})
assert put_resp.status_code == 200
put_body = put_resp.json()
assert put_body["summary_model"] == "google/gemini-2.5-flash"
assert put_body["nous_base_url"] == "https://inference-api.nousresearch.com/v1"
get_resp = _get("/api/settings")
assert get_resp.status_code == 200
get_body = get_resp.json()
assert get_body["summary_model"] == "google/gemini-2.5-flash"
assert get_body["nous_base_url"] == "https://inference-api.nousresearch.com/v1"
assert set(get_body.keys()) == {"summary_model", "nous_base_url"}