Feature: add jenkins-style H token support

aiocron/__init__.py

@@ -8,6 +8,8 @@
 import asyncio
 import sys
 import inspect
+import hashlib
+import re
 
 
 async def null_callback(*args):
@@ -27,6 +29,55 @@ async def wrapper(*args, **kwargs):
     return wrapper
 
 
+def _resolve_spec_for_func(spec, func_identifier):
+    """Resolve jenkins-like H literals in minute/hour fields using a hash of func_identifier.
+
+    Supports `H` and `H/<step>` tokens in the minute (field 0) and hour (field 1) 
+    positions. Other fields are left untouched.
+    """
+    def _hash(name, salt):
+        h = hashlib.md5((name + ":" + salt).encode("utf-8")).digest()
+        return int.from_bytes(h, "big")
+
+    def resolve_field(field_text, field_index):
+        # only apply to minute (0) and hour (1)
+        if field_index not in (0, 1):
+            return field_text
+
+        max_val = 59 if field_index == 0 else 23
+
+        def resolve_token(tok):
+            # search for H or H/<step>
+            m = re.match(r"^H(?:/(\d+))?$", tok)
+
+            # if no match, return token as is
+            if not m: return tok
+
+            step = m.group(1)
+            hval = _hash(func_identifier, str(field_index))
+            if step:
+                step_i = int(step)
+                start = hval % step_i
+                return f"{start}/{step_i}"
+            return str(hval % (max_val + 1))
+
+        # process comma-separated tokens in the field e.g. "1,H,10" or "H/15,30"
+        tokens = field_text.split(",")
+        return ",".join(resolve_token(t) for t in tokens)
+
+    parts = spec.split()
+
+    # pass malformed specs without modification
+    if len(parts) < 2:
+        return spec
+
+    # only resolve minute and hour fields (0 and 1)
+    for i in (0, 1):
+        parts[i] = resolve_field(parts[i], i)
+
+    return " ".join(parts)
+
+
 class Cron(object):
     def __init__(
         self,
@@ -79,6 +130,20 @@ def initialize(self):
             self.time = time.time()
             self.datetime = datetime.now(self.tz)
             self.loop_time = self.loop.time()
+
+            # resolve jenkins-like H literals in minute/hour fields using
+            # a stable hash of the wrapped function name (or uuid fallback).
+            if self.func is not None and self.func is not null_callback:
+                func_identifier = (
+                    getattr(self.func, "__module__", "")
+                    + "."
+                    + getattr(self.func, "__qualname__", getattr(self.func, "__name__", ""))
+                )
+            else:
+                func_identifier = str(self.uuid)
+
+            # replace H tokens in the spec with resolved values based on func_identifier
+            self.spec = _resolve_spec_for_func(self.spec, func_identifier)
             self.cronsim = CronSim(self.spec, self.datetime)
 
     def get_next(self):

tests/test_h_jenkins.py

@@ -0,0 +1,56 @@
+# -*- coding: utf-8 -*-
+import re
+
+from aiocron import _resolve_spec_for_func
+
+
+def _is_number(tok):
+    return re.match(r"^\d+$", tok) is not None
+
+
+def test_resolve_H_plain_minute_hour():
+    spec = "H H * * *"
+    out = _resolve_spec_for_func(spec, "module.func")
+    parts = out.split()
+    assert len(parts) == 5
+    # minute and hour should be numeric
+    assert _is_number(parts[0])
+    assert 0 <= int(parts[0]) <= 59
+    assert _is_number(parts[1])
+    assert 0 <= int(parts[1]) <= 23
+
+
+def test_resolve_H_with_step():
+    spec = "H/15 H/6 * * *"
+    out = _resolve_spec_for_func(spec, "module.func")
+    parts = out.split()
+    # minute should be like N/15 where 0 <= N < 15
+    assert re.match(r"^\d+/15$", parts[0])
+    start = int(parts[0].split("/")[0])
+    assert 0 <= start < 15
+    # hour should be like N/6 where 0 <= N < 6
+    assert re.match(r"^\d+/6$", parts[1])
+    start_h = int(parts[1].split("/")[0])
+    assert 0 <= start_h < 6
+
+
+def test_resolve_H_in_list_and_mixed():
+    spec = "1,H,10 H/2,3 * * *"
+    out = _resolve_spec_for_func(spec, "another.module:Fn")
+    parts = out.split()
+    # minute tokens should preserve existing numbers and resolve H
+    minute_tokens = parts[0].split(",")
+    assert minute_tokens[0] == "1"
+    assert _is_number(minute_tokens[1])
+    assert minute_tokens[2] == "10"
+    # hour tokens should resolve H/2 into N/2 or preserve numbers
+    hour_tokens = parts[1].split(",")
+    assert re.match(r"^\d+/2$", hour_tokens[0])
+    assert hour_tokens[1] == "3"
+
+
+def test_resolve_is_stable_for_same_identifier():
+    spec = "H H * * *"
+    out1 = _resolve_spec_for_func(spec, "stable.name")
+    out2 = _resolve_spec_for_func(spec, "stable.name")
+    assert out1 == out2