Matthias JacobandCopilot
b9cee0a7fb
Scrape the official alarm level thresholds (Richtwasserstände) from the
station detail page and include them in GET /api/pegel/{pegel_id} as
alarm_thresholds, cached for 24 hours with a stale-cache fallback on
errors. null for stations without a flood alert function.
Also clarify in the flood_alert_level docs that 'Kein Hochwassermeldepegel'
means the station is not used for official flood alerts/control, not
that measurement data is missing.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
432 lines
15 KiB
Python
432 lines
15 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
from pydantic import BaseModel, Field
|
|
import feedparser
|
|
import re
|
|
import html
|
|
import datetime as dt
|
|
from urllib.request import urlopen, Request
|
|
from urllib.error import URLError, HTTPError
|
|
from email.utils import parsedate_to_datetime
|
|
from typing import TypedDict, Any
|
|
|
|
app = FastAPI(
|
|
title="Sachsen Pegel Proxy API",
|
|
description=(
|
|
"Liefert Stationslisten und Wasserstandsmessungen des sächsischen "
|
|
"Hochwasser- und Wasserstands-Informationssystems (HWIMS) als JSON.\n\n"
|
|
"Die API liest die offiziellen HWIMS-Übersichts- und RSS-Quellen aus. "
|
|
"Messwerte und Stationsnamen können `null` sein, wenn sie in der "
|
|
"Quelldatei fehlen oder nicht erkannt werden. Zeitstempel werden als "
|
|
"ISO-8601-Werte mit Zeitzoneninformation ausgegeben.\n\n"
|
|
"Die interaktive Dokumentation steht unter `/docs`, das OpenAPI-Schema "
|
|
"unter `/openapi.json` bereit."
|
|
),
|
|
version="1.0.0",
|
|
openapi_tags=[
|
|
{
|
|
"name": "Pegel",
|
|
"description": "Entdecken von Pegelstationen und Abrufen ihrer Messwerte.",
|
|
}
|
|
],
|
|
)
|
|
|
|
@app.get("/", include_in_schema=False)
|
|
def redirect_to_docs():
|
|
return RedirectResponse(url="/docs", status_code=307)
|
|
|
|
class PegelStation(BaseModel):
|
|
station_id: str
|
|
station_title: str | None
|
|
source: str
|
|
|
|
class PegelIndexResponse(BaseModel):
|
|
count: int
|
|
total_available: int
|
|
items: list[PegelStation]
|
|
source: str
|
|
fetched_at: str
|
|
warnings: list[str] | None = None
|
|
cache_fetched_at: str | None = None
|
|
|
|
class PegelEntry(BaseModel):
|
|
timestamp: str | None
|
|
title: str | None
|
|
water_level_cm: float | None
|
|
flow_m3s: float | None
|
|
flood_alert_level: str | None = Field(
|
|
description=(
|
|
"Meldestufe als Text. Beobachtete Werte: `Niedrigwasser`, `Kein Hochwasser`, "
|
|
"`Kein Hochwassermeldepegel` (der Pegel wird nicht für amtliche "
|
|
"Hochwassermeldungen bzw. -steuerung genutzt; für solche Pegel liefert "
|
|
"`alarm_thresholds` `null`). Bei Hochwasser sind laut HWIMS-Legende "
|
|
"`Alarmstufe 1` bis `Alarmstufe 4` zu erwarten; deren genaue Schreibweise "
|
|
"im Feed ist nicht verifiziert. Clients sollten unbekannte Werte tolerieren."
|
|
),
|
|
examples=[
|
|
"Niedrigwasser",
|
|
"Kein Hochwasser",
|
|
"Kein Hochwassermeldepegel",
|
|
"Alarmstufe 1",
|
|
"Alarmstufe 2",
|
|
"Alarmstufe 3",
|
|
"Alarmstufe 4",
|
|
],
|
|
)
|
|
|
|
class AlarmThreshold(BaseModel):
|
|
level: int = Field(description="Alarmstufe, 1 bis 4.")
|
|
water_level_cm: int = Field(description="Richtwasserstand dieser Alarmstufe in Zentimetern.")
|
|
|
|
class PegelResponse(BaseModel):
|
|
station_id: str
|
|
station_title: str | None
|
|
count: int
|
|
latest: PegelEntry
|
|
entries: list[PegelEntry]
|
|
alarm_thresholds: list[AlarmThreshold] | None = Field(
|
|
description=(
|
|
"Richtwasserstände der Alarmstufen 1 bis 4 in Zentimetern, von der "
|
|
"Pegel-Detailseite ermittelt. `null`, wenn der Pegel keine "
|
|
"Hochwassermeldefunktion hat (siehe `flood_alert_level`) oder die "
|
|
"Werte nicht ermittelt werden konnten."
|
|
)
|
|
)
|
|
source: str
|
|
fetched_at: str
|
|
|
|
BASE = "https://www.umwelt.sachsen.de/umwelt/infosysteme/hwims/portal/web/feed/wasserstand-pegel-{}"
|
|
DETAIL_BASE = "https://www.umwelt.sachsen.de/umwelt/infosysteme/hwims/portal/web/wasserstand-pegel-{}"
|
|
DISCOVERY_URL = "https://www.umwelt.sachsen.de/umwelt/infosysteme/hwims/portal/web/wasserstand-uebersicht"
|
|
INDEX_CACHE_TTL_SECONDS = 900
|
|
LATEST_CACHE_TTL_SECONDS = 120
|
|
ALARM_THRESHOLDS_CACHE_TTL_SECONDS = 86400
|
|
|
|
ALARM_THRESHOLD_PATTERN = re.compile(
|
|
r'Alarmstufe\s*(\d)</span>.*?<td headers="rws">\s*([\d.,]+)\s*cm\s*</td>',
|
|
re.S,
|
|
)
|
|
|
|
class IndexCache(TypedDict):
|
|
items: list[dict[str, Any]] | None
|
|
fetched_at: dt.datetime | None
|
|
|
|
|
|
_index_cache: IndexCache = {
|
|
"items": None,
|
|
"fetched_at": None,
|
|
}
|
|
|
|
_latest_cache: dict[str, dict[str, Any]] = {}
|
|
_alarm_thresholds_cache: dict[str, dict[str, Any]] = {}
|
|
|
|
def parse_number(pattern, text):
|
|
m = re.search(pattern, text, re.I)
|
|
return float(m.group(1).replace(",", ".")) if m else None
|
|
|
|
def parse_text(pattern, text):
|
|
m = re.search(pattern, text, re.I)
|
|
return m.group(1).strip() if m else None
|
|
|
|
def clean_text(text):
|
|
text = re.sub(r"<br\s*/?>", "\n", text, flags=re.I)
|
|
text = re.sub(r"<[^>]+>", " ", text)
|
|
return text
|
|
|
|
def extract_station_name(title: str | None):
|
|
if not title:
|
|
return None
|
|
|
|
m = re.search(r"Pegel\s+(.+)$", title)
|
|
return m.group(1).strip() if m else title
|
|
|
|
def _now_utc():
|
|
return dt.datetime.now(dt.UTC)
|
|
|
|
def _is_cache_fresh(fetched_at, ttl_seconds: int):
|
|
if not fetched_at:
|
|
return False
|
|
return (_now_utc() - fetched_at).total_seconds() < ttl_seconds
|
|
|
|
def _set_index_cache(items):
|
|
_index_cache["items"] = items
|
|
_index_cache["fetched_at"] = _now_utc()
|
|
|
|
def _extract_station_title_from_id(pegel_id: str):
|
|
feed = feedparser.parse(BASE.format(pegel_id))
|
|
feed_obj = getattr(feed, "feed", None)
|
|
feed_title = feed_obj.get("title") if isinstance(feed_obj, dict) else None
|
|
if isinstance(feed_title, str) and feed_title:
|
|
return extract_station_name(feed_title)
|
|
if feed.entries:
|
|
first_title = feed.entries[0].get("title")
|
|
if isinstance(first_title, str):
|
|
return extract_station_name(first_title)
|
|
return None
|
|
|
|
def _parse_stations_from_overview(body: str):
|
|
stations: dict[str, str | None] = {}
|
|
|
|
paired_pattern = re.compile(
|
|
r'<a\s+href="wasserstand-pegel-(\d{6,})"[^>]*></a>\s*'
|
|
r'<div\s+class="popUp\s+popUpMs"[^>]*>.*?'
|
|
r'<div\s+class="popUpTitle">\s*<span\s+class="popUpTitleBold">(.*?)</span>',
|
|
re.I | re.S,
|
|
)
|
|
|
|
for pegel_id, raw_title in paired_pattern.findall(body):
|
|
title = re.sub(r"\s+", " ", html.unescape(raw_title)).strip() or None
|
|
stations[pegel_id] = title
|
|
|
|
for pegel_id in re.findall(r"wasserstand-pegel-(\d{6,})", body):
|
|
stations.setdefault(pegel_id, None)
|
|
|
|
return stations
|
|
|
|
def discover_stations():
|
|
req = Request(
|
|
DISCOVERY_URL,
|
|
headers={"User-Agent": "sachsen-pegel-proxy/1.0"},
|
|
)
|
|
with urlopen(req, timeout=12) as resp:
|
|
body = resp.read().decode("utf-8", errors="ignore")
|
|
|
|
stations = _parse_stations_from_overview(body)
|
|
ids = sorted(stations.keys())
|
|
if not ids:
|
|
raise ValueError("Keine Pegel-IDs in Discovery-Quelle gefunden")
|
|
|
|
items = []
|
|
for pegel_id in ids:
|
|
station_title = stations.get(pegel_id)
|
|
items.append({
|
|
"station_id": pegel_id,
|
|
"station_title": station_title or _extract_station_title_from_id(pegel_id),
|
|
"source": BASE.format(pegel_id),
|
|
})
|
|
return items
|
|
|
|
def parse_timestamp_from_title(title: str):
|
|
if not title:
|
|
return None
|
|
|
|
m = re.search(r"(\d{2}\.\d{2}\.\d{4})\s+(\d{2}:\d{2})\s+Uhr\s+\((MESZ|MEZ)", title)
|
|
if not m:
|
|
return None
|
|
|
|
date_part, time_part, tz_hint = m.groups()
|
|
parsed = dt.datetime.strptime(f"{date_part} {time_part}", "%d.%m.%Y %H:%M")
|
|
offset = dt.timedelta(hours=2 if tz_hint == "MESZ" else 1)
|
|
return parsed.replace(tzinfo=dt.timezone(offset)).isoformat()
|
|
|
|
def parse_timestamp(entry):
|
|
ts_from_title = parse_timestamp_from_title(entry.get("title"))
|
|
if ts_from_title:
|
|
return ts_from_title
|
|
|
|
if entry.get("dc_date"):
|
|
return entry.get("dc_date")
|
|
|
|
if entry.get("published"):
|
|
return parsedate_to_datetime(entry.get("published")).isoformat()
|
|
|
|
return None
|
|
|
|
def parse_entry(entry):
|
|
text = clean_text(entry.get("description", ""))
|
|
|
|
return {
|
|
"timestamp": parse_timestamp(entry),
|
|
"title": entry.get("title"),
|
|
"water_level_cm": parse_number(
|
|
r"Wasserstand:\s*([0-9]+(?:[,.][0-9]+)?)\s*cm", text
|
|
),
|
|
"flow_m3s": parse_number(
|
|
r"Durchfluss:\s*([0-9]+(?:[,.][0-9]+)?)\s*m³/s", text
|
|
),
|
|
"flood_alert_level": parse_text(
|
|
r"Meldestufe:\s*([^\n\r]+)", text
|
|
),
|
|
}
|
|
|
|
def parse_valid_entries(feed_entries):
|
|
# Die Quelle liefert für neueste Zeitpunkte teils Platzhalter ohne Messwerte.
|
|
entries = [parse_entry(entry) for entry in feed_entries]
|
|
return [
|
|
e for e in entries
|
|
if e["water_level_cm"] is not None or e["flow_m3s"] is not None
|
|
]
|
|
|
|
def _get_latest_for_station(station_id: str):
|
|
cache_hit = _latest_cache.get(station_id)
|
|
if cache_hit and _is_cache_fresh(cache_hit.get("fetched_at"), LATEST_CACHE_TTL_SECONDS):
|
|
return cache_hit.get("latest")
|
|
|
|
feed = feedparser.parse(BASE.format(station_id))
|
|
entries = parse_valid_entries(feed.entries)
|
|
if not entries:
|
|
raise ValueError("Keine Daten gefunden")
|
|
|
|
latest = entries[0]
|
|
_latest_cache[station_id] = {
|
|
"latest": latest,
|
|
"fetched_at": _now_utc(),
|
|
}
|
|
return latest
|
|
|
|
def parse_alarm_thresholds(body: str):
|
|
# Pegel ohne Hochwassermeldefunktion ("Kein Hochwassermeldepegel") haben
|
|
# keine Richtwasserstände-Tabelle; in diesem Fall liefert das Muster
|
|
# keine Treffer und die Funktion gibt None zurück.
|
|
matches = ALARM_THRESHOLD_PATTERN.findall(body)
|
|
if not matches:
|
|
return None
|
|
return [
|
|
{"level": int(level), "water_level_cm": round(float(value.replace(",", ".")))}
|
|
for level, value in matches
|
|
]
|
|
|
|
def _fetch_alarm_thresholds(station_id: str):
|
|
req = Request(
|
|
DETAIL_BASE.format(station_id),
|
|
headers={"User-Agent": "sachsen-pegel-proxy/1.0"},
|
|
)
|
|
with urlopen(req, timeout=12) as resp:
|
|
body = resp.read().decode("utf-8", errors="ignore")
|
|
return parse_alarm_thresholds(body)
|
|
|
|
def _get_alarm_thresholds(station_id: str):
|
|
cache_hit = _alarm_thresholds_cache.get(station_id)
|
|
if cache_hit and _is_cache_fresh(
|
|
cache_hit.get("fetched_at"), ALARM_THRESHOLDS_CACHE_TTL_SECONDS
|
|
):
|
|
return cache_hit.get("thresholds")
|
|
|
|
try:
|
|
thresholds = _fetch_alarm_thresholds(station_id)
|
|
except (URLError, HTTPError, TimeoutError, OSError):
|
|
# Richtwasserstände sind ein optionales Zusatzfeld; bei Fehlern lieber
|
|
# auf einen vorhandenen (ggf. abgelaufenen) Cache zurückfallen oder
|
|
# `None` liefern, statt die ganze Anfrage scheitern zu lassen.
|
|
return cache_hit.get("thresholds") if cache_hit else None
|
|
|
|
_alarm_thresholds_cache[station_id] = {
|
|
"thresholds": thresholds,
|
|
"fetched_at": _now_utc(),
|
|
}
|
|
return thresholds
|
|
|
|
@app.get(
|
|
"/api/pegel",
|
|
response_model=PegelIndexResponse,
|
|
tags=["Pegel"],
|
|
summary="Pegelstationen auflisten",
|
|
description=(
|
|
"Liefert die auf der HWIMS-Übersichtsseite gefundenen Pegelstationen. "
|
|
"Mit `limit` lässt sich die zurückgegebene Liste begrenzen; "
|
|
"`total_available` enthält weiterhin die Gesamtzahl. Der Index wird "
|
|
"15 Minuten zwischengespeichert. Ist die Quelle nicht erreichbar, "
|
|
"wird ein vorhandener Cache verwendet und die Antwort enthält `warnings` "
|
|
"sowie `cache_fetched_at`. Ohne verfügbaren Cache wird HTTP 502 geliefert."
|
|
),
|
|
response_description="Stationsliste mit Quelle und Abrufzeitpunkt.",
|
|
responses={
|
|
400: {"description": "`limit` ist keine positive Zahl."},
|
|
502: {"description": "Die Discovery-Quelle ist nicht verfügbar und es gibt keinen Cache."},
|
|
},
|
|
)
|
|
def list_pegel(limit: int | None = None):
|
|
if limit is not None and limit <= 0:
|
|
raise HTTPException(400, "Ungültiger limit-Parameter")
|
|
|
|
warnings = []
|
|
discovered_items: list[dict[str, Any]] | None = None
|
|
source_used = DISCOVERY_URL
|
|
cache_fallback = False
|
|
|
|
if _is_cache_fresh(_index_cache.get("fetched_at"), INDEX_CACHE_TTL_SECONDS):
|
|
discovered_items = _index_cache.get("items")
|
|
else:
|
|
try:
|
|
discovered_items = discover_stations()
|
|
_set_index_cache(discovered_items)
|
|
except (URLError, HTTPError, TimeoutError, ValueError, OSError) as e:
|
|
if _index_cache.get("items"):
|
|
discovered_items = _index_cache.get("items")
|
|
source_used = "cache"
|
|
cache_fallback = True
|
|
warnings.append(f"Discovery fehlgeschlagen, nutze Cache: {str(e)}")
|
|
else:
|
|
raise HTTPException(502, "Upstream-Discovery nicht verfügbar")
|
|
|
|
if discovered_items is None:
|
|
raise HTTPException(502, "Upstream-Discovery nicht verfügbar")
|
|
|
|
total_available = len(discovered_items)
|
|
items = discovered_items[:limit] if limit is not None else list(discovered_items)
|
|
|
|
response = {
|
|
"count": len(items),
|
|
"total_available": total_available,
|
|
"items": items,
|
|
"source": source_used,
|
|
"fetched_at": _now_utc().isoformat(),
|
|
}
|
|
if warnings:
|
|
response["warnings"] = warnings
|
|
cache_fetched_at = _index_cache.get("fetched_at")
|
|
if cache_fallback and cache_fetched_at:
|
|
response["cache_fetched_at"] = cache_fetched_at.isoformat()
|
|
|
|
return JSONResponse(response)
|
|
|
|
@app.get(
|
|
"/api/pegel/{pegel_id}",
|
|
response_model=PegelResponse,
|
|
tags=["Pegel"],
|
|
summary="Messwerte eines Pegels abrufen",
|
|
description=(
|
|
"Liefert alle im RSS-Feed verfügbaren Einträge einer Station. "
|
|
"Einträge ohne Wasserstand und ohne Durchfluss (Platzhalter der Quelle "
|
|
"für noch nicht verfügbare Messwerte) werden herausgefiltert. "
|
|
"`latest` ist der neueste verbleibende Eintrag und entspricht `entries[0]`; "
|
|
"die Reihenfolge von `entries` folgt dem Feed (neueste zuerst). "
|
|
"Pegel-IDs bestehen aus mindestens sechs Ziffern. Einzelne fehlende "
|
|
"Messwerte werden als `null` ausgegeben. `alarm_thresholds` wird "
|
|
"zusätzlich von der Pegel-Detailseite ermittelt und 24 Stunden "
|
|
"zwischengespeichert; ist sie nicht erreichbar, wird ein vorhandener "
|
|
"Cache verwendet oder `null` geliefert, ohne die übrige Antwort zu "
|
|
"beeinträchtigen."
|
|
),
|
|
response_description="Stationsmetadaten, neuester Messwert und Feed-Historie.",
|
|
responses={
|
|
400: {"description": "Die Pegel-ID besteht nicht aus mindestens sechs Ziffern."},
|
|
404: {"description": "Für diese Pegel-ID enthält der RSS-Feed keine Einträge mit Messwerten."},
|
|
},
|
|
)
|
|
def get_pegel(pegel_id: str):
|
|
if not re.fullmatch(r"\d{6,}", pegel_id):
|
|
raise HTTPException(400, "Ungültige Pegel-ID")
|
|
|
|
url = BASE.format(pegel_id)
|
|
feed = feedparser.parse(url)
|
|
|
|
entries = parse_valid_entries(feed.entries)
|
|
|
|
if not entries:
|
|
raise HTTPException(404, "Keine Daten gefunden")
|
|
|
|
feed_obj = getattr(feed, "feed", None)
|
|
feed_title = feed_obj.get("title") if isinstance(feed_obj, dict) else None
|
|
|
|
return JSONResponse({
|
|
"station_id": pegel_id,
|
|
"station_title": extract_station_name(feed_title),
|
|
"count": len(entries),
|
|
"latest": entries[0],
|
|
"entries": entries,
|
|
"alarm_thresholds": _get_alarm_thresholds(pegel_id),
|
|
"source": url,
|
|
"fetched_at": dt.datetime.now(dt.UTC).isoformat(),
|
|
})
|