civil_protection: aggiunge allerte regionali Emilia-Romagna (Arpae)
Il bollettino nazionale Protezione Civile copre solo idraulico/idrogeologico/ temporali, quindi le allerte per temperature estreme non venivano notificate. - Aggiunta fonte regionale Arpae via API get-stato-allerta (documento valido ora + preavviso giorno successivo) per le categorie assenti dal nazionale: temperature estreme, vento, neve, stato del mare, mareggiate, pioggia che gela - Fonte regionale indipendente: parte anche se il bollettino nazionale fallisce - Firma anti-spam (dpc_state.json) estesa ai documenti regionali - Corretto il regex dei rischi su piu' parole (es. "TEMPERATURE ESTREME") Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Cursor
@@ -28,6 +28,10 @@ DEBUG = os.environ.get("DEBUG", "0").strip() == "1"
|
||||
|
||||
BULLETIN_URL = "https://mappe.protezionecivile.gov.it/it/mappe-rischi/bollettino-di-criticita/"
|
||||
|
||||
# Allerta meteo regionale Emilia-Romagna (Arpae): copre categorie assenti dal
|
||||
# bollettino nazionale (temperature estreme, vento, neve, mareggiate, ecc.).
|
||||
REGIONAL_ALERT_URL = "https://allertameteo.regione.emilia-romagna.it/o/get-stato-allerta"
|
||||
|
||||
# Chat IDs
|
||||
TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
|
||||
|
||||
@@ -48,10 +52,20 @@ TARGET_ZONES = {
|
||||
"EMR-D1": "Pianura bolognese",
|
||||
}
|
||||
|
||||
# Mappa codice zona regionale Arpae -> nome leggibile (deriva da TARGET_ZONES,
|
||||
# togliendo il prefisso "EMR-": es. EMR-D1 -> D1 "Pianura bolognese").
|
||||
REGIONAL_TARGET_ZONES = {code.split("-")[-1]: name for code, name in TARGET_ZONES.items()}
|
||||
|
||||
RISK_ICON = {
|
||||
"IDRAULICO": "💧",
|
||||
"IDROGEOLOGICO": "⛰️",
|
||||
"TEMPORALI": "⚡",
|
||||
"TEMPERATURE ESTREME": "🌡️",
|
||||
"VENTO": "🌬️",
|
||||
"NEVE": "❄️",
|
||||
"STATO DEL MARE": "🌊",
|
||||
"MAREGGIATE": "🌊",
|
||||
"PIOGGIA CHE GELA": "🧊",
|
||||
}
|
||||
|
||||
COLOR_ICON = {
|
||||
@@ -60,6 +74,25 @@ COLOR_ICON = {
|
||||
"ROSSA": "🔴",
|
||||
}
|
||||
|
||||
# Categorie di rischio prese SOLO dalla fonte regionale (quelle che il bollettino
|
||||
# nazionale non pubblica). Idraulico/idrogeologico/temporali restano gestiti dal
|
||||
# bollettino nazionale per evitare doppioni.
|
||||
REGIONAL_RISK_LABELS = {
|
||||
"temperature_estreme": "TEMPERATURE ESTREME",
|
||||
"vento": "VENTO",
|
||||
"neve": "NEVE",
|
||||
"stato_mare": "STATO DEL MARE",
|
||||
"mareggiate": "MAREGGIATE",
|
||||
"ghiaccio_pioggia_gela": "PIOGGIA CHE GELA",
|
||||
}
|
||||
|
||||
# Codici colore Arpae -> etichette interne (green/null = nessuna allerta).
|
||||
REGIONAL_COLOR_MAP = {
|
||||
"yellow": "GIALLA",
|
||||
"orange": "ARANCIONE",
|
||||
"red": "ROSSA",
|
||||
}
|
||||
|
||||
HTTP_HEADERS = {"User-Agent": "rpi-dpc-bollettino/1.2"}
|
||||
|
||||
# =============================================================================
|
||||
@@ -222,7 +255,7 @@ def html_to_lines(html_text: str) -> list[str]:
|
||||
# =============================================================================
|
||||
|
||||
RISK_HEADER_RE = re.compile(
|
||||
r"^(ORDINARIA|MODERATA|ELEVATA)\s+CRITICITA'\s+PER\s+RISCHIO\s+([A-ZÀ-Ü]+)\s*/\s*ALLERTA\s+(GIALLA|ARANCIONE|ROSSA)\s*:\s*$"
|
||||
r"^(ORDINARIA|MODERATA|ELEVATA)\s+CRITICITA'\s+PER\s+RISCHIO\s+([A-ZÀ-Ü]+(?:\s+[A-ZÀ-Ü]+)*)\s*/\s*ALLERTA\s+(GIALLA|ARANCIONE|ROSSA)\s*:\s*$"
|
||||
)
|
||||
DAY_START_RE = re.compile(r"^Per la giornata di (oggi|domani),\s*(.+?):\s*$", re.IGNORECASE)
|
||||
TITLE_RE = re.compile(r"^Bollettino di Criticità del (.+?) ore (\d{1,2}:\d{2})", re.IGNORECASE)
|
||||
@@ -293,8 +326,66 @@ def has_any_alert(parsed: dict) -> bool:
|
||||
day = parsed.get("days", {}).get(day_key, {})
|
||||
if day.get("alerts"):
|
||||
return True
|
||||
if parsed.get("regional"):
|
||||
return True
|
||||
return False
|
||||
|
||||
# =============================================================================
|
||||
# Allerta regionale Emilia-Romagna (Arpae)
|
||||
# =============================================================================
|
||||
|
||||
def fetch_regional_state(data: Optional[str] = None) -> dict:
|
||||
"""Stato allertamento regionale. `data` opzionale: 'YYYY-MM-DD HH:mm'."""
|
||||
params = {"data": data} if data else None
|
||||
r = requests.get(REGIONAL_ALERT_URL, params=params, headers=HTTP_HEADERS, timeout=25)
|
||||
r.raise_for_status()
|
||||
r.encoding = "utf-8"
|
||||
return r.json()
|
||||
|
||||
def parse_regional_doc(data: dict) -> dict:
|
||||
"""Estrae le allerte (solo categorie REGIONAL_RISK_LABELS) per le zone target."""
|
||||
out = {"titolo": (data.get("titolo") or "").strip(), "alerts": {}}
|
||||
for zone_code, zone_name in REGIONAL_TARGET_ZONES.items():
|
||||
zinfo = data.get(zone_code)
|
||||
if not isinstance(zinfo, dict):
|
||||
continue
|
||||
for risk_key, risk_label in REGIONAL_RISK_LABELS.items():
|
||||
color = zinfo.get(risk_key)
|
||||
color_name = REGIONAL_COLOR_MAP.get((color or "").strip().lower())
|
||||
if not color_name:
|
||||
continue
|
||||
icon = COLOR_ICON.get(color_name, "⚪")
|
||||
ricon = RISK_ICON.get(risk_label, "⚠️")
|
||||
entry = f"{icon} {ricon} {risk_label}"
|
||||
out["alerts"].setdefault(zone_name, [])
|
||||
if entry not in out["alerts"][zone_name]:
|
||||
out["alerts"][zone_name].append(entry)
|
||||
return out
|
||||
|
||||
def fetch_regional_documents() -> list:
|
||||
"""Documento valido ora + documento del giorno successivo (preavviso).
|
||||
|
||||
Restituisce solo i documenti che contengono allerte nelle zone target,
|
||||
deduplicati per titolo. Non solleva eccezioni: in caso di errore logga e
|
||||
restituisce la lista dei documenti ottenuti finora.
|
||||
"""
|
||||
docs = []
|
||||
seen_titoli = set()
|
||||
tomorrow = (now_rome() + datetime.timedelta(days=1)).strftime("%Y-%m-%d 12:00")
|
||||
for data in (None, tomorrow):
|
||||
try:
|
||||
j = fetch_regional_state(data)
|
||||
except Exception as e:
|
||||
LOGGER.exception("Errore fetch allerta regionale (data=%s): %s", data, e)
|
||||
continue
|
||||
doc = parse_regional_doc(j)
|
||||
if doc["titolo"] in seen_titoli:
|
||||
continue
|
||||
seen_titoli.add(doc["titolo"])
|
||||
if doc["alerts"]:
|
||||
docs.append(doc)
|
||||
return docs
|
||||
|
||||
def build_signature(parsed: dict) -> str:
|
||||
parts = [parsed.get("issued_date", ""), parsed.get("issued_time", ""), parsed.get("title", "")]
|
||||
for day_key in ("oggi", "domani"):
|
||||
@@ -303,6 +394,11 @@ def build_signature(parsed: dict) -> str:
|
||||
alerts = day.get("alerts", {})
|
||||
for zone in sorted(alerts.keys()):
|
||||
parts.append(zone + "=" + ",".join(sorted(alerts[zone])))
|
||||
for doc in parsed.get("regional", []):
|
||||
parts.append("REG:" + doc.get("titolo", ""))
|
||||
alerts = doc.get("alerts", {})
|
||||
for zone in sorted(alerts.keys()):
|
||||
parts.append(zone + "=" + ",".join(sorted(alerts[zone])))
|
||||
return "|".join(parts)
|
||||
|
||||
def format_message(parsed: dict) -> str:
|
||||
@@ -329,6 +425,19 @@ def format_message(parsed: dict) -> str:
|
||||
lines.append(html_lib.escape(entry))
|
||||
lines.append("")
|
||||
|
||||
regional = parsed.get("regional", [])
|
||||
for doc in regional:
|
||||
titolo = doc.get("titolo") or "Allerta meteo Emilia-Romagna"
|
||||
lines.append(f"🗺️ <b>{html_lib.escape(titolo)}</b>")
|
||||
alerts = doc.get("alerts", {})
|
||||
for zone in sorted(alerts.keys()):
|
||||
lines.append(f"📍 <b>{html_lib.escape(zone)}</b>")
|
||||
for entry in alerts[zone]:
|
||||
lines.append(html_lib.escape(entry))
|
||||
lines.append("")
|
||||
|
||||
if regional:
|
||||
lines.append("<i>Fonte: allertameteo.regione.emilia-romagna.it</i>")
|
||||
lines.append("<i>Fonte: mappe.protezionecivile.gov.it</i>")
|
||||
return "\n".join(lines)
|
||||
|
||||
@@ -339,13 +448,17 @@ def format_message(parsed: dict) -> str:
|
||||
def main(chat_ids: Optional[List[str]] = None, debug_mode: bool = False):
|
||||
LOGGER.info("--- Controllo Protezione Civile (Bollettino ufficiale) ---")
|
||||
|
||||
parsed = {"title": "", "issued_date": "", "issued_time": "", "days": {}}
|
||||
try:
|
||||
lines = fetch_bulletin_text_lines()
|
||||
parsed = parse_bulletin(lines)
|
||||
except Exception as e:
|
||||
# NESSUN Telegram in caso di errori: solo log.
|
||||
LOGGER.exception("Errore durante fetch/parse bollettino: %s", e)
|
||||
return
|
||||
# Bollettino nazionale non disponibile: prosegue comunque con la fonte
|
||||
# regionale (sotto). Nessun Telegram per il solo errore nazionale.
|
||||
LOGGER.exception("Errore durante fetch/parse bollettino nazionale: %s", e)
|
||||
|
||||
# Fonte regionale Arpae (indipendente: gestisce internamente i propri errori).
|
||||
parsed["regional"] = fetch_regional_documents()
|
||||
|
||||
if DEBUG:
|
||||
LOGGER.debug("title=%s", parsed.get("title", ""))
|
||||
@@ -353,6 +466,8 @@ def main(chat_ids: Optional[List[str]] = None, debug_mode: bool = False):
|
||||
d = parsed.get("days", {}).get(k, {})
|
||||
LOGGER.debug("%s label=%s", k, d.get("date_label", ""))
|
||||
LOGGER.debug("%s alerts=%s", k, d.get("alerts", {}))
|
||||
for doc in parsed.get("regional", []):
|
||||
LOGGER.debug("regionale titolo=%s alerts=%s", doc.get("titolo", ""), doc.get("alerts", {}))
|
||||
|
||||
# Regola: invia Telegram SOLO se esistono allerte (tranne in debug)
|
||||
if not has_any_alert(parsed):
|
||||
|
||||
Reference in New Issue
Block a user