gurix
c50f296232
- crontab_manager.get_alarms now parses --style from the cron command (defaults to 'blink' for legacy entries without --style) - GraphQL Alarm type gains a 'style' field - setAlarm accepts a 'style' arg (default 'simple') and validates it against the styles registry; unknown styles raise a GraphQL error - _default_command embeds --style <name> into the cron command - startRinging accepts 'style' (default 'simple') and passes --style to the spawned wecker.py The crontab remains the single source of truth; the style simply rides in the cron command alongside --music-file.
89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
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
|