daniele/loogle-scripts
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Backup automatico script del 2026-07-12 07:00

Browse Source
This commit is contained in:
daniele
2026-07-12 07:00:03 +02:00
parent b5a2a814e2
commit 927f0d4184
18 changed files with 2365 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/telegram-bot/previsione7.py
+173 -50
View File
@@ -8,19 +8,32 @@ import argparse
import datetime
import os
import sys
import json
from zoneinfo import ZoneInfo
from collections import defaultdict, Counter, Counter
from typing import List, Dict, Tuple, Optional
from statistics import mean, median
from open_meteo_client import open_meteo_get
from open_meteo_precip import (
CASA_LAT,
CASA_LON,
CASA_TZ,
daily_precip_from_hourly,
fetch_icon_italia,
hourly_precip_at_index,
hourly_precip_series,
is_casa,
overlay_icon_precip_on_daily,
overlay_icon_precip_on_hourly,
)
# --- CONFIGURAZIONE DEFAULT ---
DEFAULT_LAT = 43.9356
DEFAULT_LON = 12.4296
DEFAULT_LAT = CASA_LAT
DEFAULT_LON = CASA_LON
DEFAULT_NAME = "🏠 Casa (Strada Cà Toro,12 - San Marino)"
# --- TIMEZONE ---
TZ_STR = "Europe/Berlin"
TZ_STR = CASA_TZ
TZINFO = ZoneInfo(TZ_STR)
# --- TELEGRAM CONFIG ---
@@ -34,6 +47,10 @@ TOKEN_FILE_VOLUME = "/Volumes/Pi2/etc/telegram_dpc_bot_token"
SOGLIA_VENTO_KMH = 40.0
MIN_MM_PER_EVENTO = 0.1
# Giorni mostrati in tabella / WebApp (previsione 7 giorni)
DISPLAY_FORECAST_DAYS = 7
GIORNI_ITA_SHORT = ["Lun", "Mar", "Mer", "Gio", "Ven", "Sab", "Dom"]
# --- MODELLI METEO ---
# Modelli a breve termine (alta risoluzione, 48-72h)
SHORT_TERM_MODELS = ["meteofrance_seamless", "icon_d2"] # Usa seamless invece di arome_france_hd
@@ -235,6 +252,7 @@ def get_weather_multi_model(lat, lon, short_term_models, long_term_models, forec
results[model] = None
# Recupera modelli a lungo termine (3-10d): tre modelli per mediana (come Agent Irrigazione)
short_set = set(short_term_models or [])
for model in (long_term_models or []):
url = "https://api.open-meteo.com/v1/forecast"
fd_long = LONG_TERM_FORECAST_DAYS.get(model, forecast_days)
@@ -276,13 +294,16 @@ def get_weather_multi_model(lat, lon, short_term_models, long_term_models, forec
snow_depth_cm.append(None)
hourly_data["snow_depth"] = snow_depth_cm
data["hourly"] = hourly_data
results[model] = data
results[model]["model_type"] = "long_term"
lt_key = f"{model}__long" if model in short_set else model
results[lt_key] = data
results[lt_key]["model_type"] = "long_term"
else:
results[model] = None
lt_key = f"{model}__long" if model in short_set else model
results[lt_key] = None
except Exception:
results[model] = None
lt_key = f"{model}__long" if model in short_set else model
results[lt_key] = None
return results
@@ -303,9 +324,13 @@ def _median_or_single(values):
return median(nums)
# Chiavi che esistono solo su ICON Italia (no merge, si tiene il valore da quel modello)
HOURLY_KEYS_ICON_ONLY = ["snow_depth", "showers"]
DAILY_KEYS_ICON_ONLY = ["showers_sum"]
# Chiavi solo ICON Italia (precip 0–2d: niente mediana con AROME HD a San Marino)
HOURLY_KEYS_ICON_ONLY = [
"snow_depth", "showers", "precipitation", "rain", "snowfall",
]
DAILY_KEYS_ICON_ONLY = [
"showers_sum", "precipitation_sum", "rain_sum", "snowfall_sum", "precipitation_hours",
]
def _merge_hourly_median(hourly_by_model, single_source_keys=None, single_source_model=None):
@@ -614,6 +639,23 @@ def merge_multi_model_forecast(models_data, forecast_days=10):
return merged
def format_day_label(day_index: int, daily_time_list, with_relative: bool = True) -> str:
"""Etichetta calendario per indice giorno 0-based (es. Lun 12/07)."""
if daily_time_list and 0 <= day_index < len(daily_time_list):
raw = str(daily_time_list[day_index]).split("T")[0]
try:
dt = datetime.datetime.strptime(raw, "%Y-%m-%d")
cal = f"{GIORNI_ITA_SHORT[dt.weekday()]} {dt.strftime('%d/%m')}"
if with_relative:
if day_index == 0:
return f"oggi ({cal})"
if day_index == 1:
return f"domani ({cal})"
return cal
except ValueError:
pass
return f"giorno {day_index + 1}"
def analyze_temperature_trend(daily_temps_max, daily_temps_min, days=10):
"""Analizza trend temperatura per identificare fronti caldi/freddi con dettaglio completo"""
if not daily_temps_max or not daily_temps_min:
@@ -731,7 +773,7 @@ def analyze_weather_transitions(daily_weathercodes):
if code in (95, 96, 99): return "temporale"
return "variabile"
for i in range(1, min(len(daily_weathercodes), 8)):
for i in range(1, min(len(daily_weathercodes), DISPLAY_FORECAST_DAYS)):
prev_code = daily_weathercodes[i-1] if i-1 < len(daily_weathercodes) else None
curr_code = daily_weathercodes[i] if i < len(daily_weathercodes) else None
prev_cat = get_category(prev_code)
@@ -1014,13 +1056,13 @@ def generate_practical_advice(trend, transitions, events_summary, daily_data):
return advice
def format_detailed_trend_explanation(trend, daily_data_list):
"""Genera spiegazione dettagliata del trend temperatura su 10 giorni"""
def format_detailed_trend_explanation(trend, daily_time_list=None, display_days=DISPLAY_FORECAST_DAYS):
"""Genera spiegazione dettagliata del trend temperatura sui giorni in previsione."""
if not trend:
return ""
explanation = []
explanation.append(f"📊 <b>EVOLUZIONE TEMPERATURE (10 GIORNI)</b>\n")
explanation.append(f"📊 <b>EVOLUZIONE TEMPERATURE ({display_days} GIORNI)</b>\n")
# Trend principale con spiegazione chiara
trend_type = trend["type"]
@@ -1052,13 +1094,16 @@ def format_detailed_trend_explanation(trend, daily_data_list):
explanation.append(f"{trend_desc}{intensity_text}")
explanation.append(f"{desc_text}")
# Aggiungi solo picchi significativi in modo sintetico
# Aggiungi solo picchi significativi in modo sintetico (entro i giorni in tabella)
if trend.get("change_days"):
significant_changes = [c for c in trend["change_days"] if abs(c['delta']) > 3.0][:3]
significant_changes = [
c for c in trend["change_days"]
if abs(c["delta"]) > 3.0 and c["day"] < display_days
][:3]
if significant_changes:
change_texts = []
for change in significant_changes:
day_name = f"Giorno {change['day']+1}"
day_name = format_day_label(change["day"], daily_time_list or [])
direction = "↑" if change['delta'] > 0 else "↓"
change_texts.append(f"{direction} {day_name}: {change['from']:.0f}°→{change['to']:.0f}°C")
if change_texts:
@@ -1068,7 +1113,35 @@ def format_detailed_trend_explanation(trend, daily_data_list):
return "\n".join(explanation)
def format_weather_context_report(models_data, location_name, country_code):
def _apply_unified_precip(hourly: Dict, daily: Dict, casa: bool) -> Tuple[Dict, Dict]:
"""Precip oraria/giornaliera da ICON Italia (ARPAE 2i) per Casa."""
if not casa or not hourly.get("time"):
return hourly, daily
hourly = dict(hourly)
daily = dict(daily)
icon = fetch_icon_italia(CASA_LAT, CASA_LON, CASA_TZ, forecast_days=10)
if icon:
icon_h = icon.get("hourly") or {}
icon_d = icon.get("daily") or {}
if icon_h.get("time"):
hourly = overlay_icon_precip_on_hourly(hourly, icon_h)
if icon_d.get("time"):
daily = overlay_icon_precip_on_daily(daily, icon_d)
hourly["precipitation"] = hourly_precip_series(hourly)
totals = daily_precip_from_hourly(hourly)
times = daily.get("time") or []
psum = list(daily.get("precipitation_sum") or [])
while len(psum) < len(times):
psum.append(None)
for i, t in enumerate(times):
d = str(t)[:10]
if d in totals:
psum[i] = round(totals[d], 2)
daily["precipitation_sum"] = psum
return hourly, daily
def format_weather_context_report(models_data, location_name, country_code, as_json=False):
"""Genera report contestuale intelligente con ensemble multi-modello"""
# Combina modelli a breve e lungo termine
merged_data = merge_multi_model_forecast(models_data, forecast_days=10)
@@ -1079,6 +1152,10 @@ def format_weather_context_report(models_data, location_name, country_code):
hourly = merged_data.get('hourly', {})
daily = merged_data.get('daily', {})
models_used = merged_data.get('models_used', [])
casa = country_code in ("SM", "IT")
hourly, daily = _apply_unified_precip(hourly, daily, casa)
merged_data["hourly"] = hourly
merged_data["daily"] = daily
if not daily or not daily.get('time'):
return "❌ Errore: Dati meteo incompleti"
@@ -1089,21 +1166,30 @@ def format_weather_context_report(models_data, location_name, country_code):
models_text = " + ".join(models_used) if models_used else "Multi-modello"
msg_parts.append(f"🌍 <b>METEO FORECAST</b>")
msg_parts.append(f"{location_name.upper()}")
msg_parts.append(f"📡 <i>Ensemble: {models_text}</i>\n")
msg_parts.append(f"📡 <i>Ensemble: {models_text}</i>")
if casa:
msg_parts.append(f"💧 <i>Precipitazioni 0–2g: ICON Italia (ARPAE 2i)</i>\n")
else:
msg_parts.append("")
# ANALISI TREND TEMPERATURA (Fronti) - Completa su 10 giorni
# ANALISI TREND TEMPERATURA (Fronti) — allineato ai giorni mostrati in tabella
daily_temps_max = daily.get('temperature_2m_max', [])
daily_temps_min = daily.get('temperature_2m_min', [])
trend = analyze_temperature_trend(daily_temps_max, daily_temps_min, days=10)
daily_time_list = daily.get('time', [])
trend = analyze_temperature_trend(
daily_temps_max, daily_temps_min, days=DISPLAY_FORECAST_DAYS
)
trend_explanation = ""
# Spiegazione dettagliata trend (sempre, anche se stabile)
if trend:
trend_explanation = format_detailed_trend_explanation(trend, daily_data_list=[])
trend_explanation = format_detailed_trend_explanation(
trend, daily_time_list=daily_time_list, display_days=DISPLAY_FORECAST_DAYS
)
if trend_explanation:
msg_parts.append(trend_explanation)
# ANALISI TRANSIZIONI METEO - Include anche precipitazioni prossimi giorni
daily_time_list = daily.get('time', []) # Definito qui per uso successivo
daily_weathercodes = daily.get('weathercode', [])
transitions = analyze_weather_transitions(daily_weathercodes)
@@ -1115,12 +1201,10 @@ def format_weather_context_report(models_data, location_name, country_code):
if transitions:
significant_trans = [t for t in transitions if t.get("significant", False)]
for trans in significant_trans[:5]:
day_names = ["oggi", "domani", "dopodomani", "fra 3 giorni", "fra 4 giorni", "fra 5 giorni", "fra 6 giorni"]
day_idx = trans["day"] - 1
if day_idx < len(day_names):
day_ref = day_names[day_idx]
else:
day_ref = f"fra {trans['day']} giorni"
day_idx = trans["day"]
if day_idx >= DISPLAY_FORECAST_DAYS:
continue
day_ref = format_day_label(day_idx, daily_time_list)
weather_changes.append({
"day": trans["day"],
"day_ref": day_ref,
@@ -1193,17 +1277,15 @@ def format_weather_context_report(models_data, location_name, country_code):
# Aggiungi solo se supera la soglia appropriata
if precip_amount > threshold_mm:
day_names = ["oggi", "domani", "dopodomani"]
if day_idx < len(day_names):
weather_changes.append({
"day": day_num,
"day_ref": day_names[day_idx],
"from": "variabile",
"to": "precipitazioni",
"type": "precip",
"amount": precip_amount,
"precip_symbol": precip_type_symbol
})
weather_changes.append({
"day": day_num,
"day_ref": format_day_label(day_idx, daily_time_list),
"from": "variabile",
"to": "precipitazioni",
"type": "precip",
"amount": precip_amount,
"precip_symbol": precip_type_symbol,
})
if weather_changes:
# Ordina per giorno
@@ -1227,7 +1309,7 @@ def format_weather_context_report(models_data, location_name, country_code):
temp_max_list = daily.get('temperature_2m_max', [])
# Limita ai giorni per cui abbiamo dati daily validi
max_days = min(len(daily_time_list), len(temp_min_list), len(temp_max_list), 10)
max_days = min(len(daily_time_list), len(temp_min_list), len(temp_max_list), DISPLAY_FORECAST_DAYS)
# Mappa hourly per eventi dettagliati
daily_map = defaultdict(list)
@@ -1250,7 +1332,10 @@ def format_weather_context_report(models_data, location_name, country_code):
d_times = [hourly['time'][i] for i in indices if i < len(hourly.get('time', []))]
d_codes = [hourly.get('weathercode', [])[i] for i in indices if i < len(hourly.get('weathercode', []))]
d_probs = [hourly.get('precipitation_probability', [])[i] for i in indices if i < len(hourly.get('precipitation_probability', []))]
d_precip = [hourly.get('precipitation', [])[i] for i in indices if i < len(hourly.get('precipitation', []))]
d_precip = [
hourly_precip_at_index(hourly, i)
for i in indices if i < len(hourly.get('time', []))
]
d_snow = [hourly.get('snowfall', [])[i] for i in indices if i < len(hourly.get('snowfall', []))]
d_winds = [hourly.get('windspeed_10m', [])[i] for i in indices if i < len(hourly.get('windspeed_10m', []))]
d_winddir = [hourly.get('winddirection_10m', [])[i] for i in indices if i < len(hourly.get('winddirection_10m', []))]
@@ -1342,9 +1427,7 @@ def format_weather_context_report(models_data, location_name, country_code):
events_summary.append(events_list)
dt = datetime.datetime.strptime(day_date, "%Y-%m-%d")
# Nomi giorni in italiano
giorni_ita = ["Lun", "Mar", "Mer", "Gio", "Ven", "Sab", "Dom"]
day_str = f"{giorni_ita[dt.weekday()]} {dt.strftime('%d/%m')}"
day_str = f"{GIORNI_ITA_SHORT[dt.weekday()]} {dt.strftime('%d/%m')}"
# Icona meteo principale basata sul weathercode del giorno
wcode = daily.get('weathercode', [])[count] if count < len(daily.get('weathercode', [])) else None
@@ -1662,8 +1745,39 @@ def format_weather_context_report(models_data, location_name, country_code):
# Aggiorna per il prossimo giorno
prev_snow_depth_end = snow_depth_end if snow_depth_end is not None else prev_snow_depth_end
msg_parts.append("")
if as_json:
def _serialize_day(d):
out = dict(d)
out["events"] = list(d.get("events") or [])
for k in ("t_min", "t_max", "precip_sum", "wind_max", "snowfall_sum", "rain_sum", "showers_sum"):
if out.get(k) is not None:
out[k] = round(float(out[k]), 1)
for k in ("snow_depth_min", "snow_depth_max", "snow_depth_avg", "snow_depth_end"):
if out.get(k) is not None:
out[k] = round(float(out[k]), 1)
return out
return {
"location": location_name,
"country_code": country_code,
"models": models_used,
"trend_html": trend_explanation if trend else "",
"weather_changes": weather_changes,
"days": [_serialize_day(d) for d in daily_details],
"columns": [
{"key": "day_str", "label": "Giorno"},
{"key": "weather_icon", "label": ""},
{"key": "t_min", "label": "Min°C"},
{"key": "t_max", "label": "Max°C"},
{"key": "precip_sum", "label": "Precip"},
{"key": "precip_detail", "label": "Tipo"},
{"key": "wind", "label": "Vento"},
{"key": "snow_depth_end", "label": "Manto cm"},
{"key": "events", "label": "Eventi"},
],
}
return "\n".join(msg_parts)
def send_telegram(text, chat_id, token, debug_mode=False):
@@ -1689,6 +1803,8 @@ def main():
parser.add_argument("--debug", action="store_true")
parser.add_argument("--home", action="store_true")
parser.add_argument("--timezone", help="Timezone IANA (es: Europe/Rome, America/New_York)")
parser.add_argument("--stdout", action="store_true", help="Stampa report su stdout invece di Telegram")
parser.add_argument("--json", action="store_true", help="Output JSON strutturato (WebApp)")
args = parser.parse_args()
token = get_bot_token()
@@ -1719,7 +1835,7 @@ def main():
# Recupera dati multi-modello (breve + lungo termine) - selezione intelligente basata su country code
# Determina se è Casa
is_home = (abs(lat - DEFAULT_LAT) < 0.01 and abs(lon - DEFAULT_LON) < 0.01)
is_home = is_casa(lat, lon)
# Recupera dati multi-modello (breve + lungo termine)
# - Per Casa: usa AROME Seamless e ICON-D2
@@ -1741,13 +1857,20 @@ def main():
return
# Genera report
if args.json:
payload = format_weather_context_report(models_data, name, cc, as_json=True)
print(json.dumps(payload, ensure_ascii=False))
return
report = format_weather_context_report(models_data, name, cc)
if debug_mode:
report = f"🛠 <b>[DEBUG MODE]</b> 🛠\n\n{report}"
# Invia
if token:
if args.stdout:
print(report)
elif token:
success = False
for chat_id in recipients:
if send_telegram(report, chat_id, token, debug_mode):