"""ESPresso Frame server: pulls photos from Immich, pre-processes them for the panel, and serves ESP32 frames ready-to-display images. This module is assembly only -- routes live in app/routers/ (device.py for the firmware-facing /frame/* protocol, api.py for the web UI's /api/*, pages.py for setup/login/settings/admin), storage in SQLite via models.py/db.py, with migration.py importing a pre-database config.json deployment on first boot.""" from __future__ import annotations import logging from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from sqlalchemy import select from . import migration from .auth import ( browser_token_valid, current_session, current_user, management_token, users_exist, ) from .db import SessionLocal from .models import Frame from .quiet_hours import ALL_TIMEZONES from .routers import api, device, manage, pages from .routers.common import default_frame, immich_creds logger = logging.getLogger(__name__) # Schema + legacy-config import, before the first request is served. migration.run_migrations() app = FastAPI(title="ESPresso Frame Server") templates = Jinja2Templates(directory="app/templates") app.include_router(device.router) app.include_router(api.router) app.include_router(pages.router) app.include_router(manage.router) def _device_credential_redirect(request: Request, db, allow_legacy: bool) -> str | None: """The on-frame manage QR points at the server root with the device's own credentials (new firmware: ?id=&token=; deployed firmware: ?token=). Those scans get the frame's limited manage page -- never the full UI, which now requires a login. allow_legacy is False before /setup has run: at that point a bare ?token= hit is the admin coming through the token prompt to do first-run setup, not a QR scan.""" device_id = request.query_params.get("id", "").strip().lower() token = request.query_params.get("token", "") if device_id and token: frame = db.scalars(select(Frame).where(Frame.device_id == device_id)).first() if frame is not None and token == frame.device_token: return f"/m/{frame.manage_token}" if allow_legacy and token and management_token() and token == management_token(): frame = db.scalars( select(Frame).where(Frame.legacy_token_enabled == True) # noqa: E712 ).first() if frame is not None: return f"/m/{frame.manage_token}" return None @app.get("/health") def health() -> dict: return {"status": "ok"} @app.get("/", response_class=HTMLResponse) def index(request: Request): """The web UI (still the single-frame page until Phase D). Access: a user session (the normal path once /setup has run), or -- only while no users exist AND no MANAGEMENT_TOKEN is configured -- fully open, the original trusted-LAN default. A hit carrying device credentials (the on-frame manage QR) redirects to that frame's limited manage page instead.""" with SessionLocal() as db: have_users = users_exist(db) manage_redirect = _device_credential_redirect(request, db, allow_legacy=have_users) if manage_redirect is not None: return RedirectResponse(manage_redirect, status_code=303) user = current_user(request, db) session = current_session(request, db) if user else None if user is None: if not have_users: if management_token() and not browser_token_valid(request): supplied = request.query_params.get("token") return templates.TemplateResponse( "token_prompt.html", {"request": request, "wrong": supplied is not None} ) # Pre-setup: reachable (optionally token-gated), nudge setup. return RedirectResponse("/setup", status_code=303) return RedirectResponse("/login", status_code=303) frame = default_frame(db) immich_url, _ = immich_creds(frame) return templates.TemplateResponse( "index.html", { "request": request, "cfg": frame, "immich_url": immich_url, "timezones": ALL_TIMEZONES, "user": user, "csrf_token": session.csrf_token if session else None, }, )