osint-dashboard/app/main.py

"""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 asyncio
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 fastapi.staticfiles import StaticFiles
from sqlalchemy import and_, func, or_, 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, fires,
    articles, article_summaries,
)
from schemas import (
    AlertCreate, AlertOut, AlertSeverity, AlertType, AlertUpdate,
    DashboardSummary, EntityCreate, EntityKind, EntityOut,
    EventCreate, EventOut, FireOut, NewsArticleOut, NewsSummaryOut,
    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
from live_layers import (
    fetch_aircraft, fetch_fire_incidents, fetch_fire_perimeters,
    fetch_radar_meta, fetch_storms, fetch_trains, fetch_vessels,
    fetch_weather_alerts, overlay_catalog, parse_bbox,
)

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.get("source_id"),
        title=row.get("title"),
        body=row.get("body"),
        url=row.get("url"),
        sentiment_score=row.get("sentiment_score"),
        sentiment_label=row.get("sentiment_label"),
        location_lat=row.get("location_lat"),
        location_lon=row.get("location_lon"),
        location_name=row.get("location_name"),
        entities=row.get("entities"),
        tags=row.get("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 PostGIS/TimescaleDB extensions on first connection.

    Schema migrations are applied by the container entrypoint (alembic upgrade
    head) before uvicorn starts, so they don't run nested inside the event loop.
    """
    await init_extensions()
    from config import AISSTREAM_IN_APP
    if AISSTREAM_IN_APP:
        from ais_stream import run_ais_worker
        asyncio.create_task(run_ais_worker())


# ── 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),
    bbox: str | None = Query(
        None,
        description="Comma-separated 'minlon,minlat,maxlon,maxlat' to bound the "
                    "result set by event coordinates. Omit for all stored events.",
    ),
    since: datetime | None = Query(
        None,
        description="Only events ingested at/after this UTC instant "
                    "(ISO 8601, e.g. '2026-08-24T12:00:00Z').",
    ),
    has_coords: bool = Query(
        False,
        description="Only events that carry a location (location_lat/lon set). "
                    "Used by the map's event-blips layer.",
    ),
    limit: int = Query(50, ge=1, le=500),
    offset: int = Query(0, ge=0),
):
    """List recent ingested events."""
    async with async_session() as session:
        if has_coords:
            stmt = select(
                events.c.id, events.c.source_type, events.c.source_id,
                events.c.title, events.c.url,
                events.c.location_lat, events.c.location_lon, events.c.location_name,
                events.c.ingested_at, events.c.source_timestamp,
            ).order_by(events.c.ingested_at.desc())
        else:
            stmt = select(events).order_by(events.c.ingested_at.desc())
        if source_type:
            stmt = stmt.where(events.c.source_type == source_type.value)
        if since:
            stmt = stmt.where(events.c.ingested_at >= since)
        if has_coords:
            stmt = stmt.where(events.c.location_lat.isnot(None))
        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_(
                    events.c.location_lon >= minlon, events.c.location_lon <= maxlon,
                    events.c.location_lat >= minlat, events.c.location_lat <= maxlat,
                )
            )
        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)}


# ── Active Fires / Hotspots (NASA FIRMS) ─────────────────────────────────

def fire_heat_row(r) -> dict:
    """Minimal FIRMS point for the heatmap overlay."""
    return {
        "lat": r["latitude"],
        "lon": r["longitude"],
        "i": r["brightness"],
        "c": r["confidence"],
    }


def camera_map_row(r) -> dict:
    """Minimal camera pin — URLs stay off the list payload."""
    return {
        "id": str(r["id"]),
        "lat": r["location_lat"],
        "lon": r["location_lon"],
        "device_type": r["device_type"],
        "discovery_source": r["discovery_source"],
        "location_name": r["location_name"],
    }


@app.get("/api/fires", response_model=None)
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),
    format: str = Query("full", description="'full' FireOut rows or 'heat' {lat,lon,i,c}"),
):
    """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=...&format=heat` and renders the points.
    """
    fmt = (format or "full").lower()
    if fmt not in ("full", "heat"):
        raise HTTPException(422, "format must be 'full' or 'heat'")
    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()
    if fmt == "heat":
        return [fire_heat_row(r) for r in rows]
    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)
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],
    }


