125 lines
4.7 KiB
Python
Executable File
125 lines
4.7 KiB
Python
Executable File
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)
|
|
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()))
|