feat: Added the utils module and classes useful for managing task and event frequency.

app/utils/frequency.py

@@ -0,0 +1,150 @@
+import re
+
+
+class Interval:
+    def __init__(self, unit: str, value: int, exec_day: int = None):
+        if unit not in ["day", "week", "month"]:
+            raise ValueError("unit must be day, week or month")
+        if value < 1 or value > 2**17 - 1:
+            raise ValueError("value must be between 1 and 131071")
+        self.unit = unit
+        self.value = value
+        self.exec_day = exec_day
+
+    def copy(self):
+        return Interval(self.unit, self.value, self.exec_day)
+
+    def __repr__(self):
+        return f"<Interval unit={self.unit} value={self.value} exec_day={self.exec_day}>"
+
+
+class DayInterval(Interval):
+    def __init__(self, value: int):
+        super().__init__("day", value)
+
+
+class WeekInterval(Interval):
+    def __init__(self, value: int, exec_day: int = 1):
+        super().__init__("week", value)
+        if exec_day not in range(1, 8) and exec_day is not None:
+            raise ValueError("exec_day must be between 1 and 7 or None")
+        self.exec_day = exec_day
+
+
+class MonthInterval(Interval):
+    def __init__(self, value: int, exec_day: int = 1):
+        super().__init__("month", value)
+        if exec_day not in range(1, 32) and exec_day is not None:
+            raise ValueError("exec_day must be between 1 and 31 or None")
+        self.exec_day = exec_day
+
+
+class Cron:
+    def __init__(self, days: str, weeks: str, months: str):
+        assert re.fullmatch(r"^(?:\*|[1-7](?:,[1-7])*|(?:1-[2-7]|2-[3-7]|3-[4-7]|4-[5-7]|5-[6-7]|6-7))$", days)
+        assert re.fullmatch(r"^(?:\*|[1-5](?:,[1-5])*|(?:1-[2-5]|2-[3-5]|3-[4-5]|4-5))$", weeks)
+        assert re.fullmatch(r"^(?:\*|(?:[1-9]|1[0-2])(?:,(?:[1-9]|1[0-2]))*|(?:1-(?:[2-9]|1[0-2])|2-(?:[3-9]|1[0-2])|3-(?:[4-9]|1[0-2])|4-(?:[5-9]|1[0-2])|5-(?:[6-9]|1[0-2])|6-(?:[7-9]|1[0-2])|7-(?:[8-9]|1[0-2])|8-(?:9|1[0-2])|9-(1[0-2])|10-(11|12)|11-12))$", months)
+
+        self.days = days
+        self.weeks = weeks
+        self.months = months
+
+    def copy(self):
+        return Cron(self.days, self.weeks, self.months)
+
+    def __repr__(self):
+        return f"<Cron days={self.days} weeks={self.weeks} months={self.months}>"
+
+    def __str__(self):
+        return f"{self.days} {self.weeks} {self.months}"
+
+
+class Frequency:
+    def __init__(self, frequency: Cron | Interval | int):
+        self.data = 0b0
+        self.__model = {"type": None, "instance": None}
+
+        if isinstance(frequency, Cron):
+            self.__model["type"] = Cron
+            self.__model["instance"] = frequency.copy()
+            self.data = 0b1 << 24
+            _bin_days = self.__cron_bin(frequency.days, 7) << 17
+            _bin_weeks = self.__cron_bin(frequency.weeks, 5) << 12
+            _bin_months = self.__cron_bin(frequency.months, 12)
+            self.data |= _bin_days | _bin_weeks | _bin_months
+
+        elif isinstance(frequency, Interval):
+            self.__model["type"] = Interval
+            self.__model["instance"] = frequency.copy()
+            _bin_unit = {"day": 1, "week": 2, "month": 3}[frequency.unit] << 22
+            _bin_value = frequency.value << 5
+            _bin_exec_day = frequency.exec_day if frequency.exec_day is not None else 0b0
+            self.data |= _bin_unit | _bin_value | _bin_exec_day
+
+        elif isinstance(frequency, int):
+            if frequency < 1 or frequency > 2**25 - 1:
+                raise ValueError("frequency must be between 1 and 2**25 - 1")
+            self.data = frequency
+            self.__model = self.__int_model(frequency)
+
+        else:
+            raise ValueError("frequency must be Cron, Interval or int")
+
+    def __cron_bin(self, cron: str, length: int) -> int:
+        _bin = 0b0
+        if cron == "*":
+            _bin |= (1 << length) - 1
+        elif '-' in cron:
+            start, end = map(int, cron.split("-"))
+            for i in range(start, end + 1):
+                _bin |= 1 << (i - 1)
+        else:
+            for i in cron.split(","):
+                _bin |= 1 << (int(i) - 1)
+        return _bin
+
+    def __int_model(self, _int: int):
+        if _int & (1 << 24):  # Cron
+            _days = (_int >> 17) & 0b1111111
+            _weeks = (_int >> 12) & 0b11111
+            _months = _int & 0b111111111111
+
+            days = self.__bin_to_str(_days, 7)
+            weeks = self.__bin_to_str(_weeks, 5)
+            months = self.__bin_to_str(_months, 12)
+
+            return {"type": Cron, "instance": Cron(days, weeks, months)}
+        else:  # Interval
+            _unit = (_int >> 22) & 0b11
+            _value = (_int >> 5) & ((1 << 17) - 1)
+            _exec_day = _int & 0b11111
+            unit = {1: "day", 2: "week", 3: "month"}[_unit]
+            IntervalClass = {"day": DayInterval, "week": WeekInterval, "month": MonthInterval}[unit]
+
+            return {"type": Interval, "instance": IntervalClass(_value, _exec_day or None)}
+
+    def __bin_to_str(self, value: int, length: int) -> str:
+        if value == (1 << length) - 1:
+            return "*"
+
+        nums = [str(i + 1) for i in range(length) if value & (1 << i)]
+        if not nums:
+            return "*"
+        return ",".join(nums)
+
+    def __int__(self):
+        return self.data
+
+    def __len__(self):
+        return len(bin(self.data)[2:])
+
+    def __repr__(self):
+        return str(bin(self.data)[2:].zfill(25))
+
+    __str__ = __repr__
+
+    def to_model(self) -> Cron | Interval:
+        return self.__model.get("instance")
+
+    def get_type(self):
+        return self.__model.get("type")