"""Weather context for the map: NWS active alerts and nearest-METAR wind (server-26#237).

Sources -- both US Government works, public domain, no licence gate:
  * NWS alerts: https://api.weather.gov/alerts/active?area=<codes>
    (docs: https://www.weather.gov/documentation/services-web-api). NWS requires
    a User-Agent naming the app and a contact, and rate-limits, so the feed is
    fetched server-side, cached (``settings.weather_alerts_ttl_s``) and shared by
    every viewer. Credit: "National Weather Service".
  * METAR wind: https://aviationweather.gov/api/data/metar?bbox=minLat,minLon,maxLat,maxLon&format=json
    (docs: https://aviationweather.gov/data/api/). Verified 2026-10-05: JSON list,
    one latest report per station, HTTP 204 with an empty body when the bbox holds
    no station, ``wdir`` is an int or the string "VRB".

Shape and failure rules are the 511NY ones (app/internal/ny511.py): a failed
refresh keeps serving the last good data and reports the error and its age; a
payload whose records ALL fail to parse is a schema change and an error, never
an empty layer.

Many NWS alerts carry ``geometry: null`` -- they are zone-based (forecast/county
zones listed in ``affectedZones``). Each zone's polygon is fetched once and kept
24 h. An alert whose zones cannot be resolved is returned in ``zone_only`` with
its event and expiry, so nothing is silently dropped.
"""
import asyncio
import math
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple

import httpx

from app.config import settings
from app.internal.cameras.base import safe_error
from app.internal.logger import logger

NWS_ALERTS_URL = "https://api.weather.gov/alerts/active"
NWS_ZONE_PREFIX = "https://api.weather.gov/zones/"   # only zone URLs under this are ever fetched
METAR_URL = "https://aviationweather.gov/api/data/metar"
CREDIT = "National Weather Service"
WIND_CREDIT = "NOAA Aviation Weather Center (METAR)"
DEFAULT_USER_AGENT = "DRB scanner map (contact: set NWS_USER_AGENT)"

RETRY_AFTER_FAILURE_S = 30
ZONE_TTL_S = 24 * 3600
ZONE_FAILURE_RETRY_S = 15 * 60
ZONE_CONCURRENCY = 6
# A state-wide winter event can list 100+ zones. Fetches past this cap wait for
# the next refresh (the zone cache fills across refreshes); meanwhile the alert
# shows in zone_only or with the zones resolved so far.
MAX_ZONE_FETCHES_PER_REFRESH = 80
# Wall-clock cap on one refresh's zone fetches. The refresh runs under the feed
# lock, so without it a slow zone endpoint (20 s timeout x 80 fetches / 6 wide)
# could hold every /weather/alerts request for minutes. Unfinished zones are
# simply not cached and are retried next refresh.
ZONE_BUDGET_S = 15.0

WIND_TTL_S = 5 * 60
WIND_CELL_DEG = 0.5      # wind cache cell; one upstream call per cell per TTL
WIND_MAX_KM = 75.0       # a station farther than this says nothing about the incident's wind
WIND_STALE_S = 2 * 3600  # older reports are returned but flagged stale
_WIND_CACHE_MAX = 500

_warned_ua = False


def _headers(accept: str) -> Dict[str, str]:
    global _warned_ua
    ua = settings.nws_user_agent or DEFAULT_USER_AGENT
    if (ua == DEFAULT_USER_AGENT or "example.com" in ua) and not _warned_ua:
        _warned_ua = True
        logger.warning("NWS_USER_AGENT is the placeholder -- set it to 'app (contact email)'; NWS may block us")
    return {"User-Agent": ua, "Accept": accept}


# ── geometry helpers ─────────────────────────────────────────────────────────

def _polygons(geom: Any) -> List[Any]:
    """Polygon coordinate arrays out of a Polygon / MultiPolygon / GeometryCollection."""
    if not isinstance(geom, dict):
        return []
    t = geom.get("type")
    if t == "Polygon" and isinstance(geom.get("coordinates"), list):
        return [geom["coordinates"]]
    if t == "MultiPolygon" and isinstance(geom.get("coordinates"), list):
        return list(geom["coordinates"])
    if t == "GeometryCollection":
        return [p for g in geom.get("geometries") or [] for p in _polygons(g)]
    return []


def _merge(polys: List[Any]) -> Optional[Dict[str, Any]]:
    if not polys:
        return None
    if len(polys) == 1:
        return {"type": "Polygon", "coordinates": polys[0]}
    return {"type": "MultiPolygon", "coordinates": polys}


