HexaWetter/api/app/services/warnings.py

from __future__ import annotations

import json
import re
from datetime import datetime, timezone
from typing import Any

from app.cache import cache_get, cache_set
from app.config import CACHE_TTL_WARNINGS, DWD_WARNINGS_URL
from app.database import set_cached_json
from app.http_client import get_client
from app.services.warncells import resolve_warncells

WARN_LEVELS = {
    10: ("Vorabinformation", "info"),
    20: ("Wetterwarnung", "minor"),
    30: ("Markantes Wetter", "moderate"),
    40: ("Unwetterwarnung", "severe"),
    50: ("Extremes Unwetter", "extreme"),
    60: ("Extremes Unwetter", "extreme"),
    70: ("Extremes Unwetter", "extreme"),
    80: ("Extremes Unwetter", "extreme"),
    90: ("Extremes Unwetter", "extreme"),
}


def _flatten_warnings(payload: dict[str, Any]) -> list[dict[str, Any]]:
    warnings_root = payload.get("warnings", {})
    flattened: list[dict[str, Any]] = []
    for cell_id, items in warnings_root.items():
        if not isinstance(items, list):
            continue
        for item in items:
            level = int(item.get("level", 0))
            level_label, severity = WARN_LEVELS.get(level, (f"Level {level}", "unknown"))
            flattened.append(
                {
                    "cell_id": str(cell_id),
                    "state": item.get("state"),
                    "region_name": item.get("regionName"),
                    "event": item.get("event"),
                    "headline": item.get("headline"),
                    "description": (item.get("description") or "").strip(),
                    "instruction": (item.get("instruction") or "").strip(),
                    "level": level,
                    "level_label": level_label,
                    "severity": severity,
                    "type": item.get("type"),
                    "start": datetime.fromtimestamp(item["start"] / 1000, tz=timezone.utc).isoformat()
                    if item.get("start")
                    else None,
                    "end": datetime.fromtimestamp(item["end"] / 1000, tz=timezone.utc).isoformat()
                    if item.get("end")
                    else None,
                }
            )
    flattened.sort(key=lambda w: (-w["level"], w.get("start") or ""))
    return flattened


def _normalize_kreis_name(name: str) -> str:
    value = name.lower().strip()
    value = re.sub(r"^(landkreis|kreis und stadt|kreis|stadt)\s+", "", value)
    value = re.sub(r"[^a-zäöüß0-9]+", " ", value)
    return " ".join(value.split())


def _match_by_area_names(warning: dict[str, Any], area_names: list[str]) -> bool:
    region = _normalize_kreis_name(warning.get("region_name") or "")
    if not region:
        return False
    for area_name in area_names:
        normalized = _normalize_kreis_name(area_name or "")
        if not normalized or len(normalized) < 3:
            continue
        if region == normalized or normalized in region or region in normalized:
            return True
    return False


async def fetch_all_warnings() -> list[dict[str, Any]]:
    cache_key = "warnings:all"
    cached = await cache_get(cache_key)
    if cached:
        return cached

    client = get_client()
    response = await client.get(DWD_WARNINGS_URL, timeout=10.0)
    response.raise_for_status()
    text = response.text
    if text.startswith("warnWetter.loadWarnings("):
        text = text[text.index("(") + 1 : text.rindex(")")]
    payload = json.loads(text)

    warnings = _flatten_warnings(payload)
    await cache_set(cache_key, warnings, CACHE_TTL_WARNINGS)
    await set_cached_json("warnings_cache", {}, payload)
    return warnings


def _warning_identity_key(warning: dict[str, Any]) -> tuple[Any, ...]:
    return (
        warning.get("event"),
        warning.get("headline"),
        warning.get("level"),
        warning.get("start"),
        warning.get("end"),
        warning.get("description"),
        warning.get("instruction"),
    )


def _dedupe_warnings(warnings: list[dict[str, Any]]) -> list[dict[str, Any]]:
    merged: dict[tuple[Any, ...], dict[str, Any]] = {}
    for warning in warnings:
        key = _warning_identity_key(warning)
        region = warning.get("region_name")
        if key not in merged:
            merged[key] = {
                **warning,
                "_cell_ids": [warning["cell_id"]],
                "_regions": [region] if region else [],
            }
            continue
        entry = merged[key]
        if warning["cell_id"] not in entry["_cell_ids"]:
            entry["_cell_ids"].append(warning["cell_id"])
        if region and region not in entry["_regions"]:
            entry["_regions"].append(region)

    result: list[dict[str, Any]] = []
    for entry in merged.values():
        regions = entry.pop("_regions", [])
        cell_ids = entry.pop("_cell_ids", [entry.get("cell_id")])
        if len(regions) > 1:
            entry["region_name"] = " · ".join(regions)
        elif regions:
            entry["region_name"] = regions[0]
        entry["cell_id"] = cell_ids[0] if len(cell_ids) == 1 else ",".join(cell_ids)
        result.append(entry)
    result.sort(key=lambda w: (-w["level"], w.get("start") or ""))
    return result


async def fetch_warnings_for_location(lat: float, lon: float, local_only: bool = True) -> dict[str, Any]:
    warnings = await fetch_all_warnings()
    warncells = await resolve_warncells(lat, lon)
    cell_ids = set(warncells.get("cell_ids", []))
    area_names = [area.get("name") for area in warncells.get("areas", []) if area.get("name")]

    local = [
        warning
        for warning in warnings
        if warning["cell_id"] in cell_ids or _match_by_area_names(warning, area_names)
    ]

    if local_only:
        result_warnings = _dedupe_warnings(local)
    else:
        result_warnings = _dedupe_warnings(warnings[:100])

    return {
        "lat": lat,
        "lon": lon,
        "count_total": len(warnings),
        "count_local": len(result_warnings),
        "count_matched_raw": len(local),
        "warnings": result_warnings,
        "warncells": warncells.get("areas", []),
        "matched_cell_ids": sorted(cell_ids),
        "local_only": local_only,
        "all_warnings_truncated": not local_only and len(warnings) > 100,
    }