from __future__ import annotations

from copy import deepcopy
from datetime import datetime, timezone
import time
from typing import Any, Optional

from loguru import logger

from web.services.canonical_temperature import build_canonical_temperature

_RAW_AMSC_RUNWAY_HISTORY_CACHE: dict[tuple[str, bool], tuple[float, dict[str, list[dict[str, Any]]]]] = {}
_RAW_AMSC_RUNWAY_HISTORY_CACHE_TTL_SEC = 60.0
_RAW_AMSC_RUNWAY_HISTORY_RECENT_WINDOW_SEC = 24 * 60 * 60
_RAW_AMSC_RUNWAY_HISTORY_MIN_POINTS = 30
_RAW_AMSC_RUNWAY_HISTORY_MIN_SPAN_SEC = 12 * 60 * 60


def parse_observation_epoch(value: Any) -> Optional[int]:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        dt = value
    else:
        text = str(value).strip()
        if not text:
            return None
        if text.endswith("Z"):
            text = text[:-1] + "+00:00"
        try:
            dt = datetime.fromisoformat(text)
        except ValueError:
            for fmt in ("%Y-%m-%d %H:%M:%S%z", "%Y-%m-%d %H:%M:%S"):
                try:
                    dt = datetime.strptime(text, fmt)
                    break
                except ValueError:
                    continue
            else:
                return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.timestamp())


def _parse_observation_datetime(value: Any) -> Optional[datetime]:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        dt = value
    else:
        text = str(value).strip()
        if not text:
            return None
        if text.endswith("Z"):
            text = text[:-1] + "+00:00"
        try:
            dt = datetime.fromisoformat(text)
        except ValueError:
            for fmt in ("%Y-%m-%d %H:%M:%S%z", "%Y-%m-%d %H:%M:%S"):
                try:
                    dt = datetime.strptime(text, fmt)
                    break
                except ValueError:
                    continue
            else:
                return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt


def _payload_latest_epoch(payload: dict[str, Any], keys: tuple[str, ...]) -> Optional[int]:
    values = [payload.get(key) for key in keys]
    parsed = [epoch for epoch in (parse_observation_epoch(value) for value in values) if epoch is not None]
    return max(parsed) if parsed else None


def _block_epoch(block: Any) -> Optional[int]:
    if not isinstance(block, dict):
        return None
    canonical_epoch = _payload_latest_epoch(
        block,
        (
            "observed_at",
            "observation_time",
            "obs_time",
        ),
    )
    if canonical_epoch is not None:
        return canonical_epoch
    return _payload_latest_epoch(
        block,
        (
            "observed_at_local",
            "observation_time_local",
        ),
    )


def _raw_amsc_epoch(row: dict[str, Any], raw_payload: dict[str, Any]) -> Optional[int]:
    payload_values = (
        raw_payload.get("observation_time"),
        raw_payload.get("observed_at"),
    )
    parsed = [
        epoch
        for epoch in (parse_observation_epoch(value) for value in payload_values)
        if epoch is not None
    ]
    if parsed:
        return max(parsed)
    return parse_observation_epoch(row.get("observed_at"))


def _to_float(value: Any) -> Optional[float]:
    try:
        if value is None or value == "":
            return None
        return float(value)
    except (TypeError, ValueError):
        return None


def _to_int(value: Any) -> Optional[int]:
    try:
        if value is None or value == "":
            return None
        return int(float(value))
    except (TypeError, ValueError):
        return None


def _latest_jma_row(weather: Any, city: str, use_fahrenheit: bool) -> Optional[dict[str, Any]]:
    fetcher = getattr(weather, "fetch_jma_amedas_official_nearby", None)
    if callable(fetcher):
        try:
            rows = fetcher(city, use_fahrenheit=use_fahrenheit)
        except Exception as exc:
            logger.debug("latest JMA overlay read failed city={}: {}", city, exc)
            rows = []
        for row in rows if isinstance(rows, list) else []:
            if not isinstance(row, dict):
                continue
            if _to_float(row.get("temp")) is not None and row.get("obs_time"):
                return row

    current_fetcher = getattr(weather, "fetch_jma_amedas_current", None)
    if callable(current_fetcher):
        try:
            current = current_fetcher(city, use_fahrenheit=use_fahrenheit)
        except Exception as exc:
            logger.debug("latest JMA current overlay read failed city={}: {}", city, exc)
            return None
        if isinstance(current, dict):
            temp = _to_float((current.get("current") or {}).get("temp"))
            obs_time = current.get("obs_time")
            if temp is not None and obs_time:
                return {
                    "station_label": current.get("station_name"),
                    "temp": temp,
                    "icao": current.get("station_code"),
                    "source": "jma",
                    "source_label": "JMA",
                    "obs_time": obs_time,
                }
    return None


