Backup automatico script del 2025-12-02 22:16

services/telegram-bot/civil_protection.py

@@ -0,0 +1,151 @@
+import requests
+import json
+import os
+import time
+import datetime
+from zoneinfo import ZoneInfo
+
+# --- CONFIGURAZIONE UTENTE ---
+# 👇👇 INSERISCI QUI I TUOI DATI 👇👇
+TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
+TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
+
+# --- ZONE DA MONITORARE ---
+# EMR-B2 = Costa Romagnola (Rimini/Dogana)
+# EMR-A2 = Alta Collina Romagnola (San Marino/Titano)
+# EMR-D1 = Pianura Bolognese (Bologna Città)
+TARGET_ZONES = ["EMR-B2", "EMR-A2", "EMR-D1"]
+
+# URL Ufficiale DPC
+DPC_URL = "https://raw.githubusercontent.com/pcm-dpc/DPC-Bollettini-Criticita-Idrogeologica-Idraulica/master/files/geojson/today.json"
+
+# File di stato
+STATE_FILE = "/home/daniely/docker/telegram-bot/dpc_state.json"
+
+# Mappe
+RISK_MAP = {
+    1: "🟡 GIALLA",
+    2: "🟠 ARANCIONE",
+    3: "🔴 ROSSA"
+}
+
+RISK_TYPES = {
+    "idro": "💧 Idraulico",
+    "idrogeo": "⛰️ Idrogeologico",
+    "temporali": "⚡ Temporali",
+    "vento": "💨 Vento",
+    "neve": "❄️ Neve",
+    "ghiaccio": "🧊 Ghiaccio",
+    "mare": "🌊 Mareggiate"
+}
+
+def get_zone_label(zone_id):
+    if zone_id == "EMR-B2": return "Rimini / Bassa RSM"
+    if zone_id == "EMR-A2": return "Alta RSM / Carpegna"
+    if zone_id == "EMR-D1": return "Bologna Città"
+    return zone_id
+
+def send_telegram_message(message):
+    if not TELEGRAM_BOT_TOKEN or "INSERISCI" in TELEGRAM_BOT_TOKEN:
+        print(f"[TEST OUT] {message}")
+        return
+
+    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=10)
+            time.sleep(0.2)
+        except: pass
+
+def load_last_alert():
+    if os.path.exists(STATE_FILE):
+        try:
+            with open(STATE_FILE, 'r') as f: return json.load(f)
+        except: pass
+    return {}
+
+def save_current_alert(data):
+    try:
+        with open(STATE_FILE, 'w') as f: json.dump(data, f)
+    except: pass
+
+def analyze_dpc():
+    print("--- Controllo Protezione Civile ---")
+    
+    try:
+        r = requests.get(DPC_URL, timeout=10)
+        r.raise_for_status()
+        data = r.json()
+    except Exception as e:
+        print(f"Errore DPC: {e}")
+        return
+
+    today_str = datetime.datetime.now(ZoneInfo("Europe/Rome")).strftime('%d/%m/%Y')
+    
+    active_alerts = {}
+    max_global_level = 0
+    
+    # Stringa univoca per identificare se l'allerta è cambiata nel contenuto
+    current_alert_signature = ""
+
+    for feature in data['features']:
+        props = feature['properties']
+        zone_id = props.get('zone_id')
+        
+        if zone_id in TARGET_ZONES:
+            zone_risks = []
+            
+            for key, label in RISK_TYPES.items():
+                level = props.get(f"crit_{key}")
+                
+                if level and level > 0:
+                    if level > max_global_level: max_global_level = level
+                    
+                    color_icon = "🟡" if level == 1 else "🟠" if level == 2 else "🔴"
+                    risk_str = f"{color_icon} {label}"
+                    zone_risks.append(risk_str)
+            
+            if zone_risks:
+                label = get_zone_label(zone_id)
+                active_alerts[label] = zone_risks
+                # Aggiungiamo alla firma per capire se qualcosa è cambiato
+                current_alert_signature += f"{zone_id}:{','.join(zone_risks)}|"
+
+    # --- LOGICA DI INVIO ---
+    
+    last_state = load_last_alert()
+    last_date = last_state.get("date")
+    last_sig = last_state.get("signature", "")
+
+    # Se tutto verde (livello 0)
+    if max_global_level == 0:
+        print("Nessuna allerta.")
+        if last_date == today_str and last_sig != "":
+             # Opzionale: Potremmo mandare "Allerta Rientrata", ma DPC resetta a mezzanotte.
+             # Per ora resettiamo solo lo stato.
+             save_current_alert({"date": today_str, "level": 0, "signature": ""})
+        return
+
+    # Invia SE:
+    # 1. È un giorno nuovo
+    # 2. OPPURE la situazione è cambiata (es. aggiunto Bologna, o passato da Giallo a Rosso)
+    if last_date != today_str or current_alert_signature != last_sig:
+        
+        msg = f"📢 **PROTEZIONE CIVILE (Allerta)**\n"
+        msg += f"📅 {today_str}\n\n"
+        
+        for zone_name, risks in active_alerts.items():
+            msg += f"📍 **{zone_name}**\n"
+            msg += "\n".join(risks) + "\n\n"
+            
+        msg += "_Fonte: Dipartimento Protezione Civile_"
+        
+        send_telegram_message(msg)
+        print("Allerta inviata.")
+        
+        save_current_alert({"date": today_str, "level": max_global_level, "signature": current_alert_signature})
+    else:
+        print("Allerta già notificata e invariata.")
+
+if __name__ == "__main__":
+    analyze_dpc()