rules.py · BryceTRussell/tinycron at main

tinycron / rules.py

Release 1.2.0

747ffd4 verified 5 days ago

5.56 kB

"""A request that is already a cron line, read exactly before the model; standard library only.
"*/5 * * * *", "run 0 9 * * mon-fri" and "@daily" are parsed, not guessed. Seconds fields,
L, W, #, ?, H, @reboot, @every and 6 or 7 fields are UNSUPPORTED; a line with a value out
of range, or a date that never comes, is INVALID. Any other text goes to the model.
A port may skip this file: the model still answers pasted cron, less often right.
"""

import re

UNSUPPORTED, INVALID = "UNSUPPORTED", "INVALID"

_MACROS = {"@yearly": "0 0 1 1 *", "@annually": "0 0 1 1 *", "@monthly": "0 0 1 * *",
           "@weekly": "0 0 * * 0", "@daily": "0 0 * * *", "@midnight": "0 0 * * *", "@hourly": "0 * * * *"}
_MONTHS = {n: i for i, n in enumerate("jan feb mar apr may jun jul aug sep oct nov dec".split(), start=1)}
_DAYS = {n: i for i, n in enumerate("sun mon tue wed thu fri sat".split())}
_LONGEST = dict(enumerate([31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31], start=1))
_NAMES = "|".join(list(_MONTHS) + list(_DAYS))
# (low, high, names) per field; the day of the week also takes 7 for Sunday.
_BOUNDS = [(0, 59, None), (0, 23, None), (1, 31, None), (1, 12, _MONTHS), (0, 6, _DAYS)]

# Words people put around a pasted line. Anything else in the text leaves it to the model.
_WRAPPER = re.compile(r"^(?:run(?: it)?(?: at)?|cron(?:tab)?(?: line| entry| job)?:?|schedule:?|expression:?)\s+", re.I)
_TOKEN = re.compile(rf"^(?:[\d*/,\-?#]+[lw]?|[lw]|h(?:\([\d\-]+\))?(?:/\d+)?|(?:{_NAMES})(?:[\-,/](?:{_NAMES}|\d+))*)$", re.I)
_OUTSIDE = re.compile(r"[lw#?h]", re.I)       # L, W, #, ? and H, once the names are taken out


def _values(text, low, high, names):
    """The values one field fires on, or None when the field is not valid cron."""
    top = 7 if names is _DAYS else high
    values = set()
    for item in text.split(","):
        rng, slash, step = item.partition("/")
        if slash and not (step.isdigit() and int(step) >= 1):
            return None
        ends = [low, high] if rng == "*" else rng.split("-", 1)
        try:
            ends = [e if isinstance(e, int) else names[e] if names and e in names else int(e) for e in ends]
        except ValueError:
            return None
        a, b = ends[0], ends[-1] if len(ends) == 2 else (top if slash else ends[0])
        if not (low <= a <= b <= top):
            return None
        values.update(range(a, b + 1, int(step) if slash else 1))
    if names is _DAYS and 7 in values:
        values.discard(7)
        values.add(0)
    return values


def read_cron(text):
    """For text that is only a cron line or macro: (kind, lines) with kind "schedule" and
    the line, or (UNSUPPORTED or INVALID, []). None for any other text."""
    t = text.strip().strip("`'\"").strip()
    t = _WRAPPER.sub("", t).strip("`'\" .")
    low = t.lower()
    if low.startswith(("@reboot", "@every")):
        return UNSUPPORTED, []
    if low in _MACROS:
        return "schedule", [_MACROS[low]]
    fields = low.split()
    if not 5 <= len(fields) <= 7 or not all(_TOKEN.match(f) for f in fields):
        return None
    if not any(c in "*/,-" for f in fields for c in f):
        return None          # five plain numbers ("1 2 3 4 5") are more likely a list than a cron line
    if len(fields) > 5 or _OUTSIDE.search(re.sub(_NAMES, "", low)):
        return UNSUPPORTED, []
    values = [_values(f, *b) for f, b in zip(fields, _BOUNDS)]
    if None in values:
        return INVALID, []
    doms, months = values[2], values[3]
    if fields[4].startswith("*") and not any(d <= _LONGEST[m] for d in doms for m in months):
        return INVALID, []   # a date that never comes ("0 0 30 2 *")
    return "schedule", [" ".join(fields)]


