diff --git a/services/telegram-bot/civil_protection.py b/services/telegram-bot/civil_protection.py
index 92583d8..d781d47 100644
--- a/services/telegram-bot/civil_protection.py
+++ b/services/telegram-bot/civil_protection.py
@@ -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"🗺️ {html_lib.escape(titolo)}")
+ alerts = doc.get("alerts", {})
+ for zone in sorted(alerts.keys()):
+ lines.append(f"📍 {html_lib.escape(zone)}")
+ for entry in alerts[zone]:
+ lines.append(html_lib.escape(entry))
+ lines.append("")
+
+ if regional:
+ lines.append("Fonte: allertameteo.regione.emilia-romagna.it")
lines.append("Fonte: mappe.protezionecivile.gov.it")
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):