130 lines
4.8 KiB
Python
130 lines
4.8 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)
|
|
|
|
async def close(self):
|
|
await self._client.aclose()
|