"""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 starlette.middleware.base import BaseHTTPMiddleware 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(BaseHTTPMiddleware): """Set Cache-Control for /app/*.js and /app/*.html so WebView gets fresh JS (i18n, etc.).""" async def dispatch(self, request, call_next): response = await call_next(request) path = request.url.path if path.startswith("/app/") and ( path.endswith(".js") or path.endswith(".html") ): response.headers["Cache-Control"] = "no-store" return response app.add_middleware(NoCacheStaticMiddleware) @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")