"""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