Backup automatico script del 2025-12-01 17:25

services/telegram-bot/severe_weather.py

@@ -0,0 +1,128 @@
+import requests
+import datetime
+import os
+import json
+import time
+from dateutil import parser
+from zoneinfo import ZoneInfo
+
+# --- CONFIGURAZIONE ---
+TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
+TELEGRAM_CHAT_IDS = ["64463169", "132455422"]
+
+# Coordinate (San Marino)
+LAT = 43.9356
+LON = 12.4296
+
+# SOGLIE DI ALLARME
+WIND_LIMIT_WARN = 60.0  # km/h (Attenzione: Vasi, stendini)
+WIND_LIMIT_CRIT = 90.0  # km/h (Pericolo)
+RAIN_3H_LIMIT   = 25.0  # mm in 3 ore (Rischio allagamenti/disagi forti)
+
+# File di stato per evitare spam (memorizza ultimo avviso inviato)
+STATE_FILE = "/tmp/weather_alert_state.json"
+
+def send_telegram(msg):
+    if "INSERISCI" in TELEGRAM_BOT_TOKEN: return
+    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
+    for cid in TELEGRAM_CHAT_IDS:
+        try:
+            requests.post(url, json={"chat_id": cid, "text": msg, "parse_mode": "Markdown"}, timeout=5)
+            time.sleep(0.2)
+        except: pass
+
+def get_forecast():
+    url = "https://api.open-meteo.com/v1/forecast"
+    params = {
+        "latitude": LAT, "longitude": LON,
+        "hourly": "precipitation,windgusts_10m",
+        "models": "arome_france_hd", # Alta precisione
+        "timezone": "Europe/Rome",
+        "forecast_days": 2
+    }
+    try:
+        r = requests.get(url, params=params, timeout=10)
+        r.raise_for_status()
+        return r.json()
+    except: return None
+
+def load_state():
+    if os.path.exists(STATE_FILE):
+        try:
+            with open(STATE_FILE, 'r') as f: return json.load(f)
+        except: pass
+    return {"last_wind": 0, "last_rain": 0, "time": 0}
+
+def save_state(state):
+    with open(STATE_FILE, 'w') as f: json.dump(state, f)
+
+def analyze():
+    data = get_forecast()
+    if not data: return
+
+    hourly = data.get("hourly", {})
+    times = hourly.get("time", [])
+    wind = hourly.get("windgusts_10m", [])
+    rain = hourly.get("precipitation", [])
+
+    now = datetime.datetime.now(ZoneInfo("Europe/Rome"))
+    state = load_state()
+    
+    # Reset stato se sono passate più di 8 ore dall'ultimo avviso
+    if time.time() - state.get("time", 0) > (8 * 3600):
+        state = {"last_wind": 0, "last_rain": 0, "time": time.time()}
+
+    # Trova indice ora corrente
+    start_idx = -1
+    for i, t in enumerate(times):
+        if parser.isoparse(t).replace(tzinfo=ZoneInfo("Europe/Rome")) >= now:
+            start_idx = i
+            break
+    
+    if start_idx == -1: return
+
+    # Analisi prossime 12 ore
+    end_idx = min(start_idx + 12, len(times))
+    
+    max_wind = 0.0
+    max_wind_time = ""
+    sum_rain_3h = 0.0
+    
+    # Cerca picco vento nelle 12h
+    for i in range(start_idx, end_idx):
+        if wind[i] > max_wind:
+            max_wind = wind[i]
+            max_wind_time = parser.isoparse(times[i]).strftime('%H:%M')
+            
+    # Cerca picco pioggia (finestra mobile 3h)
+    for i in range(start_idx, end_idx - 3):
+        current_sum = sum(rain[i:i+3])
+        if current_sum > sum_rain_3h:
+            sum_rain_3h = current_sum
+
+    alerts = []
+
+    # LOGICA VENTO
+    if max_wind > WIND_LIMIT_WARN:
+        # Manda avviso solo se è peggiore del precedente o se non ho mandato nulla
+        if max_wind > state["last_wind"] + 10: # +10km/h di tolleranza per non ripetere
+            icon = "💨" if max_wind < WIND_LIMIT_CRIT else "🌪️ ⛔️"
+            livello = "FORTE" if max_wind < WIND_LIMIT_CRIT else "BURRASCA (Pericolo)"
+            alerts.append(f"{icon} **VENTO {livello}**\nRaffiche previste fino a **{max_wind:.0f} km/h** verso le {max_wind_time}.\n_Consiglio: Ritirare oggetti leggeri/tende._")
+            state["last_wind"] = max_wind
+            state["time"] = time.time()
+
+    # LOGICA PIOGGIA
+    if sum_rain_3h > RAIN_3H_LIMIT:
+        if sum_rain_3h > state["last_rain"] + 10:
+            alerts.append(f"🌧️ **PIOGGIA INTENSA**\nPrevisti **{sum_rain_3h:.1f} mm** in sole 3 ore.\n_Possibili disagi stradali._")
+            state["last_rain"] = sum_rain_3h
+            state["time"] = time.time()
+
+    if alerts:
+        full_msg = f"⚠️ **AVVISO METEO (Prossime 12h)**\n\n" + "\n\n".join(alerts)
+        send_telegram_message(full_msg)
+        save_state(state)
+
+if __name__ == "__main__":
+    analyze()