daniele/loogle-scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backup automatico script del 2025-12-01 17:14

Browse Source
This commit is contained in:
daniele
2025-12-01 17:15:00 +01:00
parent 273b92fc38
commit 4904465587
1 changed files with 90 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

152
services/telegram-bot/arome_snow_alert.py
View File

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