Nirodan
98bb34f094
- Fix AttributeError crash on empty request body in md5, hasher, textdiff,
jwtdecoder, timestamp, passwordgen (get_json without silent=True / or {})
- Fix memory exhaustion in ipcalc: replace list(network.hosts()) with direct
arithmetic — safe for /8 and larger networks
- Fix O(1M) loop in cronexplainer.get_next_runs: rewrite to skip by
month/day/hour instead of iterating every minute
- Fix connection leak in notes.ensure_table: add try/finally around conn.close
- Fix admin._ensure_tables / notes._ensure_table running DDL on every request:
guard with module-level flags (_tables_initialized, _table_ready)
- Fix update_website returning 200 when no row matched; delete_website returning
success when nothing was deleted; add rowcount checks for both
- Add role validation in admin create_user / update_user (_VALID_ROLES guard)
- Add delimiter length guard in csvviewer (csv.reader requires single char)
- Fix loremipsum: wrap int(count) in try/except ValueError → 400 response
- Fix auth/token: use auth_header[7:] instead of fragile .replace()
- Fix app.py: remove duplicate import sys; cache DB liveness check with 30s TTL
to avoid a new TCP connection on every frontend page load; move api/setup
path guard before DB check
- Replace FileHandler with RotatingFileHandler (5 MB / 3 backups) in logger;
fix relative log paths to absolute paths anchored to __file__
- Wrap all DB connections in try/finally conn.close() throughout admin and notes
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
237 lines
9.3 KiB
Python
237 lines
9.3 KiB
Python
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):
|
||
"""Return up to 5 upcoming run times.
|
||
|
||
Uses targeted jumps (skip month, day, hour) instead of iterating every
|
||
minute, so performance is acceptable even for rare expressions like
|
||
'0 0 29 2 *' (Feb 29 at midnight).
|
||
"""
|
||
# 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)
|
||
|
||
minute_list = sorted(minute_vals)
|
||
hour_list = sorted(hour_vals)
|
||
month_list = sorted(month_vals)
|
||
|
||
if not minute_list or not hour_list or not month_list:
|
||
return []
|
||
|
||
now = datetime.now().replace(second=0, microsecond=0) + timedelta(minutes=1)
|
||
max_date = now + timedelta(days=4 * 366)
|
||
results = []
|
||
current = now
|
||
|
||
def _advance_from_results():
|
||
"""Move current to the next candidate after a match."""
|
||
nonlocal current
|
||
next_mins = [m for m in minute_list if m > current.minute]
|
||
if next_mins:
|
||
current = current.replace(minute=next_mins[0])
|
||
return
|
||
next_hours = [h for h in hour_list if h > current.hour]
|
||
if next_hours:
|
||
current = current.replace(hour=next_hours[0], minute=minute_list[0])
|
||
return
|
||
current = (current + timedelta(days=1)).replace(hour=hour_list[0], minute=minute_list[0])
|
||
|
||
while current <= max_date and len(results) < 5:
|
||
# ── month ──────────────────────────────────────────────────────────
|
||
if current.month not in month_set:
|
||
next_months = [m for m in month_list if m > current.month]
|
||
if next_months:
|
||
current = current.replace(
|
||
month=next_months[0], day=1,
|
||
hour=hour_list[0], minute=minute_list[0]
|
||
)
|
||
else:
|
||
current = current.replace(
|
||
year=current.year + 1, month=month_list[0], day=1,
|
||
hour=hour_list[0], minute=minute_list[0]
|
||
)
|
||
continue
|
||
|
||
# ── day + weekday ──────────────────────────────────────────────────
|
||
if current.day not in day_set or current.weekday() not in py_weekdays:
|
||
current = (current + timedelta(days=1)).replace(
|
||
hour=hour_list[0], minute=minute_list[0]
|
||
)
|
||
continue
|
||
|
||
# ── hour ───────────────────────────────────────────────────────────
|
||
if current.hour not in hour_set:
|
||
next_hours = [h for h in hour_list if h > current.hour]
|
||
if next_hours:
|
||
current = current.replace(hour=next_hours[0], minute=minute_list[0])
|
||
else:
|
||
current = (current + timedelta(days=1)).replace(
|
||
hour=hour_list[0], minute=minute_list[0]
|
||
)
|
||
continue
|
||
|
||
# ── minute ─────────────────────────────────────────────────────────
|
||
if current.minute not in minute_set:
|
||
next_mins = [m for m in minute_list if m > current.minute]
|
||
if next_mins:
|
||
current = current.replace(minute=next_mins[0])
|
||
else:
|
||
next_hours = [h for h in hour_list if h > current.hour]
|
||
if next_hours:
|
||
current = current.replace(hour=next_hours[0], minute=minute_list[0])
|
||
else:
|
||
current = (current + timedelta(days=1)).replace(
|
||
hour=hour_list[0], minute=minute_list[0]
|
||
)
|
||
continue
|
||
|
||
# ── match ──────────────────────────────────────────────────────────
|
||
results.append(current.isoformat())
|
||
_advance_from_results()
|
||
|
||
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(silent=True) 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
|