def _jma_observation_update(
    city: str,
    row: dict[str, Any],
    obs_time: str,
    temp: float,
) -> dict[str, Any]:
    source_label = str(row.get("source_label") or "JMA").strip() or "JMA"
    station_code = str(row.get("icao") or row.get("istNo") or "").strip() or None
    station_name = str(row.get("station_label") or row.get("name") or source_label).strip()
    freshness = {
        "freshness_status": "fresh",
        "observed_at": obs_time,
        "source_code": "jma_amedas",
        "source_label": source_label,
    }
    return {
        "temp": round(float(temp), 1),
        "source_code": "jma_amedas",
        "source_label": source_label,
        "station_code": station_code,
        "station_name": station_name,
        "station_label": station_name,
        "observed_at": obs_time,
        "observation_time": obs_time,
        "obs_time": obs_time,
        "freshness": freshness,
        "observation_status": "live",
        "city": city,
    }


def _jma_today_point(local_time: str, obs_time: str, temp: float) -> dict[str, Any]:
    return {
        "time": local_time,
        "temp": round(float(temp), 1),
        "obs_time": obs_time,
        "source_code": "jma_amedas",
        "source_label": "JMA",
    }


def _replace_or_append_today_point(
    rows: Any,
    point: dict[str, Any],
    *,
    replace_all: bool,
) -> list[dict[str, Any]]:
    if replace_all:
        return [point]
    next_rows: list[dict[str, Any]] = []
    replaced = False
    for row in rows if isinstance(rows, list) else []:
        if not isinstance(row, dict):
            continue
        current_time = str(row.get("time") or "").strip()
        current_obs_time = str(row.get("obs_time") or row.get("observed_at") or "").strip()
        if current_time == point["time"] or current_obs_time == point["obs_time"]:
            next_rows.append(point)
            replaced = True
        else:
            next_rows.append(dict(row))
    if not replaced:
        next_rows.append(point)
    return next_rows


def _sync_jma_today_series(
    payload: dict[str, Any],
    point: dict[str, Any],
    *,
    replace_all: bool,
) -> None:
    rows = _replace_or_append_today_point(
        payload.get("metar_today_obs"),
        point,
        replace_all=replace_all,
    )
    payload["metar_today_obs"] = rows
    payload["airport_primary_today_obs"] = rows
    official = payload.get("official")
    if not isinstance(official, dict):
        official = {}
    official["airport_primary_today_obs"] = rows
    payload["official"] = official

    timeseries = payload.get("timeseries")
    if not isinstance(timeseries, dict):
        timeseries = {}
    timeseries["metar_today_obs"] = rows
    for key in ("metar_recent_obs", "settlement_today_obs"):
        if replace_all and key in timeseries:
            timeseries[key] = []
    payload["timeseries"] = timeseries
    if replace_all:
        for key in ("metar_recent_obs", "settlement_today_obs"):
            if key in payload:
                payload[key] = []


