commit 93b8b89639d52fecf7de0fcdfbdf72c014ea718a Author: sirius0xdev Date: Thu Jun 4 20:30:04 2026 -0400 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..780d19b --- /dev/null +++ b/Dockerfile @@ -0,0 +1,18 @@ +FROM python:3.13-slim AS base + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends \ + gcc libpq-dev \ + && rm -rf /var/lib/apt/lists/* + +COPY app/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY app/ ./app/ +COPY alembic.ini ./alembic.ini +COPY alembic/ ./alembic/ + +EXPOSE 8000 +ENV PYTHONPATH=/app +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..a470ca8 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,37 @@ +[alembic] +script_location = alembic +# Override at runtime via DATABASE_URL environment variable (set in ConfigMap/Deployment) +sqlalchemy.url = driver://user:pass@localhost/dbname + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..9b2fb7e --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,55 @@ +import sys +from logging.config import fileConfig +from pathlib import Path + +from sqlalchemy import pool +from sqlalchemy.ext.asyncio import async_engine_from_config + +from alembic import context + +# Add app directory to path so we can import models/database +sys.path.insert(0, str(Path(__file__).parent.parent / "app")) + +from database import metadata, DATABASE_URL + +config = context.config +if config.config_file_name is not None: + fileConfig(config.config_file_name) +target_metadata = metadata + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode.""" + url = config.get_main_option("sqlalchemy.url") + context.configure(url=url, target_metadata=target_metadata, literal_binds=True) + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection): + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations(): + connectable = async_engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + await connectable.dispose() + + +def run_migrations_online() -> None: + """Run migrations in 'online' mode.""" + import asyncio + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..3776ebd --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,28 @@ +<%%doc>Template for rendering a Multiple Migration Revision Identifier. +<%%- + from alembic import context + context.configure() +-%> +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma_n, trim} +Create Date: ${create_date} +""" +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision = ${repr(up_revision)} +down_revision = ${repr(down_revision)} +branch_labels = ${repr(branch_labels)} +depends_on = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/001_initial.py b/alembic/versions/001_initial.py new file mode 100644 index 0000000..7cc2207 --- /dev/null +++ b/alembic/versions/001_initial.py @@ -0,0 +1,149 @@ +"""initial schema + +Revision ID: 001_initial +Revises: +Create Date: 2026-05-18 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import UUID, TSVECTOR, ENUM + +# revision identifiers, used by Alembic. +revision = '001_initial' +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Enums + op.execute("CREATE TYPE feed_source_type AS ENUM ('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite')") + op.execute("CREATE TYPE event_source_type AS ENUM ('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite')") + op.execute("CREATE TYPE sentiment_label AS ENUM ('positive', 'neutral', 'negative')") + op.execute("CREATE TYPE entity_type AS ENUM ('person', 'organization', 'location', 'topic', 'asset')") + op.execute("CREATE TYPE alert_type AS ENUM ('entity_mention', 'sentiment_shift', 'geo_proximity', 'keyword_match', 'threshold', 'anomaly')") + op.execute("CREATE TYPE alert_severity AS ENUM ('low', 'medium', 'high', 'critical')") + + # Extensions + op.execute("CREATE EXTENSION IF NOT EXISTS postgis") + op.execute("CREATE EXTENSION IF NOT EXISTS timescaledb") + + # feed_sources + op.create_table( + 'feed_sources', + sa.Column('id', UUID(as_uuid=True), primary_key=True), + sa.Column('name', sa.String(256), nullable=False), + sa.Column('source_type', sa.Enum('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite', name='feed_source_type'), nullable=False), + sa.Column('url', sa.Text()), + sa.Column('config', sa.JSON()), + sa.Column('enabled', sa.Integer, server_default='1', nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + + # events (will become hypertable) + op.create_table( + 'events', + sa.Column('id', UUID(as_uuid=True), primary_key=True), + sa.Column('source_type', sa.Enum('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite', name='event_source_type'), nullable=False, index=True), + sa.Column('source_id', UUID(as_uuid=True)), + sa.Column('title', sa.Text()), + sa.Column('body', sa.Text()), + sa.Column('url', sa.Text()), + sa.Column('sentiment_score', sa.Float()), + sa.Column('sentiment_label', sa.Enum('positive', 'neutral', 'negative', name='sentiment_label')), + sa.Column('location_lat', sa.Float()), + sa.Column('location_lon', sa.Float()), + sa.Column('location_name', sa.String(512)), + sa.Column('entities', sa.JSON()), + sa.Column('tags', sa.JSON()), + sa.Column('raw', sa.JSON()), + sa.Column('ingested_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column('source_timestamp', sa.DateTime(timezone=True), nullable=False), + sa.Column('search_vector', TSVECTOR), + ) + + # Convert events to TimescaleDB hypertable + op.execute("SELECT create_hypertable('events', 'ingested_at', if_not_exists => TRUE)") + + # GIN index for full-text search + op.create_index('ix_events_search_vector', 'events', ['search_vector'], postgresql_using='gin') + # Spatial index + op.create_index('ix_events_location', 'events', ['location_lat', 'location_lon']) + + # entities + op.create_table( + 'entities', + sa.Column('id', UUID(as_uuid=True), primary_key=True), + sa.Column('name', sa.String(512), nullable=False, index=True), + sa.Column('entity_type', sa.Enum('person', 'organization', 'location', 'topic', 'asset', name='entity_type'), nullable=False), + sa.Column('aliases', sa.JSON()), + sa.Column('description', sa.Text()), + sa.Column('metadata', sa.JSON()), + sa.Column('location_lat', sa.Float()), + sa.Column('location_lon', sa.Float()), + sa.Column('event_count', sa.Integer, server_default='0'), + sa.Column('first_seen', sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column('last_seen', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + + # entity_events + op.create_table( + 'entity_events', + sa.Column('entity_id', UUID(as_uuid=True), primary_key=True), + sa.Column('event_id', UUID(as_uuid=True), primary_key=True), + sa.Column('relevance_score', sa.Float()), + sa.Column('linked_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + + # alerts + op.create_table( + 'alerts', + sa.Column('id', UUID(as_uuid=True), primary_key=True), + sa.Column('alert_type', sa.Enum('entity_mention', 'sentiment_shift', 'geo_proximity', 'keyword_match', 'threshold', 'anomaly', name='alert_type'), nullable=False), + sa.Column('entity_id', UUID(as_uuid=True)), + sa.Column('event_id', UUID(as_uuid=True)), + sa.Column('severity', sa.Enum('low', 'medium', 'high', 'critical', name='alert_severity'), nullable=False), + sa.Column('title', sa.Text(), nullable=False), + sa.Column('message', sa.Text()), + sa.Column('context', sa.JSON()), + sa.Column('acknowledged', sa.Integer, server_default='0'), + sa.Column('acknowledged_by', sa.String(256)), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column('resolved_at', sa.DateTime(timezone=True)), + ) + op.create_index('ix_alerts_severity_created', 'alerts', ['severity', 'created_at']) + op.create_index('ix_alerts_entity', 'alerts', ['entity_id']) + + # documents + op.create_table( + 'documents', + sa.Column('id', UUID(as_uuid=True), primary_key=True), + sa.Column('bucket', sa.String(256), nullable=False), + sa.Column('object_key', sa.String(1024), nullable=False), + sa.Column('content_type', sa.String(256)), + sa.Column('size_bytes', sa.Integer()), + sa.Column('description', sa.Text()), + sa.Column('tags', sa.JSON()), + sa.Column('event_id', UUID(as_uuid=True)), + sa.Column('uploaded_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + + +def downgrade() -> None: + op.drop_table('documents') + op.drop_table('alerts') + op.drop_table('entity_events') + op.drop_table('entities') + op.execute("SELECT drop_hypertable('events', cascade => TRUE)") + op.drop_table('events') + op.drop_table('feed_sources') + + # Drop enums + op.execute("DROP TYPE IF EXISTS feed_source_type") + op.execute("DROP TYPE IF EXISTS event_source_type") + op.execute("DROP TYPE IF EXISTS sentiment_label") + op.execute("DROP TYPE IF EXISTS entity_type") + op.execute("DROP TYPE IF EXISTS alert_type") + op.execute("DROP TYPE IF EXISTS alert_severity") diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..767c26a --- /dev/null +++ b/app/database.py @@ -0,0 +1,28 @@ +import os + +from sqlalchemy import MetaData, event, text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +DB_USER = os.getenv("DB_USER", "osint") +DB_PASS = os.getenv("DB_PASSWORD", "") +DB_HOST = os.getenv("DB_HOST", "osint-pgdb-rw.customer1.svc.cluster.local") +DB_PORT = os.getenv("DB_PORT", "5432") +DB_NAME = os.getenv("DB_NAME", "osint_data") + +DATABASE_URL = f"postgresql+asyncpg://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}" + +engine = create_async_engine( + DATABASE_URL, echo=False, pool_size=5, max_overflow=10, pool_recycle=300 +) +async_session = async_sessionmaker( + engine, class_=AsyncSession, expire_on_commit=False +) +metadata = MetaData() + + +async def init_extensions(): + """Initialize PostGIS and TimescaleDB extensions on first connection.""" + async with engine.connect() as conn: + await conn.execute(text("CREATE EXTENSION IF NOT EXISTS postgis")) + await conn.execute(text("CREATE EXTENSION IF NOT EXISTS timescaledb")) + await conn.commit() diff --git a/app/ingest_cron.py b/app/ingest_cron.py new file mode 100644 index 0000000..a0dcfa6 --- /dev/null +++ b/app/ingest_cron.py @@ -0,0 +1,46 @@ +"""CronJob entry point for scheduled ingestion.""" + +import asyncio +import os +import sys +from pathlib import Path + +# Add app dir to path +sys.path.insert(0, str(Path(__file__).parent)) + +from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes +from ingestor import fetch_and_process + +INGESTOR_TYPE = os.getenv("INGESTOR_TYPE", "rss") +RSS_URL = os.getenv("RSS_URL", "") + + +async def main(): + print(f"Starting ingester: {INGESTOR_TYPE}") + + if INGESTOR_TYPE == "rss": + if not RSS_URL: + print("No RSS_URL set, skipping") + return + count = await ingest_rss_feed(RSS_URL) + print(f"RSS: ingested {count} items") + + elif INGESTOR_TYPE == "gdelt": + count = await ingest_gdelt(max_articles=50) + print(f"GDELT: ingested {count} articles") + + elif INGESTOR_TYPE == "earthquake": + count = await ingest_earthquakes() + print(f"Earthquakes: ingested {count} events") + + elif INGESTOR_TYPE == "nats": + count = await fetch_and_process(batch_size=500) + print(f"NATS: processed {count} messages") + + else: + print(f"Unknown ingestor type: {INGESTOR_TYPE}") + sys.exit(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/app/ingestor.py b/app/ingestor.py new file mode 100644 index 0000000..d74d4f0 --- /dev/null +++ b/app/ingestor.py @@ -0,0 +1,107 @@ +"""NATS JetStream consumer — ingests OSINT events from NATS streams.""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timezone + +import nats +from nats.errors import TimeoutError + +from database import async_session +from models import events as events_table + +logger = logging.getLogger("osint.ingestor") + +# NATS connection settings +NATS_URLS = "nats://osint-nats.customer1.svc.cluster.local:4222" +NATS_STREAM = "events" +NATS_DURABLE = "osint-ingestor" + +# Redis cache settings +REDIS_URL = "redis://osint-redis-sentinel.customer1.svc.cluster.local:26379/0" + + +async def ingest_event(msg: dict): + """Ingest a single event from NATS into PostgreSQL.""" + event_row = { + "source_type": msg.get("source_type", "rss"), + "source_id": msg.get("source_id"), + "title": msg.get("title"), + "body": msg.get("body"), + "url": msg.get("url"), + "sentiment_score": msg.get("sentiment_score"), + "sentiment_label": msg.get("sentiment_label"), + "location_lat": msg.get("location_lat"), + "location_lon": msg.get("location_lon"), + "location_name": msg.get("location_name"), + "entities": msg.get("entities", []), + "tags": msg.get("tags", []), + "raw": msg.get("raw"), + "source_timestamp": msg.get("source_timestamp", datetime.now(timezone.utc).isoformat()), + } + + # Parse timestamp if string + if isinstance(event_row["source_timestamp"], str): + event_row["source_timestamp"] = datetime.fromisoformat(event_row["source_timestamp"]) + + async with async_session() as session: + result = await session.execute(events_table.insert().values(**event_row)) + await session.commit() + event_id = result.inserted_primary_key[0] # type: ignore[union-attr] + logger.info("Ingested event %s from source %s", event_id, msg.get("source_type")) + return event_id + + +async def start_nats_consumer(): + """Start NATS JetStream consumer for OSINT events.""" + nc = await nats.connect(NATS_URLS) + js = nc.jetstream() + + # Create stream if not exists + try: + await js.add_stream( + name=NATS_STREAM, + subjects=[ + "events.gdelt", "events.rss", "events.social", + "events.earthquake", "events.disaster", "events.weather", + "events.fire", "events.satellite", "events.new", "events.alert", + ], + retention=nats.js.api.RetentionPolicy.INTERESTS, + max_msgs=1_000_000, + ) + logger.info("Created NATS stream %s", NATS_STREAM) + except Exception: + logger.debug("Stream %s already exists", NATS_STREAM) + + # Create durable consumer + sub = await js.pull_subscribe( + subject="events.>", + durable_name=NATS_DURABLE, + ) + + logger.info("NATS consumer started, durable=%s", NATS_DURABLE) + return nc, sub + + +async def fetch_and_process(batch_size: int = 100): + """Fetch a batch of messages and process them.""" + nc, sub = await start_nats_consumer() + js = nc.jetstream() + + msgs = await sub.fetch(batch_size, timeout=5) + processed = 0 + + for msg in msgs: + try: + data = json.loads(msg.data) + await ingest_event(data) + await msg.ack() + processed += 1 + except Exception: + logger.error("Failed to process message: %s", msg.data, exc_info=True) + + await nc.close() + logger.info("Processed %d messages in batch", processed) + return processed diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..01be05b --- /dev/null +++ b/app/main.py @@ -0,0 +1,603 @@ +"""OSINT Dashboard — FastAPI backend. + +Real-time geospatial OSINT dashboard API: +- Event ingestion via NATS JetStream consumers +- Full-text search across events (PostgreSQL tsvector) +- Entity tracking and relationship mapping +- Alert management +- Document storage (MinIO-backed) +- Sentiment aggregation and timeline analytics +""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from pathlib import Path +from uuid import UUID + +import structlog +from fastapi import FastAPI, HTTPException, Query +from fastapi.responses import FileResponse, HTMLResponse +from sqlalchemy import and_, func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from database import async_session, init_extensions +from models import ( + alerts, documents, entities, entity_events, events, feed_sources +) +from schemas import ( + AlertCreate, AlertOut, AlertSeverity, AlertType, AlertUpdate, + DashboardSummary, EntityCreate, EntityKind, EntityOut, + EventCreate, EventOut, + FeedSourceCreate, FeedSourceOut, + 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 + +logging.basicConfig(level=logging.INFO) +logger = structlog.get_logger("osint.dashboard") + +app = FastAPI( + title="OSINT Dashboard", + description="Real-time geospatial OSINT intelligence dashboard", + version="0.1.0", +) + +STATIC_DIR = Path(__file__).parent / "static" + + +# ── Helpers ─────────────────────────────────────────────────────────────── + +def event_to_out(row: dict) -> EventOut: + """Convert DB row dict to EventOut schema.""" + return EventOut( + id=row["id"], + source_type=row["source_type"], + source_id=row["source_id"], + title=row["title"], + body=row["body"], + url=row["url"], + sentiment_score=row["sentiment_score"], + sentiment_label=row["sentiment_label"], + location_lat=row["location_lat"], + location_lon=row["location_lon"], + location_name=row["location_name"], + entities=row["entities"], + tags=row["tags"], + ingested_at=row["ingested_at"], + source_timestamp=row["source_timestamp"], + ) + + +def entity_to_out(row: dict) -> EntityOut: + """Convert DB row dict to EntityOut schema.""" + return EntityOut( + id=row["id"], + name=row["name"], + entity_type=row["entity_type"], + aliases=row["aliases"], + description=row["description"], + metadata=row["metadata"], + location_lat=row["location_lat"], + location_lon=row["location_lon"], + event_count=row["event_count"], + first_seen=row["first_seen"], + last_seen=row["last_seen"], + ) + + +def alert_to_out(row: dict) -> AlertOut: + """Convert DB row dict to AlertOut schema.""" + return AlertOut( + id=row["id"], + alert_type=row["alert_type"], + entity_id=row["entity_id"], + event_id=row["event_id"], + severity=row["severity"], + title=row["title"], + message=row["message"], + context=row["context"], + acknowledged=bool(row["acknowledged"]), + acknowledged_by=row["acknowledged_by"], + created_at=row["created_at"], + resolved_at=row["resolved_at"], + ) + + +# ── Health ──────────────────────────────────────────────────────────────── + +@app.get("/api/health") +async def health(): + """Health check with database connectivity.""" + async with async_session() as session: + result = await session.execute(select(func.now())) + db_time = result.scalar() + return {"status": "ok", "db_time": db_time.isoformat() if db_time else None} + + +# ── Startup ─────────────────────────────────────────────────────────────── + +@app.on_event("startup") +async def startup(): + """Initialize extensions and run migrations.""" + await init_extensions() + from alembic import command + from alembic.config import Config + alembic_cfg = Config(str(Path(__file__).parent.parent / "alembic.ini")) + command.upgrade(alembic_cfg, "head") + + +# ── Feed Sources ────────────────────────────────────────────────────────── + +@app.get("/api/sources", response_model=list[FeedSourceOut]) +async def list_sources(enabled_only: bool = Query(True)): + """List all configured feed sources.""" + async with async_session() as session: + stmt = select(feed_sources).order_by(feed_sources.c.name) + if enabled_only: + stmt = stmt.where(feed_sources.c.enabled == 1) + rows = (await session.execute(stmt)).mappings().all() + return [FeedSourceOut( + id=r["id"], name=r["name"], source_type=r["source_type"], + url=r["url"], config=r["config"], enabled=bool(r["enabled"]), + created_at=r["created_at"], + ) for r in rows] + + +@app.post("/api/sources", status_code=201) +async def create_source(payload: FeedSourceCreate): + """Add a new feed source.""" + async with async_session() as session: + values = payload.model_dump() + result = await session.execute(feed_sources.insert().values(**values)) + await session.commit() + pk = result.inserted_primary_key[0] # type: ignore + return {"id": str(pk)} + + +@app.patch("/api/sources/{source_id}") +async def update_source(source_id: UUID, payload: dict): + """Update a feed source (e.g., toggle enabled).""" + async with async_session() as session: + row = (await session.execute( + select(feed_sources).where(feed_sources.c.id == source_id) + )).mappings().one_or_none() + if not row: + raise HTTPException(404, "Source not found") + await session.execute( + feed_sources.update() + .where(feed_sources.c.id == source_id) + .values(**payload) + ) + await session.commit() + return {"ok": True} + + +# ── Events ──────────────────────────────────────────────────────────────── + +@app.get("/api/events", response_model=list[EventOut]) +async def list_events( + source_type: SourceType | None = Query(None), + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), +): + """List recent ingested events.""" + async with async_session() as session: + stmt = select(events).order_by(events.c.ingested_at.desc()) + if source_type: + stmt = stmt.where(events.c.source_type == source_type.value) + stmt = stmt.limit(limit).offset(offset) + rows = (await session.execute(stmt)).mappings().all() + return [event_to_out(r) for r in rows] + + +@app.get("/api/events/{event_id}", response_model=EventOut) +async def get_event(event_id: UUID): + """Get a single event by ID.""" + async with async_session() as session: + row = (await session.execute( + select(events).where(events.c.id == event_id) + )).mappings().one_or_none() + if not row: + raise HTTPException(404, "Event not found") + return event_to_out(row) + + +@app.post("/api/events", status_code=201) +async def create_event(payload: EventCreate): + """Manually ingest an event (bypasses NATS).""" + values = payload.model_dump(exclude_unset=True) + if not values.get("source_timestamp"): + values["source_timestamp"] = datetime.now(timezone.utc) + event_id = await ingest_event(values) + return {"id": str(event_id)} + + +# ── Search ──────────────────────────────────────────────────────────────── + +@app.post("/api/search", response_model=SearchResult) +async def search_events(query: SearchQuery): + """Full-text search across events with optional filters.""" + async with async_session() as session: + # Build query with tsvector full-text search (parameterized to avoid SQL injection) + tsquery_param = text("plainto_tsquery('english', :q)") + + base_stmt = select( + events, + func.count().over().label("total") + ).where( + events.c.search_vector.op("@@")(tsquery_param) + ) + + # Apply filters + if query.source_type: + base_stmt = base_stmt.where(events.c.source_type == query.source_type.value) + if query.entity_id: + base_stmt = base_stmt.join( + entity_events, entity_events.c.event_id == events.c.id + ).where(entity_events.c.entity_id == query.entity_id) + if query.sentiment: + base_stmt = base_stmt.where(events.c.sentiment_label == query.sentiment.value) + if query.min_date: + base_stmt = base_stmt.where(events.c.source_timestamp >= query.min_date) + if query.max_date: + base_stmt = base_stmt.where(events.c.source_timestamp <= query.max_date) + if query.min_lat is not None and query.max_lat is not None: + base_stmt = base_stmt.where( + and_( + events.c.location_lat >= query.min_lat, + events.c.location_lat <= query.max_lat, + ) + ) + if query.min_lon is not None and query.max_lon is not None: + base_stmt = base_stmt.where( + and_( + events.c.location_lon >= query.min_lon, + events.c.location_lon <= query.max_lon, + ) + ) + + base_stmt = base_stmt.order_by(events.c.ingested_at.desc()) + base_stmt = base_stmt.limit(query.limit).offset(query.offset) + + result = (await session.execute(base_stmt, {"q": query.q})).mappings().all() + if result: + total = result[0]["total"] + else: + total = 0 + evts = [event_to_out(r) for r in result] + + return SearchResult( + events=evts, + total=total, + has_more=query.offset + len(evts) < total, + ) + + +# ── Entities ────────────────────────────────────────────────────────────── + +@app.get("/api/entities", response_model=list[EntityOut]) +async def list_entities( + entity_type: EntityKind | None = Query(None), + limit: int = Query(50, ge=1, le=500), +): + """List tracked entities.""" + async with async_session() as session: + stmt = select(entities).order_by(entities.c.event_count.desc()) + if entity_type: + stmt = stmt.where(entities.c.entity_type == entity_type.value) + stmt = stmt.limit(limit) + rows = (await session.execute(stmt)).mappings().all() + return [entity_to_out(r) for r in rows] + + +@app.get("/api/entities/{entity_id}", response_model=EntityOut) +async def get_entity(entity_id: UUID): + """Get entity details with recent events.""" + async with async_session() as session: + row = (await session.execute( + select(entities).where(entities.c.id == entity_id) + )).mappings().one_or_none() + if not row: + raise HTTPException(404, "Entity not found") + return entity_to_out(row) + + +@app.post("/api/entities", status_code=201) +async def create_entity(payload: EntityCreate): + """Create or update a tracked entity.""" + async with async_session() as session: + # Check if entity already exists by name + existing = (await session.execute( + select(entities).where(entities.c.name == payload.name) + )).mappings().one_or_none() + + if existing: + # Update + updates = payload.model_dump(exclude_unset=True) + updates["last_seen"] = datetime.now(timezone.utc) + await session.execute( + entities.update() + .where(entities.c.id == existing["id"]) + .values(**updates) + ) + await session.commit() + return {"id": str(existing["id"]), "created": False} + + # Create + values = payload.model_dump() + result = await session.execute(entities.insert().values(**values)) + await session.commit() + pk = result.inserted_primary_key[0] # type: ignore + return {"id": str(pk), "created": True} + + +@app.get("/api/entities/{entity_id}/events", response_model=list[EventOut]) +async def get_entity_events( + entity_id: UUID, + limit: int = Query(50, ge=1, le=500), +): + """Get events linked to a specific entity.""" + async with async_session() as session: + stmt = ( + select(events) + .join(entity_events, entity_events.c.event_id == events.c.id) + .where(entity_events.c.entity_id == entity_id) + .order_by(events.c.source_timestamp.desc()) + .limit(limit) + ) + rows = (await session.execute(stmt)).mappings().all() + return [event_to_out(r) for r in rows] + + +# ── Alerts ──────────────────────────────────────────────────────────────── + +@app.get("/api/alerts", response_model=list[AlertOut]) +async def list_alerts( + severity: AlertSeverity | None = Query(None), + acknowledged: bool | None = Query(None), + entity_id: UUID | None = Query(None), + limit: int = Query(50, ge=1, le=500), +): + """List alerts with optional filters.""" + async with async_session() as session: + stmt = select(alerts).order_by( + alerts.c.severity.desc(), alerts.c.created_at.desc() + ) + if severity: + stmt = stmt.where(alerts.c.severity == severity.value) + if acknowledged is not None: + stmt = stmt.where(alerts.c.acknowledged == int(acknowledged)) + if entity_id: + stmt = stmt.where(alerts.c.entity_id == entity_id) + stmt = stmt.limit(limit) + rows = (await session.execute(stmt)).mappings().all() + return [alert_to_out(r) for r in rows] + + +@app.post("/api/alerts", status_code=201) +async def create_alert(payload: AlertCreate): + """Create a new alert.""" + async with async_session() as session: + values = payload.model_dump() + result = await session.execute(alerts.insert().values(**values)) + await session.commit() + pk = result.inserted_primary_key[0] # type: ignore + return {"id": str(pk)} + + +@app.patch("/api/alerts/{alert_id}") +async def update_alert(alert_id: UUID, payload: AlertUpdate): + """Update alert (acknowledge, resolve).""" + async with async_session() as session: + row = (await session.execute( + select(alerts).where(alerts.c.id == alert_id) + )).mappings().one_or_none() + if not row: + raise HTTPException(404, "Alert not found") + updates = payload.model_dump(exclude_unset=True) + if "acknowledged" in updates: + updates["acknowledged"] = int(updates["acknowledged"]) + await session.execute( + alerts.update().where(alerts.c.id == alert_id).values(**updates) + ) + await session.commit() + return {"ok": True} + + +# ── Documents ───────────────────────────────────────────────────────────── + +@app.get("/api/documents", response_model=dict) +async def list_documents( + limit: int = Query(50, ge=1, le=500), + offset: int = Query(0, ge=0), +): + """List documents indexed in MinIO.""" + async with async_session() as session: + stmt = select(documents).order_by(documents.c.uploaded_at.desc()).limit(limit).offset(offset) + rows = (await session.execute(stmt)).mappings().all() + return { + "documents": [{ + "id": str(r["id"]), "bucket": r["bucket"], "object_key": r["object_key"], + "content_type": r["content_type"], "size_bytes": r["size_bytes"], + "description": r["description"], "tags": r["tags"], + "event_id": str(r["event_id"]) if r["event_id"] else None, + "uploaded_at": r["uploaded_at"].isoformat() if r["uploaded_at"] else None, + } for r in rows], + } + + +# ── Ingestion Triggers ─────────────────────────────────────────────────── + +@app.post("/api/ingest/rss") +async def trigger_rss_ingest(feed_url: str, source_id: str | None = None): + """Trigger RSS feed ingestion.""" + count = await ingest_rss_feed(feed_url, source_id) + return {"status": "ok", "items_ingested": count} + + +@app.post("/api/ingest/gdelt") +async def trigger_gdelt_ingest(query: str = "", max_articles: int = 50): + """Trigger GDELT data ingestion.""" + count = await ingest_gdelt(query, max_articles) + return {"status": "ok", "articles_ingested": count} + + +@app.post("/api/ingest/earthquakes") +async def trigger_earthquake_ingest(): + """Trigger USGS earthquake ingestion.""" + count = await ingest_earthquakes() + return {"status": "ok", "events_ingested": count} + + +@app.post("/api/ingest/social") +async def trigger_social_ingest(query: str = "", max_items: int = 50): + """Trigger social signals ingestion.""" + count = await ingest_social_signals(query, max_items) + return {"status": "ok", "signals_ingested": count} + + +@app.post("/api/ingest/process") +async def trigger_nats_processing(batch_size: int = 100): + """Process pending NATS JetStream messages.""" + count = await fetch_and_process(batch_size) + return {"status": "ok", "processed": count} + + +# ── Analytics / Aggregation ─────────────────────────────────────────────── + +@app.get("/api/analytics/summary", response_model=DashboardSummary) +async def get_dashboard_summary(): + """Dashboard overview: event counts, sentiment, top entities, alerts.""" + async with async_session() as session: + now = datetime.now(timezone.utc) + yesterday = now - timedelta(hours=24) + + # Total events + total = (await session.execute( + select(func.count()).select_from(events) + )).scalar() or 0 + + # Events in last 24h + events_24h = (await session.execute( + select(func.count()).where(events.c.ingested_at >= yesterday) + )).scalar() or 0 + + # Active sources + active = (await session.execute( + select(func.count()).where(feed_sources.c.enabled == 1) + )).scalar() or 0 + + # Open alerts + open_alerts = (await session.execute( + select(func.count()).where(alerts.c.acknowledged == 0) + )).scalar() or 0 + + # Tracked entities + ent_count = (await session.execute( + select(func.count()).select_from(entities) + )).scalar() or 0 + + # Sentiment breakdown (last 24h) + def sentiment_query(): + return select( + func.count().where(events.c.sentiment_label == "positive").label("pos"), + func.count().where(events.c.sentiment_label == "neutral").label("neu"), + func.count().where(events.c.sentiment_label == "negative").label("neg"), + func.avg(events.c.sentiment_score).label("avg"), + ).where(events.c.ingested_at >= yesterday) + + sent_row = (await session.execute(sentiment_query())).mappings().one() + sentiment = SentimentSummary( + period="24h", + positive_count=sent_row["pos"] or 0, + neutral_count=sent_row["neu"] or 0, + negative_count=sent_row["neg"] or 0, + avg_score=float(sent_row["avg"] or 0), + ) + + # Top entities by event count + top_ent = (await session.execute( + select(entities).order_by(entities.c.event_count.desc()).limit(10) + )).mappings().all() + + return DashboardSummary( + total_events=total, + events_last_24h=events_24h, + active_sources=active, + open_alerts=open_alerts, + tracked_entities=ent_count, + sentiment=sentiment, + top_entities=[entity_to_out(r) for r in top_ent], + ) + + +@app.get("/api/analytics/timeline") +async def get_timeline( + hours: int = Query(24, ge=1, le=168), + bucket_hours: int = Query(1, ge=1, le=24), +): + """Event timeline: counts and avg sentiment per time bucket.""" + async with async_session() as session: + cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) + # Use date_trunc for bucketing + buckets = await session.execute(text(f""" + SELECT + date_trunc('hour', source_timestamp) AS ts, + COUNT(*) AS event_count, + COALESCE(AVG(sentiment_score), 0) AS avg_sentiment + FROM events + WHERE source_timestamp >= :cutoff + GROUP BY ts + ORDER BY ts + """), {"cutoff": cutoff}) + rows = buckets.mappings().all() + + return [TimelinePoint(timestamp=r["ts"], event_count=r["event_count"], + avg_sentiment=float(r["avg_sentiment"])) for r in rows] + + +@app.get("/api/analytics/sentiment/by-source") +async def sentiment_by_source(hours: int = 24): + """Sentiment breakdown grouped by source type.""" + async with async_session() as session: + cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) + result = await session.execute(text(f""" + SELECT + source_type, + COUNT(*) AS total, + COUNT(*) FILTER (WHERE sentiment_label = 'positive') AS positive, + COUNT(*) FILTER (WHERE sentiment_label = 'neutral') AS neutral, + COUNT(*) FILTER (WHERE sentiment_label = 'negative') AS negative, + COALESCE(AVG(sentiment_score), 0) AS avg_score + FROM events + WHERE ingested_at >= :cutoff + GROUP BY source_type + ORDER BY total DESC + """), {"cutoff": cutoff}) + rows = result.mappings().all() + return [{ + "source_type": r["source_type"], + "total": r["total"], + "positive": r["positive"], + "neutral": r["neutral"], + "negative": r["negative"], + "avg_score": float(r["avg_score"]), + } for r in rows] + + +# ── Frontend ────────────────────────────────────────────────────────────── + +@app.get("/", response_class=HTMLResponse) +async def index(): + return FileResponse(str(STATIC_DIR / "index.html")) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..6f052b2 --- /dev/null +++ b/app/models.py @@ -0,0 +1,147 @@ +"""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, +) +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", 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) + + +# ── 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()), +) diff --git a/app/requirements.txt b/app/requirements.txt new file mode 100644 index 0000000..3edff72 --- /dev/null +++ b/app/requirements.txt @@ -0,0 +1,13 @@ +fastapi>=0.115 +uvicorn[standard]>=0.34 +sqlalchemy[asyncio]>=2.0 +asyncpg>=0.30 +nats-py>=2.9 +redis[hiredis]>=5.2 +minio>=7.2 +pydantic>=2.10 +alembic>=1.14 +httpx>=0.28 +feedparser>=6.0 +python-dateutil>=2.9 +structlog>=24.4 diff --git a/app/schemas.py b/app/schemas.py new file mode 100644 index 0000000..1cac012 --- /dev/null +++ b/app/schemas.py @@ -0,0 +1,242 @@ +"""OSINT Dashboard — Pydantic schemas.""" + +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import Optional +from uuid import UUID + +from pydantic import BaseModel, Field + + +# ─── Enums ─────────────────────────────────────────────────────────────── + +class SourceType(str, Enum): + rss = "rss" + gdel_t2 = "gdel-t2" + social = "social" + earthquake = "earthquake" + disaster = "disaster" + weather = "weather" + fire = "fire" + satellite = "satellite" + + +class EntityKind(str, Enum): + person = "person" + organization = "organization" + location = "location" + topic = "topic" + asset = "asset" + + +class Sentiment(str, Enum): + positive = "positive" + neutral = "neutral" + negative = "negative" + + +class AlertType(str, Enum): + entity_mention = "entity_mention" + sentiment_shift = "sentiment_shift" + geo_proximity = "geo_proximity" + keyword_match = "keyword_match" + threshold = "threshold" + anomaly = "anomaly" + + +class AlertSeverity(str, Enum): + low = "low" + medium = "medium" + high = "high" + critical = "critical" + + +# ─── Feed Sources ─────────────────────────────────────────────────────── + +class FeedSourceCreate(BaseModel): + name: str + source_type: SourceType + url: Optional[str] = None + config: Optional[dict] = None + + +class FeedSourceOut(BaseModel): + id: UUID + name: str + source_type: SourceType + url: Optional[str] + config: Optional[dict] + enabled: bool + created_at: datetime + + +# ─── Events ───────────────────────────────────────────────────────────── + +class EventCreate(BaseModel): + source_type: SourceType + source_id: Optional[UUID] = None + title: Optional[str] = None + body: Optional[str] = None + url: Optional[str] = None + sentiment_score: Optional[float] = None + sentiment_label: Optional[Sentiment] = None + location_lat: Optional[float] = None + location_lon: Optional[float] = None + location_name: Optional[str] = None + entities: Optional[list[dict]] = None + tags: Optional[list[str]] = None + raw: Optional[dict] = None + source_timestamp: Optional[datetime] = None + + +class EventOut(BaseModel): + id: UUID + source_type: SourceType + source_id: Optional[UUID] + title: Optional[str] + body: Optional[str] + url: Optional[str] + sentiment_score: Optional[float] + sentiment_label: Optional[Sentiment] + location_lat: Optional[float] + location_lon: Optional[float] + location_name: Optional[str] + entities: Optional[list[dict]] + tags: Optional[list[str]] + ingested_at: datetime + source_timestamp: datetime + + +# ─── Search ────────────────────────────────────────────────────────────── + +class SearchQuery(BaseModel): + q: str = Field(..., min_length=1, max_length=500) + source_type: Optional[SourceType] = None + entity_id: Optional[UUID] = None + sentiment: Optional[Sentiment] = None + min_date: Optional[datetime] = None + max_date: Optional[datetime] = None + min_lat: Optional[float] = None + max_lat: Optional[float] = None + min_lon: Optional[float] = None + max_lon: Optional[float] = None + limit: int = Field(50, ge=1, le=500) + offset: int = Field(0, ge=0) + + +class SearchResult(BaseModel): + events: list[EventOut] + total: int + has_more: bool + + +# ─── Entities ──────────────────────────────────────────────────────────── + +class EntityCreate(BaseModel): + name: str + entity_type: EntityKind + aliases: Optional[list[str]] = None + description: Optional[str] = None + metadata: Optional[dict] = None + location_lat: Optional[float] = None + location_lon: Optional[float] = None + + +class EntityOut(BaseModel): + id: UUID + name: str + entity_type: EntityKind + aliases: Optional[list[str]] + description: Optional[str] + metadata: Optional[dict] + location_lat: Optional[float] + location_lon: Optional[float] + event_count: int + first_seen: datetime + last_seen: datetime + + +# ─── Alerts ────────────────────────────────────────────────────────────── + +class AlertCreate(BaseModel): + alert_type: AlertType + entity_id: Optional[UUID] = None + event_id: Optional[UUID] = None + severity: AlertSeverity + title: str + message: Optional[str] = None + context: Optional[dict] = None + + +class AlertUpdate(BaseModel): + acknowledged: Optional[bool] = None + acknowledged_by: Optional[str] = None + resolved_at: Optional[datetime] = None + + +class AlertOut(BaseModel): + id: UUID + alert_type: AlertType + entity_id: Optional[UUID] + event_id: Optional[UUID] + severity: AlertSeverity + title: str + message: Optional[str] + context: Optional[dict] + acknowledged: bool + acknowledged_by: Optional[str] + created_at: datetime + resolved_at: Optional[datetime] + + +# ─── Documents ─────────────────────────────────────────────────────────── + +class DocumentCreate(BaseModel): + bucket: str + object_key: str + content_type: Optional[str] = None + size_bytes: Optional[int] = None + description: Optional[str] = None + tags: Optional[list[str]] = None + event_id: Optional[UUID] = None + + +class DocumentOut(BaseModel): + id: UUID + bucket: str + object_key: str + content_type: Optional[str] + size_bytes: Optional[int] + description: Optional[str] + tags: Optional[list[str]] + event_id: Optional[UUID] + uploaded_at: datetime + + +# ─── Aggregations ──────────────────────────────────────────────────────── + +class SentimentSummary(BaseModel): + period: str + positive_count: int + neutral_count: int + negative_count: int + avg_score: float + + +class TimelinePoint(BaseModel): + timestamp: datetime + event_count: int + avg_sentiment: float + + +class DashboardSummary(BaseModel): + total_events: int + events_last_24h: int + active_sources: int + open_alerts: int + tracked_entities: int + sentiment: SentimentSummary + top_entities: list[EntityOut] + diff --git a/app/sources.py b/app/sources.py new file mode 100644 index 0000000..8ca87ff --- /dev/null +++ b/app/sources.py @@ -0,0 +1,186 @@ +"""Data source ingestors — fetch from external APIs and push to NATS.""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime + +import httpx +import feedparser +import nats + +logger = logging.getLogger("osint.sources") + + +def _parse_rfc822(date_str: object) -> str | None: + """Parse RFC-822 date string from feedparser entries.""" + if not isinstance(date_str, str): + return None + try: + return parsedate_to_datetime(date_str).astimezone(timezone.utc).isoformat() + except (ValueError, TypeError): + return None + +# NATS connection +NATS_URLS = "nats://osint-nats.customer1.svc.cluster.local:4222" + + +async def publish_event(subject: str, event: dict): + """Publish an event to NATS JetStream.""" + nc = await nats.connect(NATS_URLS) + js = nc.jetstream() + await js.publish(subject, json.dumps(event).encode()) + await nc.close() + logger.debug("Published event to %s", subject) + + +# ─── RSS Feed Ingestor ────────────────────────────────────────────────── + +async def ingest_rss_feed(feed_url: str, source_id: str = None): + """Fetch and parse an RSS feed, publish items to NATS.""" + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.get(feed_url) + resp.raise_for_status() + feed = feedparser.parse(resp.text) + + count = 0 + for entry in feed.entries[:100]: # max 100 per run + event = { + "source_type": "rss", + "source_id": source_id, + "title": entry.get("title"), + "body": entry.get("summary") or entry.get("description"), + "url": entry.get("link"), + "source_timestamp": _parse_rfc822(entry.get("published")) + or datetime.now(timezone.utc).isoformat(), + "tags": [t.get("term") for t in entry.get("tags", []) if t.get("term")], + "raw": { + "feed_title": feed.feed.get("title"), + "author": entry.get("author"), + "categories": [c.get("term") for c in entry.get("categories", [])], + }, + } + await publish_event("events.rss", event) + count += 1 + + logger.info("Ingested %d items from RSS feed %s", count, feed_url) + return count + + +# ─── GDELT 2.0 Ingestor ───────────────────────────────────────────────── + +GDELT_API = "https://api.gdeltproject.org/gdeltv2" + + +async def ingest_gdelt(query: str = "", max_articles: int = 50): + """Fetch articles from GDELT 2.0 API.""" + params = { + "mode": "artlist", + "format": "json", + "maxrecords": max_articles, + "mode": "artlist", + } + if query: + params["search"] = query + + async with httpx.AsyncClient(timeout=60) as client: + resp = await client.get(GDELT_API, params=params) + resp.raise_for_status() + data = resp.json() + + count = 0 + for article in data.get("articles", []): + event = { + "source_type": "gdel-t2", + "title": article.get("title"), + "body": article.get("articleBody"), + "url": article.get("url"), + "sentiment_score": _parse_gdelt_tone(article.get("Tone", "0")), + "location_lat": article.get("Latitude"), + "location_lon": article.get("Longitude"), + "location_name": article.get("Location"), + "source_timestamp": article.get("FirstCreated"), + "entities": [ + {"name": e.get("Topic"), "type": "topic"} + for e in article.get("Mentions", []) + if e.get("Topic") + ], + "raw": article, + } + await publish_event("events.gdelt", event) + count += 1 + + logger.info("Ingested %d articles from GDELT", count) + return count + + +def _parse_gdelt_tone(tone: str) -> float | None: + """Parse GDELT tone string to a -1..1 sentiment score.""" + try: + tone_float = float(tone) + return max(-1.0, min(1.0, tone_float / 4249.0)) # GDELT tone range + except (ValueError, TypeError): + return None + + +# ─── Earthquake Ingestor (USGS) ───────────────────────────────────────── + +USGS_API = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson" + + +async def ingest_earthquakes(): + """Fetch recent earthquakes from USGS.""" + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.get(USGS_API) + resp.raise_for_status() + data = resp.json() + + count = 0 + for feature in data.get("features", []): + props = feature.get("properties", {}) + geometry = feature.get("geometry", {}).get("coordinates", []) + event = { + "source_type": "earthquake", + "title": props.get("title"), + "body": props.get("description"), + "url": props.get("url"), + "location_lat": geometry[1] if len(geometry) > 1 else None, + "location_lon": geometry[0] if len(geometry) > 0 else None, + "location_name": props.get("place"), + "sentiment_label": "neutral", + "tags": [f"magnitude:{props.get('mag')}"] if props.get("mag") else [], + "source_timestamp": ( + datetime.utcfromtimestamp(props.get("time", 0) / 1000) + .replace(tzinfo=timezone.utc) + .isoformat() + ), + "raw": props, + } + await publish_event("events.earthquake", event) + count += 1 + + logger.info("Ingested %d earthquake events", count) + return count + + +# ─── Social Signals (Twitter/X-like placeholder) ──────────────────────── + +async def ingest_social_signals(query: str = "", max_items: int = 50): + """Placeholder for social media signal ingestion. + + In production, this would connect to Twitter API, Reddit, NewsAPI, etc. + For now, it publishes a heartbeat to signal the pipeline is active. + """ + event = { + "source_type": "social", + "title": f"Social signal scan: {query}", + "body": f"Scanned for '{query}' — placeholder connector", + "source_timestamp": datetime.now(timezone.utc).isoformat(), + "tags": [query] if query else [], + "raw": {"query": query, "max_items": max_items, "connector": "placeholder"}, + } + await publish_event("events.social", event) + logger.info("Social signal scan complete for '%s'", query) + return 1 diff --git a/app/static/index.html b/app/static/index.html new file mode 100644 index 0000000..dbf3f8b --- /dev/null +++ b/app/static/index.html @@ -0,0 +1,307 @@ + + + + + + OSINT Dashboard + + + +
+

OSINT Dashboard

+
+ Status: checking... +  |  Last update: - +
+
+ +
+ +
+

Total Events

-
+

Events (24h)

-
last 24 hours
+

Active Sources

-
+

Open Alerts

-
+

Tracked Entities

-
+
+

Sentiment (24h)

+
+
+
+
+
+
+
+
+ + +
+

Search Events

+
+ + +
+
+ + +
+ + + + +
+ + +
+

Recent Events

+ + + +
TimeSourceTitleSentimentLocation
+
+ + + + + + + + + + + + +
+ + + +