diff --git a/services/telegram-bot/freeze_alert.py b/services/telegram-bot/freeze_alert.py index 7af2c08..d2620c3 100644 --- a/services/telegram-bot/freeze_alert.py +++ b/services/telegram-bot/freeze_alert.py @@ -1,134 +1,376 @@ -import requests +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + import datetime +import html import json +import logging import os import time -from dateutil import parser +from logging.handlers import RotatingFileHandler +from typing import Dict, Optional, Tuple from zoneinfo import ZoneInfo -# --- CONFIGURAZIONE UTENTE --- -TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4" -TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"] +import requests +from dateutil import parser -# --- COORDINATE (Strada Cà Toro, 12 - San Marino) --- +# ============================================================================= +# FREEZE ALERT (next 48h) - San Marino (Casa) +# +# Scopo: +# Notificare su Telegram se nelle prossime 48 ore è prevista una temperatura +# minima < SOGLIA_GELO. +# +# Requisiti operativi applicati (come gli altri script): +# - Nessun token in chiaro (legge da env o file /etc/telegram_dpc_bot_token) +# - Log su file + modalità DEBUG (DEBUG=1) +# - Timezone robusta (naive -> Europe/Rome; offset -> conversione) +# - Nessun Telegram in caso di errori (solo log) +# - Anti-spam: notifica solo su nuovo evento o variazione significativa +# +# Esecuzione: +# DEBUG=1 python3 freeze_alert.py +# tail -n 200 freeze_alert.log +# ============================================================================= + +DEBUG = os.environ.get("DEBUG", "0").strip() == "1" + +# ----------------- TELEGRAM ----------------- +TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"] +TOKEN_FILE_HOME = os.path.expanduser("~/.telegram_dpc_bot_token") +TOKEN_FILE_ETC = "/etc/telegram_dpc_bot_token" + +# ----------------- LOCATION ----------------- LAT = 43.9356 LON = 12.4296 LOCATION_NAME = "🏠 Casa (Strada Cà Toro)" -# Soglia Gelo (°C) -SOGLIA_GELO = 0.0 +# ----------------- THRESHOLD ----------------- +SOGLIA_GELO = 0.0 # °C (allerta se min < 0.0°C) -# File di stato +# ----------------- HORIZON ----------------- +HOURS_AHEAD = 48 +FORECAST_DAYS = 3 # per coprire bene 48h + +# ----------------- TIMEZONE ----------------- +TZ = "Europe/Rome" +TZINFO = ZoneInfo(TZ) + +# ----------------- FILES ----------------- STATE_FILE = "/home/daniely/docker/telegram-bot/freeze_state.json" +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +LOG_FILE = os.path.join(BASE_DIR, "freeze_alert.log") -def load_state(): +# ----------------- OPEN-METEO ----------------- +OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast" +HTTP_HEADERS = {"User-Agent": "rpi-freeze-alert/2.0"} + + +# ============================================================================= +# LOGGING +# ============================================================================= +def setup_logger() -> logging.Logger: + logger = logging.getLogger("freeze_alert") + logger.setLevel(logging.DEBUG if DEBUG else logging.INFO) + logger.handlers.clear() + + fh = RotatingFileHandler(LOG_FILE, maxBytes=1_000_000, backupCount=5, encoding="utf-8") + fh.setLevel(logging.DEBUG) + fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s") + fh.setFormatter(fmt) + logger.addHandler(fh) + + if DEBUG: + sh = logging.StreamHandler() + sh.setLevel(logging.DEBUG) + sh.setFormatter(fmt) + logger.addHandler(sh) + + return logger + + +LOGGER = setup_logger() + + +# ============================================================================= +# UTILS +# ============================================================================= +def now_local() -> datetime.datetime: + return datetime.datetime.now(TZINFO) + + +def read_text_file(path: str) -> str: + try: + with open(path, "r", encoding="utf-8") as f: + return f.read().strip() + except FileNotFoundError: + return "" + except PermissionError: + LOGGER.debug("Permission denied reading %s", path) + return "" + except Exception as e: + LOGGER.exception("Error reading %s: %s", path, e) + return "" + + +def load_bot_token() -> str: + tok = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip() + if tok: + return tok + tok = read_text_file(TOKEN_FILE_HOME) + if tok: + return tok + tok = read_text_file(TOKEN_FILE_ETC) + return tok.strip() if tok else "" + + +def parse_time_to_local(t: str) -> datetime.datetime: + """ + Parsing robusto: + - se naive (frequente con timezone=Europe/Rome) -> interpreta come Europe/Rome + - se con offset -> converte in Europe/Rome + """ + dt = parser.isoparse(t) + if dt.tzinfo is None: + return dt.replace(tzinfo=TZINFO) + return dt.astimezone(TZINFO) + + +def fmt_dt(dt: datetime.datetime) -> str: + return dt.strftime("%d/%m alle %H:%M") + + +# ============================================================================= +# TELEGRAM +# ============================================================================= +def telegram_send_html(message_html: str) -> bool: + """ + Non solleva eccezioni. Ritorna True se almeno un invio ha successo. + IMPORTANTE: chiamare solo per allerte (mai per errori). + """ + token = load_bot_token() + if not token: + LOGGER.warning("Telegram token missing: message not sent.") + return False + + url = f"https://api.telegram.org/bot{token}/sendMessage" + base_payload = { + "text": message_html, + "parse_mode": "HTML", + "disable_web_page_preview": True, + } + + sent_ok = False + with requests.Session() as s: + for chat_id in TELEGRAM_CHAT_IDS: + payload = dict(base_payload) + payload["chat_id"] = chat_id + try: + resp = s.post(url, json=payload, timeout=15) + if resp.status_code == 200: + sent_ok = True + else: + LOGGER.error("Telegram error chat_id=%s status=%s body=%s", + chat_id, resp.status_code, resp.text[:500]) + time.sleep(0.25) + except Exception as e: + LOGGER.exception("Telegram exception chat_id=%s err=%s", chat_id, e) + + return sent_ok + + +# ============================================================================= +# STATE +# ============================================================================= +def load_state() -> Dict: + default = { + "alert_active": False, + "min_temp": 100.0, + "min_time": "", + "signature": "", + "updated": "", + } if os.path.exists(STATE_FILE): try: - with open(STATE_FILE, 'r') as f: return json.load(f) - except: pass - return {"alert_active": False, "min_temp": 100.0} + with open(STATE_FILE, "r", encoding="utf-8") as f: + data = json.load(f) or {} + default.update(data) + except Exception as e: + LOGGER.exception("State read error: %s", e) + return default -def save_state(active, min_temp): + +def save_state(state: Dict) -> None: try: - with open(STATE_FILE, 'w') as f: - json.dump({"alert_active": active, "min_temp": min_temp, "updated": str(datetime.datetime.now())}, f) - except: pass + os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True) + state["updated"] = now_local().isoformat() + with open(STATE_FILE, "w", encoding="utf-8") as f: + json.dump(state, f, ensure_ascii=False, indent=2) + except Exception as e: + LOGGER.exception("State write error: %s", e) -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 get_forecast(): - url = "https://api.open-meteo.com/v1/forecast" +# ============================================================================= +# OPEN-METEO +# ============================================================================= +def get_forecast() -> Optional[Dict]: params = { - "latitude": LAT, "longitude": LON, + "latitude": LAT, + "longitude": LON, "hourly": "temperature_2m", - "timezone": "Europe/Rome", - "forecast_days": 3 # Prendiamo 3 giorni per coprire bene le 48h + "timezone": TZ, + "forecast_days": FORECAST_DAYS, } try: - r = requests.get(url, params=params, timeout=10) + r = requests.get(OPEN_METEO_URL, params=params, headers=HTTP_HEADERS, timeout=25) + if r.status_code == 400: + try: + j = r.json() + LOGGER.error("Open-Meteo 400: %s", j.get("reason", j)) + except Exception: + LOGGER.error("Open-Meteo 400: %s", r.text[:500]) + return None r.raise_for_status() return r.json() - except: return None + except Exception as e: + LOGGER.exception("Open-Meteo request error: %s", e) + return None -def analyze_freeze(): - print("--- Controllo Gelo ---") - data = get_forecast() - if not data: return - hourly = data.get("hourly", {}) - times = hourly.get("time", []) - temps = hourly.get("temperature_2m", []) - - now = datetime.datetime.now(ZoneInfo("Europe/Rome")) - limit_time = now + datetime.timedelta(hours=48) - +def compute_min_next_48h(data: Dict) -> Optional[Tuple[float, datetime.datetime]]: + hourly = data.get("hourly", {}) or {} + times = hourly.get("time", []) or [] + temps = hourly.get("temperature_2m", []) or [] + + n = min(len(times), len(temps)) + if n == 0: + return None + + now = now_local() + limit_time = now + datetime.timedelta(hours=HOURS_AHEAD) + min_temp_val = 100.0 - min_temp_time = None - - # Cerca la minima nelle prossime 48 ore - for i, t_str in enumerate(times): - t_obj = parser.isoparse(t_str).replace(tzinfo=ZoneInfo("Europe/Rome")) - - # Filtra solo futuro prossimo (da adesso a +48h) - if t_obj > now and t_obj <= limit_time: - temp = temps[i] - if temp < min_temp_val: - min_temp_val = temp - min_temp_time = t_obj + min_temp_time: Optional[datetime.datetime] = None + + for i in range(n): + try: + t_obj = parse_time_to_local(times[i]) + except Exception: + continue + + # solo intervallo (now, now+48h] + if t_obj <= now or t_obj > limit_time: + continue + + try: + temp = float(temps[i]) + except Exception: + continue + + if temp < min_temp_val: + min_temp_val = temp + min_temp_time = t_obj + + if min_temp_time is None: + return None + + return float(min_temp_val), min_temp_time + + +# ============================================================================= +# MAIN +# ============================================================================= +def analyze_freeze() -> None: + LOGGER.info("--- Controllo Gelo (next %sh) ---", HOURS_AHEAD) + + data = get_forecast() + if not data: + # errori: solo log + return + + result = compute_min_next_48h(data) + if not result: + LOGGER.error("Impossibile calcolare minima nelle prossime %s ore.", HOURS_AHEAD) + return + + min_temp_val, min_temp_time = result + is_freezing = (min_temp_val < SOGLIA_GELO) - # --- LOGICA ALLARME --- state = load_state() - was_active = state.get("alert_active", False) - - # C'è rischio gelo? - is_freezing = min_temp_val < SOGLIA_GELO - - if is_freezing: - # Formatta orario - time_str = min_temp_time.strftime('%d/%m alle %H:%M') - - # SCENARIO A: NUOVO GELO (o peggioramento significativo di 2 gradi) - if not was_active or min_temp_val < state.get("min_temp", 0) - 2.0: - msg = ( - f"❄️ **ALLERTA GELO**\n" - f"📍 {LOCATION_NAME}\n\n" - f"Prevista temperatura minima di **{min_temp_val:.1f}°C**\n" - f"📅 Quando: {time_str}\n\n" - f"_Proteggere piante e tubature esterne._" - ) - send_telegram_message(msg) - save_state(True, min_temp_val) - print(f"Allerta inviata: {min_temp_val}°C") - else: - print(f"Gelo già notificato ({min_temp_val}°C).") - # Aggiorniamo comunque la minima registrata nel file - save_state(True, min(min_temp_val, state.get("min_temp", 100))) + was_active = bool(state.get("alert_active", False)) + last_sig = str(state.get("signature", "")) - # SCENARIO B: ALLARME RIENTRATO - elif was_active and not is_freezing: + # firma per evitare spam: temp (0.1) + timestamp + sig = f"{min_temp_val:.1f}|{min_temp_time.isoformat()}" + + if is_freezing: + # invia se: + # - prima non era attivo, oppure + # - peggiora di almeno 2°C, oppure + # - cambia la firma (es. orario minima spostato o min diversa) + prev_min = float(state.get("min_temp", 100.0) or 100.0) + + should_notify = (not was_active) or (min_temp_val < prev_min - 2.0) or (sig != last_sig) + + if should_notify: + msg = ( + "❄️ ALLERTA GELO
" + f"📍 {html.escape(LOCATION_NAME)}

