"""Fresh feeder — Pilon 4 of the cache freshness defense. Polls RSS feeds at volatility-aware intervals and POSTs new articles to ``/v1/ingest`` so the brain corpus stays current. The intervals are: - volatile feeds → every 15 min (war, breaking, daily politics) - evolving feeds → every 6 h (economy, climate, science debates) - stable feeds → every 24 h (settled science, history) Each cycle: 1. Fetch all feeds in the tier (parallel httpx). 2. Drop items outside the freshness window (default: ≥1h old, ≤3 days old). 3. Dedup by URL within the batch. 4. POST to /v1/ingest with run_extraction=true so brain extracts claim atoms in the background. Brain handles deduplication against existing atoms via canonical URL match (see brain_api/services/ingest.py), so resending the same article is idempotent — already-ingested items are skipped silently. """ from __future__ import annotations import asyncio import os from datetime import datetime, timezone # noqa: F401 (timezone used below) from scheduler.brain_client import BrainClient from scheduler.config import ( DEFAULT_EVOLVING_FEEDS, DEFAULT_STABLE_FEEDS, DEFAULT_VOLATILE_FEEDS, parse_csv, settings, ) from scheduler.feed_parser import ( FeedItem, dedup_by_url, fetch_feeds, filter_by_age, ) from shared.logging import get_logger log = get_logger(__name__) def _feed_item_to_evidence(item: FeedItem) -> dict: """Convert a FeedItem to the EvidenceItem shape that /v1/ingest expects. EvidenceItem requires ``retrieved_at`` (when we fetched it) plus optional ``published_at`` (the source's publish date). Ingest.evidence_to_markdown assembles the Document atom body from these fields. """ body = item.full_text or item.summary or "" return { "url": item.url, "title": item.title, "summary": item.summary, "full_text": body, "publisher": item.publisher, "published_at": item.published_at.isoformat(), "retrieved_at": datetime.now(tz=timezone.utc).isoformat(), "credibility_score": _publisher_credibility_score(item.publisher), "relevance_score": 0.5, # neutral; real ranking happens at gather time } # Heuristic credibility tiers — keeps brain_api/services/ingest.py's # credibility_score_to_tag_path mapping coherent. _TIER_1_PUBLISHERS = { "reuters.com", "ap.org", "apnews.com", "bbc.co.uk", "bbc.com", "afp.com", "npr.org", "aljazeera.com", "scientificamerican.com", } _TIER_2_PUBLISHERS = { "digi24.ro", "hotnews.ro", "g4media.ro", } def _publisher_credibility_score(publisher: str) -> float: """Map publisher hostname to a coarse credibility score in [0,1]. Used by brain's credibility_score_to_tag_path to assign Credibility/Tier tags to ingested Document atoms. Conservative defaults — anything we don't recognize gets the middle tier. """ p = publisher.lower().lstrip("www.") if p in _TIER_1_PUBLISHERS: return 0.90 if p in _TIER_2_PUBLISHERS: return 0.70 return 0.50 def _feeds_for_volatility(volatility: str) -> list[str]: """Resolve env override → default for a volatility tier.""" env_var = f"FEED_{volatility.upper()}_URLS" defaults = { "volatile": DEFAULT_VOLATILE_FEEDS, "evolving": DEFAULT_EVOLVING_FEEDS, "stable": DEFAULT_STABLE_FEEDS, }[volatility] return parse_csv(os.environ.get(env_var), defaults) async def _run_one_cycle( *, brain: BrainClient, volatility: str, max_items: int ) -> tuple[int, int]: """Pull feeds for one volatility tier and ingest fresh items. Returns ``(fetched, ingested)`` — fetched is items that passed the age filter, ingested is what brain accepted (could be lower if some were duplicates of existing atoms). """ feeds = _feeds_for_volatility(volatility) if not feeds: return 0, 0 items = await fetch_feeds(feeds) items = filter_by_age( items, min_age_s=settings.feeder_min_published_age_s, max_age_s=settings.feeder_max_published_age_s, ) items = dedup_by_url(items) # Newest first, cap at max_items per cycle so a single feed flood doesn't # overwhelm the LLM-backed extraction pipeline downstream. items.sort(key=lambda i: i.published_at, reverse=True) items = items[:max_items] if not items: return 0, 0 evidence = [_feed_item_to_evidence(i) for i in items] # Tag each batch with its volatility so brain's classifier has a hint # already and the resulting Document atoms can be invalidated by # topic + volatility later. response = await brain.ingest( claim=None, # this is corpus refresh, not a specific user claim evidence=evidence, default_tags=[ f"Volatility/{volatility.capitalize()}", ], run_extraction=True, ) if response is None: log.warning( "feeder_ingest_failed", volatility=volatility, attempted=len(items), ) return len(items), 0 accepted = int(response.get("accepted") or 0) skipped = int(response.get("skipped_duplicate") or 0) log.info( "feeder_cycle_done", volatility=volatility, feeds=len(feeds), fetched=len(items), accepted=accepted, skipped_dup=skipped, errors=int(response.get("errors") or 0), ) return len(items), accepted async def feeder_loop(brain: BrainClient, volatility: str) -> None: """Long-running task — one per volatility tier. Runs forever; each cycle catches its own exceptions so a single failure doesn't kill the loop. """ interval = { "volatile": settings.feeder_volatile_interval_s, "evolving": settings.feeder_evolving_interval_s, "stable": settings.feeder_stable_interval_s, }[volatility] log.info( "feeder_loop_start", volatility=volatility, interval_s=interval, ) while True: try: await _run_one_cycle( brain=brain, volatility=volatility, max_items=settings.feeder_max_items_per_run, ) except Exception as e: # noqa: BLE001 log.exception("feeder_cycle_error", volatility=volatility) await asyncio.sleep(interval) async def run_feeder(brain: BrainClient) -> None: """Launch one loop per volatility tier — runs forever.""" if not settings.feeder_enabled: log.info("feeder_disabled") return await asyncio.gather( feeder_loop(brain, "volatile"), feeder_loop(brain, "evolving"), feeder_loop(brain, "stable"), )