diff --git a/app/utils/__init__.py b/app/utils/__init__.py new file mode 100644 index 0000000..46451ee --- /dev/null +++ b/app/utils/__init__.py @@ -0,0 +1,11 @@ +""" +Utils package for MokPyo application. + +- Expose utility classes for easy imports +""" + +from .frequency import Frequency, Cron, DayInterval, WeekInterval, MonthInterval, Interval + +__all__ = [ + "Frequency", "Cron", "DayInterval", "WeekInterval", "MonthInterval", "Interval" +] \ No newline at end of file diff --git a/app/utils/frequency.py b/app/utils/frequency.py new file mode 100644 index 0000000..6fef66e --- /dev/null +++ b/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"" + + +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"" + + 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")