"""FastAPI app: /api/duties, /api/calendar-events, personal ICS, and static webapp at /app.""" import logging import re from datetime import date, timedelta import duty_teller.config as config from fastapi import Depends, FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import Response from fastapi.staticfiles import StaticFiles from sqlalchemy.orm import Session from duty_teller.api.calendar_ics import get_calendar_events from duty_teller.api.dependencies import ( fetch_duties_response, get_db_session, get_validated_dates, require_miniapp_username, ) from duty_teller.api.personal_calendar_ics import build_personal_ics, build_team_ics from duty_teller.cache import ics_calendar_cache from duty_teller.db.repository import ( get_duties, get_duties_for_user, get_user_by_calendar_token, ) from duty_teller.db.schemas import CalendarEvent, DutyWithUser log = logging.getLogger(__name__) # Calendar tokens are secrets.token_urlsafe(32) → base64url, length 43 _CALENDAR_TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{40,50}$") def _is_valid_calendar_token(token: str) -> bool: """Return True if token matches expected format (length and alphabet). Rejects invalid before DB.""" return bool(token and _CALENDAR_TOKEN_RE.match(token)) app = FastAPI(title="Duty Teller API") @app.get("/health", summary="Health check") def health() -> dict: """Return 200 when the app is up. Used by Docker HEALTHCHECK.""" return {"status": "ok"} app.add_middleware( CORSMiddleware, allow_origins=config.CORS_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class NoCacheStaticMiddleware: """ Raw ASGI middleware: Cache-Control: no-store for all /app and /app/* static files; Vary: Accept-Language on all responses so reverse proxies do not serve one user's response to another. """ def __init__(self, app, **kwargs): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return path = scope.get("path", "") is_app_path = path == "/app" or path.startswith("/app/") async def send_wrapper(message): if message["type"] == "http.response.start": headers = list(message.get("headers", [])) header_names = {h[0].lower(): i for i, h in enumerate(headers)} if is_app_path: cache_control = (b"cache-control", b"no-store") if b"cache-control" in header_names: headers[header_names[b"cache-control"]] = cache_control else: headers.append(cache_control) vary_val = b"Accept-Language" if b"vary" in header_names: idx = header_names[b"vary"] existing = headers[idx][1] tokens = [p.strip() for p in existing.split(b",")] if vary_val not in tokens: headers[idx] = (b"vary", existing + b", " + vary_val) else: headers.append((b"vary", vary_val)) message = { "type": "http.response.start", "status": message["status"], "headers": headers, } await send(message) await self.app(scope, receive, send_wrapper) app.add_middleware(NoCacheStaticMiddleware) @app.get( "/app/config.js", summary="Mini App config (language, log level)", description=( "Returns JS that sets window.__DT_LANG and window.__DT_LOG_LEVEL. Loaded before main.js." ), ) def app_config_js() -> Response: """Return JS assigning window.__DT_LANG and window.__DT_LOG_LEVEL for the webapp. No caching.""" lang = config.DEFAULT_LANGUAGE log_level = config.LOG_LEVEL_STR.lower() body = f'window.__DT_LANG = "{lang}";\nwindow.__DT_LOG_LEVEL = "{log_level}";' return Response( content=body, media_type="application/javascript; charset=utf-8", headers={"Cache-Control": "no-store"}, ) @app.get( "/api/duties", response_model=list[DutyWithUser], summary="List duties", description=( "Returns duties for the given date range. Requires Telegram Mini App initData " "(or MINI_APP_SKIP_AUTH / private IP in dev)." ), ) def list_duties( request: Request, dates: tuple[str, str] = Depends(get_validated_dates), _username: str = Depends(require_miniapp_username), session: Session = Depends(get_db_session), ): from_date_val, to_date_val = dates log.info( "GET /api/duties from %s", request.client.host if request.client else "?", ) return fetch_duties_response(session, from_date_val, to_date_val) @app.get( "/api/calendar-events", response_model=list[CalendarEvent], summary="List calendar events", description=( "Returns calendar events for the date range, including external ICS when " "EXTERNAL_CALENDAR_ICS_URL is set. Auth same as /api/duties." ), ) def list_calendar_events( dates: tuple[str, str] = Depends(get_validated_dates), _username: str = Depends(require_miniapp_username), ): from_date_val, to_date_val = dates url = config.EXTERNAL_CALENDAR_ICS_URL if not url: return [] events = get_calendar_events(url, from_date=from_date_val, to_date=to_date_val) return [CalendarEvent(date=e["date"], summary=e["summary"]) for e in events] @app.get( "/api/calendar/ical/team/{token}.ics", summary="Team calendar ICS", description=( "Returns an ICS calendar with all team duty shifts. Each event has " "DESCRIPTION set to the duty holder's name. No Telegram auth; access by token." ), ) def get_team_calendar_ical( token: str, session: Session = Depends(get_db_session), ) -> Response: """Return ICS calendar with all duties (event_type duty only). Token validates user.""" if not _is_valid_calendar_token(token): return Response(status_code=404, content="Not found") user = get_user_by_calendar_token(session, token) if user is None: return Response(status_code=404, content="Not found") cache_key = ("team_ics",) ics_bytes, found = ics_calendar_cache.get(cache_key) if not found: today = date.today() from_date = (today - timedelta(days=365)).strftime("%Y-%m-%d") to_date = (today + timedelta(days=365 * 2)).strftime("%Y-%m-%d") all_duties = get_duties(session, from_date=from_date, to_date=to_date) duties_duty_only = [ (d, name) for d, name, *_ in all_duties if (d.event_type or "duty") == "duty" ] ics_bytes = build_team_ics(duties_duty_only) ics_calendar_cache.set(cache_key, ics_bytes) return Response( content=ics_bytes, media_type="text/calendar; charset=utf-8", ) @app.get( "/api/calendar/ical/{token}.ics", summary="Personal calendar ICS", description=( "Returns an ICS calendar with the subscribing user's duty shifts only. " "No Telegram auth; access is by secret token in the URL." ), ) def get_personal_calendar_ical( token: str, session: Session = Depends(get_db_session), ) -> Response: """ Return ICS calendar with the subscribing user's duty shifts only. No Telegram auth; access is by secret token in the URL. """ if not _is_valid_calendar_token(token): return Response(status_code=404, content="Not found") user = get_user_by_calendar_token(session, token) if user is None: return Response(status_code=404, content="Not found") cache_key = ("personal_ics", user.id) ics_bytes, found = ics_calendar_cache.get(cache_key) if not found: today = date.today() from_date = (today - timedelta(days=365)).strftime("%Y-%m-%d") to_date = (today + timedelta(days=365 * 2)).strftime("%Y-%m-%d") duties_with_name = get_duties_for_user( session, user.id, from_date=from_date, to_date=to_date, event_types=["duty"] ) ics_bytes = build_personal_ics(duties_with_name) ics_calendar_cache.set(cache_key, ics_bytes) return Response( content=ics_bytes, media_type="text/calendar; charset=utf-8", ) webapp_path = config.PROJECT_ROOT / "webapp" if webapp_path.is_dir(): app.mount("/app", StaticFiles(directory=str(webapp_path), html=True), name="webapp")