211 lines
7.0 KiB
Python
Executable File
211 lines
7.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Wetterserver-Monitor: Prüft alle 5 Minuten ob Wetterdaten ankommen.
|
||
- Alert-E-Mail wenn >15 Minuten keine Daten (Wiederholung nach 2h)
|
||
- Tägliche Status-E-Mail zur Bestätigung dass alles läuft
|
||
|
||
Cron-Empfehlung: alle 5 Minuten
|
||
*/5 * * * * /pfad/zu/.venv/bin/python /pfad/zu/check_wetterserver.py >> /pfad/zu/monitor.log 2>&1
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import smtplib
|
||
import ssl
|
||
import urllib.request
|
||
from datetime import datetime, timezone, timedelta, date
|
||
from email.message import EmailMessage
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
SCRIPT_DIR = Path(__file__).parent
|
||
ENV_FILE = SCRIPT_DIR / ".env"
|
||
STATE_FILE = SCRIPT_DIR / ".monitoring_state.json"
|
||
|
||
API_URL = "https://stwwetter.fuerst-stuttgart.de/api/weather/latest"
|
||
TIMEOUT_MINUTES = 15
|
||
RESEND_AFTER_HOURS = 2
|
||
|
||
|
||
def _ssl_context() -> ssl.SSLContext:
|
||
try:
|
||
import certifi
|
||
return ssl.create_default_context(cafile=certifi.where())
|
||
except ImportError:
|
||
return ssl.create_default_context()
|
||
|
||
|
||
def load_env() -> None:
|
||
if not ENV_FILE.exists():
|
||
return
|
||
for line in ENV_FILE.read_text().splitlines():
|
||
line = line.strip()
|
||
if line and not line.startswith("#") and "=" in line:
|
||
key, _, val = line.partition("=")
|
||
val = val.strip().strip("'\"")
|
||
if key not in os.environ:
|
||
os.environ[key] = val
|
||
|
||
|
||
def load_state() -> dict:
|
||
if STATE_FILE.exists():
|
||
try:
|
||
return json.loads(STATE_FILE.read_text())
|
||
except Exception:
|
||
pass
|
||
return {"alert_active": False, "last_alert_sent": None, "last_daily_report": None}
|
||
|
||
|
||
def save_state(state: dict) -> None:
|
||
STATE_FILE.write_text(json.dumps(state, indent=2))
|
||
|
||
|
||
def get_latest_weather() -> Optional[dict]:
|
||
try:
|
||
with urllib.request.urlopen(API_URL, timeout=10, context=_ssl_context()) as resp:
|
||
return json.loads(resp.read())
|
||
except Exception as e:
|
||
print(f"API-Anfrage fehlgeschlagen: {e}")
|
||
return None
|
||
|
||
|
||
def parse_datetime(dt_str: str) -> Optional[datetime]:
|
||
try:
|
||
dt = datetime.fromisoformat(str(dt_str).replace("Z", "+00:00"))
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _send_email(subject: str, body: str) -> bool:
|
||
smtp_host = os.getenv("MONITOR_SMTP_HOST", "smtp.gmx.de")
|
||
smtp_port = int(os.getenv("MONITOR_SMTP_PORT", "587"))
|
||
smtp_user = os.getenv("MONITOR_SMTP_USER", "")
|
||
smtp_pass = os.getenv("MONITOR_SMTP_PASSWORD", "")
|
||
to_email = os.getenv("MONITOR_TO_EMAIL", "rxf@gmx.de")
|
||
from_email = os.getenv("MONITOR_FROM_EMAIL", smtp_user)
|
||
|
||
if not smtp_user or not smtp_pass:
|
||
print("FEHLER: MONITOR_SMTP_USER / MONITOR_SMTP_PASSWORD nicht in .env gesetzt")
|
||
return False
|
||
|
||
msg = EmailMessage()
|
||
msg["Subject"] = subject
|
||
msg["From"] = from_email
|
||
msg["To"] = to_email
|
||
msg.set_content(body)
|
||
|
||
try:
|
||
with smtplib.SMTP(smtp_host, smtp_port) as smtp:
|
||
smtp.ehlo()
|
||
smtp.starttls()
|
||
smtp.login(smtp_user, smtp_pass)
|
||
smtp.send_message(msg)
|
||
print(f"E-Mail gesendet: {subject!r} -> {to_email}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"FEHLER beim E-Mail-Versand: {e}")
|
||
return False
|
||
|
||
|
||
def send_alert(latest_dt: Optional[datetime], state: dict, now: datetime) -> None:
|
||
if latest_dt is not None:
|
||
age_min = int((now - latest_dt).total_seconds() / 60)
|
||
body = (
|
||
f"Achtung: Die Wetterstation sendet seit {age_min} Minuten keine Daten mehr.\n\n"
|
||
f"Letzter Datenpunkt: {latest_dt.strftime('%d.%m.%Y %H:%M:%S UTC')}\n"
|
||
f"Aktueller Zeitpunkt: {now.strftime('%d.%m.%Y %H:%M:%S UTC')}\n\n"
|
||
f"Bitte überprüfen Sie den Wetterserver.\n"
|
||
f"API: {API_URL}"
|
||
)
|
||
else:
|
||
body = (
|
||
f"Achtung: Die Wetterstation-API ist nicht erreichbar oder liefert keine Daten.\n\n"
|
||
f"Zeitpunkt: {now.strftime('%d.%m.%Y %H:%M:%S UTC')}\n\n"
|
||
f"Bitte überprüfen Sie den Wetterserver.\n"
|
||
f"API: {API_URL}"
|
||
)
|
||
if _send_email("⚠ Wetterstation: Keine Daten seit >15 Minuten", body):
|
||
state["last_alert_sent"] = now.isoformat()
|
||
|
||
|
||
def send_daily_report(data: dict, latest_dt: datetime, now: datetime, state: dict) -> None:
|
||
temp = data.get("temperature")
|
||
humidity = data.get("humidity")
|
||
pressure = data.get("pressure")
|
||
age_min = int((now - latest_dt).total_seconds() / 60)
|
||
|
||
lines = [
|
||
"✓ Die Wetterstation läuft einwandfrei.",
|
||
"",
|
||
f"Letzter Datenpunkt: {latest_dt.strftime('%d.%m.%Y %H:%M:%S UTC')} (vor {age_min} Min.)",
|
||
f"Prüfzeitpunkt: {now.strftime('%d.%m.%Y %H:%M:%S UTC')}",
|
||
"",
|
||
"Aktuelle Messwerte:",
|
||
]
|
||
if temp is not None:
|
||
lines.append(f" Temperatur: {temp:.1f} °C")
|
||
if humidity is not None:
|
||
lines.append(f" Luftfeuchte: {humidity} %")
|
||
if pressure is not None:
|
||
lines.append(f" Luftdruck: {pressure:.1f} hPa")
|
||
lines += ["", f"API: {API_URL}"]
|
||
|
||
if _send_email("✓ Wetterstation: Täglicher Status – alles in Ordnung", "\n".join(lines)):
|
||
state["last_daily_report"] = date.today().isoformat()
|
||
|
||
|
||
def main() -> None:
|
||
load_env()
|
||
state = load_state()
|
||
now = datetime.now(timezone.utc)
|
||
|
||
weather = get_latest_weather()
|
||
latest_dt: Optional[datetime] = None
|
||
if weather:
|
||
dt_str = weather.get("datetime") or weather.get("received_at")
|
||
if dt_str:
|
||
latest_dt = parse_datetime(dt_str)
|
||
|
||
outage = latest_dt is None or (now - latest_dt) > timedelta(minutes=TIMEOUT_MINUTES)
|
||
|
||
if outage:
|
||
should_send = not state["alert_active"]
|
||
if not should_send and state.get("last_alert_sent"):
|
||
last_sent = datetime.fromisoformat(state["last_alert_sent"])
|
||
should_send = (now - last_sent) > timedelta(hours=RESEND_AFTER_HOURS)
|
||
|
||
state["alert_active"] = True
|
||
if should_send:
|
||
send_alert(latest_dt, state, now)
|
||
else:
|
||
print(
|
||
f"{now.strftime('%Y-%m-%d %H:%M:%S')} Ausfall aktiv, "
|
||
f"nächste E-Mail frühestens um "
|
||
f"{(datetime.fromisoformat(state['last_alert_sent']) + timedelta(hours=RESEND_AFTER_HOURS)).strftime('%H:%M')} UTC"
|
||
)
|
||
else:
|
||
if state["alert_active"]:
|
||
print(
|
||
f"{now.strftime('%Y-%m-%d %H:%M:%S')} Wetterstation liefert wieder Daten "
|
||
f"(letzter Datenpunkt: {latest_dt.strftime('%d.%m.%Y %H:%M:%S UTC')})"
|
||
)
|
||
state["alert_active"] = False
|
||
|
||
today = date.today().isoformat()
|
||
if state.get("last_daily_report") != today:
|
||
send_daily_report(weather, latest_dt, now, state)
|
||
else:
|
||
print(
|
||
f"{now.strftime('%Y-%m-%d %H:%M:%S')} OK – "
|
||
f"letzter Datenpunkt: {latest_dt.strftime('%H:%M:%S UTC')}"
|
||
)
|
||
|
||
save_state(state)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|