++ Primo Caricamento

web/cron_helper.py

@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+# ============================================================
+# cron_helper.py - Parser di espressioni cron minimale
+# ============================================================
+# Calcola i secondi mancanti alla prossima esecuzione basata
+# su un'espressione cron standard a 5 campi.
+#
+# Formato: minuto ora giorno_mese mese giorno_settimana
+#
+# Supporta:
+#   *        = ogni valore
+#   */N      = ogni N unità (step)
+#   N        = valore esatto
+#   N,M      = lista di valori
+#   N-M      = range di valori
+#   N-M/S    = range con step
+#
+# Uso:
+#   python3 cron_helper.py "0 3 * * *"          → secondi fino alle 3:00
+#   python3 cron_helper.py "*/30 * * * *" next   → timestamp prossima esecuzione
+#   python3 cron_helper.py "0 3 * * *" human     → descrizione leggibile
+# ============================================================
+
+import sys
+from datetime import datetime, timedelta
+
+
+def parse_cron_field(field, min_val, max_val):
+    """
+    Parsa un singolo campo dell'espressione cron e ritorna
+    l'insieme dei valori validi.
+
+    Esempi:
+        "*"     → {0,1,2,...,max_val}
+        "*/15"  → {0,15,30,45} (per i minuti)
+        "1,15"  → {1,15}
+        "1-5"   → {1,2,3,4,5}
+        "1-10/2" → {1,3,5,7,9}
+    """
+    values = set()
+
+    for part in field.split(","):
+        if "/" in part:
+            # Gestione step: */N o range/N
+            range_part, step = part.split("/", 1)
+            step = int(step)
+
+            if range_part == "*":
+                start, end = min_val, max_val
+            elif "-" in range_part:
+                start, end = map(int, range_part.split("-", 1))
+            else:
+                start, end = int(range_part), max_val
+
+            values.update(range(start, end + 1, step))
+
+        elif "-" in part:
+            # Range: N-M
+            start, end = map(int, part.split("-", 1))
+            values.update(range(start, end + 1))
+
+        elif part == "*":
+            # Wildcard: tutti i valori
+            values.update(range(min_val, max_val + 1))
+
+        else:
+            # Valore singolo
+            values.add(int(part))
+
+    return values
+
+
+def parse_cron_expression(expression):
+    """
+    Parsa un'espressione cron a 5 campi e ritorna i set di valori validi.
+
+    Campi: minuto(0-59) ora(0-23) giorno_mese(1-31) mese(1-12) giorno_settimana(0-6, 0=dom)
+    """
+    parts = expression.strip().split()
+    if len(parts) != 5:
+        raise ValueError(
+            f"Espressione cron non valida: '{expression}'. "
+            f"Servono 5 campi: minuto ora giorno_mese mese giorno_settimana"
+        )
+
+    minutes = parse_cron_field(parts[0], 0, 59)
+    hours = parse_cron_field(parts[1], 0, 23)
+    days = parse_cron_field(parts[2], 1, 31)
+    months = parse_cron_field(parts[3], 1, 12)
+    weekdays = parse_cron_field(parts[4], 0, 6)
+
+    return minutes, hours, days, months, weekdays
+
+
+def next_cron_time(expression, now=None):
+    """
+    Calcola il prossimo istante che corrisponde all'espressione cron.
+    Ritorna un oggetto datetime.
+
+    Cerca fino a 366 giorni nel futuro (copre un anno intero + margine).
+    """
+    if now is None:
+        now = datetime.now()
+
+    minutes, hours, days, months, weekdays = parse_cron_expression(expression)
+
+    # Parti dal minuto successivo (non eseguire "adesso")
+    candidate = now.replace(second=0, microsecond=0) + timedelta(minutes=1)
+
+    # Cerca nei prossimi 366 giorni (massimo ~527040 minuti)
+    max_iterations = 366 * 24 * 60
+    for _ in range(max_iterations):
+        if (candidate.month in months and
+            candidate.day in days and
+            candidate.weekday() in _convert_weekdays(weekdays) and
+            candidate.hour in hours and
+            candidate.minute in minutes):
+            return candidate
+        candidate += timedelta(minutes=1)
+
+    raise ValueError(f"Nessuna esecuzione trovata per '{expression}' nei prossimi 366 giorni")
+
+
+def _convert_weekdays(cron_weekdays):
+    """
+    Converte i giorni della settimana dal formato cron (0=domenica)
+    al formato Python (0=lunedì).
+
+    Cron:   0=dom, 1=lun, 2=mar, 3=mer, 4=gio, 5=ven, 6=sab
+    Python: 0=lun, 1=mar, 2=mer, 3=gio, 4=ven, 5=sab, 6=dom
+    """
+    mapping = {0: 6, 1: 0, 2: 1, 3: 2, 4: 3, 5: 4, 6: 5}
+    return {mapping[d] for d in cron_weekdays}
+
+
+def seconds_until_next(expression):
+    """Calcola i secondi mancanti alla prossima esecuzione cron."""
+    now = datetime.now()
+    next_time = next_cron_time(expression, now)
+    delta = (next_time - now).total_seconds()
+    return max(1, int(delta))
+
+
+def human_readable(expression):
+    """
+    Genera una descrizione leggibile in italiano dell'espressione cron.
+    Copre i casi più comuni; per espressioni complesse ritorna il formato raw.
+    """
+    parts = expression.strip().split()
+    if len(parts) != 5:
+        return expression
+
+    m, h, dom, mon, dow = parts
+
+    # Casi comuni
+    if m == "*" and h == "*" and dom == "*" and mon == "*" and dow == "*":
+        return "Ogni minuto"
+
+    if m.startswith("*/") and h == "*" and dom == "*" and mon == "*" and dow == "*":
+        mins = m.split("/")[1]
+        return f"Ogni {mins} minuti"
+
+    if h.startswith("*/") and dom == "*" and mon == "*" and dow == "*":
+        hrs = h.split("/")[1]
+        if m == "0":
+            return f"Ogni {hrs} ore"
+        return f"Al minuto {m}, ogni {hrs} ore"
+
+    if dom == "*" and mon == "*" and dow == "*":
+        if m.isdigit() and h.isdigit():
+            return f"Ogni giorno alle {h.zfill(2)}:{m.zfill(2)}"
+        if m.isdigit() and "," in h:
+            ore = h.replace(",", ", ")
+            return f"Alle ore {ore}, al minuto {m}"
+
+    if dom == "*" and mon == "*" and dow != "*":
+        giorni_map = {"0": "dom", "1": "lun", "2": "mar", "3": "mer",
+                      "4": "gio", "5": "ven", "6": "sab"}
+        if "," in dow:
+            giorni = ", ".join(giorni_map.get(d.strip(), d) for d in dow.split(","))
+        elif dow in giorni_map:
+            giorni = giorni_map[dow]
+        else:
+            giorni = dow
+        if m.isdigit() and h.isdigit():
+            return f"Ogni {giorni} alle {h.zfill(2)}:{m.zfill(2)}"
+
+    if dom != "*" and mon == "*" and dow == "*":
+        if m.isdigit() and h.isdigit():
+            return f"Il giorno {dom} di ogni mese alle {h.zfill(2)}:{m.zfill(2)}"
+
+    return f"Cron: {expression}"
+
+
+# ============================================================
+# CLI: usato da sync.sh per calcolare i tempi
+# ============================================================
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Uso: cron_helper.py <espressione_cron> [seconds|next|human|validate]", file=sys.stderr)
+        sys.exit(1)
+
+    expr = sys.argv[1]
+    mode = sys.argv[2] if len(sys.argv) > 2 else "seconds"
+
+    try:
+        if mode == "seconds":
+            # Ritorna i secondi alla prossima esecuzione
+            print(seconds_until_next(expr))
+
+        elif mode == "next":
+            # Ritorna il timestamp della prossima esecuzione
+            next_time = next_cron_time(expr)
+            print(next_time.strftime("%Y-%m-%d %H:%M:%S"))
+
+        elif mode == "human":
+            # Ritorna una descrizione leggibile
+            print(human_readable(expr))
+
+        elif mode == "validate":
+            # Valida l'espressione (exit 0 se OK, exit 1 se errore)
+            parse_cron_expression(expr)
+            print("OK")
+
+        else:
+            print(f"Modalità sconosciuta: {mode}", file=sys.stderr)
+            sys.exit(1)
+
+    except ValueError as e:
+        print(f"ERRORE: {e}", file=sys.stderr)
+        sys.exit(1)