def _bbox(geom: Optional[Dict[str, Any]]) -> Optional[List[float]]:
    """[south, west, north, east] of a (Multi)Polygon, or None."""
    lats: List[float] = []
    lons: List[float] = []
    for poly in _polygons(geom):
        for ring in poly:
            for pt in ring:
                try:
                    lons.append(float(pt[0]))
                    lats.append(float(pt[1]))
                except (TypeError, ValueError, IndexError):
                    continue
    if not lats:
        return None
    return [min(lats), min(lons), max(lats), max(lons)]


def intersects(b: Optional[List[float]], south: float, west: float, north: float, east: float) -> bool:
    return b is not None and b[0] <= north and b[2] >= south and b[1] <= east and b[3] >= west


# ── NWS alerts ───────────────────────────────────────────────────────────────

def normalize_alert(feature: Any) -> Optional[Dict[str, Any]]:
    """One GeoJSON feature -> our record (geometry may still be None). None if unparseable."""
    if not isinstance(feature, dict):
        return None
    p = feature.get("properties")
    if not isinstance(p, dict):
        return None
    aid = p.get("id") or feature.get("id")
    event = p.get("event")
    if not aid or not event:
        return None
    geometry = _merge(_polygons(feature.get("geometry")))
    zones = [z for z in (p.get("affectedZones") or []) if isinstance(z, str)]
    return {
        "id": aid,
        "event": event,
        "severity": p.get("severity"),
        "urgency": p.get("urgency"),
        "headline": p.get("headline"),
        "onset": p.get("onset"),
        "expires": p.get("expires"),
        "ends": p.get("ends"),
        "sender_name": p.get("senderName"),   # an NWS office ("NWS Upton NY"), not a person
        "area_desc": p.get("areaDesc"),
        "geometry": geometry,
        "geometry_source": "alert" if geometry else None,
        "zones": zones,
    }


class _ZoneCache:
    def __init__(self):
        self._entries: Dict[str, Tuple[float, Optional[List[Any]]]] = {}   # url -> (expires_at, polygons|None)

    def get(self, url: str) -> Tuple[bool, Optional[List[Any]]]:
        e = self._entries.get(url)
        if e is None or e[0] < time.time():
            return False, None
        return True, e[1]

    def put(self, url: str, polys: Optional[List[Any]]):
        ttl = ZONE_TTL_S if polys else ZONE_FAILURE_RETRY_S
        self._entries[url] = (time.time() + ttl, polys)

    def purge(self):
        now = time.time()
        for k in [k for k, e in self._entries.items() if e[0] < now]:
            del self._entries[k]

    def clear(self):
        self._entries.clear()


zones = _ZoneCache()


async def _fetch_zone(client: "httpx.AsyncClient", url: str, sem: asyncio.Semaphore) -> None:
    async with sem:
        try:
            r = await client.get(url, headers=_headers("application/geo+json"))
            r.raise_for_status()
            polys = _polygons(r.json().get("geometry"))
        except Exception as e:
            logger.warning(f"NWS zone {url} unresolved: {safe_error(e)}")
            polys = []
        zones.put(url, polys or None)


async def _resolve_zones(client: "httpx.AsyncClient", alerts: List[Dict[str, Any]]) -> None:
    """Fill geometry for zone-based alerts in place, from the 24 h zone cache."""
    zones.purge()
    wanted: List[str] = []
    for a in alerts:
        if a["geometry"] is None:
            for z in a["zones"]:
                if z.startswith(NWS_ZONE_PREFIX) and not zones.get(z)[0] and z not in wanted:
                    wanted.append(z)
    if wanted:
        sem = asyncio.Semaphore(ZONE_CONCURRENCY)
        try:
            await asyncio.wait_for(
                asyncio.gather(*(_fetch_zone(client, z, sem) for z in wanted[:MAX_ZONE_FETCHES_PER_REFRESH])),
                timeout=ZONE_BUDGET_S)
        except asyncio.TimeoutError:
            logger.warning(f"NWS zone fetches over {ZONE_BUDGET_S:.0f}s budget; the rest resolve next refresh")
    for a in alerts:
        if a["geometry"] is not None:
            continue
        polys: List[Any] = []
        missing = 0
        for z in a["zones"]:
            hit, p = zones.get(z)
            if hit and p:
                polys.extend(p)
            else:
                missing += 1
        a["geometry"] = _merge(polys)
        if a["geometry"] is not None:
            a["geometry_source"] = "zones"
            a["zones_unresolved"] = missing   # >0: drawn area is smaller than the real one


