"""Country intelligence, risk scoring, signal convergence, and analysis sources.

Provides higher-level analytical functions that combine data from multiple
APIs (ACLED, World Bank, USGS, Ollama, Cloudflare, OpenSky, NASA) into
country briefs, risk scores, instability indices, geographic signal
convergence, focal point detection, signal summaries, temporal anomalies,
hotspot escalation, military surge, vessel tracking, and cascade analysis.
"""

import asyncio
import logging
import math
import os
import re
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
from ..analysis.instability import (
    compute_cii,
    score_unrest,
    score_conflict_v2,
    score_security,
    score_information,
)
from ..analysis.escalation import score_all_hotspots
from ..analysis.surge import detect_surges, SENSITIVE_REGIONS
from ..analysis.cascade import simulate_cascade
from ..config.countries import (
    TIER1_COUNTRIES,
    INTEL_HOTSPOTS,
    STRATEGIC_WATERWAYS,
    get_event_multiplier,
    match_country_by_name,
)

logger = logging.getLogger("world-intel-mcp.sources.intelligence")

# Shared temporal baseline instance
_temporal = TemporalBaseline()


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

_WB_BASE = "https://api.worldbank.org/v2/country"
_USGS_ENDPOINT = "https://earthquake.usgs.gov/fdsnws/event/1/query"

_BASELINES = {
    "Syria": 5000, "Yemen": 3000, "Ukraine": 8000, "Myanmar": 4000,
    "Somalia": 2500, "Nigeria": 3500, "DR Congo": 3000, "Afghanistan": 2000,
    "Iraq": 1500, "Mali": 2000, "Burkina Faso": 2500, "Ethiopia": 2000,
    "Sudan": 3000, "South Sudan": 1500, "Cameroon": 1000, "Mozambique": 800,
    "Pakistan": 1200, "India": 1000, "Colombia": 1500, "Mexico": 4000,
}

_FOCUS_COUNTRIES = [
    "SYR", "UKR", "YEM", "MMR", "SDN", "ETH", "NGA", "COD", "AFG", "IRQ",
]

_HOTSPOTS = {
    "middle_east": (33.0, 44.0),
    "east_africa": (5.0, 38.0),
    "south_asia": (30.0, 70.0),
    "eastern_europe": (48.0, 35.0),
    "sahel": (15.0, 2.0),
}

# ISO-3166 alpha-3 to country name for ACLED queries and display.
_ISO3_TO_NAME = {
    "SYR": "Syria", "UKR": "Ukraine", "YEM": "Yemen", "MMR": "Myanmar",
    "SDN": "Sudan", "ETH": "Ethiopia", "NGA": "Nigeria", "COD": "DR Congo",
    "AFG": "Afghanistan", "IRQ": "Iraq",
}


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _risk_level(score: float) -> str:
    if score > 150:
        return "critical"
    if score > 100:
        return "elevated"
    if score > 50:
        return "moderate"
    return "low"




# ---------------------------------------------------------------------------
# Function 1: Country Intelligence Brief
# ---------------------------------------------------------------------------

