118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from ansible.module_utils.basic import AnsibleModule
|
|
from caldav import DAVClient
|
|
import vobject
|
|
import requests
|
|
from urllib.parse import urlparse
|
|
|
|
def fetch_ics_events(url, user=None, password=None):
|
|
try:
|
|
response = requests.get(url, auth=(user, password) if user else None)
|
|
response.raise_for_status()
|
|
events = []
|
|
for calendar in vobject.readComponents(response.text):
|
|
for component in calendar.components():
|
|
if component.name == 'VEVENT':
|
|
events.append(component)
|
|
return events
|
|
except Exception as e:
|
|
raise Exception(f"ICS-Fehler: {str(e)}")
|
|
|
|
def connect_caldav(url, user=None, password=None, verify_ssl=True):
|
|
try:
|
|
client = DAVClient(
|
|
url,
|
|
username=user,
|
|
password=password,
|
|
ssl_verify_cert=verify_ssl
|
|
)
|
|
target_path = urlparse(url).path
|
|
principal = client.principal()
|
|
|
|
for calendar in principal.calendars():
|
|
if target_path in str(calendar.url):
|
|
return calendar
|
|
raise Exception(f"Kalender mit Pfad '{target_path}' nicht gefunden")
|
|
except Exception as e:
|
|
raise Exception(f"CalDAV-Fehler: {str(e)}")
|
|
|
|
def sync_ics_to_caldav(module):
|
|
ics_events = fetch_ics_events(
|
|
module.params['source_url'],
|
|
module.params['source_user'],
|
|
module.params['source_password']
|
|
)
|
|
|
|
target_cal = connect_caldav(
|
|
module.params['target_url'],
|
|
module.params['target_user'],
|
|
module.params['target_password'],
|
|
module.params['verify_ssl']
|
|
)
|
|
|
|
existing_events = {}
|
|
for event in target_cal.events():
|
|
if event is None:
|
|
continue
|
|
data = getattr(event, 'data', None)
|
|
if not data:
|
|
continue
|
|
try:
|
|
for calendar in vobject.readComponents(data):
|
|
for component in calendar.components():
|
|
if component.name == 'VEVENT' and hasattr(component, 'uid'):
|
|
uid = str(component.uid)
|
|
existing_events[uid] = event
|
|
except Exception:
|
|
continue
|
|
|
|
changed = False
|
|
results = {'added': [], 'updated': [], 'removed': []}
|
|
|
|
for vevent in ics_events:
|
|
uid = str(vevent.uid)
|
|
ical_data = vevent.serialize()
|
|
|
|
if uid not in existing_events:
|
|
target_cal.add_event(ical_data)
|
|
results['added'].append(uid)
|
|
changed = True
|
|
elif ical_data != existing_events[uid].data:
|
|
existing_events[uid].data = ical_data
|
|
existing_events[uid].save()
|
|
results['updated'].append(uid)
|
|
changed = True
|
|
|
|
if module.params['purge']:
|
|
current_uids = {str(e.uid) for e in ics_events}
|
|
for uid in set(existing_events.keys()) - current_uids:
|
|
existing_events[uid].delete()
|
|
results['removed'].append(uid)
|
|
changed = True
|
|
|
|
return changed, results
|
|
|
|
def run_module():
|
|
module_args = dict(
|
|
source_url=dict(type='str', required=True),
|
|
source_user=dict(type='str', required=False, default=None),
|
|
source_password=dict(type='str', required=False, no_log=True, default=None),
|
|
target_url=dict(type='str', required=True),
|
|
target_user=dict(type='str', required=False, default=None),
|
|
target_password=dict(type='str', required=False, no_log=True, default=None),
|
|
verify_ssl=dict(type='bool', default=True),
|
|
purge=dict(type='bool', default=False)
|
|
)
|
|
|
|
module = AnsibleModule(argument_spec=module_args, supports_check_mode=False)
|
|
|
|
try:
|
|
changed, results = sync_ics_to_caldav(module)
|
|
module.exit_json(changed=changed, **results)
|
|
except Exception as e:
|
|
module.fail_json(msg=str(e))
|
|
|
|
if __name__ == '__main__':
|
|
run_module()
|