211 lines
8.5 KiB
Python
211 lines
8.5 KiB
Python
"""Client HTTP pour l'API de photobooth-app."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from backend.services.config_service import Config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PhotoboothService:
|
|
def __init__(self, config: Config):
|
|
self._cfg = config.photobooth
|
|
self._base = self._cfg.base_url.rstrip("/")
|
|
self._client = httpx.AsyncClient(base_url=self._base, timeout=10.0)
|
|
|
|
async def trigger_image_action(self, index: int):
|
|
"""Déclenche l'action image à l'index donné."""
|
|
r = await self._client.get(f"/api/actions/image/{index}")
|
|
r.raise_for_status()
|
|
logger.info("Action image %d déclenchée", index)
|
|
return r.json() if r.text else {}
|
|
|
|
async def trigger_share_latest(self, share_index: int = 0):
|
|
"""Déclenche l'action de partage (impression) sur la dernière photo."""
|
|
r = await self._client.get(f"/api/share/actions/latest/{share_index}")
|
|
r.raise_for_status()
|
|
logger.info("Share action %d déclenchée", share_index)
|
|
return r.json() if r.text else {}
|
|
|
|
async def get_media_collection(self, limit: int = 200) -> list[dict]:
|
|
"""Retourne la liste des photos de la galerie."""
|
|
try:
|
|
r = await self._client.get("/api/mediacollection/", params={"limit": limit})
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
# photobooth-app retourne soit une liste soit {"items": [...]}
|
|
if isinstance(data, list):
|
|
return data
|
|
return data.get("items", data.get("media_items", []))
|
|
except Exception as e:
|
|
logger.error("Erreur récupération galerie: %s", e)
|
|
return []
|
|
|
|
async def get_latest_media(self) -> dict | None:
|
|
"""Retourne les infos de la dernière photo."""
|
|
items = await self.get_media_collection(limit=1)
|
|
return items[0] if items else None
|
|
|
|
async def delete_media(self, media_id: str) -> bool:
|
|
"""Supprime une photo via l'API photobooth-app."""
|
|
try:
|
|
r = await self._client.delete(f"/api/mediacollection/{media_id}")
|
|
return r.status_code in (200, 204)
|
|
except Exception as e:
|
|
logger.error("Erreur suppression %s: %s", media_id, e)
|
|
return False
|
|
|
|
async def is_alive(self) -> bool:
|
|
"""Vérifie que photobooth-app répond."""
|
|
try:
|
|
r = await self._client.get("/api/about", timeout=3.0)
|
|
return r.status_code == 200
|
|
except Exception:
|
|
return False
|
|
|
|
def media_url(self, identifier: str) -> str:
|
|
return f"{self._base}/media/full/{identifier}"
|
|
|
|
def thumbnail_url(self, identifier: str) -> str:
|
|
return f"{self._base}/media/thumbnail/{identifier}"
|
|
|
|
async def read_pb_config(self) -> dict:
|
|
"""Lit le fichier config.json de photobooth-app."""
|
|
cfg_path = Path(self._cfg.config_file)
|
|
if not cfg_path.exists():
|
|
logger.warning("Config photobooth introuvable: %s", cfg_path)
|
|
return {}
|
|
with open(cfg_path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
async def write_pb_config(self, config: dict):
|
|
"""Écrit le fichier config.json de photobooth-app."""
|
|
cfg_path = Path(self._cfg.config_file)
|
|
# Backup avant écriture
|
|
backup = cfg_path.with_suffix(f".json_backup_jh")
|
|
if cfg_path.exists():
|
|
import shutil
|
|
shutil.copy2(cfg_path, backup)
|
|
|
|
with open(cfg_path, "w", encoding="utf-8") as f:
|
|
json.dump(config, f, indent=2, ensure_ascii=False)
|
|
logger.info("Config photobooth-app mise à jour")
|
|
|
|
async def list_userdata_frames(self) -> list[str]:
|
|
"""Liste les cadres PNG disponibles dans userdata."""
|
|
return self._list_files(self._cfg.userdata_dir, "*.png", "frames")
|
|
|
|
async def list_userdata_backgrounds(self) -> list[str]:
|
|
"""Liste les fonds disponibles dans userdata."""
|
|
exts = ["*.jpg", "*.jpeg", "*.png"]
|
|
files = []
|
|
for ext in exts:
|
|
files.extend(self._list_files(self._cfg.userdata_dir, ext, "backgrounds"))
|
|
return sorted(set(files))
|
|
|
|
def _list_files(self, base: str, pattern: str, subdir_hint: str) -> list[str]:
|
|
base_path = Path(base)
|
|
if not base_path.exists():
|
|
return []
|
|
results = []
|
|
for f in base_path.rglob(pattern):
|
|
if subdir_hint in f.parts or True: # liste tout
|
|
# Chemin relatif depuis data_dir pour passer à photobooth-app
|
|
try:
|
|
rel = f.relative_to(Path(self._cfg.data_dir))
|
|
results.append(str(rel))
|
|
except ValueError:
|
|
results.append(str(f))
|
|
return sorted(results)
|
|
|
|
# ── UI / private.css ──────────────────────────────────────────────────────
|
|
|
|
_CSS_HIDE_DELETE = """\
|
|
/* JH Photomaton — généré automatiquement, ne pas modifier manuellement */
|
|
/* Cacher le bouton Supprimer sur l'écran de review après capture */
|
|
.action-button-delete {
|
|
display: none !important;
|
|
}
|
|
"""
|
|
|
|
_CSS_SHOW_DELETE = """\
|
|
/* JH Photomaton — généré automatiquement, ne pas modifier manuellement */
|
|
/* Bouton Supprimer visible (activé depuis le dashboard JH Photomaton) */
|
|
/* .action-button-delete { display: none !important; } */
|
|
"""
|
|
|
|
async def set_delete_button_visible(self, visible: bool) -> bool:
|
|
"""Écrit private.css dans userdata pour afficher ou cacher le bouton Supprimer.
|
|
|
|
photobooth-app charge automatiquement userdata/private.css à chaque requête
|
|
— aucun redémarrage nécessaire, effectif dès la prochaine capture.
|
|
"""
|
|
css_path = Path(self._cfg.userdata_dir) / "private.css"
|
|
try:
|
|
css_content = self._CSS_SHOW_DELETE if visible else self._CSS_HIDE_DELETE
|
|
css_path.parent.mkdir(parents=True, exist_ok=True)
|
|
css_path.write_text(css_content, encoding="utf-8")
|
|
logger.info("private.css mis à jour : bouton delete %s", "visible" if visible else "caché")
|
|
return True
|
|
except Exception as e:
|
|
logger.error("Impossible d'écrire private.css : %s", e)
|
|
return False
|
|
|
|
async def get_delete_button_visible(self) -> bool:
|
|
"""Lit l'état actuel depuis le fichier private.css (ou depuis la config)."""
|
|
return self._cfg.show_delete_button
|
|
|
|
async def get_liveview_snapshot(self) -> bytes | None:
|
|
"""Tente de recuperer une frame JPEG du liveview de photobooth-app.
|
|
|
|
Essaie dans l'ordre :
|
|
1. /api/stream/snapshot — endpoint snapshot direct (si disponible)
|
|
2. /stream.mjpg — flux MJPEG classique, extrait la premiere frame
|
|
3. /api/stream — flux MJPEG alternatif
|
|
|
|
Retourne des bytes JPEG ou None si rien n'est disponible.
|
|
"""
|
|
# 1. Snapshot direct
|
|
for path in ("/api/stream/snapshot",):
|
|
try:
|
|
async with httpx.AsyncClient(timeout=2.0) as client:
|
|
r = await client.get(f"{self._base}{path}")
|
|
if r.status_code == 200:
|
|
ct = r.headers.get("content-type", "")
|
|
if "jpeg" in ct or "image" in ct:
|
|
return r.content
|
|
except Exception as e:
|
|
logger.debug("Snapshot %s: %s", path, e)
|
|
|
|
# 2. Premiere frame d'un flux MJPEG
|
|
for path in ("/stream.mjpg", "/api/stream"):
|
|
try:
|
|
async with httpx.AsyncClient(timeout=3.0) as client:
|
|
async with client.stream("GET", f"{self._base}{path}") as resp:
|
|
if resp.status_code != 200:
|
|
continue
|
|
buf = b""
|
|
async for chunk in resp.aiter_bytes(4096):
|
|
buf += chunk
|
|
start = buf.find(b"\xff\xd8")
|
|
if start >= 0:
|
|
end = buf.find(b"\xff\xd9", start)
|
|
if end >= 0:
|
|
return buf[start:end + 2]
|
|
if len(buf) > 500_000:
|
|
break
|
|
except Exception as e:
|
|
logger.debug("MJPEG %s: %s", path, e)
|
|
|
|
return None
|
|
|
|
async def close(self):
|
|
await self._client.aclose()
|