From ccf0825b3587f9118689787b302acf89ffc315b7 Mon Sep 17 00:00:00 2001 From: daniele Date: Tue, 2 Dec 2025 21:54:08 +0100 Subject: [PATCH] Backup automatico script del 2025-12-02 21:54 --- services/telegram-bot/severe_weather.py | 107 ++++++++++++++++-------- 1 file changed, 72 insertions(+), 35 deletions(-) diff --git a/services/telegram-bot/severe_weather.py b/services/telegram-bot/severe_weather.py index d22099d..1a041ee 100644 --- a/services/telegram-bot/severe_weather.py +++ b/services/telegram-bot/severe_weather.py @@ -7,6 +7,7 @@ from dateutil import parser from zoneinfo import ZoneInfo # --- CONFIGURAZIONE --- +# 👇👇 INSERISCI QUI I TUOI DATI 👇👇 TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4" TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"] @@ -15,19 +16,22 @@ 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) +WIND_LIMIT_WARN = 60.0 # km/h (Vento Forte) +WIND_LIMIT_CRIT = 90.0 # km/h (Burrasca) +RAIN_3H_LIMIT = 25.0 # mm in 3 ore (Rischio idrogeologico) -# File di stato per evitare spam (memorizza ultimo avviso inviato) -STATE_FILE = "/tmp/weather_alert_state.json" +# File di stato +STATE_FILE = "/home/daniely/docker/telegram-bot/weather_state.json" + +def send_telegram_message(message): + if not TELEGRAM_BOT_TOKEN or "INSERISCI" in TELEGRAM_BOT_TOKEN: + print(f"[TEST OUT] {message}") + return -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: + for chat_id in TELEGRAM_CHAT_IDS: try: - requests.post(url, json={"chat_id": cid, "text": msg, "parse_mode": "Markdown"}, timeout=5) + requests.post(url, json={"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}, timeout=5) time.sleep(0.2) except: pass @@ -36,7 +40,7 @@ def get_forecast(): params = { "latitude": LAT, "longitude": LON, "hourly": "precipitation,windgusts_10m", - "models": "arome_france_hd", # Alta precisione + "models": "arome_france_hd", "timezone": "Europe/Rome", "forecast_days": 2 } @@ -51,10 +55,12 @@ def load_state(): try: with open(STATE_FILE, 'r') as f: return json.load(f) except: pass - return {"last_wind": 0, "last_rain": 0, "time": 0} + return {"alert_active": False, "last_wind": 0, "last_rain": 0, "time": 0} def save_state(state): - with open(STATE_FILE, 'w') as f: json.dump(state, f) + try: + with open(STATE_FILE, 'w') as f: json.dump(state, f) + except: pass def analyze(): data = get_forecast() @@ -68,10 +74,6 @@ def analyze(): 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): @@ -88,41 +90,76 @@ def analyze(): max_wind_time = "" sum_rain_3h = 0.0 - # Cerca picco vento nelle 12h + # Cerca picco vento 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) + # Cerca picco pioggia (sliding window 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 + # --- LOGICA DI DECISIONE --- + + is_wind_alarm = max_wind > WIND_LIMIT_WARN + is_rain_alarm = sum_rain_3h > RAIN_3H_LIMIT + IS_ALARM_NOW = is_wind_alarm or is_rain_alarm + + WAS_ALARM = state.get("alert_active", False) + 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() + # SCENARIO A: NUOVO ALLARME o PEGGIORAMENTO + if IS_ALARM_NOW: + should_notify = False + + # Logica Vento + if is_wind_alarm: + # Notifica se è nuovo o se è peggiorato di 10km/h rispetto all'ultimo avviso + if not WAS_ALARM or max_wind > state.get("last_wind", 0) + 10: + icon = "💨" if max_wind < WIND_LIMIT_CRIT else "🌪️ ⛔️" + livello = "FORTE" if max_wind < WIND_LIMIT_CRIT else "BURRASCA" + 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._") + state["last_wind"] = max_wind + should_notify = True + + # Logica Pioggia + if is_rain_alarm: + # Notifica se è nuovo o se è peggiorata di 10mm + if not WAS_ALARM or sum_rain_3h > state.get("last_rain", 0) + 10: + alerts.append(f"🌧️ **PIOGGIA INTENSA**\nPrevisti **{sum_rain_3h:.1f} mm** in 3 ore.\n_Rischio disagi idraulici._") + state["last_rain"] = sum_rain_3h + should_notify = True - # 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 + if should_notify: + full_msg = f"⚠️ **AVVISO METEO (Prossime 12h)**\n\n" + "\n\n".join(alerts) + send_telegram_message(full_msg) + state["alert_active"] = True state["time"] = time.time() + save_state(state) + print("Allerta inviata.") + else: + print("Condizioni critiche ma stabili (già notificato).") - if alerts: - full_msg = f"⚠️ **AVVISO METEO (Prossime 12h)**\n\n" + "\n\n".join(alerts) - send_telegram_message(full_msg) + # SCENARIO B: ALLARME RIENTRATO + elif WAS_ALARM and not IS_ALARM_NOW: + msg = ( + f"🟢 **ALLERTA METEO RIENTRATA**\n" + f"📅 _Aggiornamento ore {now.strftime('%H:%M')}_\n\n" + f"Vento e pioggia sono scesi sotto le soglie di guardia per le prossime 12 ore." + ) + send_telegram_message(msg) + + # Reset stato + state = {"alert_active": False, "last_wind": 0, "last_rain": 0, "time": time.time()} save_state(state) + print("Allarme rientrato inviato.") + + else: + print(f"Situazione tranquilla. Vento: {max_wind:.1f}, Pioggia: {sum_rain_3h:.1f}") if __name__ == "__main__": analyze()