async def _fetch_alerts() -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    area = ",".join(s.strip().upper() for s in (settings.nws_area or "").split(",") if s.strip())
    if not area:
        raise ValueError("NWS_AREA is empty -- no state/marine area to ask NWS for")
    async with httpx.AsyncClient(timeout=20.0) as client:
        r = await client.get(NWS_ALERTS_URL, params={"area": area}, headers=_headers("application/geo+json"))
        r.raise_for_status()
        raw = r.json()
        features = raw.get("features") if isinstance(raw, dict) else None
        if not isinstance(features, list):
            raise ValueError("NWS alerts payload has no 'features' list -- schema change?")
        parsed = [a for a in (normalize_alert(f) for f in features) if a is not None]
        if features and not parsed:
            raise ValueError(f"0 of {len(features)} NWS alerts parsed -- schema change?")
        await _resolve_zones(client, parsed)
    mapped, zone_only = [], []
    for a in parsed:
        a["bbox"] = _bbox(a["geometry"])
        if a["geometry"] is None:
            zone_only.append({k: v for k, v in a.items() if k not in ("geometry", "geometry_source", "bbox")})
        else:
            mapped.append(a)
    return mapped, zone_only


class _AlertsFeed:
    def __init__(self):
        self.alerts: List[Dict[str, Any]] = []
        self.zone_only: List[Dict[str, Any]] = []
        self.fetched_at: Optional[float] = None   # epoch s of last SUCCESSFUL fetch
        self.error: Optional[str] = None
        self._next_attempt = 0.0
        self._lock = asyncio.Lock()

    async def get(self) -> "_AlertsFeed":
        if time.time() < self._next_attempt:
            return self
        async with self._lock:   # single flight: waiters see the refreshed state
            if time.time() < self._next_attempt:
                return self
            ttl = max(30, int(settings.weather_alerts_ttl_s))
            try:
                self.alerts, self.zone_only = await _fetch_alerts()
                self.fetched_at = time.time()
                self.error = None
                self._next_attempt = time.time() + ttl
            except Exception as e:
                self._next_attempt = time.time() + min(ttl, RETRY_AFTER_FAILURE_S)
                self.error = safe_error(e)
                logger.warning(f"NWS alerts refresh failed, serving {len(self.alerts)} cached: {self.error}")
        return self


alerts = _AlertsFeed()


# ── METAR wind ───────────────────────────────────────────────────────────────

def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dp, dl = p2 - p1, math.radians(lon2 - lon1)
    a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
    return 6371.0 * 2 * math.asin(min(1.0, math.sqrt(a)))


def normalize_metar(x: Any) -> Optional[Dict[str, Any]]:
    if not isinstance(x, dict) or not x.get("icaoId"):
        return None
    try:
        lat, lon = float(x["lat"]), float(x["lon"])
    except (KeyError, TypeError, ValueError):
        return None
    wspd = x.get("wspd")
    if not isinstance(wspd, (int, float)):
        return None   # no wind group in this report
    wdir = x.get("wdir")
    variable = isinstance(wdir, str) and wdir.upper() == "VRB"
    obs = x.get("obsTime")
    obs_iso = (datetime.fromtimestamp(obs, tz=timezone.utc).isoformat().replace("+00:00", "Z")
               if isinstance(obs, (int, float)) else x.get("reportTime"))
    gust = x.get("wgst")
    return {
        "station": x["icaoId"],
        "station_name": x.get("name") or "",
        "lat": lat, "lon": lon,
        "obs_epoch": obs if isinstance(obs, (int, float)) else None,
        "obs_time": obs_iso,
        "wind_dir_deg": wdir if isinstance(wdir, (int, float)) else None,
        "variable": variable,
        "wind_kt": wspd,
        "gust_kt": gust if isinstance(gust, (int, float)) else None,
        "raw": x.get("rawOb"),
    }


class _WindCell:
    def __init__(self):
        self.stations: List[Dict[str, Any]] = []
        self.fetched_at: Optional[float] = None
        self.error: Optional[str] = None
        self.next_attempt = 0.0
        self.lock = asyncio.Lock()


_wind_cells: Dict[Tuple[int, int], _WindCell] = {}


