#!/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 [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)