Simon/wetter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: app.py

Browse Source 
modified:   templates/base.html
	modified:   templates/weather.html
This commit is contained in:
simon
2026-04-22 10:38:42 +02:00
parent 164d0eb6a8
commit e59e88cafe
3 changed files with 159 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
app.py
View File

@@ -3,9 +3,12 @@ from flask import Flask, render_template, request, jsonify
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderTimedOut
import math
import traceback
import pandas as pd
import datetime as _dt
import logging
from collections import deque
from threading import Lock
from time import time
from wetterdienst.provider.dwd.mosmix import DwdMosmixRequest
from cachetools import TTLCache
@@ -14,6 +17,9 @@ from astral.sun import sun as astral_sun
import requests as _requests
app = Flask(__name__)
app.logger.setLevel(logging.INFO)
_GEOLOCATOR = Nominatim(user_agent="skywatcher-app/1.0")
# ── Zeitzone ────────────────────────────────────────────────────────────
try:
@@ -26,6 +32,32 @@ except ImportError:
# ── Cache: Forecasts 45 Min, DWD-Warnungen 15 Min ────────────────────────────
_forecast_cache = TTLCache(maxsize=64, ttl=2700)
_warn_cache = TTLCache(maxsize=32, ttl=900)
_suggest_cache = TTLCache(maxsize=256, ttl=900)
# API protection: simple in-memory rate limiting by client IP.
_suggest_rate_lock = Lock()
_suggest_rate_hits = {}
_SUGGEST_RATE_WINDOW_SECONDS = 60
_SUGGEST_RATE_MAX_REQUESTS = 30
STATE_ALIASES = {
"baden-wurttemberg": {"baden-wurttemberg", "baden wuerttemberg", "bw"},
"bayern": {"bayern", "by"},
"berlin": {"berlin", "be"},
"brandenburg": {"brandenburg", "bb"},
"bremen": {"bremen", "hb"},
"hamburg": {"hamburg", "hh"},
"hessen": {"hessen", "he"},
"mecklenburg-vorpommern": {"mecklenburg-vorpommern", "mecklenburg vorpommern", "mv"},
"niedersachsen": {"niedersachsen", "ni"},
"nordrhein-westfalen": {"nordrhein-westfalen", "nordrhein westfalen", "nrw", "nw"},
"rheinland-pfalz": {"rheinland-pfalz", "rheinland pfalz", "rp"},
"saarland": {"saarland", "sl"},
"sachsen": {"sachsen", "sn"},
"sachsen-anhalt": {"sachsen-anhalt", "sachsen anhalt", "st"},
"schleswig-holstein": {"schleswig-holstein", "schleswig holstein", "sh"},
"thuringen": {"thuringen", "thueringen", "th"},
}
# ── MOSMIX Parameter ─────────────────────────────────────────────────────────
MOSMIX_PARAMS = [
@@ -49,15 +81,78 @@ def _get_berlin():
import pytz
return pytz.timezone("Europe/Berlin")
def geocode_location(query):
geolocator = Nominatim(user_agent="skywatcher-app/1.0")
def _normalize_text(value):
if not value:
return ""
text = str(value).strip().lower()
replacements = {
"ä": "a",
"ö": "o",
"ü": "u",
"ß": "ss",
}
for src, dst in replacements.items():
text = text.replace(src, dst)
return " ".join(text.split())
def _state_tokens(state_name):
normalized = _normalize_text(state_name)
if not normalized:
return set()
for _, aliases in STATE_ALIASES.items():
if normalized in aliases:
return aliases
return {normalized}
def _warning_matches_state(state_hint, warning_state, warning_state_short):
hint_tokens = _state_tokens(state_hint)
warning_tokens = _state_tokens(warning_state)
warning_short = _normalize_text(warning_state_short)
if warning_short:
warning_tokens.add(warning_short)
return bool(hint_tokens and warning_tokens and hint_tokens.intersection(warning_tokens))
def _extract_state_from_location(loc):
try:
loc = geolocator.geocode(query, language="de", timeout=10)
return (loc.raw or {}).get("address", {}).get("state")
except Exception:
return None
def _client_ip(req):
forwarded = req.headers.get("X-Forwarded-For", "").strip()
if forwarded:
return forwarded.split(",")[0].strip()
return req.remote_addr or "unknown"
def _is_suggest_rate_limited(client_ip):
now = time()
with _suggest_rate_lock:
hits = _suggest_rate_hits.get(client_ip)
if hits is None:
hits = deque()
_suggest_rate_hits[client_ip] = hits
cutoff = now - _SUGGEST_RATE_WINDOW_SECONDS
while hits and hits[0] < cutoff:
hits.popleft()
if len(hits) >= _SUGGEST_RATE_MAX_REQUESTS:
return True
hits.append(now)
if len(hits) == 1:
stale_ips = [ip for ip, values in _suggest_rate_hits.items() if not values or values[-1] < cutoff]
for stale_ip in stale_ips:
_suggest_rate_hits.pop(stale_ip, None)
return False
def geocode_location(query):
try:
loc = _GEOLOCATOR.geocode(query, language="de", addressdetails=True, timeout=10)
if loc:
return loc.latitude, loc.longitude, loc.address
return loc.latitude, loc.longitude, loc.address, _extract_state_from_location(loc)
except GeocoderTimedOut:
pass
return None, None, None
except Exception:
app.logger.exception("Geocoding failed for query '%s'", query)
return None, None, None, None
def haversine(lat1, lon1, lat2, lon2):
R = 6371
@@ -103,8 +198,9 @@ def get_sun_times(lat, lon, date=None):
except Exception:
return None, None, None, None
def get_dwd_warnings(lat, lon):
key = (round(lat, 1), round(lon, 1))
def get_dwd_warnings(lat, lon, state_hint=None):
state_key = _normalize_text(state_hint)
key = (round(lat, 1), round(lon, 1), state_key)
if key in _warn_cache:
return _warn_cache[key]
try:
@@ -128,12 +224,27 @@ def get_dwd_warnings(lat, lon):
"type": w.get("event", ""),
"headline": w.get("headline", ""),
"description": w.get("description", ""),
"state": w.get("state", ""),
"state_short": w.get("stateShort", ""),
})
warnings.sort(key=lambda x: x["level"], reverse=True)
result = warnings[:3]
filtered = []
if state_hint:
filtered = [w for w in warnings if _warning_matches_state(state_hint, w.get("state"), w.get("state_short"))]
result_source = filtered if filtered else []
result_source.sort(key=lambda x: x["level"], reverse=True)
result = [
{
"level": w["level"],
"type": w["type"],
"headline": w["headline"],
"description": w["description"],
}
for w in result_source[:3]
]
_warn_cache[key] = result
return result
except Exception:
app.logger.exception("Could not load DWD warnings")
_warn_cache[key] = []
return []
@@ -225,7 +336,7 @@ def get_mosmix_forecast(lat, lon, hours=72):
_forecast_cache[cache_key] = result_data
return result_data
except Exception:
traceback.print_exc()
app.logger.exception("MOSMIX forecast loading failed")
return [], {}
@app.route("/")
@@ -239,23 +350,26 @@ def wetter():
lon_param = request.args.get("lon")
# Geolocation via Browser-Koordinaten
lat, lon, display_name = None, None, None
lat, lon, display_name, state_hint = None, None, None, None
if lat_param and lon_param:
try:
lat = float(lat_param)
lon = float(lon_param)
geolocator = Nominatim(user_agent="skywatcher-app/1.0")
loc = geolocator.reverse((lat, lon), language="de", timeout=10)
if not (-90 <= lat <= 90 and -180 <= lon <= 180):
raise ValueError("Invalid coordinate range")
loc = _GEOLOCATOR.reverse((lat, lon), language="de", addressdetails=True, timeout=10)
display_name = loc.address if loc else f"{lat:.2f}, {lon:.2f}"
state_hint = _extract_state_from_location(loc)
if not ort or ort == "Mein Standort":
ort = display_name.split(",")[0]
except Exception:
lat, lon, display_name = None, None, None
app.logger.info("Invalid or unusable browser coordinates for /wetter")
lat, lon, display_name, state_hint = None, None, None, None
if lat is None:
if not ort:
return render_template("index.html", error="Bitte einen Ort eingeben.")
lat, lon, display_name = geocode_location(ort)
lat, lon, display_name, state_hint = geocode_location(ort)
if lat is None:
return render_template("index.html", error=f'Ort "{ort}" konnte nicht gefunden werden.')
forecast, mosmix_station = get_mosmix_forecast(lat, lon, hours=72)
@@ -280,7 +394,7 @@ def wetter():
current = forecast[current_idx]
forecast = forecast[current_idx:]
sunrise, sunset, dawn, dusk = get_sun_times(lat, lon)
warnings = get_dwd_warnings(lat, lon)
warnings = get_dwd_warnings(lat, lon, state_hint=state_hint)
daily = {}
for h in forecast:
dt = h["datetime"]
@@ -331,12 +445,24 @@ def wetter():
@app.route("/api/suggest")
def suggest():
q = request.args.get("q","").strip()
if len(q) < 2: return jsonify([])
geolocator = Nominatim(user_agent="skywatcher-app/1.0")
if len(q) < 2:
return jsonify([])
client_ip = _client_ip(request)
if _is_suggest_rate_limited(client_ip):
return jsonify({"error": "Zu viele Anfragen. Bitte kurz warten."}), 429
cache_key = _normalize_text(q)
if cache_key in _suggest_cache:
return jsonify(_suggest_cache[cache_key])
try:
results = geolocator.geocode(q, exactly_one=False, limit=5, language="de", addressdetails=True, timeout=5)
return jsonify([{"name":r.address,"lat":r.latitude,"lon":r.longitude} for r in results] if results else [])
results = _GEOLOCATOR.geocode(q, exactly_one=False, limit=5, language="de", addressdetails=True, timeout=5)
payload = [{"name": r.address, "lat": r.latitude, "lon": r.longitude} for r in results] if results else []
_suggest_cache[cache_key] = payload
return jsonify(payload)
except Exception:
app.logger.exception("Suggest lookup failed for query '%s'", q)
return jsonify([])
if __name__ == "__main__":