PolyWeather/src/data_collection/country_networks.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from src.data_collection.city_registry import CITY_REGISTRY
from src.data_collection.city_time import get_city_utc_offset_seconds


CHINA_CMA_CITIES = {
    "beijing",
    "chengdu",
    "chongqing",
    "guangzhou",
    "qingdao",
    "shanghai",
    "wuhan",
}


def _japan_jma_cities() -> set[str]:
    return {"tokyo"}


def _safe_float(value: Any) -> Optional[float]:
    try:
        if value is None or value == "":
            return None
        return float(value)
    except Exception:
        return None


def _parse_obs_datetime(
    value: Any,
    epoch_value: Any = None,
    utc_offset_seconds: Any = None,
) -> Optional[datetime]:
    for candidate in (epoch_value, value):
        if candidate is None or candidate == "":
            continue
        try:
            if isinstance(candidate, (int, float)):
                numeric = float(candidate)
                if numeric > 1_000_000_000_000:
                    numeric /= 1000.0
                if numeric > 1_000_000_000:
                    return datetime.fromtimestamp(numeric, tz=timezone.utc)
            text = str(candidate).strip()
            if not text:
                continue
            if text.isdigit():
                numeric = float(text)
                if numeric > 1_000_000_000_000:
                    numeric /= 1000.0
                if numeric > 1_000_000_000:
                    return datetime.fromtimestamp(numeric, tz=timezone.utc)
            normalized = text.replace(" ", "T")
            if normalized.endswith("Z"):
                normalized = normalized[:-1] + "+00:00"
            parsed = datetime.fromisoformat(normalized)
            if parsed.tzinfo is None and utc_offset_seconds is not None:
                try:
                    offset = int(utc_offset_seconds)
                    parsed = parsed.replace(tzinfo=timezone(timedelta(seconds=offset)))
                except Exception:
                    pass
            return parsed
        except Exception:
            continue
    return None


def _format_obs_time_label(
    value: Any,
    epoch_value: Any = None,
    display_utc_offset_seconds: Any = None,
    source_utc_offset_seconds: Any = None,
) -> Optional[str]:
    text = str(value or "").strip()
    parsed = _parse_obs_datetime(value, epoch_value, source_utc_offset_seconds)
    if parsed is not None:
        if display_utc_offset_seconds is not None and parsed.tzinfo is not None:
            try:
                offset = int(display_utc_offset_seconds)
                local_tz = timezone(timedelta(seconds=offset))
                return parsed.astimezone(local_tz).strftime("%H:%M")
            except Exception:
                pass
        suffix = "Z" if parsed.tzinfo is not None and parsed.utcoffset() == timezone.utc.utcoffset(parsed) else ""
        return parsed.strftime("%H:%M") + suffix
    if not text:
        return None
    if "T" in text:
        return text.split("T")[-1][:5]
    if " " in text:
        return text.split()[-1][:5]
    return text[:5] if len(text) >= 5 else text


def _timing_delta_minutes(anchor_dt: Optional[datetime], station_dt: Optional[datetime]) -> Optional[int]:
    if anchor_dt is None or station_dt is None:
        return None
    try:
        if (anchor_dt.tzinfo is None) != (station_dt.tzinfo is None):
            return None
        delta = abs((station_dt - anchor_dt).total_seconds()) / 60.0
        return int(round(delta))
    except Exception:
        return None


def _station_age_minutes(station_dt: Optional[datetime]) -> Optional[int]:
    if station_dt is None or station_dt.tzinfo is None:
        return None
    try:
        return int(round((datetime.now(timezone.utc) - station_dt.astimezone(timezone.utc)).total_seconds() / 60.0))
    except Exception:
        return None


def _sync_status(delta_minutes: Optional[int], age_minutes: Optional[int]) -> str:
    if age_minutes is not None and age_minutes > 60:
        return "stale"
    reference = delta_minutes if delta_minutes is not None else age_minutes
    if reference is None:
        return "unknown"
    if reference <= 10:
        return "synced"
    if reference <= 30:
        return "near_realtime"
    if reference <= 60:
        return "lagged"
    return "stale"


