gcloud-lab/apps/osint-dashboard/app/ingest_cron.py
Sirius Devops 93754dcdf6 feat: OSINT Dashboard app + Helm templates
Add FastAPI backend for real-time geospatial OSINT dashboard:
- Full-text search via PostgreSQL tsvector (parameterized queries)
- Entity tracking, alert management, sentiment analytics
- Data ingestion: RSS feeds, GDELT, USGS earthquakes, social signals
- NATS JetStream consumer for event ingestion
- MinIO document storage integration
- Redis caching layer
- Alembic migrations with PostGIS + TimescaleDB extensions
- Single-page dashboard UI with live polling
- OpenTelemetry distributed tracing

Helm chart with infrastructure:
- CNPG PostgreSQL cluster (PostGIS + TimescaleDB)
- NATS JetStream with persistent streams
- MinIO distributed object storage (3 buckets)
- Redis Sentinel (1 primary + 2 replicas)
- NGINX Ingress with TLS and WebSocket support
- Prometheus + Grafana + Alertmanager monitoring stack
- Network policies with default deny
- ConfigMap, CronJob, Deployment, Service templates

Fixes applied during review:
- SQL injection in search endpoint (parameterized :q binding)
- Dockerfile PYTHONPATH mismatch (/app/app -> /app)
- Hardcoded DB credentials in alembic.ini
- RSS timestamp parsing (feedparser published_parsed -> parsedate_to_datetime)
- Removed dead PGVECTOR import
2026-05-21 13:39:52 +00:00

46 lines
1.2 KiB
Python

"""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())