311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""iCalendar (RFC 5545) lezen en schrijven voor de agenda en CalDAV, alleen met de standaardbibliotheek.
|
|
|
|
Een afspraak is een dict met uid, title, location, description, starts_at, ends_at, all_day, rrule en exdates.
|
|
Tijden zijn lokaal (Europe/Amsterdam) als 'YYYY-MM-DD HH:MM:SS'; een hele dag is 'YYYY-MM-DD' met een
|
|
exclusief einde (de dag ná de laatste dag), zoals in iCalendar. exdates is een kommalijst in hetzelfde formaat.
|
|
"""
|
|
import calendar
|
|
import re
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
TZ_NAME = 'Europe/Amsterdam'
|
|
TZ = ZoneInfo(TZ_NAME)
|
|
# Outlook/Exchange sturen Windows-namen mee
|
|
WINDOWS_TZ = {'W. Europe Standard Time': TZ_NAME, 'Romance Standard Time': 'Europe/Brussels',
|
|
'Central Europe Standard Time': 'Europe/Budapest', 'GMT Standard Time': 'Europe/London',
|
|
'UTC': 'UTC'}
|
|
WEEKDAYS = ['MO', 'TU', 'WE', 'TH', 'FR', 'SA', 'SU']
|
|
|
|
VTIMEZONE = """BEGIN:VTIMEZONE
|
|
TZID:Europe/Amsterdam
|
|
BEGIN:DAYLIGHT
|
|
TZOFFSETFROM:+0100
|
|
TZOFFSETTO:+0200
|
|
TZNAME:CEST
|
|
DTSTART:19700329T020000
|
|
RRULE:FREQ=YEARLY;BYMONTH=3;BYDAY=-1SU
|
|
END:DAYLIGHT
|
|
BEGIN:STANDARD
|
|
TZOFFSETFROM:+0200
|
|
TZOFFSETTO:+0100
|
|
TZNAME:CET
|
|
DTSTART:19701025T030000
|
|
RRULE:FREQ=YEARLY;BYMONTH=10;BYDAY=-1SU
|
|
END:STANDARD
|
|
END:VTIMEZONE""".split('\n')
|
|
|
|
|
|
# ---------------------------------------------------------------- lezen
|
|
|
|
def _parse_line(line):
|
|
"""'NAAM;PARAM=waarde:inhoud' → (NAAM, {PARAM: waarde}, inhoud); dubbelepunten tussen aanhalingstekens tellen niet."""
|
|
quoted = False
|
|
for i, ch in enumerate(line):
|
|
if ch == '"':
|
|
quoted = not quoted
|
|
elif ch == ':' and not quoted:
|
|
head, value = line[:i], line[i + 1:]
|
|
break
|
|
else:
|
|
return None
|
|
parts = re.findall(r'(?:[^;"]|"[^"]*")+', head)
|
|
if not parts:
|
|
return None
|
|
params = {}
|
|
for p in parts[1:]:
|
|
k, _, v = p.partition('=')
|
|
params[k.strip().upper()] = v.strip().strip('"')
|
|
return parts[0].strip().upper(), params, value
|
|
|
|
|
|
def parse(text):
|
|
"""Componentenboom: [{'name': 'VCALENDAR', 'props': [(naam, params, waarde)], 'subs': [...]}]."""
|
|
root = {'name': None, 'props': [], 'subs': []}
|
|
stack = [root]
|
|
for raw in re.sub(r'\r?\n[ \t]', '', text).splitlines():
|
|
parsed = _parse_line(raw) if raw.strip() else None
|
|
if not parsed:
|
|
continue
|
|
name, params, value = parsed
|
|
if name == 'BEGIN':
|
|
comp = {'name': value.strip().upper(), 'props': [], 'subs': []}
|
|
stack[-1]['subs'].append(comp)
|
|
stack.append(comp)
|
|
elif name == 'END':
|
|
if len(stack) > 1:
|
|
stack.pop()
|
|
else:
|
|
stack[-1]['props'].append((name, params, value))
|
|
return root['subs']
|
|
|
|
|
|
def _prop(comp, name):
|
|
return next((p for p in comp['props'] if p[0] == name), None)
|
|
|
|
|
|
def unescape(value):
|
|
return re.sub(r'\\([\\;,nN])', lambda m: '\n' if m.group(1) in 'nN' else m.group(1), value)
|
|
|
|
|
|
def _zone(tzid):
|
|
tzid = WINDOWS_TZ.get(tzid, tzid)
|
|
for candidate in (tzid, '/'.join(tzid.split('/')[-2:])): # ook '/mozilla.org/20050126_1/Europe/Amsterdam'
|
|
try:
|
|
return ZoneInfo(candidate)
|
|
except (KeyError, ValueError, OSError):
|
|
continue
|
|
return None
|
|
|
|
|
|
def parse_dt(value, params):
|
|
"""iCalendar-datum of -tijd → (date, True) of (lokale naïeve datetime, False)."""
|
|
v = value.strip()
|
|
if params.get('VALUE', '').upper() == 'DATE' or re.fullmatch(r'\d{8}', v):
|
|
return date(int(v[:4]), int(v[4:6]), int(v[6:8])), True
|
|
m = re.fullmatch(r'(\d{8})T(\d{4,6})(Z?)', v)
|
|
if not m:
|
|
raise ValueError(f'onbekende datum {value!r}')
|
|
dt = datetime.strptime(m.group(1) + m.group(2).ljust(6, '0'), '%Y%m%d%H%M%S')
|
|
zone = timezone.utc if m.group(3) else (_zone(params['TZID']) if params.get('TZID') else None)
|
|
if zone:
|
|
dt = dt.replace(tzinfo=zone).astimezone(TZ).replace(tzinfo=None)
|
|
return dt, False
|
|
|
|
|
|
def parse_duration(value):
|
|
m = re.fullmatch(r'([+-])?P(?:(\d+)W)?(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)?', value.strip())
|
|
if not m:
|
|
return timedelta(0)
|
|
sign, w, d, h, mi, s = m.groups()
|
|
td = timedelta(weeks=int(w or 0), days=int(d or 0), hours=int(h or 0), minutes=int(mi or 0), seconds=int(s or 0))
|
|
return -td if sign == '-' else td
|
|
|
|
|
|
def fmt_local(value):
|
|
return value.isoformat() if not isinstance(value, datetime) else value.strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
|
|
def from_ics(text):
|
|
"""Eerste (hoofd)afspraak uit een kalenderobject, of None als er geen VEVENT in zit."""
|
|
events = [s for c in parse(text) if c['name'] == 'VCALENDAR' for s in c['subs'] if s['name'] == 'VEVENT']
|
|
if not events:
|
|
return None
|
|
ev = next((e for e in events if not _prop(e, 'RECURRENCE-ID')), events[0])
|
|
start_p = _prop(ev, 'DTSTART')
|
|
if not start_p:
|
|
return None
|
|
start, all_day = parse_dt(start_p[2], start_p[1])
|
|
end_p, dur_p = _prop(ev, 'DTEND'), _prop(ev, 'DURATION')
|
|
if end_p:
|
|
end, _ = parse_dt(end_p[2], end_p[1])
|
|
if all_day and isinstance(end, datetime):
|
|
end = end.date()
|
|
elif not all_day and not isinstance(end, datetime):
|
|
end = datetime.combine(end, datetime.min.time())
|
|
elif dur_p:
|
|
end = start + parse_duration(dur_p[2])
|
|
else:
|
|
end = start + timedelta(days=1) if all_day else start
|
|
if all_day and end <= start:
|
|
end = start + timedelta(days=1)
|
|
exdates = []
|
|
for name, params, value in ev['props']:
|
|
if name == 'EXDATE':
|
|
for v in value.split(','):
|
|
try:
|
|
exdates.append(fmt_local(parse_dt(v, params)[0]))
|
|
except ValueError:
|
|
pass
|
|
text_of = lambda n: unescape(_prop(ev, n)[2]) if _prop(ev, n) else ''
|
|
rrule = _prop(ev, 'RRULE')
|
|
return {'uid': text_of('UID').strip(), 'title': text_of('SUMMARY').strip(), 'location': text_of('LOCATION').strip(),
|
|
'description': text_of('DESCRIPTION').strip(), 'starts_at': fmt_local(start), 'ends_at': fmt_local(end),
|
|
'all_day': 1 if all_day else 0, 'rrule': rrule[2].strip() if rrule else '', 'exdates': ','.join(exdates)}
|
|
|
|
|
|
# ---------------------------------------------------------------- schrijven
|
|
|
|
def escape(value):
|
|
return (str(value or '').replace('\\', '\\\\').replace(';', '\\;').replace(',', '\\,')
|
|
.replace('\r\n', '\\n').replace('\n', '\\n'))
|
|
|
|
|
|
def _fold(line):
|
|
"""Regels langer dan 75 octetten vouwen (zonder een UTF-8-teken te breken)."""
|
|
out, cur, size = [], '', 0
|
|
for ch in line:
|
|
n = len(ch.encode('utf-8'))
|
|
if size + n > 75:
|
|
out.append(cur)
|
|
cur, size = ' ', 1
|
|
cur += ch
|
|
size += n
|
|
out.append(cur)
|
|
return '\r\n'.join(out)
|
|
|
|
|
|
def _ical_dt(value, all_day):
|
|
if all_day:
|
|
return f";VALUE=DATE:{value.replace('-', '')[:8]}"
|
|
return f";TZID={TZ_NAME}:{datetime.strptime(value, '%Y-%m-%d %H:%M:%S').strftime('%Y%m%dT%H%M%S')}"
|
|
|
|
|
|
def to_ics(ev, url=''):
|
|
stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
|
|
lines = ['BEGIN:VCALENDAR', 'VERSION:2.0', 'PRODID:-//Van Emous//CRM//NL', 'CALSCALE:GREGORIAN']
|
|
if not ev['all_day']:
|
|
lines += VTIMEZONE
|
|
lines += ['BEGIN:VEVENT', f"UID:{ev['uid']}", f'DTSTAMP:{stamp}', f'LAST-MODIFIED:{stamp}',
|
|
'DTSTART' + _ical_dt(ev['starts_at'], ev['all_day']), 'DTEND' + _ical_dt(ev['ends_at'], ev['all_day']),
|
|
f"SUMMARY:{escape(ev['title'])}"]
|
|
if ev.get('location'):
|
|
lines.append(f"LOCATION:{escape(ev['location'])}")
|
|
if ev.get('description'):
|
|
lines.append(f"DESCRIPTION:{escape(ev['description'])}")
|
|
if url:
|
|
lines.append(f'URL:{url}')
|
|
if ev.get('rrule'):
|
|
lines.append(f"RRULE:{ev['rrule']}")
|
|
for x in filter(None, (ev.get('exdates') or '').split(',')):
|
|
lines.append('EXDATE' + _ical_dt(x, ev['all_day']))
|
|
lines += ['END:VEVENT', 'END:VCALENDAR']
|
|
return '\r\n'.join(_fold(l) for l in lines) + '\r\n'
|
|
|
|
|
|
# ---------------------------------------------------------------- herhalingen
|
|
|
|
def to_dt(value):
|
|
return datetime.strptime(value, '%Y-%m-%d %H:%M:%S') if len(value) > 10 else datetime.strptime(value, '%Y-%m-%d')
|
|
|
|
|
|
def parse_rrule(rrule):
|
|
return {k.upper(): v for k, _, v in (p.partition('=') for p in rrule.split(';') if p)}
|
|
|
|
|
|
def simple_freq(rrule):
|
|
"""'WEEKLY' voor een eenvoudige herhaling zoals het CRM-formulier hem maakt, anders '' (dan blijft hij staan)."""
|
|
r = parse_rrule(rrule or '')
|
|
return r['FREQ'] if set(r) == {'FREQ'} and r['FREQ'] in ('DAILY', 'WEEKLY', 'MONTHLY', 'YEARLY') else ''
|
|
|
|
|
|
def _nth_weekday(year, month, n, wd):
|
|
days = [d for d in range(1, calendar.monthrange(year, month)[1] + 1) if date(year, month, d).weekday() == wd]
|
|
try:
|
|
return days[n - 1] if n > 0 else days[n]
|
|
except IndexError:
|
|
return None
|
|
|
|
|
|
def _candidates(start, rule):
|
|
freq, interval = rule.get('FREQ', ''), max(1, int(rule.get('INTERVAL') or 1))
|
|
byday = [d for d in (rule.get('BYDAY') or '').split(',') if d]
|
|
bymonthday = [int(d) for d in (rule.get('BYMONTHDAY') or '').split(',') if re.fullmatch(r'-?\d+', d)]
|
|
i = 0
|
|
while True:
|
|
if freq == 'DAILY':
|
|
yield start + timedelta(days=i * interval)
|
|
elif freq == 'WEEKLY':
|
|
week = start - timedelta(days=start.weekday()) + timedelta(weeks=i * interval)
|
|
days = sorted({WEEKDAYS.index(d[-2:]) for d in byday if d[-2:] in WEEKDAYS}) or [start.weekday()]
|
|
for d in days:
|
|
c = week + timedelta(days=d)
|
|
if c >= start:
|
|
yield c
|
|
elif freq == 'MONTHLY':
|
|
y, m = divmod(start.month - 1 + i * interval, 12)
|
|
y, m = start.year + y, m + 1
|
|
last = calendar.monthrange(y, m)[1]
|
|
if byday:
|
|
found = []
|
|
for spec in byday:
|
|
mm = re.fullmatch(r'([+-]?\d+)?(MO|TU|WE|TH|FR|SA|SU)', spec)
|
|
if mm:
|
|
day = _nth_weekday(y, m, int(mm.group(1) or 1), WEEKDAYS.index(mm.group(2)))
|
|
if day:
|
|
found.append(day)
|
|
else:
|
|
found = [d if d > 0 else last + d + 1 for d in (bymonthday or [start.day])]
|
|
for d in sorted(set(found)):
|
|
if 1 <= d <= last:
|
|
c = start.replace(year=y, month=m, day=d)
|
|
if c >= start:
|
|
yield c
|
|
elif freq == 'YEARLY':
|
|
try:
|
|
yield start.replace(year=start.year + i * interval)
|
|
except ValueError: # 29 februari
|
|
pass
|
|
else:
|
|
return
|
|
i += 1
|
|
if i > 5000:
|
|
return
|
|
|
|
|
|
def occurrences(ev, range_start, range_end):
|
|
"""Begin- en eindtijden (datetimes) van een afspraak, herhalingen uitgevouwen, die [range_start, range_end) raken."""
|
|
start, end = to_dt(ev['starts_at']), to_dt(ev['ends_at'] or ev['starts_at'])
|
|
dur = max(end - start, timedelta(0))
|
|
hits = lambda s: s < range_end and (s + dur > range_start if dur else s >= range_start)
|
|
if not ev.get('rrule'):
|
|
return [(start, start + dur)] if hits(start) else []
|
|
rule = parse_rrule(ev['rrule'])
|
|
count = int(rule['COUNT']) if (rule.get('COUNT') or '').isdigit() else None
|
|
until = None
|
|
if rule.get('UNTIL'):
|
|
try:
|
|
u, is_date = parse_dt(rule['UNTIL'], {})
|
|
until = datetime.combine(u, datetime.max.time()) if is_date else u
|
|
except ValueError:
|
|
pass
|
|
skip = {to_dt(x) for x in (ev.get('exdates') or '').split(',') if x}
|
|
out, n = [], 0
|
|
for c in _candidates(start, rule):
|
|
if (until and c > until) or c >= range_end or (count is not None and n >= count):
|
|
break
|
|
n += 1
|
|
if c not in skip and hits(c):
|
|
out.append((c, c + dur))
|
|
return out
|