import uuid from odoo import models, api, fields from odoo.exceptions import UserError import caldav import logging from datetime import datetime from icalendar import Calendar, Event, vCalAddress, vText from bs4 import BeautifulSoup import re from pytz import timezone, utc from caldav.elements.cdav import CompFilter _logger = logging.getLogger(__name__) WEEKDAY_MAP = { 0: "MO", 1: "TU", 2: "WE", 3: "TH", 4: "FR", 5: "SA", 6: "SU", } def _parse_rrule_string(rrule_str): def try_to_int(part): try: return int(part) except Exception: return part regex_str = "RRULE:(.*)$" regex = re.compile(regex_str) params_match = regex.search(rrule_str) params_part = params_match.groups()[0] params = params_part.split(";") params_dict = {} for param in params: parts = param.split("=") params_dict.update({parts[0]: try_to_int(parts[1])}) return params_dict class CalendarEvent(models.Model): _inherit = "calendar.event" caldav_uid = fields.Char(string="CalDAV UID", readonly=True) caldav_recurrence_id = fields.Char(string="CalDAV Recurrence ID", readonly=True) @api.model def create(self, vals): if not vals.get("caldav_uid"): vals["caldav_uid"] = str(uuid.uuid4()) event = super(CalendarEvent, self).create(vals) if not self.env.context.get("caldav_no_sync"): try: _logger.debug(f"Creating event {event.name} in CalDAV") event.sync_create_to_caldav() except Exception as e: _logger.error(f"Failed to create event in CalDAV server: {e}") return event def write(self, vals): res = super(CalendarEvent, self).write(vals) if not self.env.context.get("caldav_no_sync") and self.ids: for rec in self: try: _logger.debug(f"Updating event {self.name} in CalDAV") rec.sync_update_to_caldav() except Exception as e: _logger.error(f"Failed to update event in CalDAV server: {e}") return res def unlink(self): if not self.env.context.get("caldav_no_sync"): for event in self: try: _logger.debug(f"Removing event {event.name} from CalDAV") event.sync_remove_from_caldav() except Exception as e: _logger.error(f"Failed to delete event from CalDAV server: {e}") return super(CalendarEvent, self).unlink() def _is_caldav_enabled(self): return self.env.user.is_caldav_enabled() def _get_caldav_client(self): user = self.env.user return caldav.DAVClient( url=user.caldav_calendar_url, username=user.caldav_username, password=user.caldav_password, ) def sync_create_to_caldav(self): if not self._is_caldav_enabled(): return client = self._get_caldav_client() calendar = client.calendar(url=self.env.user.caldav_calendar_url) for event in self: ical_event = event._get_icalendar() try: _logger.debug(f"Creating new CalDAV event for {event.name}") caldav_event = calendar.add_event(ical_event) caldav_uid = caldav_event.vobject_instance.vevent.uid.value _logger.debug(f"New CalDAV UID: {caldav_uid}") event.with_context(caldav_no_sync=True).write( {"caldav_uid": caldav_uid} ) except Exception as e: _logger.error(f"Failed to sync event to CalDAV server: {e}") def sync_update_to_caldav(self): if not self._is_caldav_enabled(): return client = self._get_caldav_client() calendar = client.calendar(url=self.env.user.caldav_calendar_url) for event in self: ical_event = event._get_icalendar() try: _logger.debug(f"Updating existing CalDAV event {event.caldav_uid}") calendar.save_event(ical=ical_event) except Exception as e: _logger.error(f"Failed to sync event to CalDAV server: {e}") def sync_remove_from_caldav(self): if not self._is_caldav_enabled(): return client = self._get_caldav_client() calendar = client.calendar(url=self.env.user.caldav_calendar_url) for event in self: if event.caldav_uid: try: _logger.debug(f"Removing CalDAV event {event.caldav_uid}") event._get_icalendar().delete() except caldav.error.NotFoundError: _logger.warning( f"CalDAV event {event.caldav_uid} not found on server." ) except Exception as e: _logger.error(f"Failed to remove event from CalDAV server: {e}") def _get_icalendar(self): calendar = Calendar() calendar.add("prodid", "-//Odoo//mxm.dk//") calendar.add("version", "2.0") for event in self: user_tz = timezone("UTC") if event.user_id.tz: user_tz = timezone(event.user_id.tz) ical_event = Event() ical_event.add("uid", event.caldav_uid) ical_event.add( "dtstamp", utc.localize(event.write_date).astimezone(user_tz) ) if event.name: ical_event.add("summary", event.name) if event.description and self._html_to_text(event.description): ical_event.add("description", self._html_to_text(event.description)) if event.location: ical_event.add("location", event.location) if event.videocall_location: ical_event.add("CONFERENCE", event.videocall_location) for partner in event.partner_ids: if partner == event.user_id.partner_id: continue attendee = vCalAddress(f"MAILTO:{partner.email}") attendee.params["cn"] = vText(partner.name) attendee_record = self.env["calendar.attendee"].search( [("event_id", "=", event.id), ("partner_id", "=", partner.id)], limit=1, ) if attendee_record: attendee.params["partstat"] = vText( self._map_attendee_status(attendee_record.state) ) ical_event.add("attendee", attendee, encode=0) organizer = vCalAddress(f"MAILTO:{event.user_id.email}") organizer.params["cn"] = event.user_id.name ical_event.add("organizer", organizer) # Add RRULE if the event is recurrent if event.recurrency and event.recurrence_id: rrule = event.recurrence_id._get_rrule() rrule_dict = _parse_rrule_string(str(rrule)) ical_event.add("rrule", rrule_dict) # Add DTSTART and DTEND ical_event.add("dtstart", utc.localize(event.start).astimezone(user_tz)) ical_event.add("dtend", utc.localize(event.stop).astimezone(user_tz)) calendar.add_component(ical_event) return calendar.to_ical() @api.model def poll_caldav_server(self): all_users = ( self.env["res.users"].search([]).filtered(lambda u: u.is_caldav_enabled()) ) for user in all_users: self.with_user(user).poll_user_caldav_server() @api.model def poll_user_caldav_server(self): if not self._is_caldav_enabled(): return client = self._get_caldav_client() try: calendar = client.calendar(url=self.env.user.caldav_calendar_url) events = calendar.events() except Exception as e: _logger.error(e) try: principal = client.principal() msg = f"""Failed to connect to the calendar, but successfully connected to the server at {client.url}. You may need to select another calendar URL from those below. Available calendars: """ for calendar in principal.calendars(): msg += f"{calendar.name}: {calendar.url}\n" raise UserError(msg) except Exception as e: _logger.error(e) return caldav_uids = set() _logger.info(f"Polling CalDAV server for user {self.env.user.name}") for caldav_event in events: ical_event = caldav_event.icalendar_instance self.sync_event_from_ical(ical_event) for component in ical_event.subcomponents: if isinstance(component, Event): uid = str(component.get("uid")) recurrence_id = str(component.get("recurrence-id")) if recurrence_id == "None": recurrence_id = "" caldav_uids.add(f"{uid}{recurrence_id}") _logger.info(f"CalDAV UIDs fetched: {caldav_uids}") # Remove Odoo events that no longer exist on the CalDAV server odoo_events = self.search([("caldav_uid", "!=", False)]) for event in odoo_events: recurrence_id = event.caldav_recurrence_id or "" event_uid = f"{event.caldav_uid}{event.caldav_recurrence_id or ''}" if event_uid not in caldav_uids: _logger.info( f"Deleting orphan event {event.name} with UID {event.caldav_uid} " f"and Recurrence ID {event.caldav_recurrence_id or ''}" ) event.with_context(caldav_no_sync=True).unlink() @api.model def _get_existing_instance(self, uid, recurrence_id): instance = self.env["calendar.event"].search( [("caldav_uid", "=", uid), ("recurrence_id", "=", recurrence_id)] ) return instance or self.env["calendar.event"].search( [ ("caldav_uid", "=", "uid"), ("recurrence_id", "=", False), ] ) def _get_recurrency_values_from_ical_event(self, component): """Match the fields from calendar.event (recurring fields) to the fields specified in RRULE at https://icalendar.org/iCalendar-RFC-5545/3-8-5-3-recurrence-rule.html""" rrule = component.get("rrule") if not rrule: if not self.recurrency: # No change, this was already not a recurring event return {} else: # This was a recurring event and has been made non-recurring if self.recurrence_id.base_event_id != self: # This is not the base event, so change its recurrency only return { "recurrence_update": "self_only", "recurrency": False, "follow_recurrence": False, } else: # This is the base event, so change all events in the list return {"recurrence_update": "all_events", "recurrency": False} rrule_str = rrule.to_ical().decode("utf-8") sequence = component.get("sequence") if sequence and sequence != 0: # This is not the base event so we can't change recurrence properties return {} caldav_recurrence_id = component.get("recurrence-id") rrule_params = self.env["calendar.recurrence"]._rrule_parse( rrule_str, component.decoded("dtstart") ) vals = { "recurrency": True, "follow_recurrence": True, "caldav_recurrence_id": caldav_recurrence_id, "recurrence_update": "all_events", "rrule_type": rrule_params.get("rrule_type"), "end_type": rrule_params.get("end_type"), "interval": rrule_params.get("interval"), "count": rrule_params.get("count"), "month_by": rrule_params.get("monty_by"), "day": rrule_params.get("day"), "byday": rrule_params.get("byday"), "until": rrule_params.get("until"), } if rrule_params.get("weekday"): vals.update(rrule_params.get("weekday")) day_list = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] vals.update( {day: rrule_params.get(day) for day in day_list if day in rrule_params} ) return vals def sync_event_from_ical(self, ical_event): email_regex = re.compile(r"[a-z0-9.\-+_]+@[a-z0-9.\-+_]+\.[a-z]+") current_user_email = self.env.user.email.lower() for component in ical_event.subcomponents: if isinstance(component, Event): uid = component.get("uid") recurrence_id = component.get( "recurrence_id" ) # Unique identifier for a single event in a recurrence set attendees = component.get("attendee", []) if isinstance(attendees, vCalAddress): attendees = [attendees] elif isinstance(attendees, str): attendees = [vCalAddress(attendees)] attendees_emails = [ email_regex.search(str(attendee)).group(0).lower().strip() for attendee in attendees if email_regex.search(str(attendee)) ] _logger.info(f"Attendees emails: {attendees_emails}") # Add current user email to attendees if not already present if current_user_email not in attendees_emails: attendees_emails.append(current_user_email) attendee_ids = self.env["res.partner"].search( [("email", "in", attendees_emails)] ) existing_instance = self._get_existing_instance(uid, recurrence_id) start = component.decoded("dtstart") if isinstance(start, datetime): start = start.astimezone(utc).replace(tzinfo=None) end = component.decoded("dtend") if isinstance(end, datetime): end = end.astimezone(utc).replace(tzinfo=None) values = { "name": str(component.get("summary")), "start": start, "stop": end, "description": self._extract_component_text( component, "description" ), "location": self._extract_component_text(component, "location"), "videocall_location": self._extract_component_text( component, "conference" ), "caldav_uid": uid, "partner_ids": [(6, 0, attendee_ids.ids)], } recurrency_vals = self._get_recurrency_values_from_ical_event(component) if recurrency_vals: values.update(recurrency_vals) if not existing_instance: _logger.info(f"Creating with vals: {values}") self.with_context(caldav_no_sync=True).create(values) else: _logger.info(f"Updating with vals: {values}") changed_vals = {} # Don't update partner_ids if no change if attendee_ids - existing_instance.partner_ids: changed_vals.update( { "partner_ids", values.pop("partner_ids"), } ) # Don't write values that haven't changed for key, val in values.items(): if getattr(existing_instance, key) != val: changed_vals.update({key: values.get(key)}) if ( recurrency_vals and recurrency_vals.get("recurrency") and ( not existing_instance.recurrency or not existing_instance.follow_recurrence ) ): existing_instance.write( { "recurrency": True, "follow_recurrence": True, } ) existing_instance.with_context( caldav_no_sync=True, ).write(changed_vals) @staticmethod def _extract_component_text(component, subcomponent_name): text = str(component.get(subcomponent_name)) text = text if text != "None" else "" @staticmethod def _html_to_text(html): return BeautifulSoup(html, "html.parser").getText() @staticmethod def _map_attendee_status(state): mapping = { "needsAction": "NEEDS-ACTION", "accepted": "ACCEPTED", "declined": "DECLINED", "tentative": "TENTATIVE", } return mapping.get(state, "NEEDS-ACTION") @staticmethod def _map_ical_status(ical_status): mapping = { "NEEDS-ACTION": "needsAction", "ACCEPTED": "accepted", "DECLINED": "declined", "TENTATIVE": "tentative", } return mapping.get(ical_status, "needsAction")