cal-funnel/app/main.py

import logging
import os
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse

import httpx
from icalendar import Calendar
from mergecal import merge_calendars

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    stream=sys.stdout,
)
logger = logging.getLogger("cal-funnel")

USER_AGENT = "cal-funnel/1.0"


def env_int(name: str, default: int) -> int:
    value = os.environ.get(name)
    if value is None or value.strip() == "":
        return default
    return int(value)


def parse_urls(raw: str) -> list[str]:
    return [url.strip() for url in raw.split(",") if url.strip()]


class Config:
    def __init__(self) -> None:
        urls_raw = os.environ.get("CALENDAR_URLS", "").strip()
        self.urls = parse_urls(urls_raw)
        if not self.urls:
            raise ValueError("CALENDAR_URLS must contain at least one URL")

        self.fetch_interval_seconds = env_int("FETCH_INTERVAL_SECONDS", 900)
        self.port = env_int("PORT", 8080)
        self.calendar_path = os.environ.get("CALENDAR_PATH", "/calendar.ics").strip() or "/calendar.ics"
        self.http_timeout_seconds = env_int("HTTP_TIMEOUT_SECONDS", 30)
        self.prodid = os.environ.get("PRODID", "").strip() or None

        if not self.calendar_path.startswith("/"):
            self.calendar_path = f"/{self.calendar_path}"


class CalendarStore:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._data: bytes | None = None
        self._event_count = 0
        self._updated_at: float | None = None

    def set(self, data: bytes, event_count: int) -> None:
        with self._lock:
            self._data = data
            self._event_count = event_count
            self._updated_at = time.time()

    def snapshot(self) -> tuple[bytes | None, int, float | None]:
        with self._lock:
            return self._data, self._event_count, self._updated_at

    def is_ready(self) -> bool:
        with self._lock:
            return self._data is not None


def fetch_calendar(client: httpx.Client, url: str) -> Calendar:
    response = client.get(url)
    response.raise_for_status()
    return Calendar.from_ical(response.content)


def count_events(calendar: Calendar) -> int:
    return len(list(calendar.walk("VEVENT")))


def merge_feeds(config: Config, client: httpx.Client) -> tuple[bytes, int]:
    calendars: list[Calendar] = []
    failed: list[str] = []

    for url in config.urls:
        try:
            calendar = fetch_calendar(client, url)
            calendars.append(calendar)
            logger.info("Fetched %s (%d events)", url, count_events(calendar))
        except Exception as exc:
            failed.append(url)
            logger.error("Failed to fetch %s: %s", url, exc)

    if not calendars:
        raise RuntimeError(f"All calendar feeds failed: {', '.join(failed)}")

    if failed:
        logger.warning("Merged %d/%d feeds; failed: %s", len(calendars), len(config.urls), ", ".join(failed))

    merged = merge_calendars(calendars)
    if config.prodid:
        merged["PRODID"] = config.prodid

    data = merged.to_ical()
    event_count = count_events(merged)
    return data, event_count


def refresh_loop(config: Config, store: CalendarStore, stop_event: threading.Event) -> None:
    headers = {"User-Agent": USER_AGENT}
    timeout = httpx.Timeout(config.http_timeout_seconds)

    with httpx.Client(headers=headers, timeout=timeout, follow_redirects=True) as client:
        while not stop_event.is_set():
            try:
                data, event_count = merge_feeds(config, client)
                store.set(data, event_count)
                logger.info("Merged calendar updated (%d events)", event_count)
            except Exception as exc:
                logger.error("Refresh failed: %s", exc)

            if stop_event.wait(config.fetch_interval_seconds):
                break


def make_handler(config: Config, store: CalendarStore):
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, format: str, *args) -> None:
            logger.info("%s - %s", self.address_string(), format % args)

        def do_GET(self) -> None:
            path = urlparse(self.path).path

            if path == "/health":
                if store.is_ready():
                    self.send_response(200)
                    self.send_header("Content-Type", "text/plain; charset=utf-8")
                    self.end_headers()
                    self.wfile.write(b"ok")
                else:
                    self.send_response(503)
                    self.send_header("Content-Type", "text/plain; charset=utf-8")
                    self.end_headers()
                    self.wfile.write(b"not ready")
                return

            if path == config.calendar_path:
                data, event_count, updated_at = store.snapshot()
                if data is None:
                    self.send_response(503)
                    self.send_header("Content-Type", "text/plain; charset=utf-8")
                    self.end_headers()
                    self.wfile.write(b"Calendar not ready yet")
                    return

                self.send_response(200)
                self.send_header("Content-Type", "text/calendar; charset=utf-8")
                self.send_header("Cache-Control", "no-cache")
                if updated_at is not None:
                    self.send_header("X-Cal-Funnel-Events", str(event_count))
                self.end_headers()
                self.wfile.write(data)
                return

            if path == "/":
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.end_headers()
                body = (
                    "<!doctype html><html><head><title>Cal-Funnel</title></head>"
                    "<body><h1>Cal-Funnel</h1>"
                    f"<p>Subscribe to the merged calendar at "
                    f"<a href=\"{config.calendar_path}\">{config.calendar_path}</a>.</p>"
                    "</body></html>"
                )
                self.wfile.write(body.encode("utf-8"))
                return

            self.send_response(404)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(b"Not found")

    return Handler


def main() -> None:
    config = Config()
    store = CalendarStore()
    stop_event = threading.Event()

    worker = threading.Thread(
        target=refresh_loop,
        args=(config, store, stop_event),
        name="refresh-loop",
        daemon=True,
    )
    worker.start()

    server = ThreadingHTTPServer(("0.0.0.0", config.port), make_handler(config, store))
    logger.info(
        "Serving merged calendar on port %d at %s (refresh every %ds)",
        config.port,
        config.calendar_path,
        config.fetch_interval_seconds,
    )

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        logger.info("Shutting down")
    finally:
        stop_event.set()
        server.shutdown()


if __name__ == "__main__":
    main()