AegisSight-Globe/src/data_push.py

"""Auto-Push: Sendet Globe-Ereignisse an den AegisSight Monitor."""
import asyncio
import logging
import os
import time

import httpx

logger = logging.getLogger("globe.push")

_MONITOR_URL = os.getenv("MONITOR_API_URL", "https://monitor.aegis-sight.de/api/public")
_MONITOR_KEY = os.getenv("MONITOR_API_KEY", "")
_DISASTER_INCIDENT_ID = int(os.getenv("DISASTER_INCIDENT_ID", "0"))

_task = None
_last_push: dict = {"eonet": 0, "usgs": 0}


async def _push_to_monitor(events: list):
    """Sendet Events an den Monitor-Ingest-Endpoint."""
    if not _DISASTER_INCIDENT_ID or not _MONITOR_KEY or not events:
        return 0
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            r = await client.post(
                f"{_MONITOR_URL}/globe-ingest",
                json={"incident_id": _DISASTER_INCIDENT_ID, "events": events},
                headers={"X-API-Key": _MONITOR_KEY},
            )
            if r.status_code == 200:
                data = r.json()
                inserted = data.get("inserted", 0)
                if inserted > 0:
                    logger.info(f"Push: {inserted} Ereignisse an Monitor gesendet")
                return inserted
            else:
                logger.warning(f"Push Fehler: {r.status_code} {r.text[:200]}")
    except Exception as e:
        logger.warning(f"Push Fehler: {e}")
    return 0


async def _push_loop():
    """Periodisch EONET + USGS Daten an Monitor pushen."""
    await asyncio.sleep(30)  # Warten bis Daten geladen
    while True:
        try:
            events = []

            # NASA EONET Katastrophen
            try:
                async with httpx.AsyncClient(timeout=10) as client:
                    r = await client.get("https://eonet.gsfc.nasa.gov/api/v3/events?status=open&limit=50")
                    if r.status_code == 200:
                        for evt in r.json().get("events", []):
                            geom = evt.get("geometry", [])
                            if not geom:
                                continue
                            latest = geom[-1]
                            coords = latest.get("coordinates", [])
                            if len(coords) < 2:
                                continue
                            cats = evt.get("categories", [])
                            cat_name = cats[0]["title"] if cats else "Unbekannt"
                            events.append({
                                "title": f"[{cat_name}] {evt.get('title', '?')}",
                                "source": "NASA EONET",
                                "url": evt.get("link", ""),
                                "description": f"Naturereignis: {evt.get('title', '')}. "
                                              f"Kategorie: {cat_name}. "
                                              f"Quellen: {', '.join(s.get('id','') for s in evt.get('sources',[]))}",
                                "lat": coords[1],
                                "lon": coords[0],
                                "location": evt.get("title", "")[:50],
                                "category": "primary",
                            })
            except Exception as e:
                logger.warning(f"EONET fetch: {e}")

            # USGS Erdbeben (nur M4.5+ fuer Monitor — kleinere sind zu viele)
            try:
                async with httpx.AsyncClient(timeout=10) as client:
                    r = await client.get("https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/4.5_day.geojson")
                    if r.status_code == 200:
                        for f in r.json().get("features", []):
                            c = f["geometry"]["coordinates"]
                            p = f["properties"]
                            mag = p.get("mag", 0)
                            events.append({
                                "title": f"[Erdbeben M{mag:.1f}] {p.get('place', '?')}",
                                "source": "USGS Earthquake",
                                "url": p.get("url", ""),
                                "description": f"Erdbeben der Staerke {mag:.1f} bei {p.get('place', '?')}. "
                                              f"Tiefe: {c[2]:.0f} km. "
                                              f"Zeit: {p.get('time', '')}",
                                "lat": c[1],
                                "lon": c[0],
                                "location": p.get("place", "")[:50],
                                "category": "primary" if mag >= 6 else "secondary",
                            })
            except Exception as e:
                logger.warning(f"USGS fetch: {e}")

            if events:
                await _push_to_monitor(events)

        except Exception as e:
            logger.warning(f"Push loop error: {e}")

        await asyncio.sleep(600)  # Alle 10 Minuten


def start_push_service():
    """Startet den Auto-Push Background-Task."""
    global _task
    if not _DISASTER_INCIDENT_ID:
        logger.info("Push: DISASTER_INCIDENT_ID nicht gesetzt, Push deaktiviert")
        return
    if _task is None or _task.done():
        _task = asyncio.create_task(_push_loop())
        logger.info(f"Push-Service gestartet (Lage-ID: {_DISASTER_INCIDENT_ID})")