diff --git a/app/ingestor.py b/app/ingestor.py index a44ffd4..29eae1c 100644 --- a/app/ingestor.py +++ b/app/ingestor.py @@ -66,7 +66,7 @@ async def start_nats_consumer(): "events.earthquake", "events.disaster", "events.weather", "events.fire", "events.satellite", "events.new", "events.alert", ], - retention=nats.js.api.RetentionPolicy.INTERESTS, + retention=nats.js.api.RetentionPolicy.LIMITS, max_msgs=1_000_000, ) logger.info("Created NATS stream %s", NATS_STREAM) @@ -76,7 +76,7 @@ async def start_nats_consumer(): # Create durable consumer sub = await js.pull_subscribe( subject="events.>", - durable_name=NATS_DURABLE, + durable=NATS_DURABLE, ) logger.info("NATS consumer started, durable=%s", NATS_DURABLE) diff --git a/app/run_ingester.py b/app/run_ingester.py index 4c0084b..a63dbae 100644 --- a/app/run_ingester.py +++ b/app/run_ingester.py @@ -45,7 +45,7 @@ async def producer_loop() -> None: try: for url in RSS_URLS: try: - n = await ingest_rss_feed(url, source_id=url) + n = await ingest_rss_feed(url) logger.info("RSS %s -> %d items", url, n) except Exception: # noqa: BLE001 logger.exception("RSS fetch failed: %s", url) diff --git a/app/sources.py b/app/sources.py index e7609bd..3d76b34 100644 --- a/app/sources.py +++ b/app/sources.py @@ -40,7 +40,7 @@ async def publish_event(subject: str, event: dict): # ─── RSS Feed Ingestor ────────────────────────────────────────────────── -async def ingest_rss_feed(feed_url: str, source_id: str = None): +async def ingest_rss_feed(feed_url: str): """Fetch and parse an RSS feed, publish items to NATS.""" async with httpx.AsyncClient(timeout=30) as client: resp = await client.get(feed_url) @@ -51,7 +51,6 @@ async def ingest_rss_feed(feed_url: str, source_id: str = None): 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"), @@ -59,6 +58,7 @@ async def ingest_rss_feed(feed_url: str, source_id: str = None): or datetime.now(timezone.utc).isoformat(), "tags": [t.get("term") for t in entry.get("tags", []) if t.get("term")], "raw": { + "feed_url": feed_url, "feed_title": feed.feed.get("title"), "author": entry.get("author"), "categories": [c.get("term") for c in entry.get("categories", [])], diff --git a/docker-compose.yml b/docker-compose.yml index e9a6b11..986e5d2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -70,6 +70,7 @@ services: INGEST_INTERVAL: ${INGEST_INTERVAL:-300} INGEST_EARTHQUAKES: ${INGEST_EARTHQUAKES:-1} command: ["python", "app/run_ingester.py"] + entrypoint: ["python", "app/run_ingester.py"] app: build: