t-r.de/utils/refresh_events.py

#!/usr/bin/env python3
# -*- coding: utf8 -*-

import configparser
import caldav
import vobject
from datetime import datetime, date, time, timedelta
import yaml
import os, os.path
from urllib.parse import urlparse


# find out where events.ini lives; that's the project root dir
# check current working directory
if os.path.isfile(os.getcwd() + '/events.ini'):
  project_root = os.getcwd()
# check this script's location dir
elif os.path.isfile(os.path.dirname(os.path.realpath(__file__)) + '/events.ini'):
  project_root = os.path.dirname(os.path.realpath(__file__))
# check parent dir of this script's dir  
elif os.path.isfile(os.path.dirname(os.path.realpath(__file__)) + '/../events.ini'):
  project_root = os.path.dirname(os.path.realpath(__file__)) + '/..'
# OK no luck
else:
  print("Cannot find file 'events.ini'; aborting.")
  exit(1)

# read ini file
cp = configparser.ConfigParser()
cp.read(project_root +  '/events.ini')
server_url = cp["CalDAV"]["server_url"]
cal_url = cp["CalDAV"]["cal_url"]
cal_user = cp["CalDAV"]["cal_user"]
cal_pass = cp["CalDAV"]["cal_pass"]

# Category to filter events for. Only events that belong to this category will be processed.
cal_category = "Veranstaltung"
# Name of the file the YAML data will be written to.
result_file = project_root + '/content/pages/termine.md'
# List of icalendar properties that will be passed on to Pelican. At this time, all properties are used except those regarding alarms.
cal_properties = [ # taken from ical specification
  # descriptive
  "attach", "categories", "class", "comment", "description", "geo", "location", "percent-complete", "priority", "resources", "status", "summary",
  # date and time
  "completed", "dtend", "due", "dtstart", "duration", "freebusy", "transp", 
  # timezone components
  "tzid", "tzname", "tzoffsetfrom", "tzoffsetto", "tzurl",
  # relationship
  "attendee", "contact", "organizer", "recurrence-id", "related-to", "url", "uid",
  # recurrence
  "exdate", "exrule", "rdate", "rrule",
  # alarm
#  "action", "repeat", "trigger",
  # change management
  "created", "dtstamp", "last-modified", "sequence"
]
#List of icalendar properties that the vobject passes as either datetime.datetime or datetime.date objects. To be able to compare any two objects, the script makes each one aware and/or converts into datetime.datetime if necessary. Default timezone is the system timezone; default time is 23:59:59.  
cal_prop_datetimes = ["dtend", "due", "dtstart", "duration", "dtstamp", "last-modified"]
# Current time as an aware datetime.datetime object.
now = datetime.now().astimezone()

image_suffixes = ["jpg", "jpeg", "png", "webp", "gif"]
internal_hosts = ["", "localhost", "127.0.0.1", "127.0.1.1"]
image_link_prefix = "../images/termine/"


"""Returns an aware datetime.datetime object based on the given datetime or date. Conversion happens only if necessary, i.e. if d is of type datetime.date or a naive datetime.datetime object."""
def fixDatetime(d: datetime | date, t=None, tz=None):
  if not tz:
    tz = now.tzinfo

  # datetime object is made aware if necessary
  if type(d) == datetime:
    if d.tzinfo:
      return d
    else:
      return datetime(d.date(), d.time(), tz)

  # raise error if d is neither datetime nor date
  elif type(d) != date:
    raise TypeError("parameter must be a datetime.date or datetime.datetime object")

  # d is a date object
  if not t:
    t = time(23, 59, 59, tzinfo=tz)    # if no time parameter was passed, use 23:59:59 
  if not t.tzinfo:
    t.replace(tzinfo=tz)
  return datetime.combine(d, t)


"""Takes a list of events in vobject representation and converts each event into a dict of Python datatypes.""" 
def extractEventData(events: list) -> list:
  e_tmp = []

  # iterate over events
  for e in events:
    d = {}

    # iterate over event properties
    for k, v in e.contents.items():
      # we're only interested in a subset of properties
      if not k in cal_properties:
        continue
#      print(f"k is {k} and v is {v}") #DEBUG

      # Usually, v is a list with a single item. We only need the item's value attribute.
      if len(v) == 1:
        d[k] = v[0].value
        continue

      # but sometimes v has more than one item
      # special case: if k is "categories", each list item's value is itself a one item list (which is stupid but that's how vobject handles categories)
      if k == "categories":
        d[k] = [v[i].value[0] for i in range(len(v))]
      # all other cases: extract each list item's value as is   
      else:
        d[k] = [v[i].value for i in range(len(v))]
    e_tmp.append(d)
  return e_tmp


def handleAttachments(events: list) -> list:
  for event in events:
    # skip events without attachment(s)
    if "attach" not in event.keys():
      continue

    # turn single attachment into a single element list
    if type(event["attach"]) == str:
      event["attach"] = [event["attach"]]

    # loop over attachments
    for attachment in event["attach"]:
      parsed_url = urlparse(attachment)
      isImage = True if parsed_url.path.split(".")[-1] in image_suffixes else False
#      print(f"attachment is {attachment}, parsed_url is {parsed_url} and isImage is {isImage}") #DEBUG

      # handle image link
      if isImage:
        # make internal link relative to "/termine"
        if parsed_url.hostname in internal_hosts:
          # isolate filename 
          filename = parsed_url.path.split("/")[-1]
          # prefix filename with path and add it to the event as "image" property
          event["image"] = image_link_prefix + filename
        # use external link as is
        else:
          event["image"] = attachment
      # handle non-image link
      else:
        event["link"] = attachment

#    print(event) #DEBUG
  return events


def refresh_events():
  # create vobject calendar
  vcal = vobject.newFromBehavior("vcalendar")

  # establish connection to caldav server 
  with caldav.DAVClient(url=server_url, username=cal_user, password=cal_pass) as dav_client:
    # establish connection to calendar
    dav_cal = dav_client.calendar(url=cal_url)

    # put all events from caldav calendar into vobject calendar
    for e in dav_cal.events():
      for ev in e.vobject_instance.contents["vevent"]:
        vcal.add(ev)
    
  # we only want events belonging to a specific category 
  events = [e for e in vcal.getChildren() if "categories" in e.contents.keys() and cal_category in map(lambda x: x.value[0], e.contents["categories"])]

  # extract event data 
  events = extractEventData(events)

  # fix dates and datetimes
  for e in events:
    for k in e.keys():
      if k in cal_prop_datetimes:
        # fix a CalDAV/vobject/icalendar (?) bug where a dtend value which is a date only (no time) automatically moves one day into the future 
        if k == "dtend" and type(e[k]) == date:
          e[k] = e[k] - timedelta(days=1)
        # fix all datetimes so they can be compared and sorted 
        e[k] = fixDatetime(e[k])

  # keep only future events  
  events = [e for e in events if e["dtstart"] >= now]

  # sort attached links into page links and image links
  events = handleAttachments(events)

  # sort by start date
  events.sort(key=lambda e: e["dtstart"])

  # add start date to metadata, also start time and/or end date
  for e in events:
    # start date
    e["startdate"] = e["dtstart"].astimezone()
    # start time (if not the default time)
    if e["dtstart"].timetz() != time(23, 59, 59, tzinfo=now.tzinfo):
      e["starttime"] = e["dtstart"].astimezone()
    # end date (if different from start date)
    if e["dtstart"].date() < e["dtend"].date():
      e["enddate"] = e["dtend"].astimezone()

  # read default metadata if present
  default_metadata = None
  if os.path.isfile(result_file + ".metadata"):
    with open(result_file + ".metadata", "r") as f:
      default_metadata = yaml.safe_load(f)

  # write data as YAML
  with open(result_file, 'w') as f:
    f.write("---\n")
    if default_metadata:
      yaml.dump(default_metadata, f)
    yaml.dump({"termine": events}, f)
    # write current datetime to file
    yaml.dump({"written_at": now}, f)
    f.write("---\n")

if __name__ == "__main__":
  refresh_events()