def overlay_latest_jma_amedas_observation(
    weather: Any,
    city: str,
    payload: dict[str, Any],
) -> dict[str, Any]:
    normalized_city = str(city or payload.get("name") or payload.get("city") or "").strip().lower()
    if not normalized_city or not isinstance(payload, dict) or not payload:
        return payload
    if normalized_city != "tokyo":
        return payload

    use_fahrenheit = "F" in str(payload.get("temp_symbol") or "").upper()
    row = _latest_jma_row(weather, normalized_city, use_fahrenheit)
    if not isinstance(row, dict):
        return payload

    temp = _to_float(row.get("temp"))
    obs_time = str(row.get("obs_time") or "").strip()
    if temp is None or not obs_time:
        return payload

    raw_epoch = parse_observation_epoch(obs_time)
    local_dt = _parse_observation_datetime(obs_time)
    if raw_epoch is None or local_dt is None:
        return payload
    existing_epochs = [
        epoch
        for epoch in (
            _block_epoch(payload.get("current")),
            _block_epoch(payload.get("airport_primary")),
            _block_epoch(payload.get("airport_current")),
            _block_epoch(payload.get("canonical_temperature")),
        )
        if epoch is not None
    ]
    if existing_epochs and max(existing_epochs) >= raw_epoch:
        return payload

    update = _jma_observation_update(normalized_city, row, obs_time, temp)
    next_payload = deepcopy(payload)
    changed = False
    for key in ("current", "airport_primary", "airport_current"):
        changed = _merge_observation_block(next_payload, key, update, raw_epoch) or changed

    canonical = next_payload.get("canonical_temperature")
    canonical_epoch = _block_epoch(canonical)
    if canonical_epoch is None or raw_epoch > canonical_epoch:
        canonical_payload = build_canonical_temperature(
            normalized_city,
            {
                "name": normalized_city,
                "temp_symbol": next_payload.get("temp_symbol") or "\u00b0C",
                "updated_at": obs_time,
                "current": update,
            },
            fetched_at=obs_time,
        )
        if canonical_payload:
            next_payload["canonical_temperature"] = canonical_payload
            changed = True

    local_date = local_dt.date().isoformat()
    local_time = local_dt.strftime("%H:%M")
    previous_local_date = str(next_payload.get("local_date") or "")
    if next_payload.get("local_date") != local_date:
        next_payload["local_date"] = local_date
        changed = True
    if next_payload.get("local_time") != local_time:
        next_payload["local_time"] = local_time
        changed = True

    overview = next_payload.get("overview")
    if not isinstance(overview, dict):
        overview = {}
    next_overview = dict(overview)
    overview_updates = {
        "local_date": local_date,
        "local_time": local_time,
        "current_temp": round(float(temp), 1),
        "airport_primary": update,
    }
    for key, value in overview_updates.items():
        if next_overview.get(key) != value:
            next_overview[key] = value
            changed = True
    next_payload["overview"] = next_overview

    replace_today_series = bool(previous_local_date and previous_local_date != local_date)
    _sync_jma_today_series(
        next_payload,
        _jma_today_point(local_time, obs_time, temp),
        replace_all=replace_today_series,
    )
    changed = True

    return next_payload if changed else payload


def _amsc_payload_has_observation(raw_payload: dict[str, Any]) -> bool:
    temp = raw_payload.get("temp_c") if raw_payload.get("temp_c") is not None else raw_payload.get("temp")
    return _to_float(temp) is not None


def _latest_amsc_row(db: Any, city: str) -> tuple[Optional[dict[str, Any]], Optional[dict[str, Any]]]:
    getter = getattr(db, "get_latest_raw_observation", None)
    if not callable(getter):
        return None, None
    try:
        row = getter("amsc_awos", city)
    except Exception as exc:
        logger.debug("latest AMSC raw overlay read failed city={}: {}", city, exc)
        return None, None
    if not isinstance(row, dict):
        return None, None
    raw_payload = row.get("payload")
    if isinstance(raw_payload, dict) and raw_payload and _amsc_payload_has_observation(raw_payload):
        return row, raw_payload

    lister = getattr(db, "list_latest_raw_observations_for_city", None)
    if callable(lister):
        try:
            candidates = lister(city, limit=20)
        except Exception as exc:
            logger.debug("latest AMSC raw fallback list failed city={}: {}", city, exc)
            candidates = []
        for candidate in candidates if isinstance(candidates, list) else []:
            if not isinstance(candidate, dict):
                continue
            if str(candidate.get("source") or "").strip().lower() != "amsc_awos":
                continue
            candidate_payload = candidate.get("payload")
            if isinstance(candidate_payload, dict) and _amsc_payload_has_observation(candidate_payload):
                return candidate, candidate_payload

    return row, None


