221 lines
7.1 KiB
Python
221 lines
7.1 KiB
Python
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()
|