espresso_frame/server/app/routers/manage.py

"""The limited no-login manage surface behind the on-frame "scan to
manage" QR. The QR resolves to /m/<manage_token> (see main.index's
device-credential redirect); the token grants exactly: view the current
photo + upcoming queue, promote ("show next"), advance, back, and
thumbnails. No settings, no removal, no other frames -- full control
requires logging in."""

from __future__ import annotations

import logging

import httpx
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse, Response
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session

from .. import photo_queue, quiet_hours
from ..db import frame_locked, get_db
from ..models import Frame
from .common import immich_client_for, list_assets, require_configured

logger = logging.getLogger(__name__)

router = APIRouter()
templates = Jinja2Templates(directory="app/templates")


def require_manage(manage_token: str, db: Session = Depends(get_db)) -> Frame:
    frame = db.scalars(select(Frame).where(Frame.manage_token == manage_token)).first()
    if frame is None:
        raise HTTPException(404, "Unknown manage link")
    return frame


@router.get("/m/{manage_token}", response_class=HTMLResponse)
def manage_page(manage_token: str, request: Request, db: Session = Depends(get_db)):
    frame = require_manage(manage_token, db)
    return templates.TemplateResponse(
        "manage.html",
        {"request": request, "frame": frame, "manage_token": manage_token},
    )


@router.get("/api/m/{manage_token}/queue")
def manage_queue(frame: Frame = Depends(require_manage), db: Session = Depends(get_db)):
    require_configured(frame)
    client = immich_client_for(frame)
    assets = list_assets(client, frame)

    with frame_locked(db, frame.id) as cfg:
        photo_queue.get_current(cfg, assets, in_quiet_hours=quiet_hours.in_quiet_hours(cfg))
        photo_queue.sync_queue_length(cfg, assets)
        current = cfg.current_asset_id
        queue = list(cfg.queue)

    def entry(asset_id: str) -> dict:
        return {"id": asset_id, "thumbnail_url": f"/api/m/{frame.manage_token}/thumbnail/{asset_id}"}

    return {
        "frame_name": frame.name,
        "current": entry(current) if current else None,
        "upcoming": [entry(asset_id) for asset_id in queue],
    }


class ManagePromoteRequest(BaseModel):
    asset_id: str


@router.post("/api/m/{manage_token}/promote")
def manage_promote(
    body: ManagePromoteRequest,
    frame: Frame = Depends(require_manage),
    db: Session = Depends(get_db),
):
    with frame_locked(db, frame.id) as cfg:
        if body.asset_id not in cfg.queue:
            raise HTTPException(400, "That photo is no longer in the upcoming queue")
        cfg.queue = [body.asset_id] + [a for a in cfg.queue if a != body.asset_id]
    return {"status": "saved"}


@router.post("/api/m/{manage_token}/advance")
def manage_advance(frame: Frame = Depends(require_manage), db: Session = Depends(get_db)):
    """Advances the server-side current photo; the panel itself updates
    on the device's next wake (or its next-photo button)."""
    require_configured(frame)
    client = immich_client_for(frame)
    assets = list_assets(client, frame)
    with frame_locked(db, frame.id) as cfg:
        photo_queue.advance_forced(cfg, assets)
    return {"status": "saved"}


@router.post("/api/m/{manage_token}/back")
def manage_back(frame: Frame = Depends(require_manage), db: Session = Depends(get_db)):
    require_configured(frame)
    client = immich_client_for(frame)
    assets = list_assets(client, frame)
    with frame_locked(db, frame.id) as cfg:
        photo_queue.back_forced(cfg, assets)
    return {"status": "saved"}


@router.get("/api/m/{manage_token}/thumbnail/{asset_id}")
def manage_thumbnail(asset_id: str, frame: Frame = Depends(require_manage)):
    """Thumbnails scoped to what this frame is actually showing/queuing --
    the manage token must not become a general Immich proxy."""
    if asset_id != frame.current_asset_id and asset_id not in frame.queue:
        raise HTTPException(404, "Not on this frame")
    client = immich_client_for(frame)
    try:
        content, content_type = client.download_asset_thumbnail(asset_id)
    except httpx.HTTPError as e:
        raise HTTPException(502, f"Could not download thumbnail from Immich: {e}") from e
    return Response(content=content, media_type=content_type)