Page MenuHomePhorge
Differential D785 Diff 1829 pykolab/itip/__init__.py

Changeset View

Standalone View

View Options

pykolab/itip/__init__.py

import icalendar import re
import pykolab
import traceback import traceback
import icalendar
import kolabformat import kolabformat
import re
import pykolab
from pykolab.translate import _
from pykolab.xml import to_dt from pykolab.xml import to_dt
from pykolab.xml import event_from_ical from pykolab.xml import event_from_ical
from pykolab.xml import todo_from_ical from pykolab.xml import todo_from_ical
from pykolab.xml import participant_status_label from pykolab.xml import participant_status_label
from pykolab.translate import _
from tzlocal import windows_tz from tzlocal import windows_tz
# pylint: disable=invalid-name
log = pykolab.getLogger('pykolab.wallace') log = pykolab.getLogger('pykolab.wallace')
def events_from_message(message, methods=None): def events_from_message(message, methods=None):
return objects_from_message(message, ["VEVENT"], methods) return objects_from_message(message, ["VEVENT"], methods)
def todos_from_message(message, methods=None): def todos_from_message(message, methods=None):
return objects_from_message(message, ["VTODO"], methods) return objects_from_message(message, ["VTODO"], methods)
def objects_from_message(message, objnames, methods=None):
# pylint: disable=too-many-branches
# pylint: disable=too-many-statements
def objects_from_message(message, objnames, methods=None): # noqa: C901
""" """
Obtain the iTip payload from email.message <message> Obtain the iTip payload from email.message <message>
""" """
# Placeholder for any itip_objects found in the message. # Placeholder for any itip_objects found in the message.
itip_objects = [] itip_objects = []
seen_uids = [] seen_uids = []
# iTip methods we are actually interested in. Other methods will be ignored. # iTip methods we are actually interested in. Other methods will be ignored.
if methods is None: if methods is None:
methods = [ "REQUEST", "CANCEL" ] methods = ["REQUEST", "CANCEL"]
# Are all iTip messages multipart? No! RFC 6047, section 2.4 states "A # Are all iTip messages multipart? No! RFC 6047, section 2.4 states "A
# MIME body part containing content information that conforms to this # MIME body part containing content information that conforms to this
# document MUST have (...)" but does not state whether an iTip message must # document MUST have (...)" but does not state whether an iTip message must
# therefore also be multipart. # therefore also be multipart.
# Check each part # Check each part
# pylint: disable=too-many-nested-blocks
for part in message.walk(): for part in message.walk():
# The iTip part MUST be Content-Type: text/calendar (RFC 6047, section 2.4) # The iTip part MUST be Content-Type: text/calendar (RFC 6047, section 2.4)
# But in real word, other mime-types are used as well # But in real word, other mime-types are used as well
if part.get_content_type() in [ "text/calendar", "text/x-vcalendar", "application/ics" ]: if part.get_content_type() in ["text/calendar", "text/x-vcalendar", "application/ics"]:
if not str(part.get_param('method')).upper() in methods: if str(part.get_param('method')).upper() not in methods:
log.info(_("Method %r not really interesting for us.") % (part.get_param('method'))) log.info("Method %r not really interesting for us." % (part.get_param('method')))
continue continue
# Get the itip_payload # Get the itip_payload
itip_payload = part.get_payload(decode=True) itip_payload = part.get_payload(decode=True)
log.debug(_("Raw iTip payload (%r): %r") % (part.get_param('charset'), itip_payload), level=8) log.debug(
"Raw iTip payload (%r): %r" % (part.get_param('charset'), itip_payload),
level=8
)
# Convert unsupported timezones, etc. # Convert unsupported timezones, etc.
itip_payload = _convert_itip_payload(itip_payload) itip_payload = _convert_itip_payload(itip_payload)
# Python iCalendar prior to 3.0 uses "from_string". # Python iCalendar prior to 3.0 uses "from_string".
if hasattr(icalendar.Calendar, 'from_ical'): if hasattr(icalendar.Calendar, 'from_ical'):
cal = icalendar.Calendar.from_ical(itip_payload) cal = icalendar.Calendar.from_ical(itip_payload)
elif hasattr(icalendar.Calendar, 'from_string'): elif hasattr(icalendar.Calendar, 'from_string'):
Show All 23 Lines for part in message.walk():
# - organizer # - organizer
# - attendees (if any) # - attendees (if any)
# - resources (if any) # - resources (if any)
# #
itip['type'] = 'task' if c.name == 'VTODO' else 'event' itip['type'] = 'task' if c.name == 'VTODO' else 'event'
itip['uid'] = str(c['uid']) itip['uid'] = str(c['uid'])
itip['method'] = str(cal['method']).upper() itip['method'] = str(cal['method']).upper()
itip['sequence'] = int(c['sequence']) if c.has_key('sequence') else 0 itip['sequence'] = int(c['sequence']) if 'sequence' in c else 0
itip['recurrence-id'] = c['recurrence-id'].dt if c.has_key('recurrence-id') and hasattr(c['recurrence-id'], 'dt') else None
itip['recurrence-id'] = None
if 'recurrence-id' in c:
if hasattr(c['recurrence-id'], 'dt'):
itip['recurrence-id'] = c['recurrence-id'].dt
if c.has_key('dtstart'): if 'dtstart' in c:
itip['start'] = c['dtstart'].dt itip['start'] = c['dtstart'].dt
elif itip['type'] == 'event': elif itip['type'] == 'event':
log.error(_("iTip event without a start")) log.error(_("iTip event without a start"))
continue continue
if c.has_key('dtend'): if 'dtend' in c:
itip['end'] = c['dtend'].dt itip['end'] = c['dtend'].dt
if c.has_key('duration'): if 'duration' in c:
itip['duration'] = c['duration'].dt itip['duration'] = c['duration'].dt
itip['end'] = itip['start'] + c['duration'].dt itip['end'] = itip['start'] + c['duration'].dt
# Outlook can send itip replies with no organizer property # Outlook can send itip replies with no organizer property
if c.has_key('organizer'): if 'organizer' in c:
itip['organizer'] = c['organizer'] itip['organizer'] = c['organizer']
if c.has_key('attendee'): if 'attendee' in c:
itip['attendees'] = c['attendee'] itip['attendees'] = c['attendee']
if itip.has_key('attendees') and not isinstance(itip['attendees'], list): if 'attendees' in itip and not isinstance(itip['attendees'], list):
itip['attendees'] = [c['attendee']] itip['attendees'] = [c['attendee']]
if c.has_key('resources'): if 'resources' in c:
itip['resources'] = c['resources'] itip['resources'] = c['resources']
itip['raw'] = itip_payload itip['raw'] = itip_payload
try: try:
# distinguish event and todo here # distinguish event and todo here
if itip['type'] == 'task': if itip['type'] == 'task':
itip['xml'] = todo_from_ical(c, itip_payload) itip['xml'] = todo_from_ical(c, itip_payload)
else: else:
itip['xml'] = event_from_ical(c, itip_payload) itip['xml'] = event_from_ical(c, itip_payload)
except Exception, e:
log.error("event|todo_from_ical() exception: %r; iCal: %s" % (e, itip_payload)) # pylint: disable=broad-except
except Exception as e:
log.error(
"event|todo_from_ical() exception: %r; iCal: %s" % (e, itip_payload)
)
continue continue
itip_objects.append(itip) itip_objects.append(itip)
seen_uids.append(c['uid']) seen_uids.append(c['uid'])
# end if c.name in objnames # end if c.name in objnames
# end for c in cal.walk() # end for c in cal.walk()
# end if part.get_content_type() == "text/calendar" # end if part.get_content_type() == "text/calendar"
# end for part in message.walk() # end for part in message.walk()
if not len(itip_objects) and not message.is_multipart(): if not len(itip_objects) and not message.is_multipart():
log.debug(_("Message is not an iTip message (non-multipart message)"), level=5) log.debug(_("Message is not an iTip message (non-multipart message)"), level=5)
return itip_objects return itip_objects
def check_event_conflict(kolab_event, itip_event): def check_event_conflict(kolab_event, itip_event):
""" """
Determine whether the given kolab event conflicts with the given itip event Determine whether the given kolab event conflicts with the given itip event
""" """
conflict = False conflict = False
# don't consider conflict with myself # don't consider conflict with myself
if kolab_event.uid == itip_event['uid']: if kolab_event.uid == itip_event['uid']:
return conflict return conflict
# don't consider conflict if event has TRANSP:TRANSPARENT # don't consider conflict if event has TRANSP:TRANSPARENT
if _is_transparent(kolab_event): if _is_transparent(kolab_event):
return conflict return conflict
if _is_transparent(itip_event['xml']): if _is_transparent(itip_event['xml']):
return conflict return conflict
_es = to_dt(kolab_event.get_start()) _es = to_dt(kolab_event.get_start())
_ee = to_dt(kolab_event.get_ical_dtend()) # use iCal style end date: next day for all-day events # use iCal style end date: next day for all-day events
_ev = kolab_event _ee = to_dt(kolab_event.get_ical_dtend())
_ei = 0
_is = to_dt(itip_event['start']) _is = to_dt(itip_event['start'])
_ie = to_dt(itip_event['end']) _ie = to_dt(itip_event['end'])
_iv = itip_event['xml']
_ii = 0
# Escape looping through anything if neither of the events is recurring. # Escape looping through anything if neither of the events is recurring.
if not itip_event['xml'].is_recurring() and not kolab_event.is_recurring(): if not itip_event['xml'].is_recurring() and not kolab_event.is_recurring():
return check_date_conflict(_es, _ee, _is, _ie) return check_date_conflict(_es, _ee, _is, _ie)
loop = 0 loop = 0
done = False done = False
# naive loops to check for collisions in (recurring) events # naive loops to check for collisions in (recurring) events
# TODO: compare recurrence rules directly (e.g. matching time slot or weekday or monthday) # TODO: compare recurrence rules directly (e.g. matching time slot or weekday or monthday)
while not conflict and not done: while not conflict and not done:
loop += 1 loop += 1
# Scroll forward the kolab event recurrence until we're in the prime # Scroll forward the kolab event recurrence until we're in the prime
# spot. We choose to start with the Kolab event because that is likely # spot. We choose to start with the Kolab event because that is likely
# the older one. # the older one.
if _ee < _is: if _ee < _is:
while _ee < _is and _es is not None and kolab_event.is_recurring(): while _ee < _is and _es is not None and kolab_event.is_recurring():
log.debug("Attempt to move forward kolab event recurrence from %s closer to %s" % (_ee, _is), level=8) log.debug(
"Attempt to move forward kolab event recurrence from {} closer to {}".format(
_ee,
_is
),
level=8
)
__es = to_dt(kolab_event.get_next_occurence(_es)) __es = to_dt(kolab_event.get_next_occurence(_es))
if not __es is None: if __es is not None and not __es == _es:
_es = __es _es = __es
_ee = to_dt(kolab_event.get_occurence_end_date(_es)) _ee = to_dt(kolab_event.get_occurence_end_date(_es))
else: else:
done = True done = True
break break
# Scroll forward the itip event recurrence until we're in the # Scroll forward the itip event recurrence until we're in the
# prime spot, this time with the iTip event. # prime spot, this time with the iTip event.
elif _ie < _es: elif _ie < _es:
while _ie < _es and _is is not None and itip_event['xml'].is_recurring(): while _ie < _es and _is is not None and itip_event['xml'].is_recurring():
log.debug("Attempt to move forward itip event recurrence from %s closer to %s" % (_ie, _es), level=8) log.debug(
"Attempt to move forward itip event recurrence from {} closer to {}".format(
_ie,
_es
),
level=8
)
__is = to_dt(itip_event['xml'].get_next_occurence(_is)) __is = to_dt(itip_event['xml'].get_next_occurence(_is))
if not __is is None: if __is is not None and not _is == __is:
_is = __is _is = __is
_ie = to_dt(itip_event['xml'].get_occurence_end_date(_is)) _ie = to_dt(itip_event['xml'].get_occurence_end_date(_is))
else: else:
done = True done = True
break break
# Now that we have some events somewhere in the same neighborhood... # Now that we have some events somewhere in the same neighborhood...
conflict = check_date_conflict(_es, _ee, _is, _ie) conflict = check_date_conflict(_es, _ee, _is, _ie)
log.debug("* Comparing itip at %s/%s with kolab at %s/%s: %r (%d)" % (_is, _ie, _es, _ee, conflict, loop), level=8) log.debug(
"* Comparing itip at %s/%s with kolab at %s/%s: conflict - %r (occurence - %d)" % (
_is, _ie, _es, _ee, conflict, loop
),
level=8
)
if not conflict: if not conflict:
if kolab_event.is_recurring() and itip_event['xml'].is_recurring(): if kolab_event.is_recurring() and itip_event['xml'].is_recurring():
if not kolab_event.has_exceptions() and not itip_event['xml'].has_exceptions(): if not kolab_event.has_exceptions() and not itip_event['xml'].has_exceptions():
log.debug("No conflict, both recurring, but neither with exceptions", level=8) log.debug("No conflict, both recurring, but neither with exceptions", level=8)
done = True done = True
break break
_is = to_dt(itip_event['xml'].get_next_occurence(_is)) _is = to_dt(itip_event['xml'].get_next_occurence(_is))
if _is is not None: if _is is not None:
_ie = to_dt(itip_event['xml'].get_occurence_end_date(_is)) _ie = to_dt(itip_event['xml'].get_occurence_end_date(_is))
else: else:
done = True done = True
return conflict return conflict
def _is_transparent(event): def _is_transparent(event):
return event.get_transparency() or event.get_status() == kolabformat.StatusCancelled return event.get_transparency() or event.get_status() == kolabformat.StatusCancelled
def _convert_itip_payload(itip): def _convert_itip_payload(itip):
matchlist = re.findall("^((DTSTART|DTEND|DUE|EXDATE|COMPLETED)[:;][^\n]+)$", itip, re.MULTILINE) matchlist = re.findall("^((DTSTART|DTEND|DUE|EXDATE|COMPLETED)[:;][^\n]+)$", itip, re.MULTILINE)
for match in matchlist: for match in matchlist:
match = match[0] match = match[0]
search = re.search(";TZID=([^:;]+)", match) search = re.search(";TZID=([^:;]+)", match)
if search: if search:
Show All 11 Lines for match in matchlist:
# replace old with new timezone name # replace old with new timezone name
if tzorig != tzdest: if tzorig != tzdest:
replace = match.replace(search.group(0), ";TZID=" + tzdest) replace = match.replace(search.group(0), ";TZID=" + tzdest)
itip = itip.replace("\n" + match, "\n" + replace) itip = itip.replace("\n" + match, "\n" + replace)
return itip return itip
def check_date_conflict(_es, _ee, _is, _ie): def check_date_conflict(_es, _ee, _is, _ie):
""" """
Check the given event start/end dates for conflicts Check the given event start/end dates for conflicts
""" """
conflict = False conflict = False
# TODO: add margin for all-day dates (+13h; -12h) # TODO: add margin for all-day dates (+13h; -12h)
if _es < _is: if _es < _is:
if _es <= _ie: if _es <= _ie:
if _ee <= _is: if _ee <= _is:
conflict = False conflict = False
else: else:
conflict = True conflict = True
else: else:
conflict = True conflict = True
elif _es == _is: elif _es == _is:
conflict = True conflict = True
else: # _es > _is else: # _es > _is
if _es < _ie: if _es < _ie:
conflict = True conflict = True
else: else:
conflict = False conflict = False
return conflict return conflict
def send_reply(from_address, itip_events, response_text, subject=None): def send_reply(from_address, itip_events, response_text, subject=None):
""" """
Send the given iCal events as a valid iTip REPLY to the organizer. Send the given iCal events as a valid iTip REPLY to the organizer.
""" """
import smtplib import smtplib
conf = pykolab.getConf() conf = pykolab.getConf()
smtp = None smtp = None
if isinstance(itip_events, dict): if isinstance(itip_events, dict):
itip_events = [ itip_events ] itip_events = [itip_events]
for itip_event in itip_events: for itip_event in itip_events:
attendee = itip_event['xml'].get_attendee_by_email(from_address) attendee = itip_event['xml'].get_attendee_by_email(from_address)
participant_status = itip_event['xml'].get_ical_attendee_participant_status(attendee) participant_status = itip_event['xml'].get_ical_attendee_participant_status(attendee)
log.debug(_("Send iTip reply %s for %s %r") % (participant_status, itip_event['xml'].type, itip_event['xml'].uid), level=8) log.debug(
"Send iTip reply {} for {} {}".format(
participant_status,
itip_event['xml'].type,
itip_event['xml'].uid
),
level=8
)
event_summary = itip_event['xml'].get_summary() event_summary = itip_event['xml'].get_summary()
message_text = response_text % { 'summary':event_summary, 'status':participant_status_label(participant_status), 'name':attendee.get_name() } message_text = response_text % {
'summary': event_summary,
'status': participant_status_label(participant_status),
'name': attendee.get_name()
}
if subject is not None: if subject is not None:
subject = subject % { 'summary':event_summary, 'status':participant_status_label(participant_status), 'name':attendee.get_name() } subject = subject % {
'summary': event_summary,
'status': participant_status_label(participant_status),
'name': attendee.get_name()
}
try: try:
message = itip_event['xml'].to_message_itip(from_address, message = itip_event['xml'].to_message_itip(
from_address,
method="REPLY", method="REPLY",
participant_status=participant_status, participant_status=participant_status,
message_text=message_text, message_text=message_text,
subject=subject subject=subject
) )
except Exception, e:
log.error(_("Failed to compose iTip reply message: %r: %s") % (e, traceback.format_exc())) # pylint: disable=broad-except
except Exception as e:
log.error("Failed to compose iTip reply message: %r: %s" % (e, traceback.format_exc()))
return return
smtp = smtplib.SMTP("localhost", 10026) # replies go through wallace again smtp = smtplib.SMTP("localhost", 10026) # replies go through wallace again
if conf.debuglevel > 8: if conf.debuglevel > 8:
smtp.set_debuglevel(True) smtp.set_debuglevel(True)
try: try:
smtp.sendmail(message['From'], message['To'], message.as_string()) smtp.sendmail(message['From'], message['To'], message.as_string())
except Exception, e:
# pylint: disable=broad-except
except Exception as e:
log.error(_("SMTP sendmail error: %r") % (e)) log.error(_("SMTP sendmail error: %r") % (e))
if smtp: if smtp:
smtp.quit() smtp.quit()
def send_request(to_address, itip_events, request_text, subject=None, direct=False): def send_request(to_address, itip_events, request_text, subject=None, direct=False):
""" """
Send an iTip REQUEST message from the given iCal events Send an iTip REQUEST message from the given iCal events
""" """
import smtplib import smtplib
conf = pykolab.getConf() conf = pykolab.getConf()
smtp = None smtp = None
if isinstance(itip_events, dict): if isinstance(itip_events, dict):
itip_events = [ itip_events ] itip_events = [itip_events]
for itip_event in itip_events: for itip_event in itip_events:
event_summary = itip_event['xml'].get_summary() event_summary = itip_event['xml'].get_summary()
message_text = request_text % { 'summary':event_summary } message_text = request_text % {'summary': event_summary}
if subject is not None: if subject is not None:
subject = subject % { 'summary':event_summary } subject = subject % {'summary': event_summary}
try: try:
message = itip_event['xml'].to_message_itip(None, message = itip_event['xml'].to_message_itip(
None,
method="REQUEST", method="REQUEST",
message_text=message_text, message_text=message_text,
subject=subject subject=subject
) )
except Exception, e:
# pylint: disable=broad-except
except Exception as e:
log.error(_("Failed to compose iTip request message: %r") % (e)) log.error(_("Failed to compose iTip request message: %r") % (e))
return return
port = 10027 if direct else 10026 port = 10027 if direct else 10026
smtp = smtplib.SMTP("localhost", port) smtp = smtplib.SMTP("localhost", port)
if conf.debuglevel > 8: if conf.debuglevel > 8:
smtp.set_debuglevel(True) smtp.set_debuglevel(True)
try: try:
smtp.sendmail(message['From'], to_address, message.as_string()) smtp.sendmail(message['From'], to_address, message.as_string())
except Exception, e:
# pylint: disable=broad-except
except Exception as e:
log.error(_("SMTP sendmail error: %r") % (e)) log.error(_("SMTP sendmail error: %r") % (e))
if smtp: if smtp:
smtp.quit() smtp.quit()