diff --git a/app/run_ingester.py b/app/run_ingester.py new file mode 100644 index 0000000..4c0084b --- /dev/null +++ b/app/run_ingester.py @@ -0,0 +1,96 @@ +"""Long-running ingester: pulls OSINT sources into NATS JetStream and +consumes NATS messages into PostgreSQL. + +Runs forever (one process, two concurrent tasks): + * producer loop — fetch RSS / GDELT / USGS on an interval, publish to NATS + * consumer loop — pull from the NATS `events.>` stream, write to Postgres + +Env (all optional, 12-factor): + RSS_URL comma-separated feed URLs to poll (default: none) + INGEST_INTERVAL seconds between producer cycles (default: 300) + GDELT_QUERY GDELT search term (default: "") + INGEST_EARTHQUAKES "1"/"true" to enable USGS feed (default: "1") +""" + +from __future__ import annotations + +import asyncio +import logging +import os +from pathlib import Path + +sys_path = str(Path(__file__).parent) +import sys + +sys.path.insert(0, sys_path) + +from config import NATS_URL # noqa: E402 +from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes # noqa: E402 +from ingestor import ingest_event, start_nats_consumer # noqa: E402 + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") +logger = logging.getLogger("osint.ingester") + +RSS_URLS = [u.strip() for u in os.getenv("RSS_URL", "").split(",") if u.strip()] +INTERVAL = int(os.getenv("INGEST_INTERVAL", "300")) +GDELT_QUERY = os.getenv("GDELT_QUERY", "") +ENABLE_QUAKES = os.getenv("INGEST_EARTHQUAKES", "1").lower() in ("1", "true", "yes") + +NATS_STREAM = "events" + + +async def producer_loop() -> None: + """Periodically fetch external sources and publish to NATS.""" + while True: + try: + for url in RSS_URLS: + try: + n = await ingest_rss_feed(url, source_id=url) + logger.info("RSS %s -> %d items", url, n) + except Exception: # noqa: BLE001 + logger.exception("RSS fetch failed: %s", url) + try: + g = await ingest_gdelt(query=GDELT_QUERY, max_articles=50) + logger.info("GDELT -> %d articles", g) + except Exception: # noqa: BLE001 + logger.exception("GDELT fetch failed") + if ENABLE_QUAKES: + try: + q = await ingest_earthquakes() + logger.info("USGS -> %d events", q) + except Exception: # noqa: BLE001 + logger.exception("USGS fetch failed") + except Exception: # noqa: BLE001 + logger.exception("producer cycle error") + await asyncio.sleep(INTERVAL) + + +async def consumer_loop() -> None: + """Continuously pull NATS messages and persist to Postgres.""" + _, sub = await start_nats_consumer() + while True: + try: + msgs = await sub.fetch(100, timeout=5) + except Exception: # noqa: BLE001 (TimeoutError / no messages) + continue + for msg in msgs: + try: + import json + + data = json.loads(msg.data) + await ingest_event(data) + await msg.ack() + except Exception: # noqa: BLE001 + logger.exception("failed to process message") + + +async def main() -> None: + logger.info( + "ingester starting (rss=%d feeds, gdelt_q=%r, quakes=%s, interval=%ss)", + len(RSS_URLS), GDELT_QUERY, ENABLE_QUAKES, INTERVAL, + ) + await asyncio.gather(producer_loop(), consumer_loop()) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/docker-compose.yml b/docker-compose.yml index 32bc4c7..e9a6b11 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -44,6 +44,33 @@ services: - "127.0.0.1:8222:8222" command: ["-js"] + ingester: + build: + context: . + dockerfile: Dockerfile + platforms: ["linux/arm64"] + image: localhost/osint-dashboard:latest + container_name: osint-ingester + restart: unless-stopped + profiles: ["ingest"] + depends_on: + nats: + condition: service_started + db: + condition: service_healthy + environment: + DB_USER: ${DB_USER:-osint} + DB_PASSWORD: ${DB_PASSWORD:-osint} + DB_HOST: db + DB_PORT: ${DB_PORT:-5432} + DB_NAME: ${DB_NAME:-osint_data} + NATS_URL: ${NATS_URL:-nats://nats:4222} + RSS_URL: ${RSS_URL:-} + GDELT_QUERY: ${GDELT_QUERY:-} + INGEST_INTERVAL: ${INGEST_INTERVAL:-300} + INGEST_EARTHQUAKES: ${INGEST_EARTHQUAKES:-1} + command: ["python", "app/run_ingester.py"] + app: build: context: .