import shlex import uuid from crontab import CronTab, CronSlices from styles import LEGACY_STYLE class CrontabManager: COMMENT_PREFIX = "wecker-alarm:" def __init__(self, tabfile: str | None = None, user: bool | str = True): # user=True means current user, user="username" means specific user self.tabfile = tabfile self.user = user def _get_cron(self): if self.tabfile: return CronTab(tabfile=self.tabfile) return CronTab(user=self.user) def get_alarms(self): cron = self._get_cron() alarms = [] for job in cron: if job.comment.startswith(self.COMMENT_PREFIX): alarm_id = job.comment.split(self.COMMENT_PREFIX)[1].strip() # job.slices is a valid cron slice object, str(job.slices) gives the expression alarms.append( { "id": alarm_id, "cron_expression": str(job.slices), "command": job.command, "is_enabled": job.is_enabled(), "style": _parse_style(job.command), } ) return alarms def set_alarm( self, alarm_id: str | None, cron_expression: str, command: str, is_enabled: bool = True, ): if not CronSlices.is_valid(cron_expression): raise ValueError(f"Invalid cron expression: {cron_expression!r}") if not alarm_id: alarm_id = str(uuid.uuid4()) cron = self._get_cron() comment = f"{self.COMMENT_PREFIX}{alarm_id}" # Remove existing if any cron.remove_all(comment=comment) # Create new job = cron.new(command=command, comment=comment) job.setall(cron_expression) job.enable(is_enabled) cron.write() return alarm_id def delete_alarm(self, alarm_id: str): cron = self._get_cron() comment = f"{self.COMMENT_PREFIX}{alarm_id}" cron.remove_all(comment=comment) cron.write() def _parse_style(command: str) -> str: """Extract the --style value from a cron command. Returns LEGACY_STYLE ('blink') for entries that predate --style, so an upgrade never silently changes an existing alarm's behaviour. """ try: tokens = shlex.split(command) except ValueError: return LEGACY_STYLE for i, tok in enumerate(tokens): if tok == "--style" and i + 1 < len(tokens): return tokens[i + 1] if tok.startswith("--style="): return tok.split("=", 1)[1] return LEGACY_STYLE