def _raw_observation_update(
    city: str,
    row: dict[str, Any],
    raw_payload: dict[str, Any],
) -> Optional[dict[str, Any]]:
    temp = _to_float(raw_payload.get("temp_c") if raw_payload.get("temp_c") is not None else raw_payload.get("temp"))
    if temp is None:
        return None
    observed_at = str(
        raw_payload.get("observation_time")
        or raw_payload.get("observed_at")
        or row.get("observed_at")
        or ""
    ).strip()
    observed_at_local = str(
        raw_payload.get("observation_time_local")
        or raw_payload.get("observed_at_local")
        or ""
    ).strip()
    source_label = str(raw_payload.get("source_label") or "AMSC AWOS").strip()
    station_code = str(raw_payload.get("icao") or row.get("station_code") or "").strip().upper() or None
    station_name = str(raw_payload.get("station_label") or row.get("station_name") or source_label).strip()
    freshness = {
        "freshness_status": "fresh",
        "observed_at": observed_at or None,
        "observed_at_local": observed_at_local or None,
        "source_code": "amsc_awos",
        "source_label": source_label,
    }
    return {
        "temp": round(temp, 1),
        "source_code": "amsc_awos",
        "source_label": source_label,
        "settlement_source": "amsc_awos",
        "settlement_source_label": source_label,
        "station_code": station_code,
        "station_name": station_name,
        "observed_at": observed_at or None,
        "observed_at_local": observed_at_local or None,
        "obs_time": observed_at or observed_at_local,
        "freshness": freshness,
        "observation_status": "live",
        "city": city,
    }


def _merge_observation_block(
    payload: dict[str, Any],
    key: str,
    update: dict[str, Any],
    raw_epoch: int,
) -> bool:
    current = payload.get(key)
    if not isinstance(current, dict):
        current = {}
    current_epoch = _block_epoch(current)
    if current_epoch is not None and current_epoch >= raw_epoch:
        return False
    merged = dict(current)
    merged.update(update)
    payload[key] = merged
    return True


def _runway_history_temp(point: dict[str, Any]) -> Optional[float]:
    for key in ("target_runway_max", "temp", "tdz_temp", "end_temp", "mid_temp"):
        temp = _to_float(point.get(key))
        if temp is not None:
            return temp
    return None


def _runway_history_has_recent_coverage(
    history: Any,
    latest_epoch: int,
) -> bool:
    if not isinstance(history, dict):
        return False
    window_start = latest_epoch - _RAW_AMSC_RUNWAY_HISTORY_RECENT_WINDOW_SEC
    span_start = latest_epoch - _RAW_AMSC_RUNWAY_HISTORY_MIN_SPAN_SEC
    for points in history.values():
        if not isinstance(points, list):
            continue
        epochs = [
            epoch
            for epoch in (
                parse_observation_epoch(
                    point.get("time") or point.get("timestamp") or point.get("observed_at")
                )
                for point in points
                if isinstance(point, dict)
            )
            if epoch is not None and window_start <= epoch <= latest_epoch + 300
        ]
        if len(epochs) >= _RAW_AMSC_RUNWAY_HISTORY_MIN_POINTS and min(epochs) <= span_start:
            return True
    return False


def _append_latest_amsc_runway_history(
    db: Any,
    city: str,
    payload: dict[str, Any],
    row: dict[str, Any],
    raw_payload: dict[str, Any],
    *,
    persist: bool = True,
    copy_history: bool = True,
) -> bool:
    runway_obs = raw_payload.get("runway_obs")
    if not isinstance(runway_obs, dict):
        return False
    point_temperatures = runway_obs.get("point_temperatures")
    if not isinstance(point_temperatures, list) or not point_temperatures:
        return False
    observed_at = str(
        raw_payload.get("observation_time")
        or raw_payload.get("observed_at")
        or row.get("observed_at")
        or ""
    ).strip()
    if not observed_at:
        return False

    history = payload.get("runway_plate_history")
    if not isinstance(history, dict):
        history = {}
    elif copy_history:
        history = deepcopy(history)

    use_fahrenheit = "F" in str(payload.get("temp_symbol") or "").upper()
    appender = getattr(db, "append_runway_obs", None) if persist else None
    icao = str(raw_payload.get("icao") or row.get("station_code") or "").strip().upper()
    changed = False
    for point in point_temperatures:
        if not isinstance(point, dict):
            continue
        runway = str(point.get("runway") or "").strip().upper()
        temp = _runway_history_temp(point)
        if not runway or temp is None:
            continue
        if use_fahrenheit:
            temp = temp * 9.0 / 5.0 + 32.0
        entry = {
            "time": observed_at,
            "temp": round(float(temp), 1),
        }
        existing = history.get(runway)
        runway_history = list(existing) if isinstance(existing, list) else []
        replaced = False
        for index, current in enumerate(runway_history):
            if not isinstance(current, dict):
                continue
            current_time = str(
                current.get("time")
                or current.get("timestamp")
                or current.get("observed_at")
                or ""
            ).strip()
            if current_time == observed_at:
                if current != entry:
                    runway_history[index] = entry
                    changed = True
                replaced = True
                break
        if not replaced:
            runway_history.append(entry)
            changed = True
        history[runway] = runway_history
        if callable(appender) and icao:
            try:
                appender(
                    icao=icao,
                    city=city,
                    runway=runway,
                    tdz_temp=_to_float(point.get("tdz_temp")),
                    mid_temp=_to_float(point.get("mid_temp")),
                    end_temp=_to_float(point.get("end_temp")),
                    target_runway_max=_to_float(point.get("target_runway_max")),
                    wind_dir=_to_int(point.get("wind_dir")),
                    wind_speed=_to_float(point.get("wind_speed")),
                    rvr=_to_int(point.get("rvr")),
                    mor=_to_float(point.get("mor")),
                    humidity=_to_float(point.get("humidity")),
                    otime_utc=observed_at,
                )
            except Exception as exc:
                logger.debug(
                    "latest AMSC runway history persist skipped city={} runway={}: {}",
                    city,
                    runway,
                    exc,
                )

    if changed:
        payload["runway_plate_history"] = history
    return changed


