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

Backup automatico script del 2025-12-10 20:23

Browse Source
This commit is contained in:
daniele
2025-12-10 20:23:13 +01:00
parent d06a8eddc3
commit 1ccb77c7ae
1 changed files with 88 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
services/telegram-bot/severe_weather.py
View File

@@ -14,14 +14,18 @@ TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
LAT = 43.9356
LON = 12.4296
# --- SOGLIE UFFICIALI EMILIA-ROMAGNA (Vento) ---
WIND_YELLOW = 62.0 # km/h (Burrasca Moderata)
WIND_ORANGE = 75.0 # km/h (Burrasca Forte)
WIND_RED = 88.0 # km/h (Tempesta)
# --- SOGLIE DI ALLARME ---
# Vento (Protezione Civile Emilia-Romagna)
WIND_YELLOW = 62.0 # km/h
WIND_ORANGE = 75.0 # km/h
WIND_RED = 88.0 # km/h
# Soglia Pioggia (Invariata)
# Pioggia
RAIN_3H_LIMIT = 25.0 # mm in 3 ore
# Codici WMO per Gelicidio (Freezing Rain)
FREEZING_CODES = [56, 57, 66, 67]
# File di stato
STATE_FILE = "/home/daniely/docker/telegram-bot/weather_state.json"
@@ -33,7 +37,7 @@ def send_telegram_message(message):
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)
requests.post(url, json={"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}, timeout=10)
time.sleep(0.2)
except: pass
@@ -41,7 +45,7 @@ def get_forecast():
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": LAT, "longitude": LON,
"hourly": "precipitation,windgusts_10m",
"hourly": "precipitation,windgusts_10m,weathercode",
"models": "arome_france_hd",
"timezone": "Europe/Rome",
"forecast_days": 2
@@ -57,7 +61,7 @@ def load_state():
try:
with open(STATE_FILE, 'r') as f: return json.load(f)
except: pass
return {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0}
return {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0, "freezing_active": False}
def save_state(state):
try:
@@ -65,7 +69,7 @@ def save_state(state):
except: pass
def analyze():
print("--- Controllo Meteo Severo ---")
print("--- Controllo Meteo Severo (Wind/Rain/Ice) ---")
data = get_forecast()
if not data: return
@@ -73,6 +77,7 @@ def analyze():
times = hourly.get("time", [])
wind = hourly.get("windgusts_10m", [])
rain = hourly.get("precipitation", [])
wcode = hourly.get("weathercode", [])
now = datetime.datetime.now(ZoneInfo("Europe/Rome"))
state = load_state()
@@ -93,91 +98,93 @@ def analyze():
max_wind_time = ""
sum_rain_3h = 0.0
# Cerca picco vento
freezing_rain_detected = False
freezing_time = ""
# Cerca picchi
for i in range(start_idx, end_idx):
# Vento
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)
# Gelicidio
if wcode[i] in FREEZING_CODES:
freezing_rain_detected = True
if not freezing_time: # Prendi il primo orario
freezing_time = parser.isoparse(times[i]).strftime('%H:%M')
# 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
# --- CLASSIFICAZIONE VENTO ---
wind_level_curr = 0
wind_msg_header = ""
wind_msg_desc = ""
if max_wind > WIND_RED:
wind_level_curr = 3
wind_msg_header = "🔴 **TEMPESTA (Burrasca Fortissima)**"
wind_msg_desc = f"Previste raffiche > {WIND_RED:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\n_Possibili danni a tetti e alberi._"
elif max_wind > WIND_ORANGE:
wind_level_curr = 2
wind_msg_header = "🟠 **VENTO MOLTO FORTE (Burrasca Forte)**"
wind_msg_desc = f"Previste raffiche > {WIND_ORANGE:.0f} km/h (Picco: **{max_wind:.0f}** km/h)."
elif max_wind > WIND_YELLOW:
wind_level_curr = 1
wind_msg_header = "🟡 **VENTO FORTE (Burrasca Moderata)**"
wind_msg_desc = f"Previste raffiche > {WIND_YELLOW:.0f} km/h (Picco: **{max_wind:.0f}** km/h)."
# --- LOGICA DI ALLARME ---
is_rain_alarm = sum_rain_3h > RAIN_3H_LIMIT
IS_ALARM_NOW = (wind_level_curr > 0) or is_rain_alarm
WAS_ALARM = state.get("alert_active", False)
LAST_WIND_LEVEL = state.get("wind_level", 0)
# --- CLASSIFICAZIONE ---
alerts = []
should_notify = False
# SCENARIO A: C'È UN ALLARME
if IS_ALARM_NOW:
# 1. Gestione Vento
if wind_level_curr > 0:
# Notifica se è un NUOVO allarme o se il livello è AUMENTATO (es. Giallo -> Arancione)
if not WAS_ALARM or wind_level_curr > LAST_WIND_LEVEL:
alerts.append(f"{wind_msg_header}\n{wind_msg_desc}\n🕒 Orario picco: {max_wind_time}")
should_notify = True
state["wind_level"] = wind_level_curr # Aggiorna livello salvato
state["last_wind"] = max_wind
# 2. Gestione Pioggia
if is_rain_alarm:
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 SEVERO**\n\n" + "\n\n".join(alerts)
send_telegram_message(full_msg)
state["alert_active"] = True
save_state(state)
print("Allerta inviata.")
else:
print(f"Condizioni severe stabili (Livello {wind_level_curr}). 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 completo
state = {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0}
save_state(state)
print("Allarme rientrato inviato.")
WAS_ALARM = state.get("alert_active", False)
# 1. GELO (Priorità Massima)
if freezing_rain_detected:
if not state.get("freezing_active", False):
alerts.append(f"🧊 **ALLARME GELICIDIO**\nPrevista pioggia che gela (Freezing Rain).\n🕒 Inizio previsto: {freezing_time}\n_Pericolo ghiaccio su strada!_")
should_notify = True
state["freezing_active"] = True
else:
# Aggiorna lo stato per dire "tutto ok" ma non resetta se non necessario
print(f"Situazione tranquilla. Vento Max: {max_wind:.0f} km/h")
state["freezing_active"] = False
# 2. VENTO
wind_level_curr = 0
if max_wind > WIND_RED:
wind_level_curr = 3
msg = f"🔴 **TEMPESTA (Burrasca Fortissima)**\nRaffiche > {WIND_RED:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\n🕒 Ore {max_wind_time}"
elif max_wind > WIND_ORANGE:
wind_level_curr = 2
msg = f"🟠 **VENTO MOLTO FORTE**\nRaffiche > {WIND_ORANGE:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\n🕒 Ore {max_wind_time}"
elif max_wind > WIND_YELLOW:
wind_level_curr = 1
msg = f"🟡 **VENTO FORTE**\nRaffiche > {WIND_YELLOW:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\n🕒 Ore {max_wind_time}"
if wind_level_curr > 0:
if not WAS_ALARM or wind_level_curr > state.get("wind_level", 0):
alerts.append(msg)
should_notify = True
state["wind_level"] = wind_level_curr
state["last_wind"] = max_wind
# 3. PIOGGIA
if sum_rain_3h > RAIN_3H_LIMIT:
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.")
state["last_rain"] = sum_rain_3h
should_notify = True
# --- INVIO ---
IS_ALARM_NOW = (wind_level_curr > 0) or (sum_rain_3h > RAIN_3H_LIMIT) or freezing_rain_detected
if should_notify and alerts:
full_msg = f"⚠️ **AVVISO METEO SEVERO**\n\n" + "\n\n".join(alerts)
send_telegram_message(full_msg)
state["alert_active"] = True
save_state(state)
print("Allerta inviata.")
elif WAS_ALARM and not IS_ALARM_NOW:
# Allarme rientrato
msg = f"🟢 **ALLERTA METEO RIENTRATA**\nLe condizioni sono tornate sotto le soglie di guardia."
send_telegram_message(msg)
state = {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0, "freezing_active": False}
save_state(state)
print("Allarme rientrato.")
else:
# Aggiorna stato parziale se serve (es. vento calato ma ancora presente)
if not IS_ALARM_NOW:
state["alert_active"] = False
save_state(state)
print(f"Nessun nuovo allarme. W:{max_wind:.0f} R:{sum_rain_3h:.1f} Ice:{freezing_rain_detected}")
if __name__ == "__main__":
analyze()