diff --git a/.env.example b/.env.example index 04e1658..f736246 100644 --- a/.env.example +++ b/.env.example @@ -17,3 +17,26 @@ MINIO_ACCESS_KEY= MINIO_SECRET_KEY= # "true" for TLS endpoints (e.g. S3-compatible prod); "false" for local HTTP. MINIO_SECURE=false + +# ── NASA FIRMS (active fire / hotspot ingest) ────────────────────────────── +# MAP_KEY is FREE — get one at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/ +# (1-minute signup, no payment). Leave blank to keep fire ingest idle. +FIRMS_MAP_KEY= +# NRT VIIRS S-NPP 375m active fire detection (near-real-time). +FIRMS_DATASET=VIIRS_SNPP_NRT +# Area to poll as "minlon,minlat,maxlon,maxlat". Narrow it to reduce payload +# (e.g. CONUS "-125,24,-66,50"). Default covers most of the inhabited globe. +FIRMS_BBOX=-180,-60,180,75 +# Poll cadence in seconds (~15 min). FIRMS NRT updates every ~5-10 min. +FIRMS_INTERVAL=900 +# Set to 0 to disable the fire loop entirely. +INGEST_FIRES=1 + +# ── API keys (managed from the dashboard UI) ────────────────────────────── +# Ingest-service keys such as FIRMS_MAP_KEY, GEMINI_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}). You do NOT need to put them in +# .env — the UI writes straight to the DB, and ingest services read them from +# there (keystore.get_api_key), picking up changes on the next poll without a +# container restart. Keys set here in .env are only a fallback for the FIRMS +# ingestor until a value is saved via the UI. diff --git a/.gitignore b/.gitignore index 8acdcaa..0e48ba4 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,6 @@ __pycache__/ *.egg-info/ .venv/ venv/ +.venv*/ .env +.pytest_cache/ diff --git a/alembic/versions/002_fires.py b/alembic/versions/002_fires.py new file mode 100644 index 0000000..5db4d49 --- /dev/null +++ b/alembic/versions/002_fires.py @@ -0,0 +1,60 @@ +"""fires hypertable: NASA FIRMS active fire/hotspot detections + +Revision ID: 002_fires +Revises: 001_initial +Create Date: 2026-08-24 +""" + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = '002_fires' +down_revision = '001_initial' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # The PK IS the idempotency natural key (latitude, longitude, acq_time, + # satellite). acq_time is the hypertable partitioning column and is part of + # the PK, which satisfies TimescaleDB's requirement that every unique index + # include all partitioning columns. + op.create_table( + 'fires', + sa.Column('latitude', sa.Float(), nullable=False), + sa.Column('longitude', sa.Float(), nullable=False), + sa.Column('brightness', sa.Float(), nullable=False), # bright_ti4, K + sa.Column('confidence', sa.String(10), nullable=False), # n/l/h or % + sa.Column('acq_time', sa.DateTime(timezone=True), nullable=False), + sa.Column('satellite', sa.String(16), nullable=False), + sa.Column('instrument', sa.String(16)), + sa.Column('bright_ti5', sa.Float()), + sa.Column('frp', sa.Float()), + sa.Column('daynight', sa.String(1)), + sa.Column('scan', sa.Float()), + sa.Column('track', sa.Float()), + sa.Column('version', sa.String(32)), + sa.Column('raw', sa.JSON()), + sa.Column('ingested_at', sa.DateTime(timezone=True), + server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint( + 'latitude', 'longitude', 'acq_time', 'satellite', + name='pk_fires_natural_key', + ), + ) + + # 1-day chunks keep retention/drop_chunks operations cheap and scans local. + # (TimescaleDB auto-creates an index on the partition column acq_time.) + op.execute( + "SELECT create_hypertable('fires', 'acq_time', " + "if_not_exists => TRUE, chunk_time_interval => INTERVAL '1 day')" + ) + op.create_index('ix_fires_bbox', 'fires', ['longitude', 'latitude']) + + +def downgrade() -> None: + op.drop_index('ix_fires_bbox', table_name='fires') + op.execute("SELECT drop_hypertable('fires', if_exists => TRUE)") + op.drop_table('fires') diff --git a/app/config.py b/app/config.py index da2684e..d349010 100644 --- a/app/config.py +++ b/app/config.py @@ -40,3 +40,19 @@ MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "") MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "") # "false" / "0" / "no" (case-insensitive) → plain HTTP (e.g. local compose). MINIO_SECURE = os.getenv("MINIO_SECURE", "false").lower() not in ("false", "0", "no") + +# ── NASA FIRMS (active fire / hotspot ingest) ─────────────────────────────── +# MAP_KEY is free; obtain one at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/ +# and set FIRMS_MAP_KEY in .env. Until it is set, the fire ingestor logs a +# warning and stays idle (no crash). +FIRMS_MAP_KEY = os.getenv("FIRMS_MAP_KEY", "") +# NRT VIIRS S-NPP active fire/hotspot detection (375m). +FIRMS_DATASET = os.getenv("FIRMS_DATASET", "VIIRS_SNPP_NRT") +# Area bounding box as "minlon,minlat,maxlon,maxlat". Default covers most of +# the inhabited globe; narrow it (e.g. CONUS "-125,24,-66,50") to shrink +# payloads and the Postgres write volume. +FIRMS_BBOX = os.getenv("FIRMS_BBOX", "-180,-60,180,75") +# Poll cadence in seconds. FIRMS NRT updates every ~5-10 min; 900 = 15 min. +FIRMS_INTERVAL = int(os.getenv("FIRMS_INTERVAL", "900")) +# Outbound HTTP timeout for the FIRMS CSV download. +FIRMS_TIMEOUT = float(os.getenv("FIRMS_TIMEOUT", "60")) diff --git a/app/database.py b/app/database.py index 262de31..f0d79eb 100644 --- a/app/database.py +++ b/app/database.py @@ -1,10 +1,22 @@ +import os + from sqlalchemy import MetaData, event, text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool from config import DATABASE_URL +# Pooled connections are bound to the event loop they were created on. Long-lived +# servers (uvicorn, the ingester) want a pool; short-lived / test processes that +# open a fresh event loop per unit (asyncio.run, pytest) must not reuse pooled +# connections across loops, so allow a NullPool (new connection per session). +_NULL_POOL = os.getenv("DB_NULL_POOL", "").lower() in ("1", "true", "yes") + engine = create_async_engine( - DATABASE_URL, echo=False, pool_size=5, max_overflow=10, pool_recycle=300 + DATABASE_URL, + echo=False, + **({"poolclass": NullPool} if _NULL_POOL else + {"pool_size": 5, "max_overflow": 10, "pool_recycle": 300}), ) async_session = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False diff --git a/app/fire_sources.py b/app/fire_sources.py new file mode 100644 index 0000000..192b28a --- /dev/null +++ b/app/fire_sources.py @@ -0,0 +1,192 @@ +"""NASA FIRMS active fire / hotspot ingestor. + +Fetches the VIIRS active-fire detection CSV for an area from NASA FIRMS and +publishes each hotspot to NATS JetStream on the ``events.fire`` subject. The +NATS consumer (app/ingestor.py) routes those messages to the ``fires`` +TimescaleDB hypertable, deduped by the natural key (lat, lon, acq_time, +satellite). + +Zero-cost data: the FIRMS MAP_KEY is free (https://firms.modaps.eosdis.nasa.gov/api/map_key_info/) +and the CSV area endpoint is unlimited for personal/research use. + +Endpoint (documented at https://firms.modaps.eosdis.nasa.gov/api/area/csv/): + https://firms.modaps.eosdis.nasa.gov/api/area/csv/{MAP_KEY}/{DATASET}/{bbox} + bbox = "minlon,minlat,maxlon,maxlat" (e.g. "-125,24,-66,50") + +CSV columns (VIIRS): latitude, longitude, bright_ti4, scan, track, acq_date, +acq_time, satellite, instrument, confidence, version, bright_ti5, frp, daynight. +``acq_time`` is an integer HHMM in UTC (seconds are 0); ``acq_date`` is the +UTC date (YYYY-MM-DD). +""" + +from __future__ import annotations + +import csv +import io +import json +import logging +from datetime import datetime, timezone + +import httpx +import nats + +from config import FIRMS_MAP_KEY, FIRMS_DATASET, FIRMS_BBOX, FIRMS_TIMEOUT, NATS_URL + +logger = logging.getLogger("osint.firms") + +# ── FIRMS API ───────────────────────────────────────────────────────────── + +FIRMS_AREA_CSV = ( + "https://firms.modaps.eosdis.nasa.gov/api/area/csv/{key}/{dataset}/{bbox}" +) + +# The canonical VIIRS CSV header FIRMS returns. Used to (a) locate the real +# header row if FIRMS ever prepends a legend line and (b) validate a download. +FIRMS_CSV_COLUMNS = ( + "latitude", "longitude", "bright_ti4", "scan", "track", "acq_date", + "acq_time", "satellite", "instrument", "confidence", "version", + "bright_ti5", "frp", "daynight", +) + + +def _to_float(value: object) -> float | None: + """Best-effort float conversion; returns None on empty/garbage values.""" + if value is None: + return None + try: + return float(value) + except (ValueError, TypeError): + return None + + +def normalize_acq_time(acq_date: object, acq_time: object) -> datetime | None: + """Combine FIRMS acq_date + acq_time into a timezone-aware UTC datetime. + + acq_date is 'YYYY-MM-DD'; acq_time is an integer HHMM in UTC with seconds + truncated. Returns None when the values can't be parsed (dropped). + """ + if not acq_date or acq_time is None or acq_time == "": + return None + try: + hhmm = str(int(acq_time)).zfill(4) + return datetime.strptime( + f"{acq_date} {hhmm}", "%Y-%m-%d %H%M" + ).replace(tzinfo=timezone.utc) + except (ValueError, TypeError): + return None + + +def parse_firms_csv(text: str) -> list[dict]: + """Parse a FIRMS area CSV payload into normalized fire messages. + + Returns one dict per hotspot with the fields stored in the ``fires`` table + (acq_time already combined into a UTC ISO timestamp). Rows that don't look + like valid VIIRS detections are skipped rather than failing the whole poll. + """ + rows = list(csv.reader(io.StringIO(text))) + if not rows: + return [] + + # Locate the real header row. FIRMS normally returns the CSV header first, + # but occasionally prepends a legend/info line, so scan until we see the + # canonical header. + header_idx = 0 + for i, row in enumerate(rows): + if row and row[0].strip().lower() == "latitude" and len(row) >= 4: + header_idx = i + break + header = [c.strip().lower() for c in rows[header_idx]] + # Guard against a header that isn't actually the FIRMS one. + if "latitude" not in header or "longitude" not in header: + logger.warning("FIRMS payload does not look like a hotspot CSV (first row: %r)", header[:6]) + return [] + + points: list[dict] = [] + for row in rows[header_idx + 1:]: + if len(row) < len(header): + continue + rec = dict(zip(header, row)) + lat = _to_float(rec.get("latitude")) + lon = _to_float(rec.get("longitude")) + if lat is None or lon is None: + continue + acq_time = normalize_acq_time(rec.get("acq_date"), rec.get("acq_time")) + if acq_time is None: + continue + points.append({ + "latitude": lat, + "longitude": lon, + "brightness": _to_float(rec.get("bright_ti4")), + "confidence": str(rec.get("confidence") or "").strip(), + "acq_time": acq_time.isoformat(), + "satellite": str(rec.get("satellite") or "").strip(), + "instrument": str(rec.get("instrument") or "").strip(), + "bright_ti5": _to_float(rec.get("bright_ti5")), + "frp": _to_float(rec.get("frp")), + "daynight": str(rec.get("daynight") or "").strip(), + "scan": _to_float(rec.get("scan")), + "track": _to_float(rec.get("track")), + "version": str(rec.get("version") or "").strip(), + }) + return points + + +async def publish_fire_batch(points: list[dict]) -> int: + """Publish a batch of hotspot messages to NATS JetStream in one connection. + + Fires are bulk data (thousands of detections per poll), so opening a NATS + connection per message — as the RSS/GDELT path does — would be wasteful. + One connection, one JetStream context, one close, with the batch flushed + before returning so a crash mid-write can't silently drop half a poll. + """ + if not points: + return 0 + nc = await nats.connect(NATS_URL) + js = nc.jetstream() + try: + for pt in points: + msg = {"source_type": "fire", **pt} + await js.publish("events.fire", json.dumps(msg).encode()) + finally: + await nc.close() + return len(points) + + +async def ingest_fires(bbox: str | None = None) -> int: + """Fetch the FIRMS hotspot CSV for an area and publish it to NATS. + + Returns the number of hotspot messages published. Idle (0) and a logged + warning when FIRMS_MAP_KEY is not set, so the ingester keeps running for + the other sources. + """ + if not FIRMS_MAP_KEY: + logger.warning( + "FIRMS_MAP_KEY not set — fire ingest disabled. " + "Get a free key at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/" + ) + return 0 + + area = bbox or FIRMS_BBOX + url = FIRMS_AREA_CSV.format( + key=FIRMS_MAP_KEY, dataset=FIRMS_DATASET, bbox=area + ) + + async with httpx.AsyncClient(timeout=FIRMS_TIMEOUT) as client: + resp = await client.get(url) + resp.raise_for_status() + text = resp.text + + # FIRMS returns HTTP 200 with a plain-text error for some failure modes + # (bad key, invalid bbox); surface the first line for debuggability. + if "latitude" not in text.lower()[:4096]: + first_line = text.strip().splitlines()[0][:200] if text.strip() else "(empty)" + logger.warning("FIRMS CSV download returned no hotspot data (%s)", first_line) + return 0 + + points = parse_firms_csv(text) + published = await publish_fire_batch(points) + logger.info( + "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d", + len(points), area, FIRMS_DATASET, published, + ) + return published diff --git a/app/ingestor.py b/app/ingestor.py index 6000aab..2d628b9 100644 --- a/app/ingestor.py +++ b/app/ingestor.py @@ -9,9 +9,11 @@ from datetime import datetime, timezone import nats from nats.errors import TimeoutError +from sqlalchemy.dialects.postgresql import insert as pg_insert from database import async_session from models import events as events_table +from models import fires as fires_table from config import NATS_URL logger = logging.getLogger("osint.ingestor") @@ -22,8 +24,86 @@ NATS_STREAM = "events" NATS_DURABLE = "osint-ingestor" +# ── Active fire / hotspot routing ───────────────────────────────────────── + +def _fire_row_from_msg(msg: dict) -> dict | None: + """Map a NATS fire message to a ``fires`` table row (pre-DB, unit-testable). + + Returns None (dropped) when the idempotency key fields — latitude, + longitude, acq_time, satellite — are missing or unparseable. + """ + row = { + "latitude": msg.get("latitude"), + "longitude": msg.get("longitude"), + "brightness": msg.get("brightness"), + "confidence": msg.get("confidence"), + "acq_time": msg.get("acq_time"), + "satellite": msg.get("satellite"), + "instrument": msg.get("instrument"), + "bright_ti5": msg.get("bright_ti5"), + "frp": msg.get("frp"), + "daynight": msg.get("daynight"), + "scan": msg.get("scan"), + "track": msg.get("track"), + "version": msg.get("version"), + "raw": msg.get("raw"), + } + # Idempotency key must be complete; brightness/confidence may be absent in + # malformed feeds but lat/lon/time/satellite are required for dedupe. + if ( + row["latitude"] is None or row["longitude"] is None + or row["acq_time"] is None or not row["satellite"] + ): + logger.warning( + "dropping malformed fire message (missing natural key): %s", + {k: msg.get(k) for k in ("latitude", "longitude", "acq_time", "satellite")}, + ) + return None + if isinstance(row["acq_time"], str): + try: + row["acq_time"] = datetime.fromisoformat(row["acq_time"]) + except ValueError: + logger.warning("dropping fire message with bad acq_time %r", row["acq_time"]) + return None + if not isinstance(row["acq_time"], datetime): + row["acq_time"] = datetime.fromisoformat(str(row["acq_time"])) + return row + + +async def ingest_fire_row(msg: dict) -> bool: + """Persist one FIRMS hotspot, idempotently. + + The (latitude, longitude, acq_time, satellite) primary key doubles as the + dedupe key: ON CONFLICT DO NOTHING means a hotspot re-delivered on a later + 15-minute poll is silently ignored. Returns True if a new row was inserted, + False if it was a duplicate (or dropped). + """ + row = _fire_row_from_msg(msg) + if row is None: + return False + async with async_session() as session: + stmt = ( + pg_insert(fires_table) + .values(**row) + .on_conflict_do_nothing(constraint="pk_fires_natural_key") + ) + result = await session.execute(stmt) + await session.commit() + inserted = bool(result.rowcount) + if inserted: + logger.info( + "ingested fire %.5f,%.5f %s satellite=%s", + row["latitude"], row["longitude"], row["acq_time"], row["satellite"], + ) + return inserted + + async def ingest_event(msg: dict): """Ingest a single event from NATS into PostgreSQL.""" + # Active fire/hotspot messages carry a dedicated schema and land in the + # `fires` hypertable (idempotent natural key), not the generic events feed. + if msg.get("source_type") == "fire": + return await ingest_fire_row(msg) # source_id links to a feed_sources UUID; tolerate non-UUID / missing values # (e.g. legacy messages that carried a URL) by coercing to None. raw_source_id = msg.get("source_id") diff --git a/app/keystore.py b/app/keystore.py new file mode 100644 index 0000000..1f84924 --- /dev/null +++ b/app/keystore.py @@ -0,0 +1,212 @@ +"""OSINT Dashboard — API key store (keyv-style Postgres table). + +Keys live in the ``api_keys`` table, shared between the dashboard app and the +ingest services (both connect to the same Postgres). Only the app WRITES keys +via the management API; ingest services read them with :func:`get_api_key`. + +Storage: the table is created lazily with ``CREATE TABLE IF NOT EXISTS`` on +first use in each process (no alembic migration, so it can never fork the +migration chain or block container startup). DDL is idempotent and safe when +the app and ingester containers race on first boot. + +Security contract: + * Values are stored in Postgres, never in the container image or frontend. + * ``GET /api/keys`` returns only set/missing status plus a masked + "****last4" hint. Full values are NEVER returned by the API. + * Registered keys get cheap format validation on save (regex), e.g. the + FIRMS map key must be a 32-char hex string. +""" + +from __future__ import annotations + +import asyncio +import re +from datetime import datetime, timezone + +from sqlalchemy import Column, DateTime, String, Table, Text, func, select, text + +from database import async_session, engine, metadata + +# ── Table definition (bound to the shared metadata; created lazily) ─────── +api_keys = Table( + "api_keys", + metadata, + Column("name", String(128), primary_key=True), + Column("value", Text, nullable=False), + Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False), + Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False), +) + +_CREATE_TABLE_SQL = text( + """ + CREATE TABLE IF NOT EXISTS api_keys ( + name VARCHAR(128) PRIMARY KEY, + value TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + """ +) + +# ── Registry of keys the ingest services understand ─────────────────────── +# Each entry: human description (shown in the UI) + optional ``pattern`` for +# cheap format validation and an ``example`` for error/placeholder text. +KEY_REGISTRY: dict[str, dict] = { + "FIRMS_MAP_KEY": { + "description": "NASA FIRMS API key — active fire / satellite ingest.", + "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)", + }, + "TELEGRAM_TOKEN": { + "description": "Telegram bot token — push alert notifications to a channel.", + "pattern": r"^\d{8,10}:[0-9A-Za-z_-]{35}$", + "example": "123456789:AA… (bot token from @BotFather)", + }, +} + +# Any stored key must at least be a sane UPPER_SNAKE name. +_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$") + + +class KeyFormatError(ValueError): + """Raised when a key name or value fails format validation.""" + + +# ── Lazy table bootstrap ────────────────────────────────────────────────── + +_ensure_lock = asyncio.Lock() +_ensured = False + + +async def ensure_api_keys_table() -> None: + """Create the api_keys 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 + + +# ── Validation ──────────────────────────────────────────────────────────── + +def is_registered(name: str) -> bool: + """True if ``name`` has an entry in the registry (gets format validation).""" + return name in KEY_REGISTRY + + +def validate_name(name: str) -> None: + """Reject malformed key names (must be UPPER_SNAKE).""" + if not _NAME_RE.match(name or ""): + raise KeyFormatError( + "Invalid key name — use UPPER_SNAKE_CASE (e.g. FIRMS_MAP_KEY)." + ) + + +def validate_value(name: str, value: str) -> None: + """Cheap format validation for registered keys; rejects empty values.""" + if not value or not value.strip(): + raise KeyFormatError("Value must not be empty.") + info = KEY_REGISTRY.get(name) + pattern = (info or {}).get("pattern") + if pattern and not re.match(pattern, value.strip()): + example = (info or {}).get("example", "a matching value") + raise KeyFormatError(f"'{name}' has an invalid format — expected {example}.") + + +def mask_value(value: str | None) -> str: + """Mask a stored value as '****last4' (or '****' when shorter than 4).""" + if not value: + return "" + v = value.strip() + tail = v[-4:] if len(v) >= 4 else "" + return f"****{tail}" + + +# ── Store operations ────────────────────────────────────────────────────── + +async def list_keys() -> list[dict]: + """Registry keys + any extra stored keys, with masked set-status only.""" + await ensure_api_keys_table() + async with async_session() as session: + rows = (await session.execute(select(api_keys))).mappings().all() + stored = {r["name"]: r["value"] for r in rows} + + names = list(KEY_REGISTRY) + [n for n in stored if n not in KEY_REGISTRY] + out = [] + for name in names: + value = stored.get(name) + info = KEY_REGISTRY.get(name, {}) + out.append( + { + "name": name, + "set": value is not None, + "masked": mask_value(value) if value is not None else None, + "description": info.get("description", ""), + "example": info.get("example", ""), + "validated": name in KEY_REGISTRY, + } + ) + out.sort(key=lambda k: k["name"].lower()) + return out + + +async def set_key(name: str, value: str) -> dict: + """Upsert a key value. Validates registered formats; rejects bad names. + + Returns the masked status (never the raw value). + """ + validate_name(name) + validate_value(name, value) + value = value.strip() + now = datetime.now(timezone.utc) + + await ensure_api_keys_table() + async with async_session() as session: + existing = ( + await session.execute(select(api_keys).where(api_keys.c.name == name)) + ).mappings().one_or_none() + if existing: + await session.execute( + api_keys.update() + .where(api_keys.c.name == name) + .values(value=value, updated_at=now) + ) + else: + await session.execute( + api_keys.insert().values(name=name, value=value, updated_at=now) + ) + await session.commit() + return {"name": name, "set": True, "masked": mask_value(value)} + + +async def delete_key(name: str) -> bool: + """Remove a stored key. Returns True if something was deleted.""" + await ensure_api_keys_table() + async with async_session() as session: + result = await session.execute(api_keys.delete().where(api_keys.c.name == name)) + await session.commit() + return bool(result.rowcount) + + +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 + key set via the dashboard is picked up on the next poll — no restart. + """ + await ensure_api_keys_table() + async with async_session() as session: + row = ( + await session.execute(select(api_keys).where(api_keys.c.name == name)) + ).mappings().one_or_none() + return row["value"] if row else None diff --git a/app/main.py b/app/main.py index 16072d9..029d7b7 100644 --- a/app/main.py +++ b/app/main.py @@ -26,18 +26,21 @@ from sqlalchemy.ext.asyncio import AsyncSession from database import async_session, init_extensions from models import ( - alerts, documents, entities, entity_events, events, feed_sources + alerts, documents, entities, entity_events, events, feed_sources, fires ) from schemas import ( AlertCreate, AlertOut, AlertSeverity, AlertType, AlertUpdate, DashboardSummary, EntityCreate, EntityKind, EntityOut, - EventCreate, EventOut, + EventCreate, EventOut, FireOut, FeedSourceCreate, FeedSourceOut, + KeyOut, KeyValueIn, SearchResult, SentimentSummary, SourceType, SearchQuery, TimelinePoint, ) 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 logging.basicConfig(level=logging.INFO) logger = structlog.get_logger("osint.dashboard") @@ -218,6 +221,64 @@ async def create_event(payload: EventCreate): return {"id": str(event_id)} +# ── Active Fires / Hotspots (NASA FIRMS) ───────────────────────────────── + +@app.get("/api/fires", response_model=list[FireOut]) +async def list_fires( + bbox: str | None = Query( + None, + description="Comma-separated 'minlon,minlat,maxlon,maxlat' to bound the " + "result set (e.g. '-125,24,-66,50'). Omit for all stored " + "detections (most recent first).", + ), + since: datetime | None = Query( + None, + description="Only hotspots acquired at/after this UTC instant " + "(ISO 8601, e.g. '2026-08-24T12:00:00Z').", + ), + limit: int = Query(2000, ge=1, le=10000), +): + """List stored FIRMS active fire/hotspot detections as JSON. + + This is the data contract for the map's fire heatmap overlay: the frontend + calls `GET /api/fires?bbox=...&since=...` and renders the returned points. + """ + async with async_session() as session: + stmt = select(fires).order_by(fires.c.acq_time.desc()) + if since: + stmt = stmt.where(fires.c.acq_time >= since) + 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_( + fires.c.longitude >= minlon, fires.c.longitude <= maxlon, + fires.c.latitude >= minlat, fires.c.latitude <= maxlat, + ) + ) + stmt = stmt.limit(limit) + rows = (await session.execute(stmt)).mappings().all() + return [ + FireOut( + latitude=r["latitude"], longitude=r["longitude"], + brightness=r["brightness"], confidence=r["confidence"], + acq_time=r["acq_time"], satellite=r["satellite"], + instrument=r["instrument"], bright_ti5=r["bright_ti5"], + frp=r["frp"], daynight=r["daynight"], + ) + for r in rows + ] + + # ── Search ──────────────────────────────────────────────────────────────── @app.post("/api/search", response_model=SearchResult) @@ -432,6 +493,43 @@ async def list_documents( } +# ── API Keys ──────────────────────────────────────────────────────────── + +@app.get("/api/keys", response_model=list[KeyOut]) +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) + are always included. Any extra stored keys are appended. + """ + return await list_keys() + + +@app.post("/api/keys/{name}", status_code=200) +async def save_api_key(name: str, payload: KeyValueIn): + """Save (upsert) an API key value. + + Registered key formats are validated (e.g. FIRMS_MAP_KEY must be a + 32-char hex string). Unregistered names are accepted as long as they + use UPPER_SNAKE_CASE. The raw value is stored in Postgres and never + returned by any API endpoint — only the masked status is exposed. + """ + try: + result = await set_key(name, payload.value) + except KeyFormatError as exc: + raise HTTPException(status_code=422, detail=str(exc)) + return result + + +@app.delete("/api/keys/{name}") +async def remove_api_key(name: str): + """Delete a stored API key (unsets it).""" + removed = await delete_key(name) + if not removed: + raise HTTPException(status_code=404, detail="Key not found") + return {"ok": True} + + # ── Ingestion Triggers ─────────────────────────────────────────────────── @app.post("/api/ingest/rss") @@ -455,6 +553,13 @@ async def trigger_earthquake_ingest(): return {"status": "ok", "events_ingested": count} +@app.post("/api/ingest/fires") +async def trigger_fire_ingest(bbox: str | None = None): + """Trigger a NASA FIRMS active-fire poll (uses FIRMS_BBOX if bbox omitted).""" + count = await ingest_fires(bbox) + return {"status": "ok", "hotspots_published": count} + + @app.post("/api/ingest/social") async def trigger_social_ingest(query: str = "", max_items: int = 50): """Trigger social signals ingestion.""" diff --git a/app/models.py b/app/models.py index 6f052b2..2128e82 100644 --- a/app/models.py +++ b/app/models.py @@ -5,7 +5,7 @@ from uuid import uuid4 from sqlalchemy import ( Column, Enum, Float, Index, Integer, String, Text, - DateTime, JSON, func, Table, + DateTime, JSON, func, Table, PrimaryKeyConstraint, ) from sqlalchemy.dialects.postgresql import UUID, TSVECTOR @@ -145,3 +145,40 @@ documents = Table( Column("event_id", UUID(as_uuid=True)), Column("uploaded_at", DateTime(timezone=True), server_default=func.now()), ) + + +# ── Active Fires / Hotspots (NASA FIRMS) ────────────────────────────────── + +# VIIRS active fire/hotspot detections from NASA FIRMS. The primary key IS the +# natural key (latitude, longitude, acq_time, satellite) so repeated 15-minute +# polls of the same detection are idempotent (INSERT ... ON CONFLICT DO NOTHING +# silently ignores duplicates). acq_time is the partitioning column of the +# TimescaleDB hypertable, and since it is part of the PK, TimescaleDB's +# "all unique indexes must include the partitioning column" rule is satisfied. +# No surrogate id is needed — the natural key is the identity of a detection. +fires = Table( + "fires", + metadata, + Column("latitude", Float, nullable=False), + Column("longitude", Float, nullable=False), + Column("brightness", Float, nullable=False), # bright_ti4, Kelvin + Column("confidence", String(10), nullable=False), # VIIRS: n/l/h, MODIS: % + Column("acq_time", DateTime(timezone=True), nullable=False), # UTC acquisition + Column("satellite", String(16), nullable=False), # e.g. N (S-NPP), N20, N21 + Column("instrument", String(16)), + Column("bright_ti5", Float), # 12µm brightness, Kelvin + Column("frp", Float), # fire radiative power, MW + Column("daynight", String(1)), # D / N + Column("scan", Float), + Column("track", Float), + Column("version", String(32)), # e.g. 2.0NRT / 2.0URT + Column("raw", JSON), + Column("ingested_at", DateTime(timezone=True), server_default=func.now(), nullable=False), + PrimaryKeyConstraint( + "latitude", "longitude", "acq_time", "satellite", + name="pk_fires_natural_key", + ), +) + +# Bounding-box index for `bbox=` filtering (lon first for max/min-lon scans). +Index("ix_fires_bbox", fires.c.longitude, fires.c.latitude) diff --git a/app/run_ingester.py b/app/run_ingester.py index a63dbae..1dfdd66 100644 --- a/app/run_ingester.py +++ b/app/run_ingester.py @@ -24,8 +24,9 @@ import sys sys.path.insert(0, sys_path) -from config import NATS_URL # noqa: E402 +from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET # noqa: E402 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes # noqa: E402 +from fire_sources import ingest_fires # noqa: E402 from ingestor import ingest_event, start_nats_consumer # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") @@ -35,6 +36,7 @@ RSS_URLS = [u.strip() for u in os.getenv("RSS_URL", "").split(",") if u.strip()] INTERVAL = int(os.getenv("INGEST_INTERVAL", "300")) GDELT_QUERY = os.getenv("GDELT_QUERY", "") ENABLE_QUAKES = os.getenv("INGEST_EARTHQUAKES", "1").lower() in ("1", "true", "yes") +ENABLE_FIRES = os.getenv("INGEST_FIRES", "1").lower() in ("1", "true", "yes") NATS_STREAM = "events" @@ -84,12 +86,33 @@ async def consumer_loop() -> None: logger.exception("failed to process message") +async def fire_loop() -> None: + """Poll NASA FIRMS active fires on the ~15-minute cadence. + + Runs as its own task so its slower cadence (FIRMS_INTERVAL, default 900s) + is independent of the RSS/GDELT/USGS producer cycle (INGEST_INTERVAL). + """ + logger.info("fire loop starting (interval=%ss, dataset=%s)", FIRMS_INTERVAL, FIRMS_DATASET) + while True: + try: + n = await ingest_fires() + logger.info("FIRMS -> %d hotspots", n) + except Exception: # noqa: BLE001 — keep the loop alive across transient failures + logger.exception("FIRMS fetch failed") + await asyncio.sleep(FIRMS_INTERVAL) + + async def main() -> None: logger.info( - "ingester starting (rss=%d feeds, gdelt_q=%r, quakes=%s, interval=%ss)", - len(RSS_URLS), GDELT_QUERY, ENABLE_QUAKES, INTERVAL, + "ingester starting (rss=%d feeds, gdelt_q=%r, quakes=%s, fires=%s, interval=%ss)", + len(RSS_URLS), GDELT_QUERY, ENABLE_QUAKES, ENABLE_FIRES, INTERVAL, ) - await asyncio.gather(producer_loop(), consumer_loop()) + tasks: list[asyncio.Task] = [] + if ENABLE_FIRES: + # Fire ingest only starts once FIRMS_MAP_KEY is set (ingest_fires logs + # and idles otherwise). + tasks.append(asyncio.create_task(fire_loop())) + await asyncio.gather(producer_loop(), consumer_loop(), *tasks) if __name__ == "__main__": diff --git a/app/schemas.py b/app/schemas.py index 1cac012..2021983 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -215,6 +215,44 @@ class DocumentOut(BaseModel): uploaded_at: datetime +# ─── API Keys ──────────────────────────────────────────────────────────── + +class KeyValueIn(BaseModel): + """Body for POST /api/keys/{name} — the value to store. + + The raw value is accepted only on write; it is never read back out. + """ + value: str = Field(..., min_length=1, max_length=4096) + + +class KeyOut(BaseModel): + """One key's status as exposed by GET /api/keys. + + ``masked`` is the "****last4" hint; the full value is NEVER included. + """ + name: str + set: bool + masked: Optional[str] = None + description: Optional[str] = None + example: Optional[str] = None + validated: bool = True + + +# ─── Active fires / hotspots (NASA FIRMS) ──────────────────────────────── + +class FireOut(BaseModel): + latitude: float + longitude: float + brightness: float # bright_ti4 brightness temperature, Kelvin + confidence: str # VIIRS: n (nominal) / l (low) / h (high) + acq_time: datetime + satellite: str + instrument: Optional[str] = None + bright_ti5: Optional[float] = None # 12µm brightness temperature, Kelvin + frp: Optional[float] = None # fire radiative power, MW + daynight: Optional[str] = None # D / N + + # ─── Aggregations ──────────────────────────────────────────────────────── class SentimentSummary(BaseModel): diff --git a/app/static/index.html b/app/static/index.html index dbf3f8b..0f7a0ca 100644 --- a/app/static/index.html +++ b/app/static/index.html @@ -40,6 +40,26 @@ .sentiment-bar div { transition: width 0.3s; } .tab-bar { display: flex; gap: 0.5rem; margin-bottom: 1rem; } .tab-bar .btn.active { background: var(--accent); color: #0f172a; } + .key-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(320px, 1fr)); gap: 1rem; } + .key-card { background: var(--surface); border: 1px solid var(--border); border-radius: 10px; padding: 1rem 1.1rem; box-shadow: 0 0 18px rgba(56,189,248,0.06); transition: box-shadow .2s, border-color .2s; } + .key-card:hover { box-shadow: 0 0 22px rgba(56,189,248,0.14); } + .key-card.set { border-color: rgba(74,222,128,0.5); box-shadow: 0 0 18px rgba(74,222,128,0.14); } + .key-head { display: flex; justify-content: space-between; align-items: center; gap: 0.5rem; margin-bottom: 0.35rem; } + .key-name { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; font-size: 0.9rem; font-weight: 600; color: var(--accent); word-break: break-all; } + .key-status { font-size: 0.72rem; font-family: ui-monospace, monospace; padding: 0.2rem 0.55rem; border-radius: 9999px; white-space: nowrap; } + .key-status.set { background: rgba(74,222,128,0.12); color: var(--green); box-shadow: 0 0 10px rgba(74,222,128,0.35); } + .key-status.missing { background: rgba(248,113,113,0.12); color: var(--red); box-shadow: 0 0 10px rgba(248,113,113,0.25); } + .key-desc { font-size: 0.82rem; color: var(--muted); margin-bottom: 0.7rem; } + .key-hint { font-size: 0.72rem; color: var(--muted); opacity: 0.8; margin-top: 0.15rem; } + .key-form { display: flex; gap: 0.5rem; } + .key-form input { flex: 1; background: var(--bg); border: 1px solid var(--border); border-radius: 6px; padding: 0.5rem 0.6rem; color: var(--text); font-size: 0.85rem; font-family: ui-monospace, monospace; } + .key-form input:focus { outline: none; border-color: var(--accent); box-shadow: 0 0 8px rgba(56,189,248,0.35); } + .key-form .btn { white-space: nowrap; } + .key-form .btn-danger { border-color: var(--red); color: var(--red); } + .key-form .btn-danger:hover { background: rgba(248,113,113,0.12); } + .key-msg { margin-top: 0.5rem; font-size: 0.8rem; min-height: 1rem; } + .key-msg.ok { color: var(--green); } + .key-msg.err { color: var(--red); } @@ -85,6 +105,7 @@ + @@ -144,6 +165,13 @@
+ + +