def _append_amsc_runway_history_from_raw_store(
    db: Any,
    city: str,
    payload: dict[str, Any],
    latest_epoch: int,
) -> bool:
    existing_history = payload.get("runway_plate_history")
    if _runway_history_has_recent_coverage(existing_history, latest_epoch):
        return False

    use_fahrenheit = "F" in str(payload.get("temp_symbol") or "").upper()
    cache_key = (city, use_fahrenheit)
    cached = _RAW_AMSC_RUNWAY_HISTORY_CACHE.get(cache_key)
    now = time.monotonic()
    if cached and now - cached[0] <= _RAW_AMSC_RUNWAY_HISTORY_CACHE_TTL_SEC:
        payload["runway_plate_history"] = deepcopy(cached[1])
        return True

    lister = getattr(db, "list_raw_observation_history", None)
    if not callable(lister):
        return False
    try:
        rows = lister("amsc_awos", city, minutes=24 * 60, limit=1500)
    except Exception as exc:
        logger.debug("latest AMSC raw history overlay skipped city={}: {}", city, exc)
        return False
    payload["runway_plate_history"] = {}
    changed = False
    for row in rows if isinstance(rows, list) else []:
        if not isinstance(row, dict):
            continue
        raw_payload = row.get("payload")
        if not isinstance(raw_payload, dict) or not _amsc_payload_has_observation(raw_payload):
            continue
        changed = _append_latest_amsc_runway_history(
            db,
            city,
            payload,
            row,
            raw_payload,
            persist=False,
            copy_history=False,
        ) or changed
    if changed and isinstance(payload.get("runway_plate_history"), dict):
        _RAW_AMSC_RUNWAY_HISTORY_CACHE[cache_key] = (
            now,
            deepcopy(payload["runway_plate_history"]),
        )
    return changed


def overlay_latest_amsc_observation(
    db: Any,
    city: str,
    payload: dict[str, Any],
) -> dict[str, Any]:
    normalized_city = str(city or payload.get("name") or payload.get("city") or "").strip().lower()
    if not normalized_city or not isinstance(payload, dict) or not payload:
        return payload
    row, raw_payload = _latest_amsc_row(db, normalized_city)
    if not isinstance(row, dict) or not isinstance(raw_payload, dict):
        return payload
    raw_epoch = _raw_amsc_epoch(row, raw_payload)
    if raw_epoch is None:
        return payload
    update = _raw_observation_update(normalized_city, row, raw_payload)
    if not update:
        return payload

    next_payload = deepcopy(payload)
    changed = False
    amos = next_payload.get("amos")
    amos_epoch = _block_epoch(amos)
    if amos_epoch is None or raw_epoch > amos_epoch:
        next_payload["amos"] = dict(raw_payload)
        changed = True

    for key in ("current", "airport_primary", "airport_current"):
        changed = _merge_observation_block(next_payload, key, update, raw_epoch) or changed

    changed = _append_amsc_runway_history_from_raw_store(
        db,
        normalized_city,
        next_payload,
        raw_epoch,
    ) or changed
    changed = _append_latest_amsc_runway_history(db, normalized_city, next_payload, row, raw_payload) or changed

    canonical = next_payload.get("canonical_temperature")
    canonical_epoch = _block_epoch(canonical)
    if canonical_epoch is None or raw_epoch > canonical_epoch:
        canonical_payload = build_canonical_temperature(
            normalized_city,
            {
                "name": normalized_city,
                "temp_symbol": next_payload.get("temp_symbol") or "\u00b0C",
                "updated_at": row.get("fetched_at"),
                "current": update,
            },
            fetched_at=str(row.get("fetched_at") or ""),
        )
        if canonical_payload:
            next_payload["canonical_temperature"] = canonical_payload
            changed = True

    return next_payload if changed else payload


