diff --git a/services/telegram-bot/severe_weather_circondario.py b/services/telegram-bot/severe_weather_circondario.py index 4cdb41b..5ac71b1 100755 --- a/services/telegram-bot/severe_weather_circondario.py +++ b/services/telegram-bot/severe_weather_circondario.py @@ -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"📍 {html.escape(location_name)}\n" - f"📊 Score: {max_score:.0f}/100 | {threats_str}\n" + f"📊 Score: {max_score:.0f}/100 | {html.escape(reasons_str)}\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: {max_precip_24h:.1f} mm ⚠️" + msg += f"\n💧 Accumulo 24h: {max_precip_24h:.1f} mm ⚠️ 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 = "⛈️ ALLERTA TEMPORALI SEVERI - CIRCONDARIO" + if category == "imminent": + headline = "⛈️ ALLERTA TEMPORALI SEVERI - CIRCONDARIO (imminente ≤24h)" + window_str = f"prossime {IMMINENT_HOURS} ore" + else: + headline = "⛈️ ALLERTA TEMPORALI SEVERI - CIRCONDARIO (48h)" + 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"📍 {total_locations} località con rischio temporali severi\n" + f"📍 {total_locations} località con temporali significativi\n" f"📊 Storm Severity Score max: {max_score_overall:.0f}/100\n" - f"🕒 Periodo: {period_str}\n" + f"🕒 Periodo evento: {period_str}\n" + f"🔭 Finestra analisi: {window_str}\n" f"🛰️ Modelli: AROME Seamless + ICON Italia\n" ) @@ -550,77 +705,99 @@ 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) + # --- Nessun evento significativo --- + if not significant_locations: + if was_alert_active: + msg = ( + "🟢 ALLERTA TEMPORALI SEVERI - RIENTRATA\n" + "Condizioni rientrate sotto le soglie di guardia per tutte le località del circondario." + ) + telegram_send_html(msg) + LOGGER.info("All-clear inviato") elif debug_mode: - # In modalità debug, invia messaggio anche senza rischi msg = ( "ℹ️ ANALISI CIRCONDARIO - Nessun Rischio\n" f"📍 Analizzate {len(LOCALITA_CIRCONDARIO)} località\n" f"🕒 Finestra: prossime {HOURS_AHEAD} ore\n" - "Nessun temporale severo previsto nel circondario." + "Nessun temporale significativo previsto nel circondario." ) telegram_send_html(msg) LOGGER.info("Messaggio debug inviato (nessun rischio)") + state["alert_active"] = False + state["last_signature"] = None + state["last_signature_date"] = None + save_state(state) + return - # All-clear se era attivo e ora non c'è più rischio - if was_alert_active and not locations_with_risk: - msg = ( - "🟢 ALLERTA TEMPORALI SEVERI - RIENTRATA\n" - "Condizioni rientrate sotto le soglie di guardia per tutte le località del circondario." - ) - telegram_send_html(msg) - LOGGER.info("All-clear inviato") - - state["alert_active"] = False - state["locations"] = {} - save_state(state) - elif not locations_with_risk: - state["alert_active"] = False + # --- 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) if __name__ == "__main__":