diff --git a/src/world_intel_mcp/sources/conflict.py b/src/world_intel_mcp/sources/conflict.py index 79ad9df..f711afe 100644 --- a/src/world_intel_mcp/sources/conflict.py +++ b/src/world_intel_mcp/sources/conflict.py @@ -7,18 +7,104 @@ and humanitarian dataset metadata (HDX) for the world-intel-mcp server. import asyncio import logging import os +import time from datetime import datetime, timezone, timedelta +import httpx + from ..fetcher import Fetcher logger = logging.getLogger("world-intel-mcp.sources.conflict") # --------------------------------------------------------------------------- -# ACLED: Armed Conflict Location & Event Data +# ACLED: Armed Conflict Location & Event Data (OAuth2) # --------------------------------------------------------------------------- -_ACLED_URL = "https://api.acleddata.com/acled/read" +_ACLED_API_URL = "https://acleddata.com/api/acled/read" +_ACLED_TOKEN_URL = "https://acleddata.com/oauth/token" + +# Module-level token cache (refreshed automatically) +_acled_token: str | None = None +_acled_token_expires: float = 0.0 +_acled_refresh_token: str | None = None +_acled_token_lock = asyncio.Lock() + + +async def _acled_get_token() -> str | None: + """Obtain or refresh an ACLED OAuth2 Bearer token. + + Uses password grant on first call, then refresh_token grant + before expiry. Tokens are cached module-wide. + """ + global _acled_token, _acled_token_expires, _acled_refresh_token + + async with _acled_token_lock: + # Still valid (with 5-min buffer) + if _acled_token and time.time() < _acled_token_expires - 300: + return _acled_token + + email = os.environ.get("ACLED_EMAIL") + password = os.environ.get("ACLED_PASSWORD") + + # Try refresh_token grant first + if _acled_refresh_token: + body = { + "grant_type": "refresh_token", + "refresh_token": _acled_refresh_token, + "client_id": "acled", + } + elif email and password: + body = { + "grant_type": "password", + "username": email, + "password": password, + "client_id": "acled", + } + else: + return None + + try: + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.post( + _ACLED_TOKEN_URL, + data=body, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + resp.raise_for_status() + data = resp.json() + + _acled_token = data["access_token"] + _acled_token_expires = time.time() + data.get("expires_in", 86400) + _acled_refresh_token = data.get("refresh_token", _acled_refresh_token) + logger.info("ACLED OAuth token acquired (expires in %ds)", data.get("expires_in", 0)) + return _acled_token + except Exception as exc: + logger.warning("ACLED OAuth token request failed: %s", exc) + # If refresh failed, try password grant as fallback + if _acled_refresh_token and email and password: + _acled_refresh_token = None + return await _acled_get_token() + return None + + +async def acled_query(fetcher: Fetcher, params: dict, cache_key: str, cache_ttl: int = 900) -> dict | None: + """Shared ACLED API query with OAuth2 auth. + + Returns parsed JSON or None on failure. Used by both conflict and + intelligence modules to avoid duplicating OAuth logic. + """ + token = await _acled_get_token() + if not token: + return None + return await fetcher.get_json( + _ACLED_API_URL, + source="acled", + cache_key=cache_key, + cache_ttl=cache_ttl, + params=params, + headers={"Authorization": f"Bearer {token}"}, + ) async def fetch_acled_events( @@ -27,10 +113,10 @@ async def fetch_acled_events( days: int = 7, limit: int = 100, ) -> dict: - """Fetch recent armed conflict events from the ACLED API v3. + """Fetch recent armed conflict events from the ACLED API. - Requires ``ACLED_ACCESS_TOKEN`` in the environment. Free academic - access can be obtained at https://acleddata.com. + Requires ``ACLED_EMAIL`` and ``ACLED_PASSWORD`` in the environment. + Free access can be obtained at https://acleddata.com. Args: fetcher: Shared HTTP fetcher with caching and circuit breaking. @@ -43,19 +129,10 @@ async def fetch_acled_events( """ now = datetime.now(timezone.utc) - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return { - "error": "ACLED_ACCESS_TOKEN not configured", - "note": "Free academic access at acleddata.com", - } - start_date = (now - timedelta(days=days)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") params: dict = { - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), "limit": limit, "event_date": f"{start_date}|{end_date}", "event_date_where": "BETWEEN", @@ -64,13 +141,16 @@ async def fetch_acled_events( params["country"] = country cache_country = country or "global" - data = await fetcher.get_json( - _ACLED_URL, - source="acled", + data = await acled_query( + fetcher, params, cache_key=f"conflict:acled:{cache_country}:{days}", cache_ttl=900, - params=params, ) + if data is None and not await _acled_get_token(): + return { + "error": "ACLED credentials not configured (ACLED_EMAIL + ACLED_PASSWORD)", + "note": "Free access at acleddata.com", + } if data is None: logger.warning("ACLED API returned no data") diff --git a/src/world_intel_mcp/sources/intelligence.py b/src/world_intel_mcp/sources/intelligence.py index 9c58d69..6129144 100644 --- a/src/world_intel_mcp/sources/intelligence.py +++ b/src/world_intel_mcp/sources/intelligence.py @@ -17,6 +17,7 @@ from datetime import datetime, timezone, timedelta import httpx from ..fetcher import Fetcher +from .conflict import acled_query from ..analysis.focal_points import detect_focal_points from ..analysis.signals import aggregate_country_signals from ..analysis.temporal import TemporalBaseline @@ -48,7 +49,6 @@ _temporal = TemporalBaseline() # Constants # --------------------------------------------------------------------------- -_ACLED_URL = "https://api.acleddata.com/acled/read" _WB_BASE = "https://api.worldbank.org/v2/country" _USGS_ENDPOINT = "https://earthquake.usgs.gov/fdsnws/event/1/query" @@ -186,27 +186,19 @@ async def fetch_country_brief( return values async def _fetch_acled_count() -> int: - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return 0 - start_date = (now - timedelta(days=30)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") - params: dict = { - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), - "limit": 0, - "event_date": f"{start_date}|{end_date}", - "event_date_where": "BETWEEN", - "country": country_code, - } - data = await fetcher.get_json( - _ACLED_URL, - source="acled", + data = await acled_query( + fetcher, + params={ + "limit": 0, + "event_date": f"{start_date}|{end_date}", + "event_date_where": "BETWEEN", + "country": country_code, + }, cache_key=f"intel:acled:count:{country_code}", cache_ttl=900, - params=params, ) if data is None: return 0 @@ -296,39 +288,24 @@ async def fetch_risk_scores( """ now = datetime.now(timezone.utc) - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return { - "error": "ACLED_ACCESS_TOKEN not configured", - "note": "Free academic access at acleddata.com", - "source": "risk-analysis", - "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), - } - start_date = (now - timedelta(days=30)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") - params: dict = { - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), - "limit": 500, - "event_date": f"{start_date}|{end_date}", - "event_date_where": "BETWEEN", - } - - data = await fetcher.get_json( - _ACLED_URL, - source="acled", + data = await acled_query( + fetcher, + params={ + "limit": 500, + "event_date": f"{start_date}|{end_date}", + "event_date_where": "BETWEEN", + }, cache_key="intel:risk:global:30d", cache_ttl=1800, - params=params, ) if data is None: - logger.warning("ACLED API returned no data for risk scoring") return { - "countries": [], - "count": 0, + "error": "ACLED credentials not configured (ACLED_EMAIL + ACLED_PASSWORD)", + "note": "Free academic access at acleddata.com", "source": "risk-analysis", "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), } @@ -419,23 +396,16 @@ async def _instability_single( async def _fetch_acled() -> list[dict]: """Fetch ACLED events for this country.""" - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return [] - - data = await fetcher.get_json( - _ACLED_URL, - source="acled", - cache_key=f"intel:cii2:acled:{country_code}", - cache_ttl=1800, + data = await acled_query( + fetcher, params={ - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), "limit": 500, "event_date": f"{start_date}|{end_date}", "event_date_where": "BETWEEN", "country": country_name, }, + cache_key=f"intel:cii2:acled:{country_code}", + cache_ttl=1800, ) if data is None: return [] @@ -561,32 +531,28 @@ async def _instability_single( async def _instability_multi(fetcher: Fetcher, now: datetime) -> dict: """Compute CII v2 instability index for focus countries using ACLED.""" - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return { - "error": "ACLED_ACCESS_TOKEN not configured", - "source": "instability-index-v2", - "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), - } - start_date = (now - timedelta(days=30)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") # Fetch global events and bucket by country - data = await fetcher.get_json( - _ACLED_URL, - source="acled", - cache_key="intel:cii2:multi:global:30d", - cache_ttl=1800, + data = await acled_query( + fetcher, params={ - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), "limit": 500, "event_date": f"{start_date}|{end_date}", "event_date_where": "BETWEEN", }, + cache_key="intel:cii2:multi:global:30d", + cache_ttl=1800, ) + if data is None: + return { + "error": "ACLED credentials not configured (ACLED_EMAIL + ACLED_PASSWORD)", + "source": "instability-index-v2", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + # Classify events by country and type country_data: dict[str, dict] = {} if data is not None: @@ -846,25 +812,18 @@ async def fetch_focal_points(fetcher: Fetcher) -> dict: return events async def _fetch_protest_events() -> list[dict]: - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return [] - start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") - data = await fetcher.get_json( - _ACLED_URL, - source="acled", - cache_key="intel:focal:acled:protests:7d", - cache_ttl=1800, + data = await acled_query( + fetcher, params={ - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), "limit": 200, "event_date": f"{start_date}|{end_date}", "event_date_where": "BETWEEN", "event_type": "Protests", }, + cache_key="intel:focal:acled:protests:7d", + cache_ttl=1800, ) events = [] if data is not None: @@ -951,24 +910,18 @@ async def fetch_signal_summary( return aircraft async def _fetch_protests() -> list[dict]: - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return [] start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") - data = await fetcher.get_json( - _ACLED_URL, - source="acled", - cache_key="intel:signals:acled:protests:7d", - cache_ttl=1800, + data = await acled_query( + fetcher, params={ - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), "limit": 200, "event_date": f"{start_date}|{end_date}", "event_date_where": "BETWEEN", "event_type": "Protests", }, + cache_key="intel:signals:acled:protests:7d", + cache_ttl=1800, ) if data is None: return [] @@ -1062,36 +1015,31 @@ async def fetch_temporal_anomalies(fetcher: Fetcher) -> dict: anomalies.append(result) # ACLED events by country (top focus countries) - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if access_token: - start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d") - end_date = now.strftime("%Y-%m-%d") - data = await fetcher.get_json( - _ACLED_URL, - source="acled", - cache_key="intel:temporal:acled:global:7d", - cache_ttl=1800, - params={ - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), - "limit": 500, - "event_date": f"{start_date}|{end_date}", - "event_date_where": "BETWEEN", - }, - ) - if data is not None: - country_counts: dict[str, int] = {} - events_list = data.get("data", []) if isinstance(data, dict) else [] - for event in events_list: - c = event.get("country") - if c: - country_counts[c] = country_counts.get(c, 0) + 1 + start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d") + end_date = now.strftime("%Y-%m-%d") + data = await acled_query( + fetcher, + params={ + "limit": 500, + "event_date": f"{start_date}|{end_date}", + "event_date_where": "BETWEEN", + }, + cache_key="intel:temporal:acled:global:7d", + cache_ttl=1800, + ) + if data is not None: + country_counts: dict[str, int] = {} + events_list = data.get("data", []) if isinstance(data, dict) else [] + for event in events_list: + c = event.get("country") + if c: + country_counts[c] = country_counts.get(c, 0) + 1 - for c_name, c_count in country_counts.items(): - result = _temporal.record_and_check("acled_events", c_name, c_count) - observations_recorded += 1 - if result is not None: - anomalies.append(result) + for c_name, c_count in country_counts.items(): + result = _temporal.record_and_check("acled_events", c_name, c_count) + observations_recorded += 1 + if result is not None: + anomalies.append(result) # Sort anomalies by z_score descending anomalies.sort(key=lambda a: a.get("z_score", 0), reverse=True) @@ -1146,20 +1094,10 @@ async def fetch_unrest_events( """ now = datetime.now(timezone.utc) - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return { - "error": "ACLED_ACCESS_TOKEN not configured", - "source": "acled-unrest", - "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), - } - start_date = (now - timedelta(days=days)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") params: dict = { - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), "limit": limit, "event_date": f"{start_date}|{end_date}", "event_date_where": "BETWEEN", @@ -1170,19 +1108,16 @@ async def fetch_unrest_events( params["country"] = country cache_label = country or "global" - data = await fetcher.get_json( - _ACLED_URL, - source="acled", + data = await acled_query( + fetcher, + params=params, cache_key=f"intel:unrest:{cache_label}:{days}", cache_ttl=900, - params=params, ) if data is None: return { - "events": [], - "count": 0, - "deduplicated": 0, + "error": "ACLED credentials not configured (ACLED_EMAIL + ACLED_PASSWORD)", "source": "acled-unrest", "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), } @@ -1288,23 +1223,17 @@ async def fetch_hotspot_escalation(fetcher: Fetcher) -> dict: # Fetch global data once, then distribute to hotspots async def _fetch_global_acled() -> list[dict]: - access_token = os.environ.get("ACLED_ACCESS_TOKEN") - if not access_token: - return [] start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d") end_date = now.strftime("%Y-%m-%d") - data = await fetcher.get_json( - _ACLED_URL, - source="acled", - cache_key="intel:escalation:acled:global:7d", - cache_ttl=1800, + data = await acled_query( + fetcher, params={ - "key": access_token, - "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), "limit": 500, "event_date": f"{start_date}|{end_date}", "event_date_where": "BETWEEN", }, + cache_key="intel:escalation:acled:global:7d", + cache_ttl=1800, ) if data is None: return []