def _enrich_station_timing(
    anchor: Optional[Dict[str, Any]],
    rows: List[Dict[str, Any]],
    display_utc_offset_seconds: Any = None,
) -> List[Dict[str, Any]]:
    anchor = anchor or {}
    anchor_dt = _parse_obs_datetime(
        anchor.get("obs_time"),
        anchor.get("obs_time_epoch"),
        anchor.get("obs_time_utc_offset_seconds"),
    )
    enriched: List[Dict[str, Any]] = []
    for row in rows:
        station_dt = _parse_obs_datetime(
            row.get("obs_time"),
            row.get("obs_time_epoch"),
            row.get("obs_time_utc_offset_seconds"),
        )
        delta_minutes = _timing_delta_minutes(anchor_dt, station_dt)
        age_minutes = _station_age_minutes(station_dt)
        status = _sync_status(delta_minutes, age_minutes)
        enriched_row = dict(row)
        enriched_row["obs_time_label"] = _format_obs_time_label(
            row.get("obs_time"),
            row.get("obs_time_epoch"),
            display_utc_offset_seconds,
            row.get("obs_time_utc_offset_seconds"),
        )
        if display_utc_offset_seconds is not None and enriched_row.get("obs_time_label"):
            enriched_row["obs_time_display_tz"] = "city_local"
        enriched_row["age_minutes"] = age_minutes
        enriched_row["time_delta_vs_anchor_minutes"] = delta_minutes
        enriched_row["sync_status"] = status
        enriched_row["usable_for_intraday"] = status != "stale"
        enriched.append(enriched_row)
    return enriched


def _city_meta(city: str) -> Dict[str, Any]:
    return CITY_REGISTRY.get(str(city or "").strip().lower(), {}) or {}


def _provider_code_for_city(city: str) -> str:
    normalized = str(city or "").strip().lower()
    meta = _city_meta(normalized)
    settlement_source = str(meta.get("settlement_source") or "").strip().lower()
    if normalized in {"ankara", "istanbul"}:
        return "turkey_mgm"
    if normalized in {"busan", "seoul"}:
        return "korea_kma"
    if normalized == "tel aviv":
        return "israel_ims"
    if normalized == "jeddah":
        return "saudi_ncm"
    if normalized == "paris":
        return "fr_aeroweb"
    if settlement_source == "hko":
        return "hongkong_hko"
    if settlement_source == "cwa":
        return "taiwan_cwa"
    if normalized in _japan_jma_cities():
        return "japan_jma"
    if normalized in CHINA_CMA_CITIES:
        return "china_cma"
    return "global_metar"


def _bool(value: Any) -> bool:
    return bool(value)


