import datetime import os from yaml import load, FullLoader from icalendar.parser import Contentlines, Contentline from dateutil.rrule import rrule, FREQNAMES import caldav from caldav.lib.error import NotFoundError APP_ID = 'CaldavRecurringTask' class Task: def __init__(self, reference_date, name, data): self.ref_date = reference_date self.name = name data['start']['freq'] = FREQNAMES.index(data['start']['freq']) self.date_begin = rrule(count=1, dtstart=reference_date, **data['start'])[0] self.date_end = self.date_begin + datetime.timedelta(microseconds=-1, **data['duration']) self.uid = f"{APP_ID}-{self.name}-{self.date_begin.strftime('%Y%m%d')}" self.title = data['title'] self.body = data['body'] self.priority = data['priority'] def to_ical(self): current_dtstamp = datetime.datetime.now().strftime('%Y%m%dT%H%M%SZ') c = Contentlines([ Contentline('BEGIN:VCALENDAR'), Contentline('VERSION:2.0'), Contentline('PRODID:-//DorfsvaldNet//CaldavRecuringTaskClient//FR'), Contentline('BEGIN:VTODO'), Contentline(f'DTSTAMP:{current_dtstamp}'), Contentline(f'UID:{self.uid}'), Contentline(f'CREATED:{current_dtstamp}'), Contentline(f'LAST-MODIFIED:{current_dtstamp}'), Contentline(f'SEQUENCE:{"4"}'), Contentline(f'DESCRIPTION:{self.body}'), Contentline(f'SUMMARY:{self.title}'), Contentline(f'PRIORITY:{self.priority}'), Contentline(f'DUE;VALUE=DATE:{self.date_end.strftime("%Y%m%d")}'), Contentline(f'DTSTART;VALUE=DATE:{self.date_begin.strftime("%Y%m%d")}'), Contentline('PERCENT-COMPLETE:0'), Contentline('STATUS:NEEDS-ACTION'), Contentline('END:VTODO'), Contentline('END:VCALENDAR') ]) return c.to_ical() class Client: def __init__(self, url, username, password, calendar_name): self.client = caldav.DAVClient(url, username=username, password=password) self.principal = self.client.principal() self.calendar = None for c in self.principal.calendars(): if c.name == calendar_name: self.calendar = c break if self.calendar is None: raise LookupError('No calendar named "{}" found'.format(calendar_name)) def todos(self, include_completed=True): return self.calendar.todos(include_completed=include_completed) def add_todo(self, new_task): self.calendar.add_todo(new_task.to_ical()) def get_todo_by_uid(self, uid): try: todo = self.calendar.todo_by_uid(uid) # todo.percent_complete = todo.icalendar_component['PERCENT-COMPLETE'] return todo except NotFoundError: return False if __name__ == "__main__": with open('./configuration.yml', 'r') as configuration_file: conf = load(configuration_file.read(), Loader=FullLoader) last_run_file_path = './.last.run' if os.path.isfile(last_run_file_path): with open(last_run_file_path, 'r') as last_run_file: run_date_str = last_run_file.readline() run_date = datetime.datetime.strptime(run_date_str, '%Y-%m-%d') + datetime.timedelta(days=1) run_date = run_date.date() else: run_date = datetime.date.today() while run_date <= datetime.date.today(): ref_date = run_date + datetime.timedelta(days=2) with open('tasks.yml', 'r') as content_file: tasks_conf = load(content_file.read(), Loader=FullLoader) task_list = [] for task_name, task_data in tasks_conf.items(): t = Task(ref_date, task_name, task_data) if t.date_begin.date() == ref_date: task_list.append(t) if task_list: server = conf['dav_server'] client = Client(server['url'], server['user'], server['pass'], server['calendar']) previous_tasks = client.todos(include_completed=True) for task in task_list: if not client.get_todo_by_uid(task.uid): # Removing previous completed tasks for t in previous_tasks: if t.icalendar_component.get('uid').startswith(f"{APP_ID}-{task.name}")\ and t.icalendar_component.get('status') == 'COMPLETED': t.delete() client.add_todo(task) print('process finished for date {}: {} tasks created\n'.format(run_date, len(task_list))) run_date = run_date + datetime.timedelta(days=1) with open(last_run_file_path, 'w') as last_run_file: last_run_file.write(str(datetime.date.today()))