134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
"""Berechnung der naechsten Ausfuehrungszeit einer Gruppe (lokale Zeit)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import calendar
|
|
from datetime import datetime, timedelta
|
|
|
|
from .config import WEEKDAY_NAMES, parse_time_of_day
|
|
from .util import format_duration
|
|
|
|
_EPOCH = datetime(1970, 1, 1)
|
|
|
|
|
|
def _midnight(moment):
|
|
return moment.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
|
|
def _aligned_after(moment, interval):
|
|
"""Naechster an der Uhr ausgerichteter Zeitpunkt nach `moment`.
|
|
|
|
Intervalle, die glatt in einen Tag passen (5m, 15m, 1h, 6h, 12h), werden an
|
|
Mitternacht verankert - 6h ergibt also 00:00, 06:00, 12:00, 18:00.
|
|
Alles andere wird an der Epoche verankert.
|
|
"""
|
|
if interval <= 86400 and 86400 % interval == 0:
|
|
anchor = _midnight(moment)
|
|
else:
|
|
anchor = _EPOCH
|
|
elapsed = (moment - anchor).total_seconds()
|
|
steps = int(elapsed // interval) + 1
|
|
return anchor + timedelta(seconds=steps * interval)
|
|
|
|
|
|
def _at_time(day, hour, minute):
|
|
return day.replace(hour=hour, minute=minute, second=0, microsecond=0)
|
|
|
|
|
|
def _clamp_day(year, month, day):
|
|
return min(day, calendar.monthrange(year, month)[1])
|
|
|
|
|
|
def next_due(group, last_run, now):
|
|
"""Wann ist die Gruppe das naechste Mal faellig?
|
|
|
|
`last_run` ist der letzte tatsaechliche Lauf (oder None). Liegt das
|
|
Ergebnis in der Vergangenheit, ist die Gruppe sofort faellig - so werden
|
|
Laeufe nachgeholt, die waehrend eines Neustarts ausgefallen sind.
|
|
"""
|
|
if group.interval > 0:
|
|
if last_run is None:
|
|
base = now
|
|
else:
|
|
base = last_run
|
|
if group.align:
|
|
return _aligned_after(base, group.interval)
|
|
return base + timedelta(seconds=group.interval)
|
|
|
|
base = last_run if last_run is not None else now - timedelta(seconds=1)
|
|
kind = group.schedule
|
|
|
|
if kind == "hourly":
|
|
minute = max(0, min(59, group.minute))
|
|
candidate = base.replace(minute=minute, second=0, microsecond=0)
|
|
while candidate <= base:
|
|
candidate += timedelta(hours=1)
|
|
return candidate
|
|
|
|
hour, minute = parse_time_of_day(group.at)
|
|
|
|
if kind == "daily":
|
|
candidate = _at_time(base, hour, minute)
|
|
while candidate <= base:
|
|
candidate += timedelta(days=1)
|
|
return candidate
|
|
|
|
if kind == "weekly":
|
|
candidate = _at_time(base, hour, minute)
|
|
shift = (group.day_of_week - candidate.weekday()) % 7
|
|
candidate += timedelta(days=shift)
|
|
while candidate <= base:
|
|
candidate += timedelta(days=7)
|
|
return candidate
|
|
|
|
if kind == "monthly":
|
|
year, month = base.year, base.month
|
|
for _ in range(60):
|
|
day = _clamp_day(year, month, group.day_of_month)
|
|
candidate = _at_time(datetime(year, month, day), hour, minute)
|
|
if candidate > base:
|
|
return candidate
|
|
month += 1
|
|
if month > 12:
|
|
month, year = 1, year + 1
|
|
raise ValueError("kein monatlicher Termin ermittelbar")
|
|
|
|
if kind == "yearly":
|
|
year = base.year
|
|
month = max(1, min(12, group.month))
|
|
for _ in range(5):
|
|
day = _clamp_day(year, month, group.day_of_month)
|
|
candidate = _at_time(datetime(year, month, day), hour, minute)
|
|
if candidate > base:
|
|
return candidate
|
|
year += 1
|
|
raise ValueError("kein jaehrlicher Termin ermittelbar")
|
|
|
|
raise ValueError("unbekannter Zeitplan: %r" % kind)
|
|
|
|
|
|
def is_due(group, last_run, now, run_on_start=False):
|
|
if not group.enabled:
|
|
return False
|
|
if last_run is None and run_on_start:
|
|
return True
|
|
return next_due(group, last_run, now) <= now
|
|
|
|
|
|
def describe(group):
|
|
"""Zeitplan der Gruppe als kurzer, lesbarer Text."""
|
|
if group.interval > 0:
|
|
text = "alle %s" % format_duration(group.interval)
|
|
return text + (" (an der Uhr ausgerichtet)" if group.align else "")
|
|
if group.schedule == "hourly":
|
|
return "stuendlich zur Minute %02d" % group.minute
|
|
if group.schedule == "daily":
|
|
return "taeglich um %s" % group.at
|
|
if group.schedule == "weekly":
|
|
return "woechentlich %s um %s" % (WEEKDAY_NAMES[group.day_of_week], group.at)
|
|
if group.schedule == "monthly":
|
|
return "monatlich am %d. um %s" % (group.day_of_month, group.at)
|
|
if group.schedule == "yearly":
|
|
return "jaehrlich am %d.%d. um %s" % (group.day_of_month, group.month, group.at)
|
|
return "kein Zeitplan"
|