def _normalize_station_row(
    *,
    station_code: Optional[str],
    station_label: Optional[str],
    temp: Any,
    lat: Any = None,
    lon: Any = None,
    obs_time: Optional[str] = None,
    source_code: str,
    source_label: str,
    is_official: bool,
    is_airport_station: bool,
    is_settlement_anchor: bool,
    extra: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    payload = {
        "station_code": str(station_code or "").strip() or None,
        "station_label": str(station_label or "").strip() or None,
        "is_airport_station": bool(is_airport_station),
        "lat": _safe_float(lat),
        "lon": _safe_float(lon),
        "obs_time": str(obs_time or "").strip() or None,
        "temp": _safe_float(temp),
        "source_code": str(source_code or "").strip().lower() or None,
        "source_label": str(source_label or "").strip() or None,
        "is_official": bool(is_official),
        "is_settlement_anchor": bool(is_settlement_anchor),
    }
    if isinstance(extra, dict):
        for key, value in extra.items():
            if key not in payload:
                payload[key] = value
    return payload


def _airport_primary_from_raw(city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
    meta = _city_meta(city)
    metar = raw.get("metar") or {}
    current = metar.get("current") or {}

    # High-frequency realtime sources take priority over plain METAR.
    madis = raw.get("madis_hfmetar_current") or {}
    if madis.get("temp_c") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or madis.get("icao"),
            station_label=meta.get("airport_name") or meta.get("icao"),
            temp=madis["temp_c"],
            obs_time=madis.get("obs_time") or metar.get("observation_time"),
            source_code="madis_hfmetar",
            source_label="NOAA MADIS",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(madis.get("wind_kt")) or _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    amos = raw.get("amos") or {}
    if amos.get("temp_c") is not None:
        is_amsc = amos.get("source") == "amsc_awos"
        return _normalize_station_row(
            station_code=amos.get("icao") or meta.get("icao"),
            station_label=amos.get("station_label") or meta.get("airport_name") or meta.get("icao"),
            temp=amos["temp_c"],
            obs_time=amos.get("obs_time") or amos.get("observation_time") or metar.get("observation_time"),
            source_code="amsc_awos" if is_amsc else "amos",
            source_label="AMSC AWOS" if is_amsc else "AMOS",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    mgm = raw.get("mgm") or {}
    mgm_current = mgm.get("current") or {}
    if mgm_current.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or str(mgm.get("istNo") or ""),
            station_label=meta.get("airport_name") or mgm.get("station_label") or meta.get("icao"),
            temp=_safe_float(mgm_current["temp"]),
            obs_time=mgm.get("obs_time") or metar.get("observation_time"),
            source_code="mgm",
            source_label="MGM",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    jma = raw.get("jma_current") or {}
    if jma.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or str(jma.get("icao") or "RJTT"),
            station_label=meta.get("airport_name") or jma.get("station_label") or meta.get("icao"),
            temp=_safe_float(jma["temp"]),
            obs_time=str(jma.get("obs_time") or metar.get("observation_time") or ""),
            source_code="jma_amedas",
            source_label="JMA AMeDAS",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    fmi = raw.get("fmi_current") or {}
    if fmi.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or str(fmi.get("icao") or "EFHK"),
            station_label=meta.get("airport_name") or fmi.get("station_label") or meta.get("icao"),
            temp=_safe_float(fmi["temp"]),
            obs_time=str(fmi.get("obs_time") or metar.get("observation_time") or ""),
            source_code="fmi",
            source_label="FMI",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    knmi = raw.get("knmi_current") or {}
    if knmi.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or str(knmi.get("icao") or "EHAM"),
            station_label=meta.get("airport_name") or knmi.get("station_label") or meta.get("icao"),
            temp=_safe_float(knmi["temp"]),
            obs_time=str(knmi.get("obs_time") or metar.get("observation_time") or ""),
            source_code="knmi",
            source_label="KNMI",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    cowin = raw.get("cowin_current") or {}
    if cowin.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or str(cowin.get("icao") or "COWIN6087"),
            station_label=meta.get("airport_name") or cowin.get("station_label") or meta.get("icao"),
            temp=_safe_float(cowin["temp"]),
            obs_time=str(cowin.get("obs_time") or metar.get("observation_time") or ""),
            source_code="cowin_obs",
            source_label="CoWIN 6087",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    aeroweb = raw.get("aeroweb") or {}
    aw_current = aeroweb.get("current") or {}
    if aw_current.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or "LFPB",
            station_label=meta.get("airport_name") or aeroweb.get("station_label") or "Paris Le Bourget (AEROWEB)",
            temp=_safe_float(aw_current["temp"]),
            obs_time=aeroweb.get("obs_time") or metar.get("observation_time"),
            source_code="aeroweb",
            source_label="AEROWEB",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(aw_current.get("wind_speed_kt"))
                or _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(aw_current.get("wind_dir"))
                or _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "pressure": _safe_float(aw_current.get("pressure")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": aw_current.get("raw_metar") or current.get("raw_metar"),
            },
        )

    ims = raw.get("ims") or {}
    ims_current = ims.get("current") or {}
    if ims_current.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or ims.get("station_id") or "LLBG",
            station_label=meta.get("airport_name") or ims.get("station_label") or "Ben Gurion Airport",
            temp=_safe_float(ims_current["temp"]),
            obs_time=ims.get("obs_time") or metar.get("observation_time"),
            source_code="ims",
            source_label="IMS Lod Airport",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(ims_current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(ims_current.get("wind_speed_kt"))
                or _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(ims_current.get("wind_dir"))
                or _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(ims_current.get("humidity"))
                or _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    ncm = raw.get("ncm") or {}
    ncm_current = ncm.get("current") or {}
    if ncm_current.get("temp") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or "OEJN",
            station_label=meta.get("airport_name") or ncm.get("station_label") or "King Abdulaziz Intl (NCM)",
            temp=_safe_float(ncm_current["temp"]),
            obs_time=ncm.get("obs_time") or metar.get("observation_time"),
            source_code="ncm",
            source_label="Saudi NCM",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(ncm_current.get("wind_speed_kt"))
                or _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(ncm_current.get("humidity"))
                or _safe_float(current.get("humidity")),
                "pressure": _safe_float(ncm_current.get("pressure")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    sg_mss = raw.get("singapore_mss_current") or {}
    if sg_mss.get("temp_c") is not None:
        return _normalize_station_row(
            station_code=meta.get("icao") or "WSSS",
            station_label=meta.get("airport_name") or "Changi Airport",
            temp=sg_mss["temp_c"],
            obs_time=sg_mss.get("obs_time") or metar.get("observation_time"),
            source_code="sg_mss",
            source_label="Singapore MSS",
            is_official=True,
            is_airport_station=True,
            is_settlement_anchor=False,
            extra={
                "max_so_far": _safe_float(current.get("max_temp_so_far")),
                "max_temp_time": current.get("max_temp_time"),
                "obs_age_min": None,
                "report_time": metar.get("report_time"),
                "receipt_time": metar.get("receipt_time"),
                "obs_time_epoch": metar.get("obs_time_epoch"),
                "obs_time_utc_offset_seconds": 0,
                "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
                "wind_dir": _safe_float(current.get("wind_dir")),
                "humidity": _safe_float(current.get("humidity")),
                "visibility_mi": _safe_float(current.get("visibility_mi")),
                "wx_desc": current.get("wx_desc"),
                "raw_metar": current.get("raw_metar"),
            },
        )

    return _normalize_station_row(
        station_code=meta.get("icao") or metar.get("icao"),
        station_label=meta.get("airport_name") or metar.get("station_name") or metar.get("icao"),
        temp=current.get("temp"),
        obs_time=metar.get("observation_time"),
        source_code="metar",
        source_label="METAR",
        is_official=True,
        is_airport_station=True,
        is_settlement_anchor=False,
        extra={
            "max_so_far": _safe_float(current.get("max_temp_so_far")),
            "max_temp_time": current.get("max_temp_time"),
            "obs_age_min": None,
            "report_time": metar.get("report_time"),
            "receipt_time": metar.get("receipt_time"),
            "obs_time_epoch": metar.get("obs_time_epoch"),
            "obs_time_utc_offset_seconds": 0,
            "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
            "wind_dir": _safe_float(current.get("wind_dir")),
            "humidity": _safe_float(current.get("humidity")),
            "visibility_mi": _safe_float(current.get("visibility_mi")),
            "wx_desc": current.get("wx_desc"),
            "raw_metar": current.get("raw_metar"),
        },
    )


def _metar_cluster_rows(raw: Dict[str, Any]) -> List[Dict[str, Any]]:
    rows = raw.get("mgm_nearby") or []
    out: List[Dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        out.append(
            _normalize_station_row(
                station_code=row.get("icao") or row.get("istNo"),
                station_label=row.get("name"),
                temp=row.get("temp"),
                lat=row.get("lat"),
                lon=row.get("lon"),
                obs_time=row.get("obs_time"),
                source_code="metar_cluster",
                source_label="METAR cluster",
                is_official=False,
                is_airport_station=True,
                is_settlement_anchor=False,
                extra={
                    "obs_time_epoch": row.get("obs_time_epoch"),
                    "obs_time_utc_offset_seconds": 0,
                    "wind_dir": _safe_float(row.get("wind_dir")),
                    "wind_speed_kt": _safe_float(row.get("wind_speed_kt") or row.get("wind_speed")),
                    "raw_metar": row.get("raw_metar"),
                },
            )
        )
    return out


def _jma_rows(raw: Dict[str, Any], city: str) -> List[Dict[str, Any]]:
    rows = raw.get("jma_official_nearby") or []
    out: List[Dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        out.append(
            _normalize_station_row(
                station_code=row.get("icao") or row.get("istNo"),
                station_label=row.get("name"),
                temp=row.get("temp"),
                lat=row.get("lat"),
                lon=row.get("lon"),
                obs_time=row.get("obs_time"),
                source_code="jma",
                source_label="JMA",
                is_official=True,
                is_airport_station=False,
                is_settlement_anchor=False,
            )
        )
    return out


def _kma_rows(raw: Dict[str, Any], city: str) -> List[Dict[str, Any]]:
    rows = raw.get("kma_official_nearby") or []
    out: List[Dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        out.append(
            _normalize_station_row(
                station_code=row.get("station_code") or row.get("icao") or row.get("istNo"),
                station_label=row.get("station_label") or row.get("name"),
                temp=row.get("temp"),
                lat=row.get("lat"),
                lon=row.get("lon"),
                obs_time=row.get("obs_time"),
                source_code="kma",
                source_label="KMA",
                is_official=True,
                is_airport_station=False,
                is_settlement_anchor=False,
                extra={
                    "distance_km": _safe_float(row.get("distance_km")),
                    "network_type": row.get("network_type"),
                },
            )
        )
    return out


    rows = raw.get("ru_official_nearby") or []
    out: List[Dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        out.append(
            _normalize_station_row(
                station_code=row.get("station_code") or row.get("icao") or row.get("istNo"),
                station_label=row.get("station_label") or row.get("name"),
                temp=row.get("temp"),
                lat=row.get("lat"),
                lon=row.get("lon"),
                obs_time=row.get("obs_time"),
                source_label="Russia station web",
                is_official=True,
                is_airport_station=_bool(row.get("is_airport_station")),
                is_settlement_anchor=False,
                extra={
                    "distance_km": _safe_float(row.get("distance_km")),
                    "page_url": row.get("page_url"),
                },
            )
        )
    return out


def _mgm_rows(raw: Dict[str, Any], city: str) -> List[Dict[str, Any]]:
    meta = _city_meta(city)
    rows = raw.get("mgm_nearby") or []
    out: List[Dict[str, Any]] = []
    airport_code = str(meta.get("icao") or "").strip().upper()
    for row in rows:
        if not isinstance(row, dict):
            continue
        station_code = str(row.get("icao") or row.get("istNo") or "").strip() or None
        station_label = row.get("name")
        out.append(
            _normalize_station_row(
                station_code=station_code,
                station_label=station_label,
                temp=row.get("temp"),
                lat=row.get("lat"),
                lon=row.get("lon"),
                obs_time=row.get("obs_time"),
                source_code="mgm",
                source_label="MGM",
                is_official=True,
                is_airport_station=_bool(station_code and station_code.upper() == airport_code)
                or ("airport" in str(station_label or "").lower()),
                is_settlement_anchor=False,
                extra={
                    "wind_dir": _safe_float(row.get("wind_dir")),
                    "wind_speed_kt": _safe_float(row.get("wind_speed_kt") or row.get("wind_speed")),
                },
            )
        )
    return out


def _settlement_anchor_row(city: str, raw: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    meta = _city_meta(city)
    settlement_current = raw.get("settlement_current") or {}
    current = settlement_current.get("current") or {}
    if not current and not settlement_current:
        return None
    station_code = (
        settlement_current.get("station_code")
        or meta.get("settlement_station_code")
        or meta.get("icao")
    )
    station_label = (
        settlement_current.get("station_name")
        or meta.get("settlement_station_label")
        or meta.get("airport_name")
    )
    settlement_source = str(meta.get("settlement_source") or "official").strip().lower() or "official"
    return _normalize_station_row(
        station_code=station_code,
        station_label=station_label,
        temp=current.get("temp"),
        obs_time=settlement_current.get("observation_time"),
        source_code=settlement_source,
        source_label=settlement_source.upper(),
        is_official=True,
        is_airport_station=False,
        is_settlement_anchor=True,
        extra={
            "max_so_far": _safe_float(current.get("max_temp_so_far")),
            "max_temp_time": current.get("max_temp_time"),
            "humidity": _safe_float(current.get("humidity")),
            "wind_speed_kt": _safe_float(current.get("wind_speed_kt")),
            "wind_dir": _safe_float(current.get("wind_dir")),
        },
    )


def _settlement_station_metadata(city: str) -> Dict[str, Any]:
    meta = _city_meta(city)
    settlement_source = str(meta.get("settlement_source") or "metar").strip().lower() or "metar"
    station_code = (
        str(meta.get("settlement_station_code") or "").strip()
        or str(meta.get("icao") or "").strip()
        or None
    )
    station_label = (
        str(meta.get("settlement_station_label") or "").strip()
        or str(meta.get("airport_name") or "").strip()
        or None
    )
    airport_code = str(meta.get("icao") or "").strip()
    is_explicit_official_anchor = settlement_source in {"hko", "cwa"}
    return {
        "provider_code": _provider_code_for_city(city),
        "settlement_source": settlement_source,
        "settlement_station_code": station_code,
        "settlement_station_label": station_label,
        "airport_code": airport_code or None,
        "airport_name": str(meta.get("airport_name") or "").strip() or None,
        "is_airport_anchor": not is_explicit_official_anchor,
        "is_official_station_anchor": is_explicit_official_anchor,
    }


def _network_signals(
    airport_primary: Optional[Dict[str, Any]],
    official_rows: List[Dict[str, Any]],
) -> Dict[str, Any]:
    airport_temp = _safe_float((airport_primary or {}).get("temp"))
    valid_rows = [
        row
        for row in official_rows
        if _safe_float(row.get("temp")) is not None and row.get("usable_for_intraday") is not False
    ]
    if not valid_rows:
        return {
            "network_lead_signal": {"available": False},
            "network_spread_signal": {"available": False},
            "center_station_candidate": None,
            "airport_vs_network_delta": None,
        }

    hottest = max(valid_rows, key=lambda row: float(row.get("temp") or -999))
    coolest = min(valid_rows, key=lambda row: float(row.get("temp") or 999))
    hottest_temp = _safe_float(hottest.get("temp"))
    coolest_temp = _safe_float(coolest.get("temp"))
    spread = None
    airport_delta = None
    if hottest_temp is not None and coolest_temp is not None:
        spread = round(hottest_temp - coolest_temp, 1)
    if airport_temp is not None and hottest_temp is not None:
        airport_delta = round(hottest_temp - airport_temp, 1)
    return {
        "network_lead_signal": {
            "available": airport_delta is not None,
            "delta": airport_delta,
            "leader_station_code": hottest.get("station_code"),
            "leader_station_label": hottest.get("station_label"),
            "leader_temp": hottest_temp,
            "leader_obs_time": hottest.get("obs_time"),
            "leader_obs_time_label": hottest.get("obs_time_label"),
            "leader_sync_status": hottest.get("sync_status"),
            "leader_time_delta_vs_anchor_minutes": hottest.get("time_delta_vs_anchor_minutes"),
        },
        "network_spread_signal": {
            "available": spread is not None,
            "spread": spread,
            "hottest_station_code": hottest.get("station_code"),
            "coolest_station_code": coolest.get("station_code"),
        },
        "center_station_candidate": hottest,
        "airport_vs_network_delta": airport_delta,
    }


@dataclass
class CountryNetworkProvider:
    provider_code: str
    provider_label: str

    def airport_primary_current(self, city: str, raw: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        return _airport_primary_from_raw(city, raw)

    def airport_primary_history(self, city: str, target_date: str) -> Optional[Dict[str, Any]]:
        return None

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        return []

    def official_nearby_history(self, city: str, target_date: str) -> List[Dict[str, Any]]:
        return []

    def settlement_station_metadata(self, city: str) -> Dict[str, Any]:
        return _settlement_station_metadata(city)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        rows = self.official_nearby_current(city, raw)
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "available": bool(rows),
            "mode": "active" if rows else "unavailable",
            "row_count": len(rows),
        }


class GlobalMetarNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("global_metar", "METAR")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        return _metar_cluster_rows(raw)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        rows = self.official_nearby_current(city, raw)
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "available": bool(rows),
            "mode": "fallback_metar_cluster" if rows else "no_official_network",
            "row_count": len(rows),
        }


class TurkeyMgmNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("turkey_mgm", "MGM")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        return _mgm_rows(raw, city)


class ChinaCmaNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("china_cma", "CMA")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        return _metar_cluster_rows(raw)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        rows = self.official_nearby_current(city, raw)
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "row_count": len(rows),
        }


class JapanJmaNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("japan_jma", "JMA")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        rows = _jma_rows(raw, city)
        if rows:
            return rows
        return _metar_cluster_rows(raw)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        rows = self.official_nearby_current(city, raw)
        has_jma = bool(_jma_rows(raw, city))
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "available": has_jma,
            "mode": "official_active" if has_jma else ("fallback_metar_cluster" if rows else "reference_only"),
            "row_count": len(rows),
        }


class KoreaKmaNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("korea_kma", "KMA")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        rows = _kma_rows(raw, city)
        if rows:
            return rows
        return _metar_cluster_rows(raw)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        rows = self.official_nearby_current(city, raw)
        has_kma = bool(_kma_rows(raw, city))
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "available": has_kma,
            "mode": "official_active" if has_kma else ("fallback_metar_cluster" if rows else "reference_only"),
            "row_count": len(rows),
        }


class IsraelImsNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("israel_ims", "IMS Lod Airport")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        ims = raw.get("ims") or {}
        ims_current = ims.get("current") or {}
        if ims_current.get("temp") is not None:
            row = _normalize_station_row(
                station_code="LLBG",
                station_label=ims.get("station_label") or "Lod Airport",
                temp=ims_current.get("temp"),
                lat=ims.get("lat"),
                lon=ims.get("lon"),
                obs_time=ims.get("obs_time"),
                source_code="ims",
                source_label="IMS Lod Airport",
                is_official=True,
                is_airport_station=True,
                is_settlement_anchor=False,
            )
            return [row]
        return _metar_cluster_rows(raw)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        ims = raw.get("ims") or {}
        has_ims = (ims.get("current") or {}).get("temp") is not None
        if has_ims:
            return {
                "provider_code": self.provider_code,
                "provider_label": self.provider_label,
                "available": True,
                "mode": "official_active",
                "row_count": 1,
            }
        rows = _metar_cluster_rows(raw)
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "available": bool(rows),
            "mode": "fallback_metar_cluster" if rows else "no_official_network",
            "row_count": len(rows),
        }


class AerowebNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("fr_aeroweb", "AEROWEB")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        aw = raw.get("aeroweb") or {}
        aw_current = aw.get("current") or {}
        if aw_current.get("temp") is not None:
            row = _normalize_station_row(
                station_code="LFPB",
                station_label=aw.get("station_label") or "Paris Le Bourget (AEROWEB)",
                temp=aw_current.get("temp"),
                lat=aw.get("lat"),
                lon=aw.get("lon"),
                obs_time=aw.get("obs_time"),
                source_code="aeroweb",
                source_label="AEROWEB",
                is_official=True,
                is_airport_station=True,
                is_settlement_anchor=False,
            )
            return [row]
        return _metar_cluster_rows(raw)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        aw = raw.get("aeroweb") or {}
        has_aw = (aw.get("current") or {}).get("temp") is not None
        if has_aw:
            return {
                "provider_code": self.provider_code,
                "provider_label": self.provider_label,
                "available": True,
                "mode": "official_active",
                "row_count": 1,
            }
        rows = _metar_cluster_rows(raw)
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "available": bool(rows),
            "mode": "fallback_metar_cluster" if rows else "no_official_network",
            "row_count": len(rows),
        }


class SaudiNcmNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("saudi_ncm", "Saudi NCM")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        ncm = raw.get("ncm") or {}
        ncm_current = ncm.get("current") or {}
        if ncm_current.get("temp") is not None:
            row = _normalize_station_row(
                station_code="OEJN",
                station_label=ncm.get("station_label") or "Jeddah OEJN (NCM)",
                temp=ncm_current.get("temp"),
                lat=ncm.get("lat"),
                lon=ncm.get("lon"),
                obs_time=ncm.get("obs_time"),
                source_code="ncm",
                source_label="Saudi NCM",
                is_official=True,
                is_airport_station=True,
                is_settlement_anchor=False,
            )
            return [row]
        return _metar_cluster_rows(raw)

    def official_network_status(self, city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
        ncm = raw.get("ncm") or {}
        has_ncm = (ncm.get("current") or {}).get("temp") is not None
        if has_ncm:
            return {
                "provider_code": self.provider_code,
                "provider_label": self.provider_label,
                "available": True,
                "mode": "official_active",
                "row_count": 1,
            }
        rows = _metar_cluster_rows(raw)
        return {
            "provider_code": self.provider_code,
            "provider_label": self.provider_label,
            "available": bool(rows),
            "mode": "fallback_metar_cluster" if rows else "no_official_network",
            "row_count": len(rows),
        }


class HongKongHkoNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("hongkong_hko", "HKO")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        anchor = _settlement_anchor_row(city, raw)
        return [anchor] if anchor else []


class TaiwanCwaNetworkProvider(CountryNetworkProvider):
    def __init__(self) -> None:
        super().__init__("taiwan_cwa", "CWA")

    def official_nearby_current(self, city: str, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
        anchor = _settlement_anchor_row(city, raw)
        return [anchor] if anchor else []


def get_country_network_provider(city: str) -> CountryNetworkProvider:
    provider_code = _provider_code_for_city(city)
    if provider_code == "turkey_mgm":
        return TurkeyMgmNetworkProvider()
    if provider_code == "korea_kma":
        return KoreaKmaNetworkProvider()
    if provider_code == "japan_jma":
        return JapanJmaNetworkProvider()
    if provider_code == "china_cma":
        return ChinaCmaNetworkProvider()
    if provider_code == "israel_ims":
        return IsraelImsNetworkProvider()
    if provider_code == "saudi_ncm":
        return SaudiNcmNetworkProvider()
    if provider_code == "fr_aeroweb":
        return AerowebNetworkProvider()
    if provider_code == "hongkong_hko":
        return HongKongHkoNetworkProvider()
    if provider_code == "taiwan_cwa":
        return TaiwanCwaNetworkProvider()
    return GlobalMetarNetworkProvider()


def build_country_network_snapshot(city: str, raw: Dict[str, Any]) -> Dict[str, Any]:
    provider = get_country_network_provider(city)
    city_offset = get_city_utc_offset_seconds(city)
    metadata = provider.settlement_station_metadata(city)
    airport_primary = provider.airport_primary_current(city, raw) or {}
    official_nearby = _enrich_station_timing(
        airport_primary,
        provider.official_nearby_current(city, raw),
        city_offset,
    )
    status = provider.official_network_status(city, raw)
    signals = _network_signals(airport_primary, official_nearby)
    usable_count = len([row for row in official_nearby if row.get("usable_for_intraday") is not False])
    stale_count = len([row for row in official_nearby if row.get("sync_status") == "stale"])
    unknown_count = len([row for row in official_nearby if row.get("sync_status") == "unknown"])
    status = {
        **status,
        "usable_row_count": usable_count,
        "stale_row_count": stale_count,
        "unknown_timing_count": unknown_count,
    }
    airport_primary_today_obs = ((raw.get("metar") or {}).get("today_obs") or [])
    if airport_primary.get("source_code") == "cowin_obs":
        try:
            from src.database.runtime_state import OfficialIntradayObservationRepository
            repo = OfficialIntradayObservationRepository()
            local_now = datetime.now(timezone.utc) + timedelta(seconds=city_offset or 0)
            local_date_str = local_now.strftime("%Y-%m-%d")
            points = repo.load_points(
                source_code="cowin_obs",
                station_code=airport_primary.get("station_code") or "6087",
                target_date=local_date_str,
            )
            if points:
                airport_primary_today_obs = [{"time": p["time"], "temp": p["temp"]} for p in points]
        except Exception:
            pass

    return {
        "provider_code": provider.provider_code,
        "provider_label": provider.provider_label,
        "settlement_station": metadata,
        "airport_primary_current": airport_primary,
        "airport_primary_today_obs": airport_primary_today_obs,
        "official_nearby": official_nearby,
        "official_network_source": status.get("provider_code"),
        "official_network_status": status,
        **signals,
    }


def provider_coverage_summary() -> Dict[str, Any]:
    providers: Dict[str, Dict[str, Any]] = {}
    for city in CITY_REGISTRY:
        provider_code = _provider_code_for_city(city)
        entry = providers.setdefault(
            provider_code,
            {
                "cities": [],
                "cities_count": 0,
            },
        )
        entry["cities"].append(city)
        entry["cities_count"] += 1
    return {
        "providers": providers,
        "airport_anchor_coverage": sum(
            1
            for city, meta in CITY_REGISTRY.items()
            if str(meta.get("icao") or "").strip()
        ),
        "official_station_anchor_coverage": sum(
            1
            for city in CITY_REGISTRY
            if _settlement_station_metadata(city).get("is_official_station_anchor")
        ),
    }