From 49044655877eefffa14632d009f9299af5383390 Mon Sep 17 00:00:00 2001 From: daniele Date: Mon, 1 Dec 2025 17:15:00 +0100 Subject: [PATCH] Backup automatico script del 2025-12-01 17:14 --- services/telegram-bot/arome_snow_alert.py | 152 +++++++++++++--------- 1 file changed, 90 insertions(+), 62 deletions(-) diff --git a/services/telegram-bot/arome_snow_alert.py b/services/telegram-bot/arome_snow_alert.py index 72eed29..3b3d799 100644 --- a/services/telegram-bot/arome_snow_alert.py +++ b/services/telegram-bot/arome_snow_alert.py @@ -1,110 +1,138 @@ import requests import datetime +import time from dateutil import parser from zoneinfo import ZoneInfo # --- CONFIGURAZIONE UTENTE --- +# ๐Ÿ‘‡๐Ÿ‘‡ INSERISCI QUI I TUOI DATI ๐Ÿ‘‡๐Ÿ‘‡ TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4" -TELEGRAM_CHAT_ID = "64463169" +TELEGRAM_CHAT_IDS = ["64463169", "132455422"] -# Coordinate San Marino -LATITUDE = 43.9356 -LONGITUDE = 12.4296 +# --- PUNTI DI MONITORAGGIO --- +# Il primo punto DEVE essere Casa tua +POINTS = [ + {"name": "๐Ÿ  Casa", "lat": 43.9356, "lon": 12.4296}, + {"name": "โ›ฐ๏ธ Titano", "lat": 43.9360, "lon": 12.4460}, + {"name": "๐Ÿข Dogana", "lat": 43.9800, "lon": 12.4900}, + {"name": "๐Ÿฐ San Leo", "lat": 43.8900, "lon": 12.3400} +] -# Soglia minima per notifica (es. 0.0 per qualsiasi nevicata) +# Soglia notifica (cm) SOGLIA_NOTIFICA = 0.0 def send_telegram_message(message): if not TELEGRAM_BOT_TOKEN or "INSERISCI" in TELEGRAM_BOT_TOKEN: - print(f"[TEST - NO TELEGRAM] {message}") + print(f"[TEST OUT] {message}") return url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" - payload = { - "chat_id": TELEGRAM_CHAT_ID, - "text": message, - "parse_mode": "Markdown" - } - try: - requests.post(url, json=payload, timeout=10) - except Exception as e: - print(f"Errore invio Telegram: {e}") + for chat_id in TELEGRAM_CHAT_IDS: + payload = {"chat_id": chat_id, "text": message, "parse_mode": "Markdown"} + try: + requests.post(url, json=payload, timeout=10) + time.sleep(0.2) + except Exception as e: + print(f"Errore invio a {chat_id}: {e}") -def get_arome_forecast(): +def get_forecast(lat, lon): url = "https://api.open-meteo.com/v1/forecast" params = { - "latitude": LATITUDE, - "longitude": LONGITUDE, + "latitude": lat, + "longitude": lon, "hourly": "snowfall", "models": "arome_france_hd", "timezone": "Europe/Rome", "forecast_days": 2 } - try: - response = requests.get(url, params=params, timeout=10) + response = requests.get(url, params=params, timeout=5) response.raise_for_status() return response.json() - except Exception as e: - print(f"Errore richiesta API: {e}") + except: return None -def analyze_snow(): - data = get_arome_forecast() - if not data: - return +def calculate_sums(data): + if not data: return None + + hourly = data.get("hourly", {}) + times = hourly.get("time", []) + snow = hourly.get("snowfall", []) + + if not times or not snow: return None - hourly_data = data.get("hourly", {}) - times = hourly_data.get("time", []) - snowfall = hourly_data.get("snowfall", []) - - if not times or not snowfall: - print("Dati meteo incompleti.") - return - - # Otteniamo l'ora corrente now = datetime.datetime.now(ZoneInfo("Europe/Rome")) - # Sincronizzazione orario - start_index = -1 + start_idx = -1 for i, t_str in enumerate(times): try: t_obj = parser.isoparse(t_str).replace(tzinfo=ZoneInfo("Europe/Rome")) if t_obj >= now.replace(minute=0, second=0, microsecond=0): - start_index = i + start_idx = i break - except Exception: - continue + except: continue + + if start_idx == -1: return None + + end = len(snow) + s3 = sum(x for x in snow[start_idx:min(start_idx+3, end)] if x) + s6 = sum(x for x in snow[start_idx:min(start_idx+6, end)] if x) + s12 = sum(x for x in snow[start_idx:min(start_idx+12, end)] if x) + s24 = sum(x for x in snow[start_idx:min(start_idx+24, end)] if x) - if start_index == -1: - print("Impossibile sincronizzare l'orario dei dati.") - return + return {"3h": s3, "6h": s6, "12h": s12, "24h": s24} - # Slice dati 12h e 24h - end_12h = min(start_index + 12, len(snowfall)) - end_24h = min(start_index + 24, len(snowfall)) +def analyze_snow(): + now_str = datetime.datetime.now(ZoneInfo("Europe/Rome")).strftime('%H:%M') + print(f"--- Check Meteo {now_str} ---") - snow_next_12h = snowfall[start_index:end_12h] - snow_next_24h = snowfall[start_index:end_24h] + home_stats = None + max_area_snow = 0.0 + area_details = "" + + for p in POINTS: + data = get_forecast(p["lat"], p["lon"]) + stats = calculate_sums(data) + + if not stats: continue + + if p["name"] == "๐Ÿ  Casa": + home_stats = stats + + if stats["24h"] > max_area_snow: + max_area_snow = stats["24h"] - # --- CORREZIONE ERRORE --- - # Convertiamo eventuali None in 0.0 prima di sommare - sum_12h = sum(x if x is not None else 0.0 for x in snow_next_12h) - sum_24h = sum(x if x is not None else 0.0 for x in snow_next_24h) + # Aggiungi ai dettagli area se c'รจ neve (>0) + if stats["24h"] > 0: + area_details += f"{p['name']}: {stats['24h']:.1f}cm (12h: {stats['12h']:.1f})\n" + + time.sleep(1) - print(f"Check ore {now.strftime('%H:%M')}: 12h={sum_12h}cm, 24h={sum_24h}cm") + home_max = home_stats["24h"] if home_stats else 0.0 + + if home_max > SOGLIA_NOTIFICA or max_area_snow > SOGLIA_NOTIFICA: + + def f(v): return f"**{v:.1f}**" if v > 0 else f"{v:.1f}" + + msg = f"โ„๏ธ **ALLERTA NEVE (AROME HD)**\n๐Ÿ“… _Aggiornamento ore {now_str}_\n\n" + + if home_stats: + msg += f"๐Ÿ  **CASA:**\n" + msg += f"โ€ข 03h: {f(home_stats['3h'])} cm\n" + msg += f"โ€ข 06h: {f(home_stats['6h'])} cm\n" + msg += f"โ€ข 12h: {f(home_stats['12h'])} cm\n" + msg += f"โ€ข 24h: {f(home_stats['24h'])} cm\n\n" + + if area_details: + msg += f"๐ŸŒ **NEL CIRCONDARIO (24h):**\n" + msg += f"`{area_details}`" + else: + msg += "๐ŸŒ Nessuna neve rilevante nei dintorni." - if sum_24h > SOGLIA_NOTIFICA: - msg = ( - f"โš ๏ธ **ALLERTA NEVE SAN MARINO** โš ๏ธ\n" - f"Modello: AROME-HD (1.3km)\n\n" - f"โ„๏ธ **Prossime 12 ore:** {sum_12h:.1f} cm\n" - f"๐ŸŒจ๏ธ **Prossime 24 ore:** {sum_24h:.1f} cm\n\n" - f"๐Ÿ“… _Previsto a partire dalle {now.strftime('%H:%M')}_" - ) send_telegram_message(msg) + print("Neve rilevata. Notifica inviata.") else: - print("Nessuna allerta necessaria (sotto soglia).") + print(f"Nessuna neve. Casa: {home_max}cm, Area Max: {max_area_snow}cm") if __name__ == "__main__": analyze_snow()