aacc6b27b4
Add a type check to the values being compared when iterating over current and changed values in a calendar event. Also, rewrite to use dictionary comprehension for readability.
450 lines
18 KiB
Python
450 lines
18 KiB
Python
import uuid
|
|
from odoo import models, api, fields, Command
|
|
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_create_multi
|
|
def create(self, vals_list):
|
|
events = self.env['calendar.event']
|
|
for vals in vals_list:
|
|
if not vals.get("caldav_uid"):
|
|
vals["caldav_uid"] = str(uuid.uuid4())
|
|
events = super(CalendarEvent, self).create(vals_list)
|
|
if not self.env.context.get("caldav_no_sync"):
|
|
events.sync_create_to_caldav()
|
|
return events
|
|
|
|
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
|
|
changed_vals = {
|
|
key: val
|
|
for key, val in values.items()
|
|
if (cur_val := getattr(existing_instance, key))
|
|
and isinstance(cur_val, type(val))
|
|
and cur_val != val
|
|
|
|
}
|
|
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")
|