fix: conflict zones always visible via INTEL_HOTSPOTS fallback

When both ACLED (no API key) and UCDP (timeout/empty) fail, the
dashboard now falls back to 22 pre-coded conflict hotspots from
INTEL_HOTSPOTS with severity levels (critical/high/moderate/low).
Frontend updated with 3-tier fallback: ACLED → UCDP → conflict_zones.
Marker sizing uses escalation level for hotspots, tooltips show severity.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Marc Shade
2026-02-23 22:25:05 -05:00
co-authored by Claude Opus 4.6
parent 2d46b064dd
commit dd3c4f6951
2 changed files with 55 additions and 4 deletions
+30
View File
@@ -42,6 +42,7 @@ from world_intel_mcp.sources import (
nuclear,
)
from world_intel_mcp.analysis.alerts import fetch_alert_digest, fetch_weekly_trends
from world_intel_mcp.config.countries import INTEL_HOTSPOTS
logger = logging.getLogger(__name__)
@@ -119,6 +120,35 @@ async def _fetch_overview() -> dict:
else:
result[name] = data
# Conflict zone fallback: when both ACLED and UCDP fail, provide
# static hotspot data so the conflict layer is never empty.
acled = result.get("acled_events", {})
ucdp = result.get("ucdp_events", {})
acled_ok = not acled.get("error") and (acled.get("count") or 0) > 0
ucdp_ok = not ucdp.get("error") and (ucdp.get("count") or 0) > 0
if not acled_ok and not ucdp_ok:
escalation_labels = {1: "low", 2: "low", 3: "moderate", 4: "high", 5: "critical"}
hotspot_events = [
{
"latitude": h["lat"],
"longitude": h["lon"],
"country": name.replace("_", " ").title(),
"event_type": "conflict zone",
"type_of_violence_label": "active hotspot",
"fatalities": 0,
"best": 0,
"escalation": h["baseline_escalation"],
"severity": escalation_labels.get(h["baseline_escalation"], "unknown"),
"associated_countries": h.get("associated_countries", []),
}
for name, h in INTEL_HOTSPOTS.items()
]
result["conflict_zones"] = {
"events": hotspot_events,
"count": len(hotspot_events),
"source": "intel-hotspots",
}
# Attach source health + timestamp
result["source_health"] = _breaker.status() if _breaker else {}
result["cache_stats"] = _cache.stats() if _cache else {}