"""Galerie publique -- accessible sans authentification.""" import asyncio import logging from datetime import datetime from pathlib import Path import httpx from fastapi import APIRouter, Request, Query from fastapi.responses import HTMLResponse, RedirectResponse, Response from fastapi.templating import Jinja2Templates logger = logging.getLogger(__name__) router = APIRouter() _templates = Jinja2Templates(directory=Path(__file__).parent.parent.parent / "frontend" / "templates") @router.get("/", response_class=HTMLResponse) async def index(request: Request): return RedirectResponse(url="/gallery") @router.get("/gallery", response_class=HTMLResponse) async def gallery_page(request: Request): cfg = request.app.state.config if not cfg.gallery.public_enabled: return HTMLResponse("

Galerie desactivee

", status_code=403) return _templates.TemplateResponse(request, "public/gallery.html", {"config": cfg}) @router.get("/api/gallery/photos") async def api_gallery_photos( request: Request, page: int = Query(default=1, ge=1), limit: int = Query(default=24, ge=1, le=100), event_slug: str = Query(default=None, description="Filtrer par slug d'événement"), ): pb = request.app.state.photobooth_service cfg = request.app.state.config all_photos = await pb.get_media_collection(limit=500) photos = [p for p in all_photos if _is_image(p)] # Filtre par événement si demandé if event_slug: event_svc = getattr(request.app.state, "event_service", None) if event_svc: ev = await event_svc.get_event_by_slug(event_slug) if ev: started_at = ev["started_at"] or 0.0 ended_at = ev["ended_at"] or datetime.now().timestamp() # Construire un index mtime par stem ET par nom complet mtime_index: dict[str, float] = {} for media_dir_candidate in [ Path(cfg.photobooth.media_dir), Path("/home/pi/photobooth-data/media/processed_full"), Path("/home/pi/photobooth-data/media"), ]: if media_dir_candidate.exists(): for f in media_dir_candidate.iterdir(): if f.is_file(): mt = f.stat().st_mtime mtime_index[f.name] = mt # uuid.jpg mtime_index[f.stem] = mt # uuid break def _in_event(p: dict) -> bool: pid = _get_id(p) stem = pid.rsplit(".", 1)[0] if "." in pid else pid mt = mtime_index.get(pid) or mtime_index.get(stem) or 0.0 return mt > 0 and started_at <= mt <= ended_at photos = [p for p in photos if _in_event(p)] total = len(photos) start = (page - 1) * limit end = start + limit page_photos = photos[start:end] for p in page_photos: pid = _get_id(p) p["full_url"] = pb.media_url(pid) p["thumb_url"] = pb.thumbnail_url(pid) p["download_url"] = f"/api/gallery/download/{pid}" return { "photos": page_photos, "total": total, "page": page, "pages": max(1, (total + limit - 1) // limit), "event_slug": event_slug, } @router.get("/api/gallery/download/{photo_id}") async def download_photo(request: Request, photo_id: str): """Proxy la photo full-res avec Content-Disposition et nom basé sur l'événement + date photo.""" pb = request.app.state.photobooth_service cfg = request.app.state.config event_svc = getattr(request.app.state, "event_service", None) # Toujours fetcher depuis localhost (base_url), pas l'URL publique local_url = f"{pb._base.rstrip('/')}/media/full/{photo_id}" # Nom de l'événement (lisible) + date de la photo depuis le disque event_name = (cfg.event.name or cfg.event.slug or "Photomaton").replace(" ", "_") slug = cfg.event.slug or "photomaton" # Tenter de récupérer la vraie date depuis le fichier photo_date = datetime.now().strftime("%Y-%m-%d_%H-%M") try: media_dir = Path(cfg.photobooth.media_dir) candidates = list(media_dir.glob(f"{photo_id}*")) if candidates: mtime = candidates[0].stat().st_mtime photo_date = datetime.fromtimestamp(mtime).strftime("%Y-%m-%d_%H-%M") except Exception: pass safe_name = "".join(c if c.isalnum() or c in "-_." else "_" for c in event_name) filename = f"{safe_name}_{photo_date}.jpg" try: async with httpx.AsyncClient(timeout=30.0) as client: r = await client.get(local_url) r.raise_for_status() if event_svc: asyncio.create_task(event_svc.increment(slug, "downloads")) content_type = r.headers.get("content-type", "image/jpeg") return Response( content=r.content, media_type=content_type, headers={ "Content-Disposition": f'attachment; filename="{filename}"', "Content-Length": str(len(r.content)), "Cache-Control": "no-cache", }, ) except Exception as e: logger.warning("Download proxy échoue (%s): %s", photo_id, e) return RedirectResponse(url=pb.media_url(photo_id)) def _is_image(item: dict) -> bool: t = item.get("type", item.get("mediaitem_type", "image")) return str(t).lower() in ("image", "still", "photo") def _get_id(item: dict) -> str: return str(item.get("id", item.get("filename", item.get("uid", ""))))