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

Backup automatico script del 2026-06-28 07:00

Browse Source
This commit is contained in:
daniele
2026-06-28 07:00:02 +02:00
parent d12058b2d1
commit 4377b8c9ea
1 changed files with 243 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/telegram-bot/severe_weather_circondario.py
+244 -67
View File
@@ -59,9 +59,11 @@ LOCALITA_CIRCONDARIO = [
]
# ----------------- THRESHOLDS -----------------
HOURS_AHEAD = 24 # Analisi 24 ore
HOURS_AHEAD = 48 # Analisi estesa a 48 ore (finestra previsionale completa)
IMMINENT_HOURS = 24 # Finestra "imminente": eventi entro le prossime 24 ore
# ----------------- CONVECTIVE STORM THRESHOLDS -----------------
# Soglie di RILEVAMENTO (per la scansione interna degli eventi convettivi)
CAPE_LIGHTNING_THRESHOLD = 800.0 # J/kg - Soglia per rischio fulminazioni
CAPE_SEVERE_THRESHOLD = 1500.0 # J/kg - Soglia per temporali violenti
WIND_GUST_DOWNBURST_THRESHOLD = 60.0 # km/h - Soglia vento per downburst
@@ -70,6 +72,23 @@ RAIN_INTENSE_THRESHOLD_3H = 40.0 # mm/3h - Soglia per nubifragio su 3 ore
RAIN_FLOOD_THRESHOLD_24H = 100.0 # mm/24h - Soglia per rischio alluvioni
STORM_SCORE_THRESHOLD = 40.0 # Storm Severity Score minimo per allerta
# ----------------- SOGLIE DI SIGNIFICATIVITÀ (per NOTIFICA) -----------------
# Una località viene notificata SOLO se ha almeno un evento "significativo":
# fulminazioni forti, bomba d'acqua/nubifragio forte, alluvione o downburst.
# Soglie volutamente alte per filtrare i temporali marginali estivi.
SIGNIFICANT_CAPE_LIGHTNING = 1500.0 # J/kg - fulminazioni rilevanti (oltre il marginale)
SIGNIFICANT_PRECIP_H = 30.0 # mm/h - bomba d'acqua (nubifragio forte)
SIGNIFICANT_PRECIP_3H = 50.0 # mm/3h - nubifragio forte su 3 ore
SIGNIFICANT_SCORE = 55.0 # Storm Severity Score minimo per considerare l'evento significativo
# ----------------- RATE LIMITING NOTIFICHE -----------------
# Notifiche "molto contingentate":
# - eventi imminenti (entro IMMINENT_HOURS): massimo 2 al giorno
# - eventi solo nella finestra estesa (24-48h): massimo 1 al giorno
MAX_ALERTS_IMMINENT_PER_DAY = 2
MAX_ALERTS_EXTENDED_PER_DAY = 1
MIN_GAP_HOURS_IMMINENT = 6.0 # distanza minima (ore) fra i due alert imminenti giornalieri
# ----------------- FILES -----------------
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATE_FILE = os.path.join(BASE_DIR, "weather_state_circondario.json")
@@ -213,7 +232,13 @@ def telegram_send_html(message_html: str, chat_ids: Optional[List[str]] = None)
def load_state() -> Dict:
default = {
"alert_active": False,
"locations": {}, # {location_name: {"last_score": 0.0, "last_storm_time": None}}
# Conteggi giornalieri delle notifiche inviate, per finestra:
# {"YYYY-MM-DD": {"imminent": int, "extended": int}}
"daily": {},
"last_sent_imminent": None, # ISO timestamp ultimo alert imminente
"last_sent_extended": None, # ISO timestamp ultimo alert esteso
"last_signature": None, # firma del contenuto dell'ultimo alert inviato
"last_signature_date": None, # data (YYYY-MM-DD) della firma
}
if os.path.exists(STATE_FILE):
try:
@@ -234,6 +259,130 @@ def save_state(state: Dict) -> None:
LOGGER.exception("State write error: %s", e)
# =============================================================================
# SIGNIFICATIVITÀ EVENTI (gating notifiche)
# =============================================================================
def event_significance(event: Dict) -> List[str]:
"""
Ritorna la lista dei motivi per cui un evento è "significativo" (degno di notifica).
Lista vuota = evento marginale, da NON notificare.
Significativo se almeno uno tra:
- Fulminazioni forti: CAPE >= SIGNIFICANT_CAPE_LIGHTNING e LPI > 0
- Bomba d'acqua: precip oraria >= SIGNIFICANT_PRECIP_H o 3h >= SIGNIFICANT_PRECIP_3H
- Rischio Alluvioni: già marcato in threats (accumulo 24h > soglia)
- Downburst/Temporale violento: già marcato in threats
- Storm Severity Score molto elevato (>= SIGNIFICANT_SCORE)
"""
reasons: List[str] = []
threats = event.get("threats", []) or []
cape = float(event.get("cape", 0.0) or 0.0)
lpi = float(event.get("lpi", 0.0) or 0.0)
precip = float(event.get("precip", 0.0) or 0.0)
precip_3h = float(event.get("precip_3h", 0.0) or 0.0)
score = float(event.get("score", 0.0) or 0.0)
if "Fulminazioni" in threats and cape >= SIGNIFICANT_CAPE_LIGHTNING and lpi > 0:
reasons.append("Fulminazioni forti")
if precip >= SIGNIFICANT_PRECIP_H or precip_3h >= SIGNIFICANT_PRECIP_3H:
reasons.append("Bomba d'acqua")
if "Rischio Alluvioni" in threats:
reasons.append("Rischio Alluvioni")
if "Downburst/Temporale violento" in threats:
reasons.append("Downburst")
if score >= SIGNIFICANT_SCORE and not reasons:
reasons.append("Temporale severo")
return reasons
def filter_significant_events(storm_events: List[Dict], now: datetime.datetime) -> List[Dict]:
"""Filtra gli eventi mantenendo solo quelli significativi e arricchendoli con
'lead_hours' (ore di anticipo rispetto a ora) e 'significance' (motivi)."""
significant: List[Dict] = []
for ev in storm_events or []:
reasons = event_significance(ev)
if not reasons:
continue
try:
lead = (parse_time_to_local(ev["timestamp"]) - now).total_seconds() / 3600.0
except Exception:
lead = 0.0
enriched = dict(ev)
enriched["lead_hours"] = lead
enriched["significance"] = reasons
significant.append(enriched)
return significant
# =============================================================================
# RATE LIMITING NOTIFICHE
# =============================================================================
def _today_key(now: datetime.datetime) -> str:
return now.strftime("%Y-%m-%d")
def _parse_iso_local(s: Optional[str]) -> Optional[datetime.datetime]:
if not s:
return None
try:
dt = parser.isoparse(s)
if dt.tzinfo is None:
return dt.replace(tzinfo=TZINFO)
return dt.astimezone(TZINFO)
except Exception:
return None
def prune_daily(state: Dict, now: datetime.datetime) -> None:
"""Mantiene solo i conteggi di oggi e ieri per evitare crescita illimitata."""
keep = {
_today_key(now),
_today_key(now - datetime.timedelta(days=1)),
}
daily = state.get("daily", {}) or {}
state["daily"] = {k: v for k, v in daily.items() if k in keep}
def can_notify(category: str, now: datetime.datetime, state: Dict) -> Tuple[bool, str]:
"""Verifica i limiti giornalieri per categoria ('imminent' | 'extended')."""
key = _today_key(now)
daily = state.setdefault("daily", {})
day = daily.setdefault(key, {"imminent": 0, "extended": 0})
if category == "imminent":
if int(day.get("imminent", 0)) >= MAX_ALERTS_IMMINENT_PER_DAY:
return False, "cap giornaliero imminenti raggiunto"
last = _parse_iso_local(state.get("last_sent_imminent"))
if last and (now - last).total_seconds() < MIN_GAP_HOURS_IMMINENT * 3600:
return False, "cooldown imminenti attivo"
return True, ""
# extended
if int(day.get("extended", 0)) >= MAX_ALERTS_EXTENDED_PER_DAY:
return False, "cap giornaliero estesi raggiunto"
return True, ""
def record_notify(category: str, now: datetime.datetime, state: Dict) -> None:
key = _today_key(now)
day = state.setdefault("daily", {}).setdefault(key, {"imminent": 0, "extended": 0})
day[category] = int(day.get(category, 0)) + 1
state["last_sent_" + category] = now.isoformat()
def build_signature(category: str, significant_locations: Dict[str, List[Dict]]) -> str:
"""Firma del contenuto: categoria + (località, set motivi). Cambia se la
situazione peggiora (nuova località o nuovo tipo di minaccia)."""
parts = []
for name in sorted(significant_locations.keys()):
reasons = set()
for ev in significant_locations[name]:
reasons.update(ev.get("significance", []))
parts.append(f"{name}:{'+'.join(sorted(reasons))}")
return f"{category}|" + ";".join(parts)
# =============================================================================
# OPEN-METEO
# =============================================================================
@@ -423,7 +572,7 @@ def analyze_convective_risk(icon_data: Dict, arome_data: Dict, times_base: List[
# MESSAGE FORMATTING
# =============================================================================
def format_location_alert(location_name: str, storm_events: List[Dict]) -> str:
"""Formatta alert per una singola località."""
"""Formatta alert per una singola località (riceve solo eventi significativi)."""
if not storm_events:
return ""
@@ -432,35 +581,40 @@ def format_location_alert(location_name: str, storm_events: List[Dict]) -> str:
last_time = parse_time_to_local(storm_events[-1]["timestamp"])
duration_hours = len(storm_events)
# Raggruppa minacce
all_threats = set()
# Motivi di significatività (perché è scattata la notifica)
reasons = set()
for event in storm_events:
all_threats.update(event.get("threats", []))
threats_str = ", ".join(all_threats) if all_threats else "Temporali severi"
reasons.update(event.get("significance", []))
reasons_str = ", ".join(sorted(reasons)) if reasons else "Temporale severo"
max_cape = max(e["cape"] for e in storm_events)
max_precip_h = max((e.get("precip", 0) for e in storm_events), default=0)
max_precip_24h = max((e.get("precip_24h", 0) for e in storm_events), default=0)
msg = (
f"📍 <b>{html.escape(location_name)}</b>\n"
f"📊 Score: <b>{max_score:.0f}/100</b> | {threats_str}\n"
f"📊 Score: <b>{max_score:.0f}/100</b> | <b>{html.escape(reasons_str)}</b>\n"
f"🕒 {ddmmyyhhmm(first_time)} - {ddmmyyhhmm(last_time)} (~{duration_hours}h)\n"
f"⚡ CAPE max: {max_cape:.0f} J/kg"
f"⚡ CAPE max: {max_cape:.0f} J/kg | 🌧️ Pioggia max: {max_precip_h:.1f} mm/h"
)
if max_precip_24h > RAIN_FLOOD_THRESHOLD_24H:
msg += f" | 💧 Accumulo 24h: <b>{max_precip_24h:.1f} mm</b> ⚠️"
msg += f"\n💧 <b>Accumulo 24h: {max_precip_24h:.1f} mm</b> ⚠️ rischio alluvioni"
return msg
def format_circondario_alert(locations_data: Dict[str, List[Dict]]) -> str:
"""Formatta alert aggregato per tutto il circondario."""
def format_circondario_alert(locations_data: Dict[str, List[Dict]], category: str = "imminent") -> str:
"""Formatta alert aggregato per tutto il circondario (solo eventi significativi)."""
if not locations_data:
return ""
headline = "⛈️ <b>ALLERTA TEMPORALI SEVERI - CIRCONDARIO</b>"
if category == "imminent":
headline = "⛈️ <b>ALLERTA TEMPORALI SEVERI - CIRCONDARIO (imminente ≤24h)</b>"
window_str = f"prossime {IMMINENT_HOURS} ore"
else:
headline = "⛈️ <b>ALLERTA TEMPORALI SEVERI - CIRCONDARIO (48h)</b>"
window_str = f"prossime {HOURS_AHEAD} ore"
# Statistiche aggregate
total_locations = len(locations_data)
@@ -483,9 +637,10 @@ def format_circondario_alert(locations_data: Dict[str, List[Dict]]) -> str:
period_str = "N/A"
meta = (
f"📍 <b>{total_locations} località</b> con rischio temporali severi\n"
f"📍 <b>{total_locations} località</b> con temporali significativi\n"
f"📊 <b>Storm Severity Score max:</b> <b>{max_score_overall:.0f}/100</b>\n"
f"🕒 <b>Periodo:</b> {period_str}\n"
f"🕒 <b>Periodo evento:</b> {period_str}\n"
f"🔭 <b>Finestra analisi:</b> {window_str}\n"
f"🛰️ <b>Modelli:</b> AROME Seamless + ICON Italia\n"
)
@@ -550,76 +705,98 @@ def analyze_location(location: Dict) -> Optional[List[Dict]]:
def analyze_all_locations(debug_mode: bool = False) -> None:
"""Analizza tutte le località del circondario."""
"""Analizza tutte le località del circondario con anti-spam a finestre.
Notifica SOLO eventi significativi (fulminazioni forti / bombe d'acqua /
alluvioni / downburst) e in modo contingentato:
- eventi imminenti (≤24h): max 2/giorno (con cooldown minimo)
- eventi solo estesi (24-48h): max 1/giorno
"""
LOGGER.info("=== Analisi Temporali Severi - Circondario ===")
now = now_local()
state = load_state()
prune_daily(state, now)
was_alert_active = bool(state.get("alert_active", False))
locations_with_risk = {}
# Raccoglie SOLO eventi significativi per località
significant_locations: Dict[str, List[Dict]] = {}
for location in LOCALITA_CIRCONDARIO:
name = location["name"]
storm_events = analyze_location(location)
if storm_events:
locations_with_risk[name] = storm_events
max_score = max(e["score"] for e in storm_events)
# Controlla se è un nuovo evento o peggioramento
loc_state = state.get("locations", {}).get(name, {})
prev_score = float(loc_state.get("last_score", 0.0) or 0.0)
if debug_mode or not loc_state.get("alert_sent", False) or (max_score >= prev_score + 15.0):
# Aggiorna stato
if "locations" not in state:
state["locations"] = {}
state["locations"][name] = {
"last_score": float(max_score),
"alert_sent": True,
"last_storm_time": storm_events[0]["timestamp"]
}
sig_events = filter_significant_events(storm_events, now)
if sig_events:
significant_locations[name] = sig_events
time.sleep(0.5) # Rate limiting per API
# Invia alert se ci sono località a rischio
if locations_with_risk or debug_mode:
if locations_with_risk:
msg = format_circondario_alert(locations_with_risk)
if msg:
ok = telegram_send_html(msg)
if ok:
LOGGER.info("Alert inviato per %d località", len(locations_with_risk))
else:
LOGGER.warning("Alert NON inviato (token missing o errore Telegram)")
state["alert_active"] = True
save_state(state)
elif debug_mode:
# In modalità debug, invia messaggio anche senza rischi
msg = (
"ℹ️ <b>ANALISI CIRCONDARIO - Nessun Rischio</b>\n"
f"📍 Analizzate {len(LOCALITA_CIRCONDARIO)} località\n"
f"🕒 Finestra: prossime {HOURS_AHEAD} ore\n"
"<i>Nessun temporale severo previsto nel circondario.</i>"
)
telegram_send_html(msg)
LOGGER.info("Messaggio debug inviato (nessun rischio)")
# All-clear se era attivo e ora non c'è più rischio
if was_alert_active and not locations_with_risk:
# --- Nessun evento significativo ---
if not significant_locations:
if was_alert_active:
msg = (
"🟢 <b>ALLERTA TEMPORALI SEVERI - RIENTRATA</b>\n"
"<i>Condizioni rientrate sotto le soglie di guardia per tutte le località del circondario.</i>"
)
telegram_send_html(msg)
LOGGER.info("All-clear inviato")
elif debug_mode:
msg = (
"ℹ️ <b>ANALISI CIRCONDARIO - Nessun Rischio</b>\n"
f"📍 Analizzate {len(LOCALITA_CIRCONDARIO)} località\n"
f"🕒 Finestra: prossime {HOURS_AHEAD} ore\n"
"<i>Nessun temporale significativo previsto nel circondario.</i>"
)
telegram_send_html(msg)
LOGGER.info("Messaggio debug inviato (nessun rischio)")
state["alert_active"] = False
state["locations"] = {}
state["last_signature"] = None
state["last_signature_date"] = None
save_state(state)
elif not locations_with_risk:
state["alert_active"] = False
return
# --- Determina la categoria (imminente vs esteso) ---
has_imminent = any(
any(ev.get("lead_hours", 0.0) <= IMMINENT_HOURS for ev in events)
for events in significant_locations.values()
)
category = "imminent" if has_imminent else "extended"
# --- Dedup: stessa situazione già notificata oggi -> non re-inviare ---
signature = build_signature(category, significant_locations)
today = _today_key(now)
if not debug_mode:
if state.get("last_signature") == signature and state.get("last_signature_date") == today:
LOGGER.info("Alert soppresso (dedup): situazione invariata già notificata oggi [%s]", category)
state["alert_active"] = True
save_state(state)
return
# --- Rate limiting per finestra ---
if not debug_mode:
allowed, reason = can_notify(category, now, state)
if not allowed:
LOGGER.info("Alert soppresso (rate-limit %s): %s", category, reason)
state["alert_active"] = True
save_state(state)
return
# --- Invio ---
msg = format_circondario_alert(significant_locations, category=category)
if not msg:
save_state(state)
return
ok = telegram_send_html(msg, chat_ids=[TELEGRAM_CHAT_IDS[0]] if debug_mode else None)
if ok:
LOGGER.info("Alert inviato (%s) per %d località significative", category, len(significant_locations))
else:
LOGGER.warning("Alert NON inviato (token missing o errore Telegram)")
if ok and not debug_mode:
record_notify(category, now, state)
state["last_signature"] = signature
state["last_signature_date"] = today
state["alert_active"] = True
save_state(state)