async def fetch_country_brief(
    fetcher: Fetcher,
    country_code: str = "US",
) -> dict:
    """Generate a country intelligence brief using local LLM and public data.

    Gathers economic indicators from World Bank and conflict data from ACLED
    in parallel, then optionally enriches with an Ollama-generated analytical
    brief.  Falls back to a data-only summary when Ollama is unavailable.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.
        country_code: ISO 3166-1 alpha-2 country code (e.g. ``US``, ``UA``).

    Returns:
        Dict with brief text, supporting data, LLM availability flag,
        source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    # --- Gather background data in parallel --------------------------------
    async def _fetch_gdp() -> list:
        url = f"{_WB_BASE}/{country_code}/indicator/NY.GDP.MKTP.CD"
        params = {
            "format": "json",
            "per_page": 5,
            "date": "2020:2025",
        }
        data = await fetcher.get_json(
            url,
            source="world-bank",
            cache_key=f"intel:wb:gdp:{country_code}",
            cache_ttl=86400,
            params=params,
        )
        if data is None:
            return []

        values = []
        try:
            if isinstance(data, list) and len(data) >= 2 and isinstance(data[1], list):
                for rec in data[1]:
                    year = rec.get("date")
                    value = rec.get("value")
                    if year is not None and value is not None:
                        try:
                            values.append({"year": year, "value": float(value)})
                        except (ValueError, TypeError):
                            pass
        except (KeyError, TypeError, IndexError) as exc:
            logger.warning("Failed to parse World Bank GDP for %s: %s", country_code, exc)
        return values

    async def _fetch_inflation() -> list:
        url = f"{_WB_BASE}/{country_code}/indicator/FP.CPI.TOTL.ZG"
        params = {
            "format": "json",
            "per_page": 5,
            "date": "2020:2025",
        }
        data = await fetcher.get_json(
            url,
            source="world-bank",
            cache_key=f"intel:wb:inflation:{country_code}",
            cache_ttl=86400,
            params=params,
        )
        if data is None:
            return []

        values = []
        try:
            if isinstance(data, list) and len(data) >= 2 and isinstance(data[1], list):
                for rec in data[1]:
                    year = rec.get("date")
                    value = rec.get("value")
                    if year is not None and value is not None:
                        try:
                            values.append({"year": year, "value": float(value)})
                        except (ValueError, TypeError):
                            pass
        except (KeyError, TypeError, IndexError) as exc:
            logger.warning("Failed to parse World Bank inflation for %s: %s", country_code, exc)
        return values

    async def _fetch_acled_count() -> int:
        start_date = (now - timedelta(days=30)).strftime("%Y-%m-%d")
        end_date = now.strftime("%Y-%m-%d")

        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,
        )
        if data is None:
            return 0

        # ACLED returns a count field when limit=0
        try:
            return int(data.get("count", len(data.get("data", []))))
        except (ValueError, TypeError):
            return len(data.get("data", []))

    gdp_values, inflation_values, event_count = await asyncio.gather(
        _fetch_gdp(),
        _fetch_inflation(),
        _fetch_acled_count(),
    )

    # --- Attempt Ollama-generated brief ------------------------------------
    llm_available = False
    brief_text = "LLM brief unavailable. Data summary below."

    prompt = (
        f"Provide a concise 3-paragraph intelligence brief for {country_code}. "
        "Cover: (1) current political stability and governance, "
        "(2) economic outlook and risks, "
        "(3) security concerns and regional dynamics. "
        "Be factual and analytical."
    )

    ollama_url = os.environ.get("OLLAMA_API_URL", "http://localhost:11434")
    model = os.environ.get("OLLAMA_MODEL", "llama3.2:latest")

    try:
        async with httpx.AsyncClient(timeout=30.0, proxy=None) as client:
            resp = await client.post(
                f"{ollama_url}/api/generate",
                json={
                    "model": model,
                    "prompt": prompt,
                    "stream": False,
                },
            )
            resp.raise_for_status()
            resp_data = resp.json()
            generated = resp_data.get("response", "")
            if generated and generated.strip():
                brief_text = generated.strip()
                llm_available = True
    except (httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError) as exc:
        logger.info("Ollama unavailable for country brief (%s): %s", country_code, exc)
    except Exception as exc:
        logger.warning("Unexpected error calling Ollama: %s", exc)

    return {
        "country_code": country_code,
        "brief": brief_text,
        "data": {
            "gdp": gdp_values,
            "inflation": inflation_values,
            "recent_events": event_count,
        },
        "llm_available": llm_available,
        "source": "country-intelligence",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 2: Country Risk Scores
# ---------------------------------------------------------------------------

async def fetch_risk_scores(
    fetcher: Fetcher,
    limit: int = 20,
) -> dict:
    """Compute country risk scores from ACLED conflict data and baselines.

    Fetches recent global conflict events, counts per country, and computes
    a risk score as ``(events_30d / monthly_baseline) * 100``.  Higher
    scores indicate conflict above historical norms.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.
        limit: Maximum number of countries to return (sorted by risk).

    Returns:
        Dict with ranked country list, count, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    start_date = (now - timedelta(days=30)).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:risk:global:30d",
        cache_ttl=1800,
    )

    if data is None:
        return {
            "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"),
        }

    # Count events per country
    country_counts: dict[str, int] = {}
    for event in data.get("data", []):
        country_name = event.get("country")
        if country_name:
            country_counts[country_name] = country_counts.get(country_name, 0) + 1

    # Compute risk scores
    countries: list[dict] = []
    for country_name, events_30d in country_counts.items():
        baseline_annual = _BASELINES.get(country_name, 500)
        monthly_baseline = baseline_annual / 12.0
        risk_score = (events_30d / monthly_baseline) * 100 if monthly_baseline > 0 else 0.0

        countries.append({
            "country": country_name,
            "events_30d": events_30d,
            "monthly_baseline": round(monthly_baseline, 1),
            "risk_score": round(risk_score, 1),
            "risk_level": _risk_level(risk_score),
        })

    # Sort by risk_score descending, take top N
    countries.sort(key=lambda c: c["risk_score"], reverse=True)
    countries = countries[:limit]

    return {
        "countries": countries,
        "count": len(countries),
        "source": "risk-analysis",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 3: Country Instability Index
# ---------------------------------------------------------------------------

async def fetch_instability_index(
    fetcher: Fetcher,
    country_code: str | None = None,
) -> dict:
    """Compute a Country Instability Index (CII) from multiple signals.

    Combines conflict intensity, economic stress, humanitarian crisis data,
    internet disruption indicators, and military activity into a 0-100
    composite score.  Higher values indicate greater instability.

    When *country_code* is ``None``, returns a simplified index for 10
    focus countries using ACLED data as the primary signal.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.
        country_code: Optional ISO 3166-1 alpha-3 code (e.g. ``UKR``).

    Returns:
        Dict with instability index, component scores, risk level, source,
        and timestamp.
    """
    now = datetime.now(timezone.utc)

    if country_code is not None:
        return await _instability_single(fetcher, country_code, now)

    return await _instability_multi(fetcher, now)


async def _instability_single(
    fetcher: Fetcher,
    country_code: str,
    now: datetime,
) -> dict:
    """Compute CII v2 instability index for a single country.

    Uses 4 weighted domains: unrest, conflict, security, information.
    Applies country-specific event multiplier and UCDP/displacement boosts.
    """
    country_name = _ISO3_TO_NAME.get(country_code, country_code)
    event_multiplier = get_event_multiplier(country_code)
    start_date = (now - timedelta(days=30)).strftime("%Y-%m-%d")
    end_date = now.strftime("%Y-%m-%d")

    # --- Parallel data gathering -------------------------------------------

    async def _fetch_acled() -> list[dict]:
        """Fetch ACLED events for this country."""
        data = await acled_query(
            fetcher,
            params={
                "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 []
        return data.get("data", []) if isinstance(data, dict) else []

    async def _fetch_outages() -> int:
        """Count internet outages mentioning this country."""
        from . import infrastructure
        result = await infrastructure.fetch_internet_outages(fetcher)
        count = 0
        for outage in result.get("outages", []):
            countries_list = outage.get("countries", [])
            if isinstance(countries_list, list):
                for c in countries_list:
                    if isinstance(c, str) and country_code.lower() in c.lower():
                        count += 1
        return count

    async def _fetch_military() -> int:
        """Count military aircraft near this country."""
        _COUNTRY_BBOX = {
            "SYR": "32,35,37,42", "UKR": "44,22,52,40",
            "YEM": "12,42,19,55", "MMR": "10,92,28,101",
            "SDN": "8,21,23,39", "ETH": "3,33,15,48",
            "NGA": "4,3,14,15", "COD": "-13,12,5,31",
            "AFG": "29,60,38,75", "IRQ": "29,39,37,49",
            "IRN": "25,44,40,63", "ISR": "29,34,33,36",
            "PSE": "31,34,32,35", "LBN": "33,35,34,37",
            "TWN": "21,119,26,122", "PRK": "37,124,43,131",
        }
        bbox = _COUNTRY_BBOX.get(country_code)
        if bbox is None:
            return 0

        from . import military as mil_mod
        result = await mil_mod.fetch_military_flights(fetcher, bbox=bbox)
        return result.get("count", 0)

    async def _fetch_news_velocity() -> int:
        """Estimate news velocity from GDELT."""
        from . import news
        result = await news.fetch_gdelt_search(
            fetcher, query=country_name, mode="artlist", limit=100,
        )
        return result.get("count", 0)

    acled_events, outage_count, mil_count, news_vel = await asyncio.gather(
        _fetch_acled(),
        _fetch_outages(),
        _fetch_military(),
        _fetch_news_velocity(),
    )

    # Classify ACLED events into protests/riots vs armed conflict
    protest_count = 0
    riot_count = 0
    conflict_count = 0
    total_fatalities = 0
    for ev in acled_events:
        event_type = (ev.get("event_type") or "").lower()
        fat = 0
        try:
            fat = int(ev.get("fatalities", 0))
        except (ValueError, TypeError):
            pass
        total_fatalities += fat

        if "protest" in event_type:
            protest_count += 1
        elif "riot" in event_type:
            riot_count += 1
        else:
            conflict_count += 1

    # Score each domain (0-25)
    unrest_val = score_unrest(protest_count, riot_count)
    conflict_val = score_conflict_v2(conflict_count, total_fatalities)
    security_val = score_security(mil_count, outage_count)
    info_val = score_information(news_vel)

    # UCDP floor: active wars get a minimum score
    ucdp_floor = None
    country_cfg = TIER1_COUNTRIES.get(country_code)
    if country_cfg and country_cfg.get("baseline_risk", 0) >= 80:
        ucdp_floor = 70.0
    elif country_cfg and country_cfg.get("baseline_risk", 0) >= 60:
        ucdp_floor = 50.0

    # Displacement boost
    displacement_boost = 0.0
    # (Would require UNHCR fetch; simplified: use baseline_risk as proxy)
    if country_cfg and country_cfg.get("baseline_risk", 0) >= 70:
        displacement_boost = 3.0

    cii = compute_cii(
        unrest=unrest_val,
        conflict=conflict_val,
        security=security_val,
        information=info_val,
        event_multiplier=event_multiplier,
        ucdp_floor=ucdp_floor,
        displacement_boost=displacement_boost,
    )

    return {
        "country_code": country_code,
        "country_name": country_name,
        **cii,
        "raw_data": {
            "acled_events": len(acled_events),
            "protests": protest_count,
            "riots": riot_count,
            "conflict_events": conflict_count,
            "fatalities": total_fatalities,
            "military_aircraft": mil_count,
            "internet_outages": outage_count,
            "news_articles": news_vel,
        },
        "source": "instability-index-v2",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


async def _instability_multi(fetcher: Fetcher, now: datetime) -> dict:
    """Compute CII v2 instability index for focus countries using ACLED."""
    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 acled_query(
        fetcher,
        params={
            "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:
        events_list = data.get("data", []) if isinstance(data, dict) else []
        for event in events_list:
            country_name = event.get("country")
            if not country_name:
                continue
            if country_name not in country_data:
                country_data[country_name] = {
                    "protests": 0, "riots": 0, "conflict": 0,
                    "fatalities": 0, "total": 0,
                }
            cd = country_data[country_name]
            cd["total"] += 1
            event_type = (event.get("event_type") or "").lower()
            fat = 0
            try:
                fat = int(event.get("fatalities", 0))
            except (ValueError, TypeError):
                pass
            cd["fatalities"] += fat
            if "protest" in event_type:
                cd["protests"] += 1
            elif "riot" in event_type:
                cd["riots"] += 1
            else:
                cd["conflict"] += 1

    # Compute CII v2 for each focus country
    results: list[dict] = []
    for code in _FOCUS_COUNTRIES:
        name = _ISO3_TO_NAME.get(code, code)
        cd = country_data.get(name, {
            "protests": 0, "riots": 0, "conflict": 0,
            "fatalities": 0, "total": 0,
        })
        multiplier = get_event_multiplier(code)

        unrest_val = score_unrest(cd["protests"], cd["riots"])
        conflict_val = score_conflict_v2(cd["conflict"], cd["fatalities"])
        # Security and information not available in multi-country mode
        security_val = 0.0
        info_val = 0.0

        # UCDP floor from countries config
        country_cfg = TIER1_COUNTRIES.get(code)
        ucdp_floor = None
        if country_cfg and country_cfg.get("baseline_risk", 0) >= 80:
            ucdp_floor = 70.0
        elif country_cfg and country_cfg.get("baseline_risk", 0) >= 60:
            ucdp_floor = 50.0

        cii = compute_cii(
            unrest=unrest_val,
            conflict=conflict_val,
            security=security_val,
            information=info_val,
            event_multiplier=multiplier,
            ucdp_floor=ucdp_floor,
        )

        results.append({
            "country_code": code,
            "country_name": name,
            **cii,
            "events_30d": cd["total"],
        })

    results.sort(key=lambda r: r["instability_index"], reverse=True)

    return {
        "countries": results,
        "count": len(results),
        "note": "Multi-country CII v2 using ACLED unrest + conflict. "
                "Use country_code for full 4-domain analysis.",
        "source": "instability-index-v2",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 4: Signal Convergence
# ---------------------------------------------------------------------------

async def fetch_signal_convergence(
    fetcher: Fetcher,
    lat: float | None = None,
    lon: float | None = None,
    radius_deg: float = 5.0,
) -> dict:
    """Detect geographic convergence of signals in hotspot regions.

    Checks for overlapping seismic activity and other observable signals
    within a radius of known or specified hotspot coordinates.  Higher
    convergence scores indicate multiple signal types in close proximity,
    which may warrant deeper investigation.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.
        lat: Latitude of center point.  If ``None``, scans 5 global
             hotspot regions.
        lon: Longitude of center point.
        radius_deg: Radius in degrees for bounding box queries.

    Returns:
        Dict with hotspot list, convergence scores, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    if lat is not None and lon is not None:
        regions = {"custom": (lat, lon)}
    else:
        regions = dict(_HOTSPOTS)

    async def _assess_hotspot(name: str, center: tuple[float, float]) -> dict:
        center_lat, center_lon = center

        # Earthquake count within bounding box
        min_lat = center_lat - radius_deg
        max_lat = center_lat + radius_deg
        min_lon = center_lon - radius_deg
        max_lon = center_lon + radius_deg

        starttime = (now - timedelta(days=30)).strftime("%Y-%m-%dT%H:%M:%S")

        quake_data = await fetcher.get_json(
            _USGS_ENDPOINT,
            source="usgs",
            cache_key=f"intel:convergence:usgs:{name}:{radius_deg}",
            cache_ttl=600,
            params={
                "format": "geojson",
                "minmagnitude": 2.5,
                "starttime": starttime,
                "minlatitude": min_lat,
                "maxlatitude": max_lat,
                "minlongitude": min_lon,
                "maxlongitude": max_lon,
                "limit": 100,
            },
        )

        earthquake_count = 0
        if quake_data is not None:
            earthquake_count = len(quake_data.get("features", []))

        # Convergence score heuristic (0-10)
        # Each signal type present adds to the score.
        score = 0.0

        # Earthquakes: 0-5 points based on count
        if earthquake_count > 0:
            score += min(5.0, (earthquake_count / 20.0) * 5.0)

        # Hotspot presence bonus (known conflict zones get a baseline)
        if name in _HOTSPOTS:
            score += 2.0

        score = min(10.0, round(score, 1))

        return {
            "name": name,
            "lat": center_lat,
            "lon": center_lon,
            "signals": {
                "earthquakes": earthquake_count,
            },
            "convergence_score": score,
        }

    tasks = [_assess_hotspot(name, center) for name, center in regions.items()]
    results = await asyncio.gather(*tasks)

    # Sort by convergence score descending
    hotspots = sorted(results, key=lambda h: h["convergence_score"], reverse=True)

    return {
        "hotspots": hotspots,
        "source": "signal-convergence",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 5: Focal Point Detection
# ---------------------------------------------------------------------------

async def fetch_focal_points(fetcher: Fetcher) -> dict:
    """Gather multi-source events and detect focal points.

    Fetches news headlines, military flights, internet outages, and ACLED
    protests in parallel, normalizes them into events, and runs focal point
    detection to find entities where multiple signals converge.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.

    Returns:
        Dict with focal_points list, count, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    # Import source modules for parallel data gathering
    from . import news, military, infrastructure, conflict

    async def _fetch_news_events() -> list[dict]:
        result = await news.fetch_news_feed(fetcher, limit=100)
        events = []
        for item in result.get("items", []):
            title = item.get("title", "")
            # Extract entity: try to match country names from title
            matched_iso = match_country_by_name(title)
            if matched_iso:
                country_cfg = TIER1_COUNTRIES.get(matched_iso)
                entity = country_cfg["name"] if country_cfg else matched_iso
                events.append({
                    "entity": entity,
                    "type": "news",
                    "timestamp": item.get("published") or now.strftime("%Y-%m-%dT%H:%M:%SZ"),
                    "country": entity,
                    "weight": 1.0,
                })
        return events

    async def _fetch_military_events() -> list[dict]:
        result = await military.fetch_theater_posture(fetcher)
        events = []
        for theater_name, theater_data in result.get("theaters", {}).items():
            count = theater_data.get("count", 0)
            if count > 0:
                for country in theater_data.get("countries", []):
                    events.append({
                        "entity": country,
                        "type": "military",
                        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
                        "country": country,
                        "weight": min(3.0, count / 10.0),
                    })
        return events

    async def _fetch_outage_events() -> list[dict]:
        result = await infrastructure.fetch_internet_outages(fetcher)
        events = []
        for outage in result.get("outages", []):
            countries_list = outage.get("countries", [])
            if isinstance(countries_list, list):
                for c in countries_list:
                    if c:
                        events.append({
                            "entity": c,
                            "type": "infrastructure",
                            "timestamp": outage.get("start") or now.strftime("%Y-%m-%dT%H:%M:%SZ"),
                            "country": c,
                            "weight": 2.0 if outage.get("is_ongoing") else 1.0,
                        })
        return events

    async def _fetch_protest_events() -> list[dict]:
        start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d")
        end_date = now.strftime("%Y-%m-%d")
        data = await acled_query(
            fetcher,
            params={
                "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:
            acled_list = data.get("data", []) if isinstance(data, dict) else []
            for ev in acled_list:
                country = ev.get("country")
                if country:
                    events.append({
                        "entity": country,
                        "type": "protest",
                        "timestamp": ev.get("event_date") or now.strftime("%Y-%m-%dT%H:%M:%SZ"),
                        "country": country,
                        "weight": 1.0,
                    })
        return events

    news_events, mil_events, outage_events, protest_events = await asyncio.gather(
        _fetch_news_events(),
        _fetch_military_events(),
        _fetch_outage_events(),
        _fetch_protest_events(),
    )

    all_events = news_events + mil_events + outage_events + protest_events
    focal_points = detect_focal_points(all_events)

    return {
        "focal_points": focal_points,
        "count": len(focal_points),
        "total_events_analyzed": len(all_events),
        "source": "focal-point-analysis",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 6: Signal Summary
# ---------------------------------------------------------------------------

async def fetch_signal_summary(
    fetcher: Fetcher,
    country: str | None = None,
) -> dict:
    """Run signal aggregator v2 across all domains.

    Fetches ACLED conflict, USGS earthquakes, NASA FIRMS wildfires,
    Cloudflare outages, military flights, and UNHCR displacement in parallel,
    then aggregates signals by country with convergence scoring.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.
        country: Optional country name to filter results.

    Returns:
        Dict with countries list, count, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    from . import conflict, infrastructure, military, displacement

    async def _fetch_conflict() -> list[dict]:
        result = await conflict.fetch_acled_events(fetcher, days=7, limit=200)
        return result.get("events", [])

    async def _fetch_earthquakes() -> list[dict]:
        from . import seismology
        result = await seismology.fetch_earthquakes(fetcher, min_magnitude=4.5, hours=168, limit=100)
        return result.get("earthquakes", [])

    async def _fetch_outages() -> list[dict]:
        result = await infrastructure.fetch_internet_outages(fetcher)
        return result.get("outages", [])

    async def _fetch_military() -> list[dict]:
        result = await military.fetch_theater_posture(fetcher)
        aircraft = []
        for theater_data in result.get("theaters", {}).values():
            # Theater posture returns summary, not individual aircraft
            for c in theater_data.get("countries", []):
                aircraft.append({
                    "origin_country": c,
                    "count": theater_data.get("count", 0),
                })
        return aircraft

    async def _fetch_protests() -> list[dict]:
        start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d")
        end_date = now.strftime("%Y-%m-%d")
        data = await acled_query(
            fetcher,
            params={
                "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 []
        acled_list = data.get("data", []) if isinstance(data, dict) else []
        return [
            {"country": ev.get("country"), "event_type": ev.get("event_type")}
            for ev in acled_list
            if ev.get("country")
        ]

    async def _fetch_displacement() -> list[dict]:
        result = await displacement.fetch_displacement_summary(fetcher)
        return result.get("by_origin", [])

    (
        conflict_events, earthquake_data, outage_data,
        military_data, protest_data, displacement_data,
    ) = await asyncio.gather(
        _fetch_conflict(),
        _fetch_earthquakes(),
        _fetch_outages(),
        _fetch_military(),
        _fetch_protests(),
        _fetch_displacement(),
    )

    aggregated = aggregate_country_signals(
        conflict_events=conflict_events,
        displacement_data=displacement_data,
        earthquake_data=earthquake_data,
        outage_data=outage_data,
        military_data=military_data,
        protest_data=protest_data,
    )

    # Filter to specific country if requested
    if country:
        filtered = {}
        lower_country = country.lower()
        for c_name, c_data in aggregated.items():
            if lower_country in c_name.lower():
                filtered[c_name] = c_data
        aggregated = filtered

    # Convert to list format
    countries_list = [
        {"country": name, **data}
        for name, data in aggregated.items()
    ]

    return {
        "countries": countries_list[:50],
        "count": len(countries_list),
        "source": "signal-aggregation-v2",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 7: Temporal Anomaly Detection
# ---------------------------------------------------------------------------

async def fetch_temporal_anomalies(fetcher: Fetcher) -> dict:
    """Record observations and check for temporal anomalies.

    Fetches current counts of military flights (by theater), ACLED events
    (by country), and fires (by region), records each as a temporal
    observation, and reports any that deviate significantly from baselines.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.

    Returns:
        Dict with anomalies list, observations_recorded count, source,
        and timestamp.
    """
    now = datetime.now(timezone.utc)

    from . import military

    anomalies: list[dict] = []
    observations_recorded = 0

    # Military flights by theater
    posture = await military.fetch_theater_posture(fetcher)
    for theater_name, theater_data in posture.get("theaters", {}).items():
        count = theater_data.get("count", 0)
        result = _temporal.record_and_check("military_flights", theater_name, count)
        observations_recorded += 1
        if result is not None:
            anomalies.append(result)

    # ACLED events by country (top focus countries)
    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)

    # Sort anomalies by z_score descending
    anomalies.sort(key=lambda a: a.get("z_score", 0), reverse=True)

    return {
        "anomalies": anomalies,
        "anomaly_count": len(anomalies),
        "observations_recorded": observations_recorded,
        "source": "temporal-anomaly-detection",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 8: Social Unrest Events (Protests + Riots)
# ---------------------------------------------------------------------------

def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Great-circle distance between two points in km."""
    R = 6371.0
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    a = (
        math.sin(dlat / 2) ** 2
        + math.cos(math.radians(lat1))
        * math.cos(math.radians(lat2))
        * math.sin(dlon / 2) ** 2
    )
    return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))


async def fetch_unrest_events(
    fetcher: Fetcher,
    country: str | None = None,
    days: int = 7,
    limit: int = 100,
) -> dict:
    """Fetch social unrest events (protests + riots) from ACLED.

    Filters by event_type in (Protests, Riots).
    Applies Haversine deduplication: merges events within 50 km on the
    same day to remove redundant reports.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.
        country: Optional country name filter.
        days: Lookback period in days.
        limit: Maximum results from ACLED.

    Returns:
        Dict with events list, count, dedup stats, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    start_date = (now - timedelta(days=days)).strftime("%Y-%m-%d")
    end_date = now.strftime("%Y-%m-%d")

    params: dict = {
        "limit": limit,
        "event_date": f"{start_date}|{end_date}",
        "event_date_where": "BETWEEN",
        "event_type": "Protests:Riots",
        "event_type_where": "IN",
    }
    if country:
        params["country"] = country

    cache_label = country or "global"
    data = await acled_query(
        fetcher,
        params=params,
        cache_key=f"intel:unrest:{cache_label}:{days}",
        cache_ttl=900,
    )

    if data is None:
        return {
            "error": "ACLED credentials not configured (ACLED_EMAIL + ACLED_PASSWORD)",
            "source": "acled-unrest",
            "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
        }

    raw_events = data.get("data", []) if isinstance(data, dict) else []

    # Parse events
    parsed: list[dict] = []
    for ev in raw_events:
        lat_raw = ev.get("latitude")
        lon_raw = ev.get("longitude")
        lat = None
        lon = None
        try:
            lat = float(lat_raw) if lat_raw is not None else None
            lon = float(lon_raw) if lon_raw is not None else None
        except (ValueError, TypeError):
            pass

        fat = 0
        try:
            fat = int(ev.get("fatalities", 0))
        except (ValueError, TypeError):
            pass

        parsed.append({
            "event_date": ev.get("event_date"),
            "event_type": ev.get("event_type"),
            "sub_event_type": ev.get("sub_event_type"),
            "country": ev.get("country"),
            "admin1": ev.get("admin1"),
            "location": ev.get("location"),
            "latitude": lat,
            "longitude": lon,
            "fatalities": fat,
            "actor1": ev.get("actor1"),
            "notes": ev.get("notes"),
        })

    # Haversine deduplication: merge events within 50km on same day
    DEDUP_RADIUS_KM = 50.0
    deduped: list[dict] = []
    original_count = len(parsed)

    for event in parsed:
        lat = event.get("latitude")
        lon = event.get("longitude")
        edate = event.get("event_date")

        is_dup = False
        if lat is not None and lon is not None:
            for existing in deduped:
                if existing.get("event_date") != edate:
                    continue
                ex_lat = existing.get("latitude")
                ex_lon = existing.get("longitude")
                if ex_lat is None or ex_lon is None:
                    continue
                dist = _haversine_km(lat, lon, ex_lat, ex_lon)
                if dist < DEDUP_RADIUS_KM:
                    # Merge: keep higher fatality count
                    if event["fatalities"] > existing["fatalities"]:
                        existing["fatalities"] = event["fatalities"]
                    is_dup = True
                    break

        if not is_dup:
            deduped.append(event)

    return {
        "events": deduped,
        "count": len(deduped),
        "deduplicated": original_count - len(deduped),
        "query": {"country": country, "days": days},
        "source": "acled-unrest",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 9: Hotspot Escalation Scoring
# ---------------------------------------------------------------------------

async def fetch_hotspot_escalation(fetcher: Fetcher) -> dict:
    """Score all 22 intel hotspots using multi-source signals.

    For each hotspot:
    - Fetch GDELT mentions (news velocity near lat/lon)
    - Count military aircraft near hotspot (+/- 2 deg)
    - Count ACLED events near hotspot (+/- 2 deg, last 7 days)

    Runs analysis.escalation.score_all_hotspots().

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.

    Returns:
        Dict with scored hotspots, count, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    from . import military as mil_mod

    # Fetch global data once, then distribute to hotspots
    async def _fetch_global_acled() -> list[dict]:
        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:escalation:acled:global:7d",
            cache_ttl=1800,
        )
        if data is None:
            return []
        return data.get("data", []) if isinstance(data, dict) else []

    async def _fetch_global_military() -> list[dict]:
        # Use theater posture for global coverage
        result = await mil_mod.fetch_theater_posture(fetcher)
        aircraft = []
        for theater_data in result.get("theaters", {}).values():
            for ac in theater_data.get("countries", []):
                # Create pseudo-aircraft entries at theater bbox center
                bbox = theater_data.get("bbox", "")
                parts = bbox.split(",")
                if len(parts) == 4:
                    try:
                        lat = (float(parts[0]) + float(parts[2])) / 2
                        lon = (float(parts[1]) + float(parts[3])) / 2
                        for _ in range(theater_data.get("count", 0) // max(1, len(theater_data.get("countries", [1])))):
                            aircraft.append({"lat": lat, "lon": lon, "origin_country": ac})
                    except (ValueError, TypeError):
                        pass
        return aircraft

    acled_events, military_data = await asyncio.gather(
        _fetch_global_acled(),
        _fetch_global_military(),
    )

    # Build signal dict for each hotspot
    RADIUS_DEG = 2.0
    hotspot_signals: dict[str, dict] = {}

    for hs_name, hs_config in INTEL_HOTSPOTS.items():
        hs_lat = hs_config["lat"]
        hs_lon = hs_config["lon"]

        # Count ACLED events near hotspot
        conflict_count = 0
        protest_count = 0
        fatality_count = 0
        for ev in acled_events:
            try:
                ev_lat = float(ev.get("latitude", 0))
                ev_lon = float(ev.get("longitude", 0))
            except (ValueError, TypeError):
                continue
            if abs(ev_lat - hs_lat) <= RADIUS_DEG and abs(ev_lon - hs_lon) <= RADIUS_DEG:
                event_type = (ev.get("event_type") or "").lower()
                if "protest" in event_type:
                    protest_count += 1
                else:
                    conflict_count += 1
                try:
                    fatality_count += int(ev.get("fatalities", 0))
                except (ValueError, TypeError):
                    pass

        # Count military aircraft near hotspot
        mil_count = 0
        for ac in military_data:
            ac_lat = ac.get("lat", 0)
            ac_lon = ac.get("lon", 0)
            if abs(ac_lat - hs_lat) <= RADIUS_DEG and abs(ac_lon - hs_lon) <= RADIUS_DEG:
                mil_count += 1

        hotspot_signals[hs_name] = {
            "news_mentions": 0,  # Would require per-hotspot GDELT queries (expensive); baseline 0
            "military_count": mil_count,
            "conflict_events": conflict_count,
            "convergence_score": 0,
            "fatalities": fatality_count,
            "protests": protest_count,
        }

    scored = score_all_hotspots(INTEL_HOTSPOTS, hotspot_signals)

    return {
        "hotspots": scored,
        "count": len(scored),
        "source": "hotspot-escalation",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 10: Military Surge Detection
# ---------------------------------------------------------------------------

async def fetch_military_surge(fetcher: Fetcher) -> dict:
    """Detect military surge anomalies across sensitive regions.

    1. Fetch theater posture (existing)
    2. Build temporal baselines for each region
    3. Run analysis.surge.detect_surges()

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.

    Returns:
        Dict with surges list, regions checked, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    from . import military as mil_mod

    posture = await mil_mod.fetch_theater_posture(fetcher)
    theater_data = posture.get("theaters", {})

    # Build temporal baselines for each region
    temporal_baselines: dict[str, dict] = {}
    for region_name in SENSITIVE_REGIONS:
        # Record total aircraft count in the region's matching theaters
        total = 0
        from ..analysis.surge import _THEATER_REGION_MAP
        for theater_name, mapped_regions in _THEATER_REGION_MAP.items():
            if region_name in mapped_regions:
                total += theater_data.get(theater_name, {}).get("count", 0)

        result = _temporal.record_and_check("surge_aircraft", region_name, total)
        if result is not None:
            temporal_baselines[region_name] = {
                "z_score": result["z_score"],
                "multiplier": result.get("multiplier"),
            }

    surges = detect_surges(theater_data, temporal_baselines)

    return {
        "surges": surges,
        "surge_count": len(surges),
        "regions_checked": len(SENSITIVE_REGIONS),
        "source": "military-surge-detection",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 11: Vessel Snapshot at Strategic Waterways
# ---------------------------------------------------------------------------

_NAVAL_KEYWORDS = re.compile(
    r"\b(naval|warship|destroyer|frigate|carrier|submarine|fleet|military\s+vessel|"
    r"exercise|mine|ordnance|firing|weapons)\b",
    re.IGNORECASE,
)


async def fetch_vessel_snapshot(fetcher: Fetcher) -> dict:
    """Naval activity snapshot at strategic waterways using NGA warnings.

    Uses NGA MSI (existing fetch_nav_warnings) filtered for naval/vessel
    keywords near STRATEGIC_WATERWAYS from config.
    Scores each waterway: clear/advisory/elevated/critical.

    Note: Real-time AIS requires paid API.  This uses NGA MSI as a
    free proxy for naval activity indicators.

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.

    Returns:
        Dict with waterways list, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    from . import maritime

    nav_data = await maritime.fetch_nav_warnings(fetcher)
    all_warnings = nav_data.get("warnings", [])

    waterways: list[dict] = []

    for ww in STRATEGIC_WATERWAYS:
        ww_lat = ww["lat"]
        ww_lon = ww["lon"]

        naval_warnings: list[dict] = []
        total_nearby = 0

        for warning in all_warnings:
            text = warning.get("text", "")
            # Simple proximity: check if warning text mentions coordinates
            # near the waterway (NGA warnings have lat/lon in text parsed elsewhere)
            # Use navarea as rough filter and keyword matching
            if _NAVAL_KEYWORDS.search(text):
                naval_warnings.append({
                    "id": warning.get("id"),
                    "text_snippet": text[:200],
                    "navarea": warning.get("navarea"),
                })

            # Count all warnings in the general vicinity (any topic)
            total_nearby += 1

        naval_count = len(naval_warnings)

        if naval_count >= 3:
            status = "critical"
        elif naval_count >= 2:
            status = "elevated"
        elif naval_count >= 1:
            status = "advisory"
        else:
            status = "clear"

        waterways.append({
            "name": ww["name"],
            "lat": ww_lat,
            "lon": ww_lon,
            "throughput": ww.get("throughput"),
            "naval_warnings": naval_count,
            "status": status,
            "warning_details": naval_warnings[:5],
        })

    return {
        "waterways": waterways,
        "count": len(waterways),
        "total_nav_warnings": len(all_warnings),
        "source": "nga-msi-vessel-snapshot",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }


# ---------------------------------------------------------------------------
# Function 12: Infrastructure Cascade Analysis
# ---------------------------------------------------------------------------

async def fetch_cascade_analysis(
    fetcher: Fetcher,
    corridor: str | None = None,
) -> dict:
    """Simulate infrastructure cascade from corridor disruption.

    1. Fetch current cable health (existing fetch_cable_health)
    2. If corridor specified, simulate that corridor disrupted
    3. If not, simulate each at_risk/disrupted corridor
    4. Run analysis.cascade.simulate_cascade()

    Args:
        fetcher: Shared HTTP fetcher with caching and circuit breaking.
        corridor: Optional specific corridor to simulate disruption of.

    Returns:
        Dict with scenarios, current health, source, and timestamp.
    """
    now = datetime.now(timezone.utc)

    from . import infrastructure

    health_data = await infrastructure.fetch_cable_health(fetcher)
    corridors_health = health_data.get("corridors", {})

    scenarios: list[dict] = []

    if corridor:
        # Simulate specific corridor disruption
        result = simulate_cascade([corridor], current_health=corridors_health)
        scenarios.append({
            "scenario": f"Disruption of {corridor}",
            "corridors": [corridor],
            **result,
        })
    else:
        # Simulate each at_risk or disrupted corridor
        at_risk_corridors = [
            name
            for name, info in corridors_health.items()
            if info.get("status_score", 0) >= 2
        ]

        if at_risk_corridors:
            # Individual scenarios
            for c in at_risk_corridors:
                result = simulate_cascade([c], current_health=corridors_health)
                scenarios.append({
                    "scenario": f"Disruption of {c}",
                    "corridors": [c],
                    **result,
                })

            # Combined worst-case scenario
            if len(at_risk_corridors) >= 2:
                result = simulate_cascade(at_risk_corridors, current_health=corridors_health)
                scenarios.append({
                    "scenario": "Combined disruption (worst case)",
                    "corridors": at_risk_corridors,
                    **result,
                })
        else:
            # No at-risk corridors; simulate red_sea as a common scenario
            result = simulate_cascade(["red_sea"], current_health=corridors_health)
            scenarios.append({
                "scenario": "Hypothetical: Red Sea corridor disruption",
                "corridors": ["red_sea"],
                **result,
            })

    return {
        "scenarios": scenarios,
        "scenario_count": len(scenarios),
        "current_health": {
            name: {
                "status_score": info.get("status_score"),
                "status_label": info.get("status_label"),
            }
            for name, info in corridors_health.items()
        },
        "source": "cascade-analysis",
        "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
