osint-dashboard/app/models.py
Sirius DevOps d81ea68dec fix: pipeline reliability — WS, summarizer, ingest, health
Stop the live HUD reconnect storm (nginx WS snippet + backoff), copy
intel/nous_client into the summarizer image, and make event ingest
idempotent on URL. GDELT uses the DOC API; NWS no longer sends bbox;
FIRMS is one ON CONFLICT batch; GET /api/aircraft serves last-known.
Health reports freshness without 503ing docker. EONET + CISA KEV added.
2026-08-28 21:49:05 -04:00

243 lines
9.7 KiB
Python

"""OSINT Dashboard — SQLAlchemy models (async, declarative)."""
from datetime import datetime, timezone
from uuid import uuid4
from sqlalchemy import (
Column, Enum, Float, Index, Integer, String, Text,
DateTime, JSON, func, Table, PrimaryKeyConstraint,
)
from sqlalchemy.dialects.postgresql import UUID, TSVECTOR
from database import metadata
# ── Feed Sources ──────────────────────────────────────────────────────────
feed_sources = Table(
"feed_sources",
metadata,
Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4),
Column("name", String(256), nullable=False),
Column("source_type", Enum(
"rss", "gdel-t2", "social", "earthquake", "disaster",
"weather", "fire", "satellite", name="feed_source_type"
), nullable=False),
Column("url", Text),
Column("config", JSON),
Column("enabled", Integer, server_default="1", nullable=False),
Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
Column("updated_at", DateTime(timezone=True), server_default=func.now(), onupdate=func.now()),
)
# ── Events (hypertable via TimescaleDB) ──────────────────────────────────
events = Table(
"events",
metadata,
Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4),
Column("source_type", Enum(
"rss", "gdel-t2", "social", "earthquake", "disaster",
"weather", "fire", "satellite", "camera", name="event_source_type"
), nullable=False, index=True),
Column("source_id", UUID(as_uuid=True)),
Column("title", Text),
Column("body", Text),
Column("url", Text),
Column("sentiment_score", Float),
Column("sentiment_label", Enum("positive", "neutral", "negative", name="sentiment_label")),
Column("location_lat", Float),
Column("location_lon", Float),
Column("location_name", String(512)),
Column("entities", JSON),
Column("tags", JSON),
Column("raw", JSON),
Column("ingested_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
Column("source_timestamp", DateTime(timezone=True), nullable=False),
# Full-text search vector
Column(
"search_vector",
TSVECTOR,
nullable=True,
),
)
# GIN index for full-text search
Index("ix_events_search_vector", events.c.search_vector, postgresql_using="gin")
# Spatial index on location
Index("ix_events_location", events.c.location_lat, events.c.location_lon)
# Timescale unique indexes must include the partition column, so URL
# idempotency lives on a regular table — not the events hypertable.
event_dedup = Table(
"event_dedup",
metadata,
Column("url", Text, primary_key=True),
Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
)
# ── Entities (people, organizations, locations of interest) ──────────────
entities = Table(
"entities",
metadata,
Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4),
Column("name", String(512), nullable=False, index=True),
Column("entity_type", Enum(
"person", "organization", "location", "topic", "asset",
name="entity_type"
), nullable=False),
Column("aliases", JSON),
Column("description", Text),
Column("metadata", JSON),
Column("location_lat", Float),
Column("location_lon", Float),
Column("event_count", Integer, server_default="0"),
Column("first_seen", DateTime(timezone=True), server_default=func.now()),
Column("last_seen", DateTime(timezone=True), server_default=func.now()),
)
# ── Entity-Event Link ────────────────────────────────────────────────────
entity_events = Table(
"entity_events",
metadata,
Column("entity_id", UUID(as_uuid=True), primary_key=True),
Column("event_id", UUID(as_uuid=True), primary_key=True),
Column("relevance_score", Float),
Column("linked_at", DateTime(timezone=True), server_default=func.now()),
)
# ── Alerts ───────────────────────────────────────────────────────────────
alerts = Table(
"alerts",
metadata,
Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4),
Column("alert_type", Enum(
"entity_mention", "sentiment_shift", "geo_proximity",
"keyword_match", "threshold", "anomaly",
name="alert_type"
), nullable=False),
Column("entity_id", UUID(as_uuid=True)),
Column("event_id", UUID(as_uuid=True)),
Column("severity", Enum("low", "medium", "high", "critical", name="alert_severity"), nullable=False),
Column("title", Text, nullable=False),
Column("message", Text),
Column("context", JSON),
Column("acknowledged", Integer, server_default="0"),
Column("acknowledged_by", String(256)),
Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
Column("resolved_at", DateTime(timezone=True)),
)
Index("ix_alerts_severity_created", alerts.c.severity, alerts.c.created_at.desc())
Index("ix_alerts_entity", alerts.c.entity_id)
# ── Documents (stored in MinIO, indexed here) ────────────────────────────
documents = Table(
"documents",
metadata,
Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4),
Column("bucket", String(256), nullable=False),
Column("object_key", String(1024), nullable=False),
Column("content_type", String(256)),
Column("size_bytes", Integer),
Column("description", Text),
Column("tags", JSON),
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)
# ── News pipeline (scraper + summarizer) ──────────────────────────────────
# 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",
metadata,
Column("id", Integer, primary_key=True, autoincrement=True),
Column("title", Text),
Column("url", Text, unique=True), # dedup key
Column("content", Text), # full extracted article text
Column("domain", Text), # source domain
Column("timestamp", DateTime(timezone=True)), # capture time
)
Index("ix_articles_timestamp", articles.c.timestamp)
article_summaries = Table(
"article_summaries",
metadata,
Column("id", Integer, primary_key=True, autoincrement=True),
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)