# Hours the model left out of a run that the text never excludes. Each hour is its own
# output bit, so a few bits can drop below the threshold: "every 5 minutes" comes out as
# "*/5 0-16,19-21,23 * * *". A hole of at most _MAX_GAP hours in a field of _MIN_HOURS or
# more is filled, unless the text has a word that leaves hours out; a field of 22 or more
# hours becomes "*". On the release's design sets (2026-10-25) this fixed 13 rows and
# changed no right answer. A port may skip it.
_EXCLUDE = re.compile(r"\b(?:except|exclud\w*|not|skip\w*|but|other than|apart from|avoid\w*|besides|outside|"
                      r"without|n't|no)\b", re.I)
_MAX_GAP, _MIN_HOURS = 2, 6


def _fill(hours):
    """The hour field with its small holes filled, or the field as it was."""
    if hours == "*" or "/" in hours:
        return hours
    on = sorted(_values(hours, 0, 23, None) or ())
    if len(on) < _MIN_HOURS:
        return hours
    steps = [b - a for a, b in zip(on, on[1:])]
    if max(steps) == 1 or max(steps) - 1 > _MAX_GAP or len(set(steps)) == 1:
        return hours         # no hole, a hole too wide, or an even progression
    cuts = [0] + [i + 1 for i, s in enumerate(steps) if s > 1] + [len(on)]
    if len(cuts) == 3 and min(b - a for a, b in zip(cuts, cuts[1:])) >= 3 and on[-1] - on[0] < 20:
        return hours         # two windows of 3 hours or more ("8 to 12 and 2 to 6")
    if on[-1] - on[0] >= 21:
        return "*"
    return f"{on[0]}-{on[-1]}"


