Backup automatico script del 2025-12-10 19:48
This commit is contained in:
134
services/telegram-bot/freeze_alert.py
Normal file
134
services/telegram-bot/freeze_alert.py
Normal file
@@ -0,0 +1,134 @@
|
||||
import requests
|
||||
import datetime
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from dateutil import parser
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
# --- CONFIGURAZIONE UTENTE ---
|
||||
TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
|
||||
TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
|
||||
|
||||
# --- COORDINATE (Strada Cà Toro, 12 - San Marino) ---
|
||||
LAT = 43.9356
|
||||
LON = 12.4296
|
||||
LOCATION_NAME = "🏠 Casa (Strada Cà Toro)"
|
||||
|
||||
# Soglia Gelo (°C)
|
||||
SOGLIA_GELO = 0.0
|
||||
|
||||
# File di stato
|
||||
STATE_FILE = "/home/daniely/docker/telegram-bot/freeze_state.json"
|
||||
|
||||
def load_state():
|
||||
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}
|
||||
|
||||
def save_state(active, min_temp):
|
||||
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
|
||||
|
||||
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"
|
||||
params = {
|
||||
"latitude": LAT, "longitude": LON,
|
||||
"hourly": "temperature_2m",
|
||||
"timezone": "Europe/Rome",
|
||||
"forecast_days": 3 # Prendiamo 3 giorni per coprire bene le 48h
|
||||
}
|
||||
try:
|
||||
r = requests.get(url, params=params, timeout=10)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
except: 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)
|
||||
|
||||
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
|
||||
|
||||
# --- 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)))
|
||||
|
||||
# SCENARIO B: ALLARME RIENTRATO
|
||||
elif 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."
|
||||
)
|
||||
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")
|
||||
|
||||
if __name__ == "__main__":
|
||||
analyze_freeze()
|
||||
Reference in New Issue
Block a user