From 658b7c13cbf310fc7f534ee2081e51e11e611da3 Mon Sep 17 00:00:00 2001 From: daniele Date: Tue, 2 Dec 2025 21:39:13 +0100 Subject: [PATCH] Backup automatico script del 2025-12-02 21:39 --- services/telegram-bot/arome_snow_alert.py | 99 ++++++++++++-- services/telegram-bot/severe_weather.py | 2 +- services/telegram-bot/student_alert.py | 153 ++++++++++++++++++++++ 3 files changed, 239 insertions(+), 15 deletions(-) create mode 100644 services/telegram-bot/student_alert.py diff --git a/services/telegram-bot/arome_snow_alert.py b/services/telegram-bot/arome_snow_alert.py index 3b3d799..7d6591b 100644 --- a/services/telegram-bot/arome_snow_alert.py +++ b/services/telegram-bot/arome_snow_alert.py @@ -1,26 +1,61 @@ import requests import datetime import time +import json +import os from dateutil import parser from zoneinfo import ZoneInfo # --- CONFIGURAZIONE UTENTE --- # 👇👇 INSERISCI QUI I TUOI DATI 👇👇 TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4" -TELEGRAM_CHAT_IDS = ["64463169", "132455422"] +TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"] # --- PUNTI DI MONITORAGGIO --- -# Il primo punto DEVE essere Casa tua +# Sostituito San Leo con Carpegna 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} + {"name": "🏔️ Carpegna", "lat": 43.7819, "lon": 12.3346} ] # Soglia notifica (cm) SOGLIA_NOTIFICA = 0.0 +# File di stato per ricordare l'ultima allerta +STATE_FILE = "/home/daniely/docker/telegram-bot/snow_state.json" + +# --- FUNZIONI DI UTILITÀ --- + +def is_winter_season(): + """Ritorna True se oggi è tra il 1 Novembre e il 15 Aprile""" + now = datetime.datetime.now() + month = now.month + day = now.day + + if month >= 11: return True # Nov, Dic + if month <= 3: return True # Gen, Feb, Mar + if month == 4 and day <= 15: return True # Fino al 15 Apr + return False + +def load_last_state(): + """Legge se c'era un allerta attiva""" + if not os.path.exists(STATE_FILE): return False + try: + with open(STATE_FILE, 'r') as f: + data = json.load(f) + return data.get("alert_active", False) + except: return False + +def save_current_state(is_active): + """Salva lo stato corrente""" + try: + with open(STATE_FILE, 'w') as f: + json.dump({"alert_active": is_active, "updated": str(datetime.datetime.now())}, f) + except Exception as e: + print(f"Errore salvataggio stato: {e}") + def send_telegram_message(message): if not TELEGRAM_BOT_TOKEN or "INSERISCI" in TELEGRAM_BOT_TOKEN: print(f"[TEST OUT] {message}") @@ -38,8 +73,7 @@ def send_telegram_message(message): def get_forecast(lat, lon): url = "https://api.open-meteo.com/v1/forecast" params = { - "latitude": lat, - "longitude": lon, + "latitude": lat, "longitude": lon, "hourly": "snowfall", "models": "arome_france_hd", "timezone": "Europe/Rome", @@ -75,14 +109,25 @@ def calculate_sums(data): 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) - - return {"3h": s3, "6h": s6, "12h": s12, "24h": s24} + # Calcola somme sugli orizzonti temporali + def get_sum(hours): + return sum(x for x in snow[start_idx:min(start_idx+hours, end)] if x) + + return { + "3h": get_sum(3), + "6h": get_sum(6), + "12h": get_sum(12), + "24h": get_sum(24) + } + +# --- LOGICA PRINCIPALE --- def analyze_snow(): + # 1. Controllo Stagionale + if not is_winter_season(): + print("Stagione estiva. Script in pausa.") + return + now_str = datetime.datetime.now(ZoneInfo("Europe/Rome")).strftime('%H:%M') print(f"--- Check Meteo {now_str} ---") @@ -90,6 +135,7 @@ def analyze_snow(): max_area_snow = 0.0 area_details = "" + # 2. Raccolta Dati for p in POINTS: data = get_forecast(p["lat"], p["lon"]) stats = calculate_sums(data) @@ -99,18 +145,26 @@ def analyze_snow(): if p["name"] == "🏠 Casa": home_stats = stats + # Aggiorna il massimo rilevato in zona if stats["24h"] > max_area_snow: max_area_snow = stats["24h"] - # Aggiungi ai dettagli area se c'è neve (>0) + # Costruisci dettaglio se c'è neve if stats["24h"] > 0: area_details += f"{p['name']}: {stats['24h']:.1f}cm (12h: {stats['12h']:.1f})\n" time.sleep(1) + # 3. Decisione Alert + # C'è neve se a casa o nei dintorni l'accumulo è > soglia home_max = home_stats["24h"] if home_stats else 0.0 + SNOW_DETECTED = (home_max > SOGLIA_NOTIFICA or max_area_snow > SOGLIA_NOTIFICA) - if home_max > SOGLIA_NOTIFICA or max_area_snow > SOGLIA_NOTIFICA: + # Leggi stato precedente + WAS_ACTIVE = load_last_state() + + # --- SCENARIO A: C'È NEVE (Nuova o Continua) --- + if SNOW_DETECTED: def f(v): return f"**{v:.1f}**" if v > 0 else f"{v:.1f}" @@ -130,9 +184,26 @@ def analyze_snow(): msg += "🌍 Nessuna neve rilevante nei dintorni." send_telegram_message(msg) + save_current_state(True) # Salva che l'allerta è attiva print("Neve rilevata. Notifica inviata.") + + # --- SCENARIO B: ALLARME RIENTRATO (Neve 0, ma prima c'era) --- + elif not SNOW_DETECTED and WAS_ACTIVE: + msg = ( + f"🟢 **PREVISIONE NEVE ANNULLATA**\n" + f"📅 _Aggiornamento ore {now_str}_\n\n" + f"Le ultime previsioni AROME non indicano più accumuli nevosi rilevanti nelle prossime 24 ore.\n" + f"Situazione tornata alla normalità." + ) + send_telegram_message(msg) + save_current_state(False) # Resetta lo stato + print("Allarme rientrato. Notifica inviata.") + + # --- SCENARIO C: TUTTO TRANQUILLO (E lo era anche prima) --- else: - print(f"Nessuna neve. Casa: {home_max}cm, Area Max: {max_area_snow}cm") + # Aggiorna timestamp ma mantieni false + save_current_state(False) + print(f"Nessuna neve. Casa: {home_max}cm, Area: {max_area_snow}cm") if __name__ == "__main__": analyze_snow() diff --git a/services/telegram-bot/severe_weather.py b/services/telegram-bot/severe_weather.py index d7d71e8..d22099d 100644 --- a/services/telegram-bot/severe_weather.py +++ b/services/telegram-bot/severe_weather.py @@ -8,7 +8,7 @@ from zoneinfo import ZoneInfo # --- CONFIGURAZIONE --- TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4" -TELEGRAM_CHAT_IDS = ["64463169", "132455422"] +TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"] # Coordinate (San Marino) LAT = 43.9356 diff --git a/services/telegram-bot/student_alert.py b/services/telegram-bot/student_alert.py new file mode 100644 index 0000000..43562d6 --- /dev/null +++ b/services/telegram-bot/student_alert.py @@ -0,0 +1,153 @@ +import requests +import datetime +import time +from dateutil import parser +from zoneinfo import ZoneInfo + +# --- CONFIGURAZIONE UTENTE --- +TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4" +TELEGRAM_CHAT_IDS = ["64463169", "132455422"] + +# --- SOGLIE DI ALLARME (Bologna) --- +SOGLIA_NEVE = 0.0 # cm (Basta che nevichi per attivare) +SOGLIA_PIOGGIA_3H = 15.0 # mm in 3 ore (Pioggia molto forte/Bomba d'acqua) + +# --- PUNTI DEL PERCORSO --- +# Il primo punto deve essere BOLOGNA (Trigger) +POINTS = [ + {"name": "🎓 Bologna (V. Regnoli)", "lat": 44.4930, "lon": 11.3690, "type": "trigger"}, + {"name": "🏎️ Imola", "lat": 44.3590, "lon": 11.7130, "type": "route"}, + {"name": "ceramica Faenza", "lat": 44.2900, "lon": 11.8800, "type": "route"}, + {"name": "✈️ Forlì", "lat": 44.2220, "lon": 12.0410, "type": "route"}, + {"name": "🛣️ Cesena", "lat": 44.1390, "lon": 12.2430, "type": "route"}, + {"name": "🏖️ Rimini", "lat": 44.0600, "lon": 12.5600, "type": "route"}, + {"name": "🏠 San Marino", "lat": 43.9356, "lon": 12.4296, "type": "end"} +] + +def send_telegram_message(message): + if "INSERISCI" in TELEGRAM_BOT_TOKEN: return + url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" + for chat_id in TELEGRAM_CHAT_IDS: + try: + requests.post(url, json={"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}, timeout=10) + time.sleep(0.2) + except: pass + +def get_forecast(lat, lon): + url = "https://api.open-meteo.com/v1/forecast" + params = { + "latitude": lat, "longitude": lon, + "hourly": "precipitation,snowfall", + "models": "arome_france_hd", + "timezone": "Europe/Rome", + "forecast_days": 2 + } + try: + res = requests.get(url, params=params, timeout=5) + res.raise_for_status() + return res.json() + except: return None + +def get_stats(data): + if not data: return None + hourly = data.get("hourly", {}) + times = hourly.get("time", []) + snow = hourly.get("snowfall", []) + rain = hourly.get("precipitation", []) + + now = datetime.datetime.now(ZoneInfo("Europe/Rome")) + + # Indice ora corrente + start_idx = -1 + for i, t in enumerate(times): + if parser.isoparse(t).replace(tzinfo=ZoneInfo("Europe/Rome")) >= now.replace(minute=0,second=0,microsecond=0): + start_idx = i + break + if start_idx == -1: return None + + limit = min(start_idx + 24, len(times)) # Analisi max 24h + + # Calcolo finestre temporali + def sum_slice(arr, hours): + return sum(x for x in arr[start_idx:min(start_idx+hours, limit)] if x) + + return { + "snow_3h": sum_slice(snow, 3), + "snow_6h": sum_slice(snow, 6), + "snow_12h": sum_slice(snow, 12), + "snow_24h": sum_slice(snow, 24), + "rain_3h": sum_slice(rain, 3), # Per bomba d'acqua + "rain_max": max(rain[start_idx:limit]) if rain else 0 # Picco mm/h + } + +def main(): + print("--- Analisi Studente Bologna ---") + + # 1. ANALISI BOLOGNA (Il Trigger) + bo_point = POINTS[0] + bo_data = get_forecast(bo_point["lat"], bo_point["lon"]) + bo_stats = get_stats(bo_data) + + if not bo_stats: return + + # Controlla se scatta l'allarme + alarm_snow = bo_stats["snow_24h"] > SOGLIA_NEVE + alarm_rain = bo_stats["rain_3h"] > SOGLIA_PIOGGIA_3H + + if not (alarm_snow or alarm_rain): + print("Bologna tranquilla. Nessun report.") + return + + # --- C'È ALLERTA: GENERIAMO IL REPORT --- + + now_str = datetime.datetime.now(ZoneInfo("Europe/Rome")).strftime('%H:%M') + icon_main = "❄️" if alarm_snow else "🌧️" + + msg = f"{icon_main} **ALLERTA METEO BOLOGNA**\n" + msg += f"📅 _Aggiornamento ore {now_str}_\n\n" + + # Dettaglio Bologna + msg += f"🎓 **A BOLOGNA:**\n" + if alarm_snow: + msg += f"• Neve 3h: **{bo_stats['snow_3h']:.1f}** cm\n" + msg += f"• Neve 6h: **{bo_stats['snow_6h']:.1f}** cm\n" + msg += f"• Neve 12h: **{bo_stats['snow_12h']:.1f}** cm\n" + msg += f"• Neve 24h: **{bo_stats['snow_24h']:.1f}** cm\n" + + if alarm_rain: + msg += f"• Pioggia 3h: **{bo_stats['rain_3h']:.1f}** mm (Intensa!)\n" + msg += f"• Picco orario: {bo_stats['rain_max']:.1f} mm/h\n" + + msg += "\n🚗 **SITUAZIONE RIENTRO (A14):**\n" + + # 2. ANALISI PERCORSO (Loop sugli altri punti) + route_issues = False + + for p in POINTS[1:]: + stats = get_stats(get_forecast(p["lat"], p["lon"])) + if not stats: continue + + # Format della riga percorso + # Mostriamo solo se c'è qualcosa di significativo (>0 neve o >5mm pioggia) + has_snow = stats["snow_24h"] > 0 + has_rain = stats["rain_3h"] > 5.0 + + if has_snow or has_rain: + route_issues = True + line = f"**{p['name']}**: " + if has_snow: line += f"❄️ {stats['snow_12h']:.1f}cm (12h) " + if has_rain: line += f"🌧️ {stats['rain_3h']:.1f}mm " + msg += f"{line}\n" + else: + # Se è pulito, mettiamo un check leggero per far capire che è OK + # msg += f"{p['name']}: ✅ OK\n" + pass + + if not route_issues: + msg += "✅ Il percorso di ritorno sembra pulito." + + send_telegram_message(msg) + print("Allerta inviata.") + +if __name__ == "__main__": + main()