# ── 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")
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/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."""
    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().filter(events.c.sentiment_label == "positive").label("pos"),
                func.count().filter(events.c.sentiment_label == "neutral").label("neu"),
                func.count().filter(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]


# ── Cameras (open-camera discovery map) ───────────────────────────────────

@app.get("/api/cameras")
async def list_cameras(
    bbox: str | None = Query(
        None,
        description="Bounding box 'min_lon,min_lat,max_lon,max_lat'",
    ),
    source: str | None = Query(None, description="Filter by discovery_source"),
    working: bool = Query(
        True,
        description="Only cameras with a verified HTTP/MJPEG snapshot_url "
                    "(the ones that actually preview). Set false to include "
                    "unverified masscan port-554 hits.",
    ),
    limit: int = Query(500, ge=1, le=5000),
):
    """Cameras for map display, optionally filtered by geographic bbox."""
    from camera_models import cameras as cam_table

    async with async_session() as session:
        stmt = select(cam_table).order_by(cam_table.c.last_seen.desc())
        if working:
            stmt = stmt.where(
                cam_table.c.snapshot_url.isnot(None),
                or_(
                    cam_table.c.snapshot_url.startswith("http://"),
                    cam_table.c.snapshot_url.startswith("https://"),
                ),
            )
        if bbox:
            try:
                min_lon, min_lat, max_lon, max_lat = (
                    float(v) for v in bbox.split(",")
                )
            except ValueError:
                raise HTTPException(
                    422, "bbox must be 'min_lon,min_lat,max_lon,max_lat'"
                )
            if not (-180 <= min_lon <= 180 and -180 <= max_lon <= 180
                    and -90 <= min_lat <= 90 and -90 <= max_lat <= 90):
                raise HTTPException(422, "bbox coordinates out of range")
            # Inverted/empty box (wrapped world view) → do not filter.
            if min_lon < max_lon and min_lat < max_lat:
                stmt = stmt.where(
                    and_(cam_table.c.location_lat >= min_lat,
                         cam_table.c.location_lat <= max_lat,
                         cam_table.c.location_lon >= min_lon,
                         cam_table.c.location_lon <= max_lon))
        if source:
            stmt = stmt.where(cam_table.c.discovery_source == source)
        rows = (await session.execute(stmt.limit(limit))).mappings().all()

    return [camera_map_row(r) for r in rows]


def camera_detail_row(r) -> dict:
    return {
        "id": str(r["id"]),
        "source_url": r["source_url"],
        "snapshot_url": r["snapshot_url"],
        "discovery_source": r["discovery_source"],
        "lat": r["location_lat"],
        "lon": r["location_lon"],
        "location_name": r["location_name"],
        "vendor": r["vendor"],
        "device_type": r["device_type"],
        "first_seen": r["first_seen"].isoformat() if r["first_seen"] else None,
        "last_seen": r["last_seen"].isoformat() if r["last_seen"] else None,
    }


@app.get("/api/cameras/{camera_id}")
async def get_camera(camera_id: UUID):
    """Full camera row for a map popup. List endpoint stays slim."""
    from camera_models import cameras as cam_table

    async with async_session() as session:
        row = (await session.execute(
            select(cam_table).where(cam_table.c.id == camera_id)
        )).mappings().one_or_none()
    if not row:
        raise HTTPException(404, "Camera not found")
    return camera_detail_row(row)


@app.get("/api/cameras/{camera_id}/snapshot")
async def camera_snapshot(camera_id: UUID):
    """Still image for one camera.

    HTTP cameras go through the TTL cache. masscan/RTSP finds have no HTTP
    snapshot_url — we probe common still-image paths and, failing that, grab
    one JPEG frame from RTSP via ffmpeg. No credentials are tried.
    """
    from camera_models import cameras as cam_table
    from camera_preview import resolve_preview
    from fastapi.responses import Response

    async with async_session() as session:
        row = (await session.execute(
            select(cam_table).where(cam_table.c.id == camera_id)
        )).mappings().one_or_none()
    if not row:
        raise HTTPException(404, "Camera not found")
    try:
        data, _url = await asyncio.wait_for(resolve_preview(row), timeout=10)
    except asyncio.TimeoutError:
        raise HTTPException(502, "Snapshot unavailable")
    if not data:
        raise HTTPException(502, "Snapshot unavailable")
    return Response(content=data, media_type="image/jpeg")


@app.get("/api/cameras/{camera_id}/stream")
async def camera_stream(camera_id: UUID):
    """Live MJPEG passthrough for one camera.

    Browsers render multipart/x-mixed-replace responses natively inside an
    <img> tag, so proxying the camera's own MJPEG stream through here gives a
    true live preview in the map popup (no player, no JS). Single-frame JPEG
    endpoints also work — they render as a static image.
    """
    from camera_models import cameras as cam_table
    from fastapi.responses import StreamingResponse
    import httpx

    async with async_session() as session:
        row = (await session.execute(
            select(cam_table).where(cam_table.c.id == camera_id)
        )).mappings().one_or_none()
    if not row:
        raise HTTPException(404, "Camera not found")
    url = row["snapshot_url"] or row["source_url"]
    low = str(url or "").lower()
    if low.startswith("rtsp://") or ".m3u8" in low or row.get("device_type") == "hls":
        from camera_preview import ffmpeg_mjpeg_stream, _FFMPEG
        if not _FFMPEG:
            raise HTTPException(502, "Live preview requires ffmpeg")
        if not low.startswith("rtsp://") and not low.startswith(("http://", "https://")):
            raise HTTPException(404, "Camera or snapshot not found")
        return StreamingResponse(
            ffmpeg_mjpeg_stream(url),
            media_type="multipart/x-mixed-replace; boundary=ffmpeg",
        )
    if not str(url or "").lower().startswith(("http://", "https://")):
        raise HTTPException(404, "Camera or snapshot not found")

    # connect timeout short so dead cams fail fast; read timeout None because
    # an MJPEG stream legitimately idles between frames.
    client = httpx.AsyncClient(
        timeout=httpx.Timeout(5.0, read=None), follow_redirects=True,
        headers={"User-Agent": "osint-dashboard-camera-view/1.0"},
    )
    try:
        req = client.build_request("GET", url)
        resp = await client.send(req, stream=True)
        if resp.status_code != 200 or len(resp.headers.get("content-type", "")) == 0:
            await resp.aclose()
            await client.aclose()
            raise HTTPException(502, "Stream unavailable")
    except HTTPException:
        raise
    except Exception:
        await client.aclose()
        raise HTTPException(502, "Stream unavailable")

    async def gen():
        try:
            async for chunk in resp.aiter_bytes():
                yield chunk
        finally:
            await resp.aclose()
            await client.aclose()

    ctype = resp.headers.get("content-type", "")
    media = ctype if "multipart" in ctype.lower() else (
        ctype if "image/" in ctype.lower()
        else "multipart/x-mixed-replace; boundary=frame"
    )
    return StreamingResponse(gen(), media_type=media)


@app.get("/api/cameras/{camera_id}/hls.m3u8")
async def camera_hls_playlist(camera_id: UUID):
    """CORS-safe rewritten HLS playlist for the in-page player."""
    from camera_hls import fetch_playlist
    from camera_models import cameras as cam_table

    async with async_session() as session:
        row = (await session.execute(
            select(cam_table).where(cam_table.c.id == camera_id)
        )).mappings().one_or_none()
    if not row:
        raise HTTPException(404, "Camera not found")
    url = row["snapshot_url"] or row["source_url"] or ""
    if ".m3u8" not in url.lower() and row.get("device_type") != "hls":
        raise HTTPException(404, "Camera is not an HLS feed")
    return await fetch_playlist(str(camera_id), url)


@app.get("/api/cameras/{camera_id}/hlsseg")
async def camera_hls_segment(camera_id: UUID, u: str = Query(..., min_length=8)):
    """Proxy one HLS segment/playlist URI rewritten by hls.m3u8."""
    from camera_hls import fetch_segment
    from camera_models import cameras as cam_table

    async with async_session() as session:
        exists = (await session.execute(
            select(cam_table.c.id).where(cam_table.c.id == camera_id)
        )).scalar_one_or_none()
    if not exists:
        raise HTTPException(404, "Camera not found")
    return await fetch_segment(str(camera_id), u)


# ── News pipeline (scraper + summarizer) ──────────────────────────────────
# Backing data for the frontend news panel. Written by the vendored
# news-scraper (hourly Scrapy crawl) and news-summarizer (hourly Gemini
# map-reduce) services into the shared osint-db.

@app.get("/api/news", response_model=list[NewsArticleOut])
async def list_news(
    domain: str | None = Query(
        None, description="Filter by source domain (e.g. 'www.reuters.com')"
    ),
    since: datetime | None = Query(
        None,
        description="Only articles captured at/after this UTC instant "
                    "(ISO 8601, e.g. '2026-08-24T12:00:00Z').",
    ),
    limit: int = Query(50, ge=1, le=500),
    offset: int = Query(0, ge=0),
    include_content: bool = Query(
        False,
        description="Include full article body. Default false — ticker/list only need title/url.",
    ),
):
    """Most recent scraped news articles (newest first)."""
    async with async_session() as session:
        stmt = select(articles).order_by(
            articles.c.timestamp.desc().nullslast()
        )
        if domain:
            stmt = stmt.where(articles.c.domain == domain)
        if since:
            stmt = stmt.where(articles.c.timestamp >= since)
        stmt = stmt.limit(limit).offset(offset)
        rows = (await session.execute(stmt)).mappings().all()
    return [
        NewsArticleOut(
            id=r["id"], title=r["title"], url=r["url"],
            content=r["content"] if include_content else None, domain=r["domain"],
            timestamp=r["timestamp"],
        )
        for r in rows
    ]


@app.get("/api/news/summaries", response_model=list[NewsSummaryOut])
async def list_news_summaries(
    since: datetime | None = Query(
        None,
        description="Only summaries generated at/after this UTC instant.",
    ),
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
):
    """Most recent master LLM summaries (newest first)."""
    async with async_session() as session:
        stmt = select(article_summaries).order_by(
            article_summaries.c.batch_timestamp.desc().nullslast()
        )
        if since:
            stmt = stmt.where(article_summaries.c.batch_timestamp >= since)
        stmt = stmt.limit(limit).offset(offset)
        rows = (await session.execute(stmt)).mappings().all()
    return [
        NewsSummaryOut(
            id=r["id"], summary_text=r["summary_text"],
            batch_timestamp=r["batch_timestamp"],
        )
        for r in rows
    ]


# ── Frontend ──────────────────────────────────────────────────────────────

@app.get("/", response_class=HTMLResponse)
async def index():
    # Never let browsers serve a stale copy of the app shell: no Cache-Control
    # means heuristic caching, and a stale index.html froze older bugs in users'
    # tabs after deploys. Revalidate every load (ETag still returns 304).
    resp = FileResponse(str(STATIC_DIR / "index.html"))
    resp.headers["Cache-Control"] = "no-cache"
    return resp


# ── NASA GIBS basemap map tab ─────────────────────────────────────────────

def _parse_bbox_query(bbox: str) -> tuple[float, float, float, float]:
    try:
        return parse_bbox(bbox)
    except ValueError as exc:
        raise HTTPException(422, str(exc)) from exc


@app.get("/api/map/layers")
async def map_layers():
    """Curated NASA GIBS raster basemap layers for the map tab.

    Each entry has everything the browser needs to render the WMTS tiles:
      id        — GIBS layer identifier (used in the tile URL path)
      title     — human-readable display name
      tms       — GIBS tile matrix set (GoogleMapsCompatible_LevelN)
      format    — tile image extension (jpeg|png)
      has_time  — whether the layer has a Time dimension (=> date selector)
      max_zoom  — highest native zoom served by that tile matrix set

    ``overlays`` lists toggleable live feeds (radar tiles, aircraft, …).
    """
    from gibs_map import MAP_LAYERS
    return {"layers": MAP_LAYERS, "overlays": overlay_catalog()}


def _upstream_or_502(exc: Exception, name: str):
    logger.warning("live_layer_upstream_failed", layer=name, error=str(exc))
    raise HTTPException(502, f"{name} upstream unavailable: {exc}") from exc


@app.get("/api/map/radar")
async def map_radar():
    """RainViewer frame list + IEM NEXRAD tile template. Browser fetches tiles."""
    try:
        return await fetch_radar_meta()
    except Exception as exc:
        _upstream_or_502(exc, "radar")


@app.get("/api/aircraft")
async def list_aircraft(
    bbox: str = Query(..., description="minlon,minlat,maxlon,maxlat"),
    limit: int = Query(2000, ge=1, le=5000),
):
    """Viewport ADS-B last-known (ADSB.lol). Requires bbox; radius clamped ≤ 150 nm."""
    _parse_bbox_query(bbox)
    try:
        return await fetch_aircraft(bbox, limit)
    except ValueError as exc:
        raise HTTPException(422, str(exc)) from exc
    except Exception as exc:
        _upstream_or_502(exc, "aircraft")


@app.get("/api/trains")
async def list_trains(
    bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
    limit: int = Query(2000, ge=1, le=5000),
):
    """Amtrak / Brightline / VIA last-known (Amtraker). Bbox optional (~200 rows)."""
    if bbox:
        _parse_bbox_query(bbox)
    try:
        return await fetch_trains(bbox, limit)
    except ValueError as exc:
        raise HTTPException(422, str(exc)) from exc
    except Exception as exc:
        _upstream_or_502(exc, "trains")


@app.get("/api/vessels")
async def list_vessels(
    bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
    limit: int = Query(2000, ge=1, le=5000),
):
    """AIS last-known from the server-side AISStream worker. Empty without a key."""
    if bbox:
        _parse_bbox_query(bbox)
    try:
        return await fetch_vessels(bbox, limit)
    except ValueError as exc:
        raise HTTPException(422, str(exc)) from exc


@app.get("/api/fire-incidents")
async def list_fire_incidents(
    bbox: str | None = Query(None),
    limit: int = Query(2000, ge=1, le=5000),
):
    """NIFC WFIGS current incident points."""
    if bbox:
        _parse_bbox_query(bbox)
    try:
        return await fetch_fire_incidents(bbox, limit)
    except Exception as exc:
        _upstream_or_502(exc, "fire-incidents")


@app.get("/api/fire-perimeters")
async def list_fire_perimeters(bbox: str | None = Query(None)):
    """NIFC WFIGS current wildfire perimeters (GeoJSON)."""
    if bbox:
        _parse_bbox_query(bbox)
    try:
        return await fetch_fire_perimeters(bbox)
    except Exception as exc:
        _upstream_or_502(exc, "fire-perimeters")


@app.get("/api/weather-alerts")
async def list_weather_alerts(
    area: str | None = Query(None, description="US state two-letter code, e.g. NC"),
    bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
):
    """Cached NWS active alerts + IEM storm-based warning polygons.

    Named ``/api/weather-alerts`` so it does not collide with dashboard
    ``/api/alerts`` (entity/keyword alert records).
    """
    if bbox:
        _parse_bbox_query(bbox)
    try:
        return await fetch_weather_alerts(area, bbox)
    except Exception as exc:
        _upstream_or_502(exc, "weather-alerts")


@app.get("/api/storms")
async def list_storms():
    """NHC active tropical cyclones."""
    try:
        return await fetch_storms()
    except Exception as exc:
        _upstream_or_502(exc, "storms")


@app.get("/api/map/times")
async def map_layer_times(
    layer: str = Query(..., description="GIBS layer identifier, e.g. VIIRS_SNPP_CorrectedReflectance_TrueColor"),
):
    """Available date windows for a time-aware GIBS layer.

    Parsed from the layer's Domains XML (GIBS serves nearest-time tiles even
    for dates slightly outside a window, but we surface the real ranges so
    the UI can clamp the picker and flag out-of-range picks).
    """
    from gibs_map import _LAYER_BY_ID, fetch_layer_domain

    meta = _LAYER_BY_ID.get(layer)
    if not meta:
        raise HTTPException(404, f"Unknown GIBS layer: {layer}")
    if not meta["has_time"]:
        raise HTTPException(422, f"Layer '{layer}' is static (no Time dimension)")
    try:
        domain = await fetch_layer_domain(meta["id"], meta["tms"])
    except Exception as exc:  # network / GIBS hiccup → degrade gracefully
        logger.warning("gibs_domain_fetch_failed", layer=layer, error=str(exc))
        raise HTTPException(502, f"GIBS time domain unavailable: {exc}")
    return {"layer": layer, **domain}


app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)