def _cell_key(lat: float, lon: float) -> Tuple[int, int]:
    return (math.floor(lat / WIND_CELL_DEG), math.floor(lon / WIND_CELL_DEG))


async def _fetch_metars(key: Tuple[int, int]) -> List[Dict[str, Any]]:
    # Search box = the cell padded by WIND_MAX_KM, so every point in the cell sees
    # every station within WIND_MAX_KM of it.
    s, w = key[0] * WIND_CELL_DEG, key[1] * WIND_CELL_DEG
    n, e = s + WIND_CELL_DEG, w + WIND_CELL_DEG
    pad_lat = WIND_MAX_KM / 111.0
    pad_lon = WIND_MAX_KM / (111.0 * max(0.2, math.cos(math.radians(max(abs(s), abs(n))))))
    bbox = f"{s - pad_lat:.3f},{w - pad_lon:.3f},{n + pad_lat:.3f},{e + pad_lon:.3f}"
    async with httpx.AsyncClient(timeout=15.0) as client:
        r = await client.get(METAR_URL, params={"bbox": bbox, "format": "json"}, headers=_headers("application/json"))
        r.raise_for_status()
        if r.status_code == 204 or not r.content.strip():
            return []   # no station in the box
        raw = r.json()
    if not isinstance(raw, list):
        raise ValueError(f"METAR API: expected a JSON list, got {type(raw).__name__}")
    latest: Dict[str, Dict[str, Any]] = {}
    for m in (normalize_metar(x) for x in raw):
        if m is None:
            continue
        cur = latest.get(m["station"])
        if cur is None or (m["obs_epoch"] or 0) > (cur["obs_epoch"] or 0):
            latest[m["station"]] = m
    # Schema canary: reports without a wind group are normal, but if NO record even
    # has a station id and position, the field names changed under us.
    if raw and not any(isinstance(x, dict) and x.get("icaoId") and "lat" in x for x in raw):
        raise ValueError(f"0 of {len(raw)} METAR records parsed -- schema change?")
    return list(latest.values())


async def _cell(lat: float, lon: float) -> _WindCell:
    key = _cell_key(lat, lon)
    cell = _wind_cells.get(key)
    if cell is None:
        if len(_wind_cells) >= _WIND_CACHE_MAX:   # bound memory: drop expired cells, then the oldest
            for k in [k for k, c in _wind_cells.items() if c.next_attempt < time.time() and not c.lock.locked()]:
                del _wind_cells[k]
            if len(_wind_cells) >= _WIND_CACHE_MAX:
                del _wind_cells[min(_wind_cells, key=lambda k: _wind_cells[k].next_attempt)]
        cell = _wind_cells[key] = _WindCell()
    if time.time() < cell.next_attempt:
        return cell
    async with cell.lock:
        if time.time() < cell.next_attempt:
            return cell
        try:
            cell.stations = await _fetch_metars(key)
            cell.fetched_at = time.time()
            cell.error = None
            cell.next_attempt = time.time() + WIND_TTL_S
        except Exception as e:
            cell.next_attempt = time.time() + RETRY_AFTER_FAILURE_S
            cell.error = safe_error(e)
            logger.warning(f"METAR refresh for cell {key} failed, serving {len(cell.stations)} cached: {cell.error}")
    return cell


async def nearest_wind(lat: float, lon: float) -> Dict[str, Any]:
    """Wind from the nearest METAR station within WIND_MAX_KM, or a clear reason why not."""
    cell = await _cell(lat, lon)
    base = {"fetched_at": cell.fetched_at, "error": cell.error, "credit": WIND_CREDIT}
    ranked = sorted(((haversine_km(lat, lon, s["lat"], s["lon"]), s) for s in cell.stations), key=lambda t: t[0])
    if not ranked or ranked[0][0] > WIND_MAX_KM:
        if cell.fetched_at is None:
            reason = f"METAR feed unavailable: {cell.error}" if cell.error else "METAR feed not fetched yet"
        else:
            reason = f"no METAR station reporting wind within {WIND_MAX_KM:.0f} km"
        return {**base, "available": False, "reason": reason, "wind": None}
    dist, s = ranked[0]
    stale = s["obs_epoch"] is not None and time.time() - s["obs_epoch"] > WIND_STALE_S
    wind = {k: v for k, v in s.items() if k not in ("obs_epoch",)}
    wind["distance_km"] = round(dist, 1)
    wind["stale"] = stale
    return {**base, "available": True, "reason": None, "wind": wind}