loogle-scripts/services/telegram-bot/severe_weather.py

import requests
import datetime
import os
import json
import time
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"]

# Coordinate (San Marino)
LAT = 43.9356
LON = 12.4296

# SOGLIE DI ALLARME
WIND_LIMIT_WARN = 60.0  # km/h (Vento Forte)
WIND_LIMIT_CRIT = 90.0  # km/h (Burrasca)
RAIN_3H_LIMIT   = 30.0  # mm in 3 ore (Rischio idrogeologico)

# 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

    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=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",
        "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 {"alert_active": False, "last_wind": 0, "last_rain": 0, "time": 0}

def save_state(state):
    try:
        with open(STATE_FILE, 'w') as f: json.dump(state, f)
    except: pass

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()
    
    # 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
    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 (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 = []

    # 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

        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).")

    # 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()