# ═══════════════════════════════════════════════════════════════════════════════
# AMOS (Korean runway sensors — Seoul, Busan)
# ═══════════════════════════════════════════════════════════════════════════════


def _latest_amos_row(db: Any, city: str) -> tuple:
    getter = getattr(db, "get_latest_raw_observation", None)
    if not callable(getter):
        return None, None
    try:
        row = getter("amos", city)
    except Exception as exc:
        logger.debug("latest AMOS raw overlay read failed city={}: {}", city, exc)
        return None, None
    if not isinstance(row, dict):
        return None, None
    raw_payload = row.get("payload")
    if isinstance(raw_payload, dict) and raw_payload and raw_payload.get("temp") is not None:
        return row, raw_payload
    return None, None


def _raw_amos_epoch(row, raw_payload):
    payload_values = (
        raw_payload.get("observation_time"),
        raw_payload.get("observed_at"),
    )
    parsed = [
        epoch
        for epoch in (parse_observation_epoch(value) for value in payload_values)
        if epoch is not None
    ]
    if parsed:
        return max(parsed)
    return parse_observation_epoch(row.get("observed_at"))


def overlay_latest_amos_observation(db, city, payload):
    normalized_city = str(city or payload.get("name") or payload.get("city") or "").strip().lower()
    if not normalized_city or not isinstance(payload, dict) or not payload:
        return payload
    row, raw_payload = _latest_amos_row(db, normalized_city)
    if not isinstance(row, dict) or not isinstance(raw_payload, dict):
        return payload
    raw_epoch = _raw_amos_epoch(row, raw_payload)
    if raw_epoch is None:
        return payload
    update = _raw_observation_update(normalized_city, row, raw_payload)
    if not update:
        return payload

    next_payload = deepcopy(payload)
    changed = False
    amos = next_payload.get("amos")
    amos_epoch = _block_epoch(amos)
    if amos_epoch is None or raw_epoch > amos_epoch:
        next_payload["amos"] = dict(raw_payload)
        changed = True

    for key in ("current", "airport_primary", "airport_current"):
        changed = _merge_observation_block(next_payload, key, update, raw_epoch) or changed

    canonical = next_payload.get("canonical_temperature")
    canonical_epoch = _block_epoch(canonical)
    if canonical_epoch is None or raw_epoch > canonical_epoch:
        canonical_payload = build_canonical_temperature(
            normalized_city,
            {
                "name": normalized_city,
                "temp_symbol": next_payload.get("temp_symbol") or "\u00b0C",
                "updated_at": row.get("fetched_at"),
                "current": update,
            },
            fetched_at=str(row.get("fetched_at") or ""),
        )
        if canonical_payload:
            next_payload["canonical_temperature"] = canonical_payload
            changed = True

    return next_payload if changed else payload


# ═══════════════════════════════════════════════════════════════════════════════
# HKO (Hong Kong Observatory — Hong Kong, Shenzhen)
# ═══════════════════════════════════════════════════════════════════════════════


def _latest_hko_row(db, city):
    getter = getattr(db, "get_latest_raw_observation", None)
    if not callable(getter):
        return None, None
    try:
        row = getter("hko_obs", city)
    except Exception as exc:
        logger.debug("latest HKO raw overlay read failed city={}: {}", city, exc)
        return None, None
    if not isinstance(row, dict):
        return None, None
    raw_payload = row.get("payload")
    if isinstance(raw_payload, dict) and raw_payload:
        temp = raw_payload.get("temp")
        if temp is not None:
            return row, raw_payload
    return None, None


