154 lines
5.4 KiB
Python
154 lines
5.4 KiB
Python
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()
|