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, }