def _raw_hko_epoch(row, raw_payload):
    payload_values = (raw_payload.get("obs_time"),)
    parsed = [
        epoch
        for epoch in (parse_observation_epoch(value) for value in payload_values)
        if epoch is not None
    ]
    if parsed:
        return max(parsed)
    return parse_observation_epoch(row.get("observed_at"))


def overlay_latest_hko_observation(db, city, payload):
    normalized_city = str(city or payload.get("name") or payload.get("city") or "").strip().lower()
    if not normalized_city or not isinstance(payload, dict) or not payload:
        return payload
    row, raw_payload = _latest_hko_row(db, normalized_city)
    if not isinstance(row, dict) or not isinstance(raw_payload, dict):
        return payload
    raw_epoch = _raw_hko_epoch(row, raw_payload)
    if raw_epoch is None:
        return payload
    update = _raw_observation_update(normalized_city, row, raw_payload)
    if not update:
        return payload

    next_payload = deepcopy(payload)
    changed = False

    for key in ("current", "airport_primary", "airport_current"):
        changed = _merge_observation_block(next_payload, key, update, raw_epoch) or changed

    canonical = next_payload.get("canonical_temperature")
    canonical_epoch = _block_epoch(canonical)
    if canonical_epoch is None or raw_epoch > canonical_epoch:
        canonical_payload = build_canonical_temperature(
            normalized_city,
            {
                "name": normalized_city,
                "temp_symbol": next_payload.get("temp_symbol") or "\u00b0C",
                "updated_at": row.get("fetched_at"),
                "current": update,
            },
            fetched_at=str(row.get("fetched_at") or ""),
        )
        if canonical_payload:
            next_payload["canonical_temperature"] = canonical_payload
            changed = True

    return next_payload if changed else payload
# ═══════════════════════════════════════════════════════════════════════════════


def _latest_mgm_row(db, city):
    getter = getattr(db, "get_latest_raw_observation", None)
    if not callable(getter):
        return None, None
    try:
        row = getter("mgm", city)
    except Exception as exc:
        logger.debug("latest MGM raw overlay read failed city={}: {}", city, exc)
        return None, None
    if not isinstance(row, dict):
        return None, None
    raw_payload = row.get("payload")
    if isinstance(raw_payload, dict) and raw_payload:
        temp = raw_payload.get("temp")
        if temp is not None:
            return row, raw_payload
    return None, None


def _raw_mgm_epoch(row, raw_payload):
    payload_values = (
        raw_payload.get("obs_time"),
    )
    parsed = [
        epoch
        for epoch in (parse_observation_epoch(value) for value in payload_values)
        if epoch is not None
    ]
    if parsed:
        return max(parsed)
    return parse_observation_epoch(row.get("observed_at"))


def overlay_latest_mgm_observation(db, city, payload):
    normalized_city = str(city or payload.get("name") or payload.get("city") or "").strip().lower()
    if not normalized_city or not isinstance(payload, dict) or not payload:
        return payload
    row, raw_payload = _latest_mgm_row(db, normalized_city)
    if not isinstance(row, dict) or not isinstance(raw_payload, dict):
        return payload
    raw_epoch = _raw_mgm_epoch(row, raw_payload)
    if raw_epoch is None:
        return payload
    update = _raw_observation_update(normalized_city, row, raw_payload)
    if not update:
        return payload

    next_payload = deepcopy(payload)
    changed = False

    for key in ("current", "airport_primary", "airport_current"):
        changed = _merge_observation_block(next_payload, key, update, raw_epoch) or changed

    canonical = next_payload.get("canonical_temperature")
    canonical_epoch = _block_epoch(canonical)
    if canonical_epoch is None or raw_epoch > canonical_epoch:
        canonical_payload = build_canonical_temperature(
            normalized_city,
            {
                "name": normalized_city,
                "temp_symbol": next_payload.get("temp_symbol") or "\u00b0C",
                "updated_at": row.get("fetched_at"),
                "current": update,
            },
            fetched_at=str(row.get("fetched_at") or ""),
        )
        if canonical_payload:
            next_payload["canonical_temperature"] = canonical_payload
            changed = True

    return next_payload if changed else payload