fundi-scraper-api/api/calendar_feed.py

"""Builds an RFC 5545 (iCalendar) feed from events, for webcal subscriptions.

Point a calendar app at ``webcal://<host>/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"