def fill_hours(text, lines):
    """The cron line with unasked-for hour holes filled (see _fill). Several lines are left
    as they are: their hours often fit together on purpose ("hourly on weekdays, every 3
    hours on weekends")."""
    if len(lines) != 1 or _EXCLUDE.search(text):
        return lines
    f = lines[0].split()
    f[1] = _fill(f[1])
    return [" ".join(f)]

	"""A request that is already a cron line, read exactly before the model; standard library only.

	"/5 * * ", "run 0 9 * mon-fri" and "@daily" are parsed, not guessed. Seconds fields,
	L, W, #, ?, H, @reboot, @every and 6 or 7 fields are UNSUPPORTED; a line with a value out
	of range, or a date that never comes, is INVALID. Any other text goes to the model.

	A port may skip this file: the model still answers pasted cron, less often right.
	"""

	import re

	UNSUPPORTED, INVALID = "UNSUPPORTED", "INVALID"

	_MACROS = {"@yearly": "0 0 1 1 ", "@annually": "0 0 1 1 ", "@monthly": "0 0 1 * *",
	"@weekly": "0 0 * * 0", "@daily": "0 0 * * ", "@midnight": "0 0 * ", "@hourly": "0 * * *"}
	_MONTHS = {n: i for i, n in enumerate("jan feb mar apr may jun jul aug sep oct nov dec".split(), start=1)}
	_DAYS = {n: i for i, n in enumerate("sun mon tue wed thu fri sat".split())}
	_LONGEST = dict(enumerate([31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31], start=1))
	_NAMES = "\|".join(list(_MONTHS) + list(_DAYS))
	# (low, high, names) per field; the day of the week also takes 7 for Sunday.
	_BOUNDS = [(0, 59, None), (0, 23, None), (1, 31, None), (1, 12, _MONTHS), (0, 6, _DAYS)]

	# Words people put around a pasted line. Anything else in the text leaves it to the model.
	_WRAPPER = re.compile(r"^(?:run(?: it)?(?: at)?\|cron(?:tab)?(?: line\| entry\| job)?:?\|schedule:?\|expression:?)\s+", re.I)
	_TOKEN = re.compile(rf"^(?:[\d/,\-?#]+[lw]?\|[lw]\|h(?:\([\d\-]+\))?(?:/\d+)?\|(?:{_NAMES})(?:[\-,/](?:{_NAMES}\|\d+)))$", re.I)
	_OUTSIDE = re.compile(r"[lw#?h]", re.I) # L, W, #, ? and H, once the names are taken out


	def _values(text, low, high, names):
	"""The values one field fires on, or None when the field is not valid cron."""
	top = 7 if names is _DAYS else high
	values = set()
	for item in text.split(","):
	rng, slash, step = item.partition("/")
	if slash and not (step.isdigit() and int(step) >= 1):
	return None
	ends = [low, high] if rng == "*" else rng.split("-", 1)
	try:
	ends = [e if isinstance(e, int) else names[e] if names and e in names else int(e) for e in ends]
	except ValueError:
	return None
	a, b = ends[0], ends[-1] if len(ends) == 2 else (top if slash else ends[0])
	if not (low <= a <= b <= top):
	return None
	values.update(range(a, b + 1, int(step) if slash else 1))
	if names is _DAYS and 7 in values:
	values.discard(7)
	values.add(0)
	return values


	def read_cron(text):
	"""For text that is only a cron line or macro: (kind, lines) with kind "schedule" and
	the line, or (UNSUPPORTED or INVALID, []). None for any other text."""
	t = text.strip().strip("`'\"").strip()
	t = _WRAPPER.sub("", t).strip("`'\" .")
	low = t.lower()
	if low.startswith(("@reboot", "@every")):
	return UNSUPPORTED, []
	if low in _MACROS:
	return "schedule", [_MACROS[low]]
	fields = low.split()
	if not 5 <= len(fields) <= 7 or not all(_TOKEN.match(f) for f in fields):
	return None
	if not any(c in "*/,-" for f in fields for c in f):
	return None # five plain numbers ("1 2 3 4 5") are more likely a list than a cron line
	if len(fields) > 5 or _OUTSIDE.search(re.sub(_NAMES, "", low)):
	return UNSUPPORTED, []
	values = [_values(f, *b) for f, b in zip(fields, _BOUNDS)]
	if None in values:
	return INVALID, []
	doms, months = values[2], values[3]
	if fields[4].startswith("*") and not any(d <= _LONGEST[m] for d in doms for m in months):
	return INVALID, [] # a date that never comes ("0 0 30 2 *")
	return "schedule", [" ".join(fields)]


	# Hours the model left out of a run that the text never excludes. Each hour is its own
	# output bit, so a few bits can drop below the threshold: "every 5 minutes" comes out as
	# "/5 0-16,19-21,23 * *". A hole of at most _MAX_GAP hours in a field of _MIN_HOURS or
	# more is filled, unless the text has a word that leaves hours out; a field of 22 or more
	# hours becomes "*". On the release's design sets (2026-10-25) this fixed 13 rows and
	# changed no right answer. A port may skip it.
	_EXCLUDE = re.compile(r"\b(?:except\|exclud\w\|not\|skip\w\|but\|other than\|apart from\|avoid\w*\|besides\|outside\|"
	r"without\|n't\|no)\b", re.I)
	_MAX_GAP, _MIN_HOURS = 2, 6


	def _fill(hours):
	"""The hour field with its small holes filled, or the field as it was."""
	if hours == "*" or "/" in hours:
	return hours
	on = sorted(_values(hours, 0, 23, None) or ())
	if len(on) < _MIN_HOURS:
	return hours
	steps = [b - a for a, b in zip(on, on[1:])]
	if max(steps) == 1 or max(steps) - 1 > _MAX_GAP or len(set(steps)) == 1:
	return hours # no hole, a hole too wide, or an even progression
	cuts = [0] + [i + 1 for i, s in enumerate(steps) if s > 1] + [len(on)]
	if len(cuts) == 3 and min(b - a for a, b in zip(cuts, cuts[1:])) >= 3 and on[-1] - on[0] < 20:
	return hours # two windows of 3 hours or more ("8 to 12 and 2 to 6")
	if on[-1] - on[0] >= 21:
	return "*"
	return f"{on[0]}-{on[-1]}"


	def fill_hours(text, lines):
	"""The cron line with unasked-for hour holes filled (see _fill). Several lines are left
	as they are: their hours often fit together on purpose ("hourly on weekdays, every 3
	hours on weekends")."""
	if len(lines) != 1 or _EXCLUDE.search(text):
	return lines
	f = lines[0].split()
	f[1] = _fill(f[1])
	return [" ".join(f)]