mark calendar email responses as READ on EWS Exchange server

mark.py

#!/usr/bin/python3
"""
Cron task to mark calendar email responses as READ on EWS Exchange server
(content-type: text/calendar; method=REPLY) 
Martin Monperrus
Sep 2023
Reference URL for latest version: https://gist.github.com/monperrus/23b694cee69ca4e023e8182c91a4f0b3
"""
from datetime import datetime, timedelta
import pytz
import exchangelib
from exchangelib import DELEGATE, IMPERSONATION, Account, Credentials, \
    EWSDate, EWSDateTime, EWSTimeZone, Configuration, NTLM, GSSAPI, CalendarItem, Message, \
    Mailbox, Attendee, Q, ExtendedProperty, FileAttachment, ItemAttachment, \
    HTMLBody, Build, Version, FolderCollection, UID
from exchangelib.properties import ID_FORMATS, EWS_ID, AlternateId
import json
import itertools
import os.path
import time
import keyring
import requests
import ics
import base64
import binascii

from email import message_from_bytes

"""Helper class to encode Calendar UIDs. See issue #453. """
class GlobalObjectId(ExtendedProperty):
    distinguished_property_set_id = 'Meeting'
    property_id = 3
    property_type = 'Binary'

CalendarItem.register('global_object_id', GlobalObjectId)

login_keyring=keyring.get_keyring() 
    
credentials = Credentials(
  username=login_keyring.get_password('login2', 'exchange-username'), 
  password=login_keyring.get_password('login2', 'kth'))

config = Configuration(
    server=login_keyring.get_password('login2', 'exchange-server'), 
    credentials=credentials, auth_type=NTLM
)

account = Account(primary_smtp_address=str(login_keyring.get_password('login2', 'exchange-primary')), credentials=credentials,
                  autodiscover=True, access_type=DELEGATE
)

today = EWSDateTime.now(tz=account.default_timezone)
some_time_ago = today - timedelta(days=30)

# unread_emails
unread_emails = [ x for x in account.inbox.filter(Q(is_read=False), Q(datetime_received__gt=some_time_ago)) ]

#print('emails unread: '+str(len(unread_emails)))

def mark_as_read(message):
  """ reference https://github.com/ecederstrand/exchangelib/issues/472#issuecomment-420166430"""
  message.is_read = True
  message.save(update_fields=['is_read'])
  
for message in unread_emails:
  #print(message.author.email_address)
  mime_message = message_from_bytes(message.mime_content)
  contenttype= [x for x in message.headers if x.name.lower()=='content-type'][0].value
  # was before a server-side filter on the email header
  if "text/calendar" in contenttype.lower() and "method=REPLY".lower() in contenttype.lower():   
    print("###################",message.author.email_address)
    print("top level text/calendar")
    mark_as_read(message)
  if mime_message.get_content_type() == "multipart/alternative":
    for part in mime_message.get_payload():
      #try:
        #print(part.get_content_type())
        if part.get_content_type() == "text/calendar":
          print("###################",message.author.email_address)
          contenttype= [x for x in part._headers if x[0].lower()=="content-type"][0][1]
          encoding= [x for x in part._headers if x[0].lower()=="content-transfer-encoding"][0][1]
          calendar_data = part.get_payload()
          if encoding.lower() == 'base64':
            calendar_data = base64.b64decode(calendar_data)
          #print(dir(part))
          #print(calendar_data.decode('utf-8'))
          
          # some MUAs add method=REPLY, that's good!
          if "method=REPLY".lower() in contenttype.lower():            
            # if one wants to parse
            #calendar = ics.Calendar(calendar_data.decode('utf-8'))
            #event = [x for x in calendar.events][0]            
            
            # if one wants to get the EWS calendar event
            #event_ews = account.calendar.get(global_object_id=binascii.unhexlify(event.uid))
            print(message.author.email_address, "method=REPLY")
            # the invitation has already been handled by EWS, we mark it as read
            mark_as_read(message)
          else:
            # but some MUAs don't 
            calendar = ics.Calendar(calendar_data.decode('utf-8'))
            event = [x for x in calendar.events][0]
            if calendar.method.lower() == "REPLY".lower():
              #print(calendar_data.decode('utf-8'))
              print(message.author.email_address, event.status)            
              mark_as_read(message)
            
            
      #except Exception as e:         print(e)