" + f"Minima prevista (entro {HOURS_AHEAD}h): {min_temp_val:.1f}°C
" + f"📅 Quando: {html.escape(fmt_dt(min_temp_time))}

" + "Proteggere piante e tubature esterne." + ) + ok = telegram_send_html(msg) + if ok: + LOGGER.info("Allerta gelo inviata. Tmin=%.1f°C at %s", min_temp_val, min_temp_time.isoformat()) + else: + LOGGER.warning("Allerta gelo NON inviata (token mancante o errore Telegram).") + + else: + LOGGER.info("Gelo già notificato (invariato o peggioramento < 2°C). Tmin=%.1f°C", min_temp_val) + + state.update({ + "alert_active": True, + "min_temp": min_temp_val, + "min_time": min_temp_time.isoformat(), + "signature": sig, + }) + save_state(state) + return + + # --- RIENTRO --- + if was_active and not is_freezing: msg = ( - f"☀️ **RISCHIO GELO RIENTRATO**\n" - f"📍 {LOCATION_NAME}\n\n" - f"Le previsioni per le prossime 48 ore indicano temperature sopra lo zero.\n" - f"Minima prevista: {min_temp_val:.1f}°C." + "☀️ RISCHIO GELO RIENTRATO
" + f"📍 {html.escape(LOCATION_NAME)}

" + f"Le previsioni nelle prossime {HOURS_AHEAD} ore indicano temperature sopra lo zero.
" + f"Minima prevista: {min_temp_val:.1f}°C (alle {html.escape(fmt_dt(min_temp_time))})." ) - send_telegram_message(msg) - save_state(False, min_temp_val) - print("Allarme rientrato.") - - else: - save_state(False, min_temp_val) - print(f"Nessun gelo. Minima: {min_temp_val}°C") + ok = telegram_send_html(msg) + if ok: + LOGGER.info("Rientro gelo notificato. Tmin=%.1f°C", min_temp_val) + else: + LOGGER.warning("Rientro gelo NON inviato (token mancante o errore Telegram).") + + state.update({ + "alert_active": False, + "min_temp": min_temp_val, + "min_time": min_temp_time.isoformat(), + "signature": "", + }) + save_state(state) + return + + # --- TRANQUILLO --- + state.update({ + "alert_active": False, + "min_temp": min_temp_val, + "min_time": min_temp_time.isoformat(), + "signature": "", + }) + save_state(state) + LOGGER.info("Nessun gelo. Tmin=%.1f°C at %s", min_temp_val, min_temp_time.isoformat()) + if __name__ == "__main__": analyze_freeze()