feat: badges imprimé ( copies) dans galeries admin+public; fix: timestamp_to_date dans print.html
🚀 Deploy — JH Photomaton / 🔍 Vérification (push) Has been cancelled
🚀 Deploy — JH Photomaton / 🍓 Deploy sur le Pi (push) Has been cancelled

This commit is contained in:
2026-07-17 16:48:48 +02:00
parent 742186ec8e
commit c32b3ba345
7 changed files with 123 additions and 29 deletions
+8
View File
@@ -1,6 +1,7 @@
"""Routes du dashboard admin -- authentification requise."""
import logging
from datetime import datetime
from pathlib import Path
from fastapi import APIRouter, Request, Form
@@ -141,6 +142,13 @@ async def admin_print(request: Request):
queue = await printer_svc.get_queue()
printers = await printer_svc.get_printers_status()
# Formate les dates pour le template Jinja2
for q in queue:
ts = q.get("requested_at")
q["requested_at_str"] = (
datetime.fromtimestamp(ts).strftime("%d/%m %H:%M") if ts else ""
)
return _templates.TemplateResponse(request, "admin/print.html", {
"config": cfg,
"queue": queue,
+13 -9
View File
@@ -93,21 +93,25 @@ async def admin_get_photos(
p["thumb_url"] = pb.thumbnail_url(pid)
p["download_url"] = f"/api/gallery/download/{pid}"
# Croise avec la file d'impression en attente
# Croise avec les stats d'impression (pending, printing, done)
try:
pending_map = await printer_svc.get_pending_by_photo_id()
stats_map = await printer_svc.get_print_stats_by_photo_id()
for p in page_photos:
pid = p.get("photo_id", "")
requests = pending_map.get(pid, [])
p["print_requests"] = requests
p["print_pending"] = len([r for r in requests if r["status"] == "pending"])
p["print_printing"] = len([r for r in requests if r["status"] == "printing"])
info = stats_map.get(pid, {})
p["print_requests"] = info.get("requests", [])
p["print_pending"] = info.get("pending", 0)
p["print_printing"] = info.get("printing", 0)
p["print_done"] = info.get("done", 0)
p["print_done_copies"] = info.get("copies_done", 0)
except Exception as e:
logger.warning("Impossible de croiser avec print_queue: %s", e)
for p in page_photos:
p["print_requests"] = []
p["print_pending"] = 0
p["print_printing"] = 0
p["print_requests"] = []
p["print_pending"] = 0
p["print_printing"] = 0
p["print_done"] = 0
p["print_done_copies"] = 0
return {
"photos": page_photos,
+16
View File
@@ -74,6 +74,22 @@ async def api_gallery_photos(
p["thumb_url"] = pb.thumbnail_url(pid)
p["download_url"] = f"/api/gallery/download/{pid}"
# Ajoute les stats d'impression (copies réussies) pour les badges
printer_svc = getattr(request.app.state, "printer_service", None)
if printer_svc:
try:
stats_map = await printer_svc.get_print_stats_by_photo_id()
for p in page_photos:
pid = _get_id(p)
info = stats_map.get(pid, {})
p["print_done_copies"] = info.get("copies_done", 0)
except Exception:
for p in page_photos:
p["print_done_copies"] = 0
else:
for p in page_photos:
p["print_done_copies"] = 0
return {
"photos": page_photos,
"total": total,
+34 -8
View File
@@ -111,18 +111,44 @@ class PrinterService:
return await self.get_queue("pending")
async def get_pending_by_photo_id(self) -> dict[str, list]:
"""Retourne un dict {photo_id: [entries]} pour toutes les demandes actives.
"""Retourne {photo_id: [entries actives]}. Conservé pour compatibilité."""
stats = await self.get_print_stats_by_photo_id()
return {pid: info["requests"] for pid, info in stats.items() if info["requests"]}
Utilise le champ photo_id (UUID photobooth-app) s'il est renseigné,
sinon fall-back sur le stem du filename.
async def get_print_stats_by_photo_id(self) -> dict[str, dict]:
"""Stats complètes d'impression par photo_id.
Retourne {photo_id: {requests, pending, printing, done, copies_done}}.
'requests' contient uniquement les entrées pending/printing (pour la lightbox).
'done' et 'copies_done' comptent les impressions réussies.
"""
rows = await self.get_queue("pending")
rows += await self.get_queue("printing")
async with self._db.execute(
"SELECT id, photo_id, filename, status, copies, requested_at, thumb_url "
"FROM print_queue WHERE status IN ('pending','printing','done') "
"ORDER BY requested_at"
) as cur:
rows = await cur.fetchall()
result: dict[str, list] = {}
result: dict[str, dict] = {}
for r in rows:
key = r.get("photo_id") or Path(r["filename"]).stem
result.setdefault(key, []).append(r)
key = r["photo_id"] or Path(r["filename"]).stem
if key not in result:
result[key] = {
"requests": [],
"pending": 0,
"printing": 0,
"done": 0,
"copies_done": 0,
}
if r["status"] in ("pending", "printing"):
result[key]["requests"].append(dict(r))
if r["status"] == "pending":
result[key]["pending"] += 1
elif r["status"] == "printing":
result[key]["printing"] += 1
elif r["status"] == "done":
result[key]["done"] += 1
result[key]["copies_done"] += r["copies"] or 1
return result
async def cancel_by_photo_id(self, photo_id: str) -> int: