Restructure project

fabcal/app.py

@@ -0,0 +1,33 @@
+import locale
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi_cache import FastAPICache
+from fastapi_cache.backends.inmemory import InMemoryBackend
+
+from fabcal.routers import api_v1, frontend
+
+
+app = FastAPI()
+
+app.mount("/static", StaticFiles(directory="static"), name="static")
+
+app.include_router(api_v1.router, prefix="/api/v1")
+app.include_router(frontend.router, prefix="")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[
+        "https://fablab-altmuehlfranken.de",
+        "https://www.fablab-altmuehlfranken.de",
+    ],
+)
+
+
+locale.setlocale(locale.LC_TIME, locale.getlocale())
+
+
+@app.on_event("startup")
+async def startup():
+    FastAPICache.init(InMemoryBackend())

fabcal/calendar_client.py

@@ -0,0 +1,137 @@
+import os
+
+from collections import OrderedDict
+from datetime import date, datetime, timedelta, timezone
+from typing import List
+
+import aiohttp
+import recurring_ical_events
+
+from fastapi_cache.decorator import cache
+from icalendar import Calendar, vText
+
+from fabcal.models import CalendarEvent
+
+
+def sanitize(data: str):
+    # sanitize input from random source
+    cal = Calendar.from_ical(data)
+
+    # name needs to be fixed
+    cal["X-WR-CALNAME"] = vText(b"FabLab-Termine")
+
+    return cal.to_ical()
+
+
+def get_calendar_url():
+    url = os.environ["CALENDAR_URL"]
+
+    # convenience feature
+    url = url.replace("webcal://", "https://")
+
+    return url
+
+
+# caching strings works better than caching calendar objects
+@cache(expire=120)
+async def get_data() -> str:
+    async with aiohttp.ClientSession() as session:
+        async with session.get(get_calendar_url()) as response:
+            response.raise_for_status()
+            assert response.content_type.lower() == "text/calendar"
+
+            return sanitize(await response.text())
+
+
+def get_tzinfo():
+    return timezone(timedelta(hours=1))
+
+
+def get_events_from_calendar_string(cal: Calendar) -> List[CalendarEvent]:
+    """
+    Generate list of events from calendar vevents.
+    Expands recurring events for +- one year.
+    Note that there is no validation, e.g., checking for values required (by other code in this application).
+    :param cal: calendar to fetch events from
+    :return: events
+    """
+
+    now = datetime.now()
+    td = timedelta(days=365)
+    past_year = now - td
+    next_year = now + td
+
+    events = []
+
+    for vevent in recurring_ical_events.of(cal).between(past_year, next_year):
+        # I'm pessimistic here. Prove me wrong!
+        all_day_event = False
+
+        start = vevent.get("DTSTART", None)
+        if start is not None:
+            start = start.dt
+
+            if not isinstance(start, datetime):
+                all_day_event = True
+                start = datetime.combine(start, datetime.min.time(), tzinfo=get_tzinfo())
+
+        end = vevent.get("DTEND", None)
+        if end is not None:
+            end = end.dt
+
+            if not isinstance(end, datetime):
+                all_day_event = True
+                end = datetime.combine(end, datetime.max.time(), tzinfo=get_tzinfo())
+
+        def get_str(key: str):
+            value = vevent.get(key, None)
+
+            if value is not None:
+                return str(value)
+
+            return value
+
+        summary = get_str("SUMMARY")
+        description = get_str("DESCRIPTION")
+        location = get_str("LOCATION")
+        color = get_str("COLOR")
+        uid = get_str("UID")
+
+        event = CalendarEvent(start, end, all_day_event, summary, description, location, color, uid)
+        events.append(event)
+
+    events.sort(key=lambda e: e.start)
+
+    return events
+
+
+async def get_future_events():
+    cal = Calendar.from_ical(await get_data())
+
+    events = get_events_from_calendar_string(cal)
+
+    today = datetime.today().date()
+
+    future_events = []
+
+    for event in events:
+        if event.start.date() < today:
+            continue
+
+        future_events.append(event)
+
+    return future_events
+
+
+def group_by_date(events: List[CalendarEvent]):
+    grouped_events: OrderedDict[date, List[CalendarEvent]] = OrderedDict()
+
+    for event in events:
+        start_date = event.start.date()
+
+        if start_date not in grouped_events:
+            grouped_events[start_date] = []
+
+        grouped_events[start_date].append(event)
+
+    return grouped_events

fabcal/models.py

@@ -0,0 +1,14 @@
+from datetime import datetime
+from typing import NamedTuple
+
+
+class CalendarEvent(NamedTuple):
+    start: datetime
+    end: datetime
+    # just a convenience thing, we want to keep start/date as datetime and save the client from guessing this themselves
+    all_day_event: bool
+    summary: str
+    description: str
+    location: str
+    color: str
+    uid: str

fabcal/routers/__init__.py

@@ -0,0 +1,4 @@
+from starlette.templating import Jinja2Templates
+
+
+templates = Jinja2Templates(directory="templates")

fabcal/routers/api_v1.py

@@ -0,0 +1,17 @@
+from fastapi import APIRouter
+from starlette.responses import Response
+
+from fabcal.calendar_client import get_data
+
+
+router = APIRouter()
+
+
+@router.get("/events.ics")
+async def events():
+    return Response(
+        await get_data(),
+        headers={
+            "content-type": "text/calendar",
+        },
+    )

fabcal/routers/frontend.py

@@ -0,0 +1,45 @@
+import base64
+
+from datetime import datetime
+
+import babel.dates
+
+from fastapi import APIRouter
+from fastapi.requests import Request
+from fastapi.responses import HTMLResponse
+
+from fabcal.calendar_client import get_future_events, group_by_date
+from fabcal.routers import templates
+
+
+router = APIRouter()
+
+
+@router.get("/embed-sidebar.html", response_class=HTMLResponse)
+async def embed(request: Request, max_width: str = None):
+    events = await get_future_events()
+
+    grouped_events = list(group_by_date(events).items())
+
+    # couple of helpers
+    def localized_abbreviated_month(dt: datetime):
+        return babel.dates.format_datetime(dt, format="%b", locale="de_DE")
+
+    # couple of helpers
+    def localized_abbreviated_weekday(dt: datetime):
+        return babel.dates.format_datetime(dt, format="%b", locale="de_DE")
+
+    def base64_encode(s: str):
+        return base64.b64encode(s.encode()).decode()
+
+    return templates.TemplateResponse(
+        "embed-sidebar.html",
+        context={
+            "request": request,
+            "grouped_events": grouped_events,
+            "dir": dir,
+            "localized_abbreviated_month": localized_abbreviated_month,
+            "localized_abbreviated_weekday": localized_abbreviated_weekday,
+            "base64_encode": base64_encode,
+        },
+    )