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

Backup automatico script del 2026-01-04 07:00

Browse Source
This commit is contained in:
daniele
2026-01-04 07:00:03 +01:00
parent 272e3cf0a5
commit 2859b95dbc
8 changed files with 880 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

352
services/telegram-bot/meteo.py Normal file
View File

@@ -0,0 +1,352 @@
#!/usr/bin/env python3
import requests
import datetime
import argparse
import sys
import logging
from zoneinfo import ZoneInfo
from dateutil import parser as date_parser # pyright: ignore[reportMissingModuleSource]
# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# --- CONFIGURAZIONE METEO ---
HOME_LAT = 43.9356
HOME_LON = 12.4296
HOME_NAME = "🏠 Casa (Wide View ±12km)"
TZ = "Europe/Rome"
TZINFO = ZoneInfo(TZ)
# Offset ~12-15km
OFFSET_LAT = 0.12
OFFSET_LON = 0.16
OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
GEOCODING_URL = "https://geocoding-api.open-meteo.com/v1/search"
HTTP_HEADERS = {"User-Agent": "loogle-bot-v10.4"}
def now_local() -> datetime.datetime:
return datetime.datetime.now(TZINFO)
def parse_time(t: str) -> datetime.datetime:
try:
dt = date_parser.isoparse(t)
if dt.tzinfo is None: return dt.replace(tzinfo=TZINFO)
return dt.astimezone(TZINFO)
except Exception as e:
logger.error(f"Time parse error: {e}")
return now_local()
def degrees_to_cardinal(d: int) -> str:
dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]
try:
return dirs[round(d / 45) % 8]
except: return "N"
def get_icon_set(prec, snow, code, is_day, cloud, vis, temp, rain, gust, cape, cloud_type):
sky = "☁️"
try:
if cloud_type == 'F':
sky = "🌫️"
elif code in (95, 96, 99): sky = "⛈️" if prec > 0 else "🌩️"
elif prec >= 0.1: sky = "🌨️" if snow > 0 else "🌧️"
else:
# LOGICA PERCEZIONE UMANA
if cloud_type == 'H':
if cloud <= 40: sky = "☀️" if is_day else "🌙"
elif cloud <= 80: sky = "🌤️" if is_day else "🌙"
else: sky = "🌥️"
else:
if cloud <= 15: sky = "☀️" if is_day else "🌙"
elif cloud <= 35: sky = "🌤️" if is_day else "🌙"
elif cloud <= 60: sky = "⛅️"
elif cloud <= 85: sky = "🌥️"
else: sky = "☁️"
sgx = "-"
if snow > 0 or (code is not None and code in (71,73,75,77,85,86)): sgx = "☃️"
elif temp < 0 or (code is not None and code in (66,67)): sgx = "🧊"
elif cape > 2000: sgx = "🌪️"
elif cape > 1000: sgx = "⚡"
elif temp > 35: sgx = "🥵"
elif rain > 4: sgx = "☔️"
elif gust > 50: sgx = "💨"
return sky, sgx
except Exception as e:
logger.error(f"Icon error: {e}")
return "❓", "-"
def get_coordinates(city_name: str):
params = {"name": city_name, "count": 1, "language": "it", "format": "json"}
try:
r = requests.get(GEOCODING_URL, params=params, headers=HTTP_HEADERS, timeout=10)
data = r.json()
if "results" in data and data["results"]:
res = data["results"][0]
cc = res.get("country_code", "IT").upper()
name = f"{res.get('name')} ({cc})"
return res["latitude"], res["longitude"], name, cc
except Exception as e:
logger.error(f"Geocoding error: {e}")
return None
def choose_best_model(lat, lon, cc):
if cc == 'JP': return "jma_msm", "JMA MSM"
if cc in ['NO', 'SE', 'FI', 'DK', 'IS']: return "metno_nordic", "Yr.no"
if cc in ['GB', 'IE']: return "ukmo_global", "UK MetOffice"
if cc == 'IT' or cc == 'SM': return "meteofrance_arome_france_hd", "AROME HD"
if cc in ['DE', 'AT', 'CH', 'LI', 'FR']: return "icon_d2", "ICON-D2"
return "gfs_global", "NOAA GFS"
def get_forecast(lat, lon, model):
lats = [lat, lat + OFFSET_LAT, lat - OFFSET_LAT, lat, lat]
lons = [lon, lon, lon, lon + OFFSET_LON, lon - OFFSET_LON]
lat_str = ",".join(map(str, lats))
lon_str = ",".join(map(str, lons))
params = {
"latitude": lat_str, "longitude": lon_str, "timezone": TZ,
"forecast_days": 3,
"models": model,
"wind_speed_unit": "kmh", "precipitation_unit": "mm",
"hourly": "temperature_2m,apparent_temperature,relative_humidity_2m,cloud_cover,cloud_cover_low,cloud_cover_mid,cloud_cover_high,windspeed_10m,winddirection_10m,windgusts_10m,precipitation,rain,snowfall,weathercode,is_day,cape,visibility,uv_index"
}
try:
r = requests.get(OPEN_METEO_URL, params=params, headers=HTTP_HEADERS, timeout=25)
if r.status_code != 200:
logger.error(f"API Error {r.status_code}: {r.text}")
return None
return r.json()
except Exception as e:
logger.error(f"Request error: {e}")
return None
def safe_get_list(hourly_data, key, length, default=None):
if key in hourly_data and hourly_data[key] is not None:
return hourly_data[key]
return [default] * length
def get_val(val, default=0.0):
if val is None: return default
return float(val)
def generate_weather_report(lat, lon, location_name, debug_mode=False, cc="IT") -> str:
model_id, model_name = choose_best_model(lat, lon, cc)
data_list = get_forecast(lat, lon, model_id)
if not data_list: return f"❌ Errore API Meteo ({model_name})."
if not isinstance(data_list, list): data_list = [data_list]
data_center = data_list[0]
hourly_c = data_center.get("hourly", {})
times = hourly_c.get("time", [])
if not times: return "❌ Dati orari mancanti."
L = len(times)
# --- DATI LOCALI (CASA) ---
l_temp = safe_get_list(hourly_c, "temperature_2m", L, 0)
l_app = safe_get_list(hourly_c, "apparent_temperature", L, 0)
l_rh = safe_get_list(hourly_c, "relative_humidity_2m", L, 50)
l_prec = safe_get_list(hourly_c, "precipitation", L, 0)
l_rain = safe_get_list(hourly_c, "rain", L, 0)
l_snow = safe_get_list(hourly_c, "snowfall", L, 0)
l_wspd = safe_get_list(hourly_c, "windspeed_10m", L, 0)
l_gust = safe_get_list(hourly_c, "windgusts_10m", L, 0)
l_wdir = safe_get_list(hourly_c, "winddirection_10m", L, 0)
l_code = safe_get_list(hourly_c, "weathercode", L, 0)
l_day = safe_get_list(hourly_c, "is_day", L, 1)
l_cape = safe_get_list(hourly_c, "cape", L, 0)
l_vis = safe_get_list(hourly_c, "visibility", L, 10000)
l_uv = safe_get_list(hourly_c, "uv_index", L, 0)
# Estraggo anche i dati nuvole LOCALI per il tipo
l_cl_low_loc = safe_get_list(hourly_c, "cloud_cover_low", L, 0)
l_cl_mid_loc = safe_get_list(hourly_c, "cloud_cover_mid", L, 0)
l_cl_hig_loc = safe_get_list(hourly_c, "cloud_cover_high", L, 0)
# --- DATI GLOBALI (MEDIA) ---
acc_cl_tot = [0.0] * L
points_cl_tot = [ [] for _ in range(L) ]
p_names = ["Casa", "Nord", "Sud", "Est", "Ovest"]
for d in data_list:
h = d.get("hourly", {})
for i in range(L):
cc = get_val(safe_get_list(h, "cloud_cover", L)[i])
cl = get_val(safe_get_list(h, "cloud_cover_low", L)[i])
cm = get_val(safe_get_list(h, "cloud_cover_mid", L)[i])
ch = get_val(safe_get_list(h, "cloud_cover_high", L)[i])
real_point_total = max(cc, cl, cm, ch)
acc_cl_tot[i] += real_point_total
points_cl_tot[i].append(real_point_total)
num_points = len(data_list)
avg_cl_tot = [x / num_points for x in acc_cl_tot]
if debug_mode:
output = f"🔍 **DEBUG 5 PUNTI (V10.4)**\n"
now_h = now_local().replace(minute=0, second=0, microsecond=0)
idx = 0
for i, t_str in enumerate(times):
if parse_time(t_str) >= now_h:
idx = i
break
# Valori Locali
loc_L = get_val(l_cl_low_loc[idx])
loc_H = get_val(l_cl_hig_loc[idx])
output += f"Ora: {parse_time(times[idx]).strftime('%H:%M')}\n"
output += f"📍 **LOCALE (Casa)**: L:{int(loc_L)}% | M:{int(get_val(l_cl_mid_loc[idx]))}% | H:{int(loc_H)}%\n"
output += f"🌍 **MEDIA GLOBALE**: {int(avg_cl_tot[idx])}%\n"
decision = "H"
if loc_L > 40: decision = "L (Priorità Locale)"
output += f"👉 **Decisione**: {decision}\n"
return output
now = now_local().replace(minute=0, second=0, microsecond=0)
blocks = []
header = f"{'LT':<2} {'T°':>4} {'h%':>3} {'mm':>3} {'Vento':<5} {'Nv%':>5} {'Sk':<2} {'Sx':<2}"
separator = "-" * 31
for (label, hours_duration, step) in [("Prime 24h", 24, 1), ("Successive 24h", 24, 2)]:
end_time = now + datetime.timedelta(hours=hours_duration)
lines = [header, separator]
count = 0
for i, t_str in enumerate(times):
try:
dt = parse_time(t_str)
if dt < now or dt >= end_time: continue
if dt.hour % step != 0: continue
T = get_val(l_temp[i], 0)
App = get_val(l_app[i], 0)
Rh = int(get_val(l_rh[i], 50))
t_suffix = ""
diff = App - T
if diff <= -2.5: t_suffix = "W"
elif diff >= 2.5: t_suffix = "H"
t_s = f"{int(round(T))}{t_suffix}"
Pr = get_val(l_prec[i], 0)
Sn = get_val(l_snow[i], 0)
Code = int(l_code[i]) if l_code[i] is not None else 0
p_suffix = ""
if Code in [96, 99]: p_suffix = "G"
elif Code in [66, 67]: p_suffix = "Z"
elif Sn > 0 or Code in [71, 73, 75, 77, 85, 86]: p_suffix = "N"
p_s = "--" if Pr < 0.2 else f"{int(round(Pr))}{p_suffix}"
# --- CLOUD LOGIC (V10.4: LOCAL PRIORITY) ---
# Usiamo la MEDIA per la quantità (Panoramica)
c_avg_tot = int(avg_cl_tot[i])
# Usiamo i dati LOCALI per il tipo (Cosa ho sulla testa)
loc_L = get_val(l_cl_low_loc[i])
loc_M = get_val(l_cl_mid_loc[i])
loc_H = get_val(l_cl_hig_loc[i])
Vis = get_val(l_vis[i], 10000)
# Step 1: Default matematico LOCALE
types = {'L': loc_L, 'M': loc_M, 'H': loc_H}
dominant_type = max(types, key=types.get)
# Quantità da mostrare: Media Globale
Cl = c_avg_tot
# Step 2: Override Tattico LOCALE
# Se LOCALMENTE le basse sono > 40%, vincono loro.
# (Soglia abbassata a 40 per catturare il 51%)
if loc_L > 40:
dominant_type = 'L'
# Se localmente è nuvoloso basso, forziamo la copertura visiva alta
# anche se la media globale è più bassa
if Cl < loc_L: Cl = int(loc_L)
# Step 3: Nebbia (F)
is_fog = False
if Vis < 2000 or Code in [45, 48]:
is_fog = True
elif Rh >= 96 and loc_L > 40:
is_fog = True
if is_fog:
dominant_type = 'F'
if Cl < 100: Cl = 100
# Check varianza spaziale
min_p = min(points_cl_tot[i])
max_p = max(points_cl_tot[i])
var_symbol = ""
if (max_p - min_p) > 20:
var_symbol = "~"
cl_str = f"{var_symbol}{Cl}{dominant_type}"
UV = get_val(l_uv[i], 0)
uv_suffix = ""
if UV >= 10: uv_suffix = "E"
elif UV >= 7: uv_suffix = "H"
Wspd = get_val(l_wspd[i], 0)
Gust = get_val(l_gust[i], 0)
Wdir = int(get_val(l_wdir[i], 0))
Cape = get_val(l_cape[i], 0)
IsDay = int(l_day[i]) if l_day[i] is not None else 1
card = degrees_to_cardinal(Wdir)
w_val = Gust if (Gust - Wspd) > 15 else Wspd
w_txt = f"{card} {int(round(w_val))}"
if (Gust - Wspd) > 15:
g_txt = f"G{int(round(w_val))}"
if len(card)+len(g_txt) <= 5: w_txt = f"{card}{g_txt}"
elif len(card)+1+len(g_txt) <= 5: w_txt = f"{card} {g_txt}"
else: w_txt = g_txt
w_fmt = f"{w_txt:<5}"
sky, sgx = get_icon_set(Pr, Sn, Code, IsDay, Cl, Vis, T, get_val(l_rain[i], 0), Gust, Cape, dominant_type)
sky_fmt = f"{sky}{uv_suffix}"
lines.append(f"{dt.strftime('%H'):<2} {t_s:>4} {Rh:>3} {p_s:>3} {w_fmt} {cl_str:>5} {sky_fmt:<2} {sgx:<2}")
count += 1
except Exception as e:
logger.error(f"Errore riga meteo {i}: {e}")
continue
if count > 0:
day_label = f"{['Lun','Mar','Mer','Gio','Ven','Sab','Dom'][now.weekday()]} {now.day}"
blocks.append(f"*{day_label} ({label})*\n```text\n" + "\n".join(lines) + "\n```")
now = end_time
return f"🌤️ *METEO REPORT*\n📍 {location_name}\n🧠 Fonte: {model_name}\n\n" + "\n\n".join(blocks)
if __name__ == "__main__":
args_parser = argparse.ArgumentParser()
args_parser.add_argument("--query", help="Nome città")
args_parser.add_argument("--home", action="store_true", help="Usa coordinate casa")
args_parser.add_argument("--debug", action="store_true", help="Mostra i valori dei 5 punti")
args = args_parser.parse_args()
if args.home:
print(generate_weather_report(HOME_LAT, HOME_LON, HOME_NAME, args.debug, "SM"))
elif args.query:
coords = get_coordinates(args.query)
if coords:
lat, lon, name, cc = coords
print(generate_weather_report(lat, lon, name, args.debug, cc))
else:
print(f"❌ Città '{args.query}' non trovata.")
else:
print("Uso: meteo.py --query 'Nome Città' oppure --home [--debug]")