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 = ( "
Subscribe to the merged calendar at " f"{config.calendar_path}.
" "" ) 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()