""" GDELT Collector — event-centric GDELT news article collection. Collects articles for window: event_date - 1d to event_date + 1d. Deduplicates by URL before inserting. Historical coverage: GDELT V2 DOC API covers from 2017-01-01 onwards. Rate limit: GDELT uses a global per-IP quota shared across all users worldwide. - _MIN_INTERVAL: process-wide minimum gap between calls (enforced via asyncio.Lock) - On 429: exponential backoff — 30s, 60s, 120s (up to _MAX_RETRIES attempts) - GDELT collection must NEVER be triggered on-demand from user-facing endpoints. Use the scheduler (POST /admin/collect/gdelt/{ticker}) only. """ import asyncio import logging from datetime import date, datetime, timedelta, timezone from typing import Optional from urllib.parse import urlparse import httpx from sqlalchemy import select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from app.models.attention import CompanyEntityMap, GdeltArticleRaw logger = logging.getLogger(__name__) _GDELT_DOC_API = "https://api.gdeltproject.org/api/v2/doc/doc" _MAX_RECORDS = 250 _MIN_INTERVAL = 10.0 # minimum seconds between any two GDELT calls (process-wide) _RETRY_BACKOFF = [30, 60, 120] # wait times on successive 429s _MAX_RETRIES = len(_RETRY_BACKOFF) # Earliest date covered by GDELT V2 DOC API GDELT_EARLIEST_DATE = date(2017, 1, 1) # Process-wide rate guard — enforced inside collect_gdelt_articles() _gdelt_lock = asyncio.Lock() _last_gdelt_call_ts: float = 0.0 def _date_to_gdelt_ts(d: date, end_of_day: bool = False) -> str: """Format date as GDELT datetime string YYYYMMDDHHMMSS.""" if end_of_day: return d.strftime("%Y%m%d235959") return d.strftime("%Y%m%d000000") async def _do_gdelt_request(params: dict, ticker: str) -> Optional[dict]: """Fire the GDELT HTTP request with retry/backoff on 429. Returns parsed JSON or None.""" for attempt in range(_MAX_RETRIES + 1): try: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.get(_GDELT_DOC_API, params=params) if resp.status_code == 429: if attempt < _MAX_RETRIES: wait = _RETRY_BACKOFF[attempt] logger.warning( "GDELT rate limit (429) for %s — retry %d/%d in %ds", ticker, attempt + 1, _MAX_RETRIES, wait, ) await asyncio.sleep(wait) continue else: logger.error( "GDELT rate limit persists after %d retries for %s — giving up", _MAX_RETRIES, ticker, ) return None resp.raise_for_status() if not resp.content.strip(): logger.warning("GDELT returned empty response for %s", ticker) return None try: return resp.json() except Exception: logger.warning("GDELT returned non-JSON for %s: %r", ticker, resp.text[:200]) return None except httpx.HTTPStatusError as exc: logger.error("GDELT API HTTP error %s: %s", exc.response.status_code, exc) raise except Exception as exc: logger.error("GDELT API request failed: %s", exc) raise return None def _extract_domain(url: str) -> Optional[str]: try: return urlparse(url).netloc or None except Exception: return None async def collect_gdelt_articles( db: AsyncSession, ticker: str, event_date: date, ) -> int: """Collect GDELT articles for event_date ± 1 day. Returns number of new records inserted. Raises ValueError if no entity mapping or gdelt_query exists. """ global _last_gdelt_call_ts ticker = ticker.upper() if event_date < GDELT_EARLIEST_DATE: logger.warning( "GDELT data not available before %s (requested %s) — skipping %s", GDELT_EARLIEST_DATE, event_date, ticker, ) return 0 result = await db.execute( select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker) ) entity = result.scalars().first() if not entity: raise ValueError(f"No entity mapping found for {ticker!r}. Run entity resolution first.") if not entity.gdelt_query: raise ValueError(f"Entity {ticker!r} has no gdelt_query.") gdelt_query = entity.gdelt_query start_date = event_date - timedelta(days=1) end_date = event_date + timedelta(days=1) start_ts = _date_to_gdelt_ts(start_date, end_of_day=False) end_ts = _date_to_gdelt_ts(end_date, end_of_day=True) logger.info( "Fetching GDELT articles for %s query=%r window=%s→%s", ticker, gdelt_query, start_ts, end_ts, ) params = { "query": gdelt_query, "mode": "ArtList", "maxrecords": str(_MAX_RECORDS), "format": "json", "startdatetime": start_ts, "enddatetime": end_ts, } # Global process-wide rate guard: at most 1 call per _MIN_INTERVAL seconds. # Uses a lock so concurrent callers queue up rather than both firing. async with _gdelt_lock: import time as _time elapsed = _time.monotonic() - _last_gdelt_call_ts wait = max(0.0, _MIN_INTERVAL - elapsed) if wait > 0: logger.debug("GDELT rate guard: sleeping %.1fs before request", wait) await asyncio.sleep(wait) data = None try: data = await _do_gdelt_request(params, ticker) finally: _last_gdelt_call_ts = _time.monotonic() if data is None: return 0 articles = data.get("articles", []) if isinstance(data, dict) else [] if not articles: logger.info("No GDELT articles returned for %s", ticker) return 0 # Collect existing URLs to deduplicate urls = [a.get("url", "") for a in articles if a.get("url")] existing_result = await db.execute( select(GdeltArticleRaw.url).where(GdeltArticleRaw.url.in_(urls)) ) existing_urls = set(existing_result.scalars().all()) rows_to_insert = [] for article in articles: url = article.get("url", "") if not url or url in existing_urls: continue # Parse published_at from seendate or socialimage timestamp published_at: Optional[datetime] = None seen_date = article.get("seendate", "") if seen_date and len(seen_date) >= 14: try: published_at = datetime( int(seen_date[0:4]), int(seen_date[4:6]), int(seen_date[6:8]), int(seen_date[8:10]), int(seen_date[10:12]), int(seen_date[12:14]), tzinfo=timezone.utc, ) except ValueError: pass rows_to_insert.append({ "url": url, "title": article.get("title"), "domain": _extract_domain(url), "published_at": published_at, "sourcecountry": article.get("sourcecountry"), "matched_ticker": ticker, "match_method": "gdelt_query", "match_confidence": 1.0, }) if not rows_to_insert: logger.info("No new GDELT articles for %s (all duplicates)", ticker) return 0 stmt = ( pg_insert(GdeltArticleRaw) .values(rows_to_insert) .on_conflict_do_nothing(index_elements=["url"]) ) await db.execute(stmt) await db.commit() logger.info("Inserted %d GDELT article records for %s", len(rows_to_insert), ticker) return len(rows_to_insert)