"""Builds an RFC 5545 (iCalendar) feed from events, for webcal subscriptions. Point a calendar app at ``webcal:///calendar.ics`` (or the plain ``https://`` URL) to subscribe - the app will periodically re-fetch it. """ import hashlib import re from datetime import date, datetime, timedelta, timezone from typing import List, Optional from datasource import iter_artist_names, parse_event_date from models import Event _STARTTIME_RE = re.compile(r"(\d{1,2}):(\d{2})") _DEFAULT_DURATION = timedelta(hours=5) _UID_DOMAIN = "fundbureau-scraper" def _escape_text(value: str) -> str: return ( value.replace("\\", "\\\\") .replace(";", "\\;") .replace(",", "\\,") .replace("\n", "\\n") ) def _fold_line(line: str) -> str: """Fold a logical line to <=75 octets per line, as RFC 5545 requires.""" data = line.encode("utf-8") if len(data) <= 75: return line chunks = [] start = 0 limit = 75 while start < len(data): end = min(start + limit, len(data)) # Don't split a multi-byte UTF-8 sequence in half. while end < len(data) and (data[end] & 0xC0) == 0x80: end -= 1 chunks.append(data[start:end].decode("utf-8")) start = end limit = 74 # continuation lines are prefixed with a single space return "\r\n ".join(chunks) def _parse_starttime(raw: str) -> Optional[tuple]: match = _STARTTIME_RE.search(raw or "") if not match: return None hour, minute = int(match.group(1)), int(match.group(2)) if 0 <= hour < 24 and 0 <= minute < 60: return hour, minute return None def _event_uid(event: Event) -> str: digest = hashlib.sha1(f"{event.event_date}|{event.event_name}".encode("utf-8")).hexdigest() return f"{digest}@{_UID_DOMAIN}" def _build_vevent(event: Event, dtstamp: str) -> List[str]: event_date = parse_event_date(event.event_date) lines = ["BEGIN:VEVENT", f"UID:{_event_uid(event)}", f"DTSTAMP:{dtstamp}"] if event_date is None: # Can't place it on the calendar without a parseable date. return [] starttime = _parse_starttime(event.event_starttime) if starttime is not None: hour, minute = starttime dtstart = datetime(event_date.year, event_date.month, event_date.day, hour, minute) dtend = dtstart + _DEFAULT_DURATION lines.append(f"DTSTART:{dtstart.strftime('%Y%m%dT%H%M%S')}") lines.append(f"DTEND:{dtend.strftime('%Y%m%dT%H%M%S')}") else: # No parseable door time - render as an all-day event instead of # guessing a time. next_day = event_date + timedelta(days=1) lines.append(f"DTSTART;VALUE=DATE:{event_date.strftime('%Y%m%d')}") lines.append(f"DTEND;VALUE=DATE:{next_day.strftime('%Y%m%d')}") lines.append(f"SUMMARY:{_escape_text(event.event_name)}") lines.append("LOCATION:" + _escape_text("Fundbureau, Hamburg")) description_parts = [] lineup = ", ".join(iter_artist_names(event)) if lineup: description_parts.append(f"Line-up: {lineup}") if event.event_starttime: description_parts.append(event.event_starttime) if event.event_free: description_parts.append("Eintritt frei") elif event.event_ticket_link and event.event_ticket_link != "n/a": description_parts.append(f"Tickets: {event.event_ticket_link}") if description_parts: lines.append(f"DESCRIPTION:{_escape_text(chr(10).join(description_parts))}") if event.event_ticket_link and event.event_ticket_link not in ("", "n/a"): lines.append(f"URL:{event.event_ticket_link}") lines.append("END:VEVENT") return lines def build_ics(events: List[Event]) -> str: """Render a list of events as a VCALENDAR document (CRLF-terminated).""" dtstamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") lines = [ "BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//fundi-scraper-api//Fundbureau Events//DE", "CALSCALE:GREGORIAN", "METHOD:PUBLISH", "X-WR-CALNAME:Fundbureau Events", "X-WR-CALDESC:Upcoming and past events at the Fundbureau, Hamburg", "X-WR-TIMEZONE:Europe/Berlin", "REFRESH-INTERVAL;VALUE=DURATION:PT12H", "X-PUBLISHED-TTL:PT12H", ] for event in events: lines.extend(_build_vevent(event, dtstamp)) lines.append("END:VCALENDAR") return "\r\n".join(_fold_line(line) for line in lines) + "\r\n"