111 lines
3.2 KiB
Python
111 lines
3.2 KiB
Python
import requests
|
|
import datetime
|
|
from dateutil import parser
|
|
from zoneinfo import ZoneInfo
|
|
|
|
# --- CONFIGURAZIONE UTENTE ---
|
|
TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
|
|
TELEGRAM_CHAT_ID = "64463169"
|
|
|
|
# Coordinate San Marino
|
|
LATITUDE = 43.9356
|
|
LONGITUDE = 12.4296
|
|
|
|
# Soglia minima per notifica (es. 0.0 per qualsiasi nevicata)
|
|
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}")
|
|
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}")
|
|
|
|
def get_arome_forecast():
|
|
url = "https://api.open-meteo.com/v1/forecast"
|
|
params = {
|
|
"latitude": LATITUDE,
|
|
"longitude": LONGITUDE,
|
|
"hourly": "snowfall",
|
|
"models": "arome_france_hd",
|
|
"timezone": "Europe/Rome",
|
|
"forecast_days": 2
|
|
}
|
|
|
|
try:
|
|
response = requests.get(url, params=params, timeout=10)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Errore richiesta API: {e}")
|
|
return None
|
|
|
|
def analyze_snow():
|
|
data = get_arome_forecast()
|
|
if not data:
|
|
return
|
|
|
|
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
|
|
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
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
if start_index == -1:
|
|
print("Impossibile sincronizzare l'orario dei dati.")
|
|
return
|
|
|
|
# Slice dati 12h e 24h
|
|
end_12h = min(start_index + 12, len(snowfall))
|
|
end_24h = min(start_index + 24, len(snowfall))
|
|
|
|
snow_next_12h = snowfall[start_index:end_12h]
|
|
snow_next_24h = snowfall[start_index:end_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)
|
|
|
|
print(f"Check ore {now.strftime('%H:%M')}: 12h={sum_12h}cm, 24h={sum_24h}cm")
|
|
|
|
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)
|
|
else:
|
|
print("Nessuna allerta necessaria (sotto soglia).")
|
|
|
|
if __name__ == "__main__":
|
|
analyze_snow()
|