from flask import Blueprint, request, jsonify from datetime import datetime, timedelta from util.logger import logger from auth.token import verify_token cron_blueprint = Blueprint('cron_tool', __name__) MONTH_NAMES = { '1': 'Januar', '2': 'Februar', '3': 'März', '4': 'April', '5': 'Mai', '6': 'Juni', '7': 'Juli', '8': 'August', '9': 'September', '10': 'Oktober', '11': 'November', '12': 'Dezember', } WEEKDAY_NAMES = { '0': 'Sonntag', '1': 'Montag', '2': 'Dienstag', '3': 'Mittwoch', '4': 'Donnerstag', '5': 'Freitag', '6': 'Samstag', '7': 'Sonntag', } def parse_field(field, min_val, max_val): values = set() for part in field.split(','): part = part.strip() if part == '*': values.update(range(min_val, max_val + 1)) elif '/' in part: base, step = part.split('/', 1) step = int(step) if base == '*': start, end = min_val, max_val elif '-' in base: s, e = base.split('-') start, end = int(s), int(e) else: start, end = int(base), max_val values.update(range(start, end + 1, step)) elif '-' in part: s, e = part.split('-') values.update(range(int(s), int(e) + 1)) else: values.add(int(part)) return sorted(v for v in values if min_val <= v <= max_val) def explain_field_text(field, unit, plural, labels=None): if field == '*': return f"Jede(n) {unit}" parts = field.split(',') described = [] for p in parts: if p.startswith('*/'): described.append(f"alle {p[2:]} {plural}") elif '-' in p and '/' not in p: a, b = p.split('-', 1) a_l = labels.get(a, a) if labels else a b_l = labels.get(b, b) if labels else b described.append(f"{a_l} bis {b_l}") elif '/' in p: base, step = p.split('/', 1) if '-' in base: a, b = base.split('-', 1) described.append(f"{a}–{b} alle {step}") else: described.append(f"ab {base} alle {step}") else: described.append(labels.get(p, p) if labels else p) return ', '.join(described) def build_summary(minute, hour, day, month, weekday): parts = [] if minute == '*' and hour == '*': parts.append("Jede Minute") elif minute.startswith('*/') and hour == '*': parts.append(f"Alle {minute[2:]} Minuten") elif hour != '*' and minute == '0': h = explain_field_text(hour, 'Stunde', 'Stunden') parts.append(f"Um {h}:00 Uhr") else: parts.append(f"Minute: {minute}, Stunde: {hour}") if weekday != '*' and day == '*': wd = explain_field_text(weekday, 'Wochentag', 'Wochentage', WEEKDAY_NAMES) parts.append(f"jeden {wd}") elif day != '*' and weekday == '*': parts.append(f"am {day}. des Monats") elif day != '*' and weekday != '*': wd = explain_field_text(weekday, 'Wochentag', 'Wochentage', WEEKDAY_NAMES) parts.append(f"am {day}. oder {wd}") if month != '*': m = explain_field_text(month, 'Monat', 'Monate', MONTH_NAMES) parts.append(f"im {m}") return ', '.join(parts) def get_next_runs(minute_vals, hour_vals, day_vals, month_vals, weekday_vals): # cron weekday 0=Sun..6=Sat,7=Sun -> python weekday 0=Mon..6=Sun py_weekdays = set((wd + 6) % 7 for wd in weekday_vals) minute_set = set(minute_vals) hour_set = set(hour_vals) day_set = set(day_vals) month_set = set(month_vals) now = datetime.now().replace(second=0, microsecond=0) + timedelta(minutes=1) results = [] current = now for _ in range(2 * 366 * 24 * 60): if len(results) >= 5: break if (current.month in month_set and current.day in day_set and current.weekday() in py_weekdays and current.hour in hour_set and current.minute in minute_set): results.append(current.isoformat()) current += timedelta(minutes=1) return results @cron_blueprint.route('/api/cron/explain', methods=['POST']) def explain_cron(): user = verify_token() if not user: return jsonify({"message": "Nicht autorisiert"}), 401 try: data = request.get_json() or {} expression = data.get("expression", "").strip() fields = expression.split() if len(fields) != 5: return jsonify({"error": "Cron-Ausdruck muss genau 5 Felder haben (Minute Stunde Tag Monat Wochentag)"}), 400 minute_f, hour_f, day_f, month_f, weekday_f = fields try: minute_vals = parse_field(minute_f, 0, 59) hour_vals = parse_field(hour_f, 0, 23) day_vals = parse_field(day_f, 1, 31) month_vals = parse_field(month_f, 1, 12) weekday_vals = parse_field(weekday_f, 0, 7) except (ValueError, ZeroDivisionError) as e: return jsonify({"error": f"Ungültiger Cron-Ausdruck: {e}"}), 400 field_descriptions = { "minute": explain_field_text(minute_f, 'Minute', 'Minuten'), "hour": explain_field_text(hour_f, 'Stunde', 'Stunden'), "day": explain_field_text(day_f, 'Tag', 'Tage'), "month": explain_field_text(month_f, 'Monat', 'Monate', MONTH_NAMES), "weekday": explain_field_text(weekday_f, 'Wochentag', 'Wochentage', WEEKDAY_NAMES), } next_runs = get_next_runs(minute_vals, hour_vals, day_vals, month_vals, weekday_vals) return jsonify({ "explanation": build_summary(minute_f, hour_f, day_f, month_f, weekday_f), "fields": field_descriptions, "next_runs": next_runs, }) except Exception as e: logger.error(f"Fehler cron explain: {e}") return jsonify({"error": "Fehler beim Verarbeiten des Ausdrucks"}), 500