midas/playbooks
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

library/my_calendar_sync.py aktualisiert

Browse Source
This commit is contained in:
midas
2025-04-30 08:43:19 +00:00
parent a149988a3d
commit aa0e33dc88
1 changed files with 16 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
library/my_calendar_sync.py
View File

@ -1,28 +1,17 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
from ansible.module_utils.basic import AnsibleModule from ansible.module_utils.basic import AnsibleModule
import subprocess
import sys
# Auto-Install, falls caldav fehlt
try:
from caldav import DAVClient from caldav import DAVClient
import vobject import vobject
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "caldav", "vobject"])
from caldav import DAVClient
import vobject
def connect_calendar(url, user=None, password=None): def connect_calendar(url, user=None, password=None):
client = DAVClient(url, username=user, password=password) if user else DAVClient(url) client = DAVClient(url, username=user, password=password) if user else DAVClient(url)
principal = client.principal() principal = client.principal()
calendars = principal.calendars() calendars = principal.calendars()
if not calendars: if not calendars:
raise Exception(f"Kein Kalender gefunden unter {url}") raise Exception(f"Kein Kalender unter {url} gefunden")
return calendars[0] return calendars[0]
def get_event_uid_map(calendar): def get_event_uid_map(calendar):
uid_map = {} uid_map = {}
for event in calendar.events(): for event in calendar.events():
@ -34,7 +23,6 @@ def get_event_uid_map(calendar):
continue continue
return uid_map return uid_map
def sync_calendars(source_cal, target_cal): def sync_calendars(source_cal, target_cal):
src_events = get_event_uid_map(source_cal) src_events = get_event_uid_map(source_cal)
tgt_events = get_event_uid_map(target_cal) tgt_events = get_event_uid_map(target_cal)
@ -61,52 +49,36 @@ def sync_calendars(source_cal, target_cal):
return changed, added, updated, removed return changed, added, updated, removed
def run_module(): def run_module():
module_args = dict( module_args = dict(
source_url=dict(type='str', required=True), source_url=dict(type='str', required=True),
source_user=dict(type='str', required=False, default=None), source_user=dict(type='str', required=False, default=None),
source_password=dict(type='str', required=False, default=None, no_log=True), source_password=dict(type='str', required=False, no_log=True, default=None),
target_url=dict(type='str', required=True), target_url=dict(type='str', required=True),
target_user=dict(type='str', required=True), target_user=dict(type='str', required=True),
target_password=dict(type='str', required=True, no_log=True), target_password=dict(type='str', required=True, no_log=True)
) )
result = dict( result = dict(changed=False)
changed=False,
added=[],
updated=[],
removed=[],
)
module = AnsibleModule(argument_spec=module_args, supports_check_mode=False) module = AnsibleModule(argument_spec=module_args, supports_check_mode=False)
try: try:
source_cal = connect_calendar( source_cal = connect_calendar(module.params['source_url'], module.params['source_user'], module.params['source_password'])
module.params['source_url'], target_cal = connect_calendar(module.params['target_url'], module.params['target_user'], module.params['target_password'])
module.params['source_user'],
module.params['source_password']
)
target_cal = connect_calendar(
module.params['target_url'],
module.params['target_user'],
module.params['target_password']
)
changed, added, updated, removed = sync_calendars(source_cal, target_cal) changed, added, updated, removed = sync_calendars(source_cal, target_cal)
result['changed'] = changed result.update({
result['added'] = added "changed": changed,
result['updated'] = updated "added": added,
result['removed'] = removed "updated": updated,
"removed": removed,
})
module.exit_json(**result)
except Exception as e: except Exception as e:
module.fail_json(msg=str(e), **result) module.fail_json(msg=str(e), **result)
module.exit_json(**result)
def main():
run_module()
if __name__ == '__main__': if __name__ == '__main__':
main() run_module()