#!/usr/bin/env python3 import copy import re import subprocess import sys def usage(retval: int) -> None: """Output a usage message and exit.""" if retval == 0: file_ = sys.stdout else: file_ = sys.stderr print(f"Usage: {sys.argv[0]} --command cmd --minute 0 --hour 0 --day-of-month 1 --month 1 --day-of-week 1", file=file_) sys.exit(retval) class Options: """Deal with command line options.""" def __init__(self) -> None: """Init.""" self.minute = "*" self.hour = "*" self.day_of_month = "*" self.month = "*" self.day_of_week = "*" self.full_command = "" while sys.argv[1:]: if sys.argv[1] == "--minute": self.minute = sys.argv[2] del sys.argv[1] elif sys.argv[1] == "--hour": self.hour = sys.argv[2] del sys.argv[1] elif sys.argv[1] == "--day-of-month": self.day_of_month = sys.argv[2] del sys.argv[1] elif sys.argv[1] == "--month": self.month = sys.argv[2] del sys.argv[1] elif sys.argv[1] == "--day-of-week": self.day_of_month = sys.argv[2] del sys.argv[1] elif sys.argv[1] == "--command": self.full_command = sys.argv[2] del sys.argv[1] elif sys.argv[1] in ("-h", "--help"): usage(0) else: print(f"{sys.argv[0]}: unrecognized option: {sys.argv[1]}", file=sys.stderr) usage(1) del sys.argv[1] all_good = True if self.full_command == "": all_good = False print(f"{sys.argv[0]}: --command is a required option", file=sys.stderr) if not all_good: print(f"{sys.argv[0]}: One or more items in preflight check failed", file=sys.stderr) sys.exit(1) def read_original_crontab() -> list[str]: """Inhale the initial crontab -l, and return it as a list of strings.""" subp = subprocess.run(["crontab", "-l"], capture_output=True, text=True, check=True) lines = subp.stdout.rstrip("\n").splitlines() w_newlines = [line + "\n" for line in lines] return w_newlines def modify_crontab( *, original_crontab, minute, hour, day_of_month, month, day_of_week, full_command, ) -> list[str]: """Modify the crontab: if the command already exists, replace it. If the command doesn't yet exist, append it.""" modified_crontab = copy.deepcopy(original_crontab) hits = 0 # This might be better with shlex? command_fields = full_command.split() base_command = command_fields[0] new_crontab_line = f"{minute} {hour} {day_of_month} {month} {day_of_week} {full_command}\n" for lineno, line in enumerate(modified_crontab): line_wo_comments = re.sub("#.*$", "", line) fields = line_wo_comments.split() if fields[5:] and fields[5] == base_command: hits += 1 modified_crontab[lineno] = new_crontab_line if hits > 1: print(f"{sys.argv[0]}: more than one {base_command} found. Terminating", file=sys.stderr) sys.exit(1) if hits == 0: modified_crontab.append(new_crontab_line) return modified_crontab def write_modified_crontab(crontab_content: list[str]) -> None: """Write the modified crontab content to the crontab command (usually /usr/bin/crontab).""" subp = subprocess.Popen( ["crontab"], # Replace 'cat' with your target command/program stdin=subprocess.PIPE, # stdout=subprocess.PIPE, # Optional: if you want to capture output # stderr=subprocess.PIPE, # Optional: if you want to capture errors text=True, encoding="utf-8", ) for line in crontab_content: subp.stdin.write(line) subp.stdin.close() def main() -> None: """Start the ball rolling.""" options = Options() original_crontab = read_original_crontab() # print(f"original_crontab: {original_crontab}") # modify crontab modified_crontab = modify_crontab( original_crontab=original_crontab, minute=options.minute, hour=options.hour, day_of_month=options.day_of_month, month=options.month, day_of_week=options.day_of_week, full_command=options.full_command, ) # print(f"modified_crontab: {modified_crontab}") write_modified_crontab(modified_crontab) if __name__ == "__main__": main()