cf7a59ad98
- Created models for client applications, application specifications and application types. - Created a small technical module for adding a self-incrementing sequence field to other models. See incrementing_sequence_mixin.
492 lines
20 KiB
Python
492 lines
20 KiB
Python
import uuid
|
|
from odoo import models, api, fields, Command
|
|
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
|
|
|
|
_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
|
|
|
|
|
|
def _extract_vcal_email(vcal_address):
|
|
email_regex = re.compile(r"[a-z0-9.\-+_]+@[a-z0-9.\-+_]+\.[a-z]+")
|
|
res = email_regex.search(str(vcal_address))
|
|
return res.group(0).lower().strip() if res else ""
|
|
|
|
|
|
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)
|
|
caldav_user_ids = fields.Many2many(
|
|
comodel_name="res.users",
|
|
compute="_compute_caldav_users",
|
|
)
|
|
|
|
@api.depends("user_id", "partner_ids", "partner_ids.user_id")
|
|
def _compute_caldav_users(self):
|
|
for rec in self:
|
|
rec.caldav_user_ids = (rec.user_id | rec.partner_ids.user_ids).filtered(
|
|
"is_caldav_enabled"
|
|
)
|
|
|
|
@api.model_create_multi
|
|
def create(self, vals_list):
|
|
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.filtered(lambda event: event._is_caldav_enabled()):
|
|
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 rec in self.filtered(lambda event: event._is_caldav_enabled()):
|
|
try:
|
|
_logger.debug(f"Removing event {rec.name} from CalDAV")
|
|
rec._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 _sync_create_to_caldav(self):
|
|
for event in self:
|
|
ical_event = event._get_icalendar()
|
|
for user in event.caldav_user_ids:
|
|
client = user._get_caldav_client()
|
|
calendar = client.calendar(url=user.caldav_calendar_url)
|
|
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):
|
|
for event in self:
|
|
ical_event = event._get_icalendar()
|
|
for user in event.caldav_user_ids:
|
|
client = user._get_caldav_client()
|
|
calendar = client.calendar(url=self.env.user.caldav_calendar_url)
|
|
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):
|
|
for event in self:
|
|
if event.caldav_uid:
|
|
for user in event.caldav_user_ids:
|
|
client = user._get_caldav_client()
|
|
calendar = client.calendar(url=self.env.user.caldav_calendar_url)
|
|
try:
|
|
_logger.debug(f"Removing CalDAV event {event.caldav_uid}")
|
|
calendar.event_by_uid(event.caldav_uid).delete()
|
|
# 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(datetime.now()).astimezone(user_tz))
|
|
ical_event.add(
|
|
"last-modified", utc.localize(self.write_date).astimezone(user_tz)
|
|
)
|
|
ical_event.add(
|
|
"created", utc.localize(self.create_date).astimezone(user_tz)
|
|
)
|
|
if event.name:
|
|
ical_event.add("summary", event.name)
|
|
# TODO: Consider using X-ALT-DESC to stick HTML into the iCal event desc.
|
|
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(name="attendee", value=attendee, encode=False)
|
|
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([("is_caldav_enabled", "=", True)])
|
|
for user in all_users:
|
|
self._poll_user_caldav_server(user)
|
|
|
|
@api.model
|
|
def _poll_user_caldav_server(self, user):
|
|
_logger.info(f"Polling CalDAV server for user {user.name}")
|
|
events = user._get_caldav_events()
|
|
caldav_uids = set()
|
|
|
|
for caldav_event in events:
|
|
ical_event = caldav_event.icalendar_instance
|
|
caldav_uids |= self._sync_event_from_ical(ical_event, user)
|
|
|
|
_logger.info(f"CalDAV UIDs fetched: {caldav_uids}")
|
|
|
|
# Remove Odoo events that no longer exist on the CalDAV server
|
|
# TODO: check if this fails when the user is deleting someone else's event
|
|
# TODO: check if we should send updates to invitees
|
|
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 {recurrence_id}"
|
|
)
|
|
event.with_context(caldav_no_sync=True).with_user(user).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 = [item[1] for item in component.property_items() if item[0] == "RRULE"]
|
|
rrule = rrule[0] if rrule else None
|
|
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() and rrule.to_ical().decode("utf-8")
|
|
sequence = component.get("sequence")
|
|
if sequence and sequence != 1:
|
|
# 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, user):
|
|
current_user_email = user.email.lower()
|
|
caldav_uids = set()
|
|
event_components = [
|
|
component for component in ical_event.walk() if component.name == "VEVENT"
|
|
]
|
|
|
|
for component in event_components:
|
|
uid = component.get("uid")
|
|
recurrence_id = component.get(
|
|
"recurrence_id"
|
|
) # Unique identifier for a single event in a recurrence set
|
|
partner_ids = self._get_attendee_partners(component, current_user_email)
|
|
|
|
existing_instance = self._get_existing_instance(uid, recurrence_id)
|
|
outdated = False
|
|
last_modified = component.get("last-modified") and component.decoded(
|
|
"last-modified"
|
|
)
|
|
if existing_instance and last_modified:
|
|
last_modified = last_modified.astimezone(utc).replace(tzinfo=None)
|
|
if last_modified < existing_instance.write_date:
|
|
# _logger.info(
|
|
# f"Last modified date {last_modified} is before most recent "
|
|
# f"write date {existing_instance.write_date}. Skipping."
|
|
# )
|
|
outdated = True
|
|
owned = (
|
|
existing_instance and existing_instance.partner_id == user.partner_id
|
|
)
|
|
values, recurrency_vals = (
|
|
self._get_vals_recurrency_vals_from_ical_component(
|
|
partner_ids, component, user
|
|
)
|
|
)
|
|
changed_vals = {}
|
|
if not existing_instance:
|
|
_logger.info(f"Creating with vals: {values}")
|
|
self.with_context(caldav_no_sync=True).create(values)
|
|
elif outdated or not owned:
|
|
_logger.info(
|
|
f"Event {existing_instance.caldav_uid} "
|
|
f"{'outdated ' if outdated else ''}"
|
|
f"{'not owned by user ' + user.name if not owned else ''}."
|
|
f" Skipping."
|
|
)
|
|
pass # Do nothing, it's not this user's event to modify or it's outdated
|
|
else:
|
|
# Don't update partner_ids if no change
|
|
if partner_ids != existing_instance.partner_ids:
|
|
changed_vals.update(partner_ids=values.pop("partner_ids"))
|
|
updated_attendees = existing_instance.attendee_ids.filtered(
|
|
lambda rec: rec.partner_id in partner_ids
|
|
)
|
|
changed_vals.update(
|
|
attendee_ids=[Command.set(updated_attendees.ids)]
|
|
)
|
|
else:
|
|
values.pop("partner_ids") # They break the equality check later
|
|
|
|
# Get just the list of values that have changed, leave the others alone
|
|
for key, val in values.items():
|
|
curr_val = getattr(existing_instance, key)
|
|
# Can't deal with x2many fields, need ID from a record
|
|
if isinstance(val, list):
|
|
continue
|
|
if curr_val and isinstance(curr_val, models.Model):
|
|
if len(curr_val) > 1:
|
|
continue
|
|
curr_val = curr_val.id
|
|
if curr_val != val:
|
|
changed_vals.update({key: val})
|
|
|
|
self._update_event_recurrence(existing_instance, recurrency_vals)
|
|
_logger.info(f"Updating with : {changed_vals}")
|
|
existing_instance.with_context(
|
|
caldav_no_sync=True,
|
|
).write(changed_vals)
|
|
recurrence_id = str(component.get("recurrence-id"))
|
|
if recurrence_id == "None":
|
|
recurrence_id = ""
|
|
caldav_uids.add(f"{uid}{recurrence_id}")
|
|
return caldav_uids
|
|
|
|
@staticmethod
|
|
def _update_event_recurrence(existing_instance, recurrency_vals):
|
|
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,
|
|
}
|
|
)
|
|
|
|
def _get_vals_recurrency_vals_from_ical_component(
|
|
self, attendee_ids, component, user
|
|
):
|
|
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)
|
|
organizer = self._get_organizer_partner(component)
|
|
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": component.get("uid"),
|
|
"partner_ids": [(6, 0, attendee_ids.ids)],
|
|
"partner_id": organizer.id if organizer else False,
|
|
"user_id": user.id,
|
|
}
|
|
recurrency_vals = self._get_recurrency_values_from_ical_event(component)
|
|
if recurrency_vals:
|
|
values.update(recurrency_vals)
|
|
return values, recurrency_vals
|
|
|
|
def _get_attendee_partners(self, component, current_user_email):
|
|
attendee_emails = self._get_ical_attendee_emails(component)
|
|
if current_user_email not in attendee_emails:
|
|
attendee_emails.append(current_user_email)
|
|
existing_partners = self.env["res.partner"].search(
|
|
[("email", "in", attendee_emails)]
|
|
)
|
|
missing_emails = [
|
|
email
|
|
for email in attendee_emails
|
|
if email not in [partner.email for partner in existing_partners]
|
|
]
|
|
added_partners = self.env["res.partner"].create(
|
|
[
|
|
{
|
|
"name": email,
|
|
"email": email,
|
|
}
|
|
for email in missing_emails
|
|
]
|
|
)
|
|
return existing_partners | added_partners
|
|
|
|
def _get_organizer_partner(self, component):
|
|
organizer = component.get("organizer")
|
|
if organizer:
|
|
partner = self.env["res.partner"].search(
|
|
[("email", "=", _extract_vcal_email(organizer))]
|
|
)
|
|
# TODO: prioritize partner with a user if there is one
|
|
return partner[0] if partner else partner # partner[0] in case many matches
|
|
else:
|
|
return self.env["res.partner"]
|
|
|
|
@staticmethod
|
|
def _get_ical_attendee_emails(component):
|
|
attendees = component.get("attendee", [])
|
|
if not isinstance(attendees, list):
|
|
attendees = [attendees]
|
|
attendee_emails = [_extract_vcal_email(attendee) for attendee in attendees]
|
|
return attendee_emails
|
|
|
|
@staticmethod
|
|
def _extract_component_text(component, subcomponent_name):
|
|
val = component.get(subcomponent_name)
|
|
text = str(val) if val else ""
|
|
return text
|
|
|
|
@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")
|