"""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 get_media_item(self, media_id: str) -> dict | None: """Retourne les détails complets d'un item (id, created_at, processed, ...).""" try: r = await self._client.get(f"/api/mediacollection/{media_id}") if r.status_code == 404: return None r.raise_for_status() return r.json() except Exception as e: logger.error("Erreur récupération item %s: %s", media_id, e) return None def media_file_path(self, item: dict) -> Path | None: """Retourne le chemin absolu du fichier traité depuis le champ 'processed' d'un item. photobooth-app retourne 'processed': 'media/processed_full/YYYYMMDD-HHMMSS-xxx.jpg' On préfixe avec data_dir pour obtenir le chemin complet. """ processed = item.get("processed") if not processed: return None return Path(self._cfg.data_dir) / processed 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: base = (self._cfg.public_url.rstrip("/") or self._base) return f"{base}/media/full/{identifier}" def thumbnail_url(self, identifier: str) -> str: base = (self._cfg.public_url.rstrip("/") or self._base) return f"{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. Endpoints snapshot direct (retour JPEG immédiat) 2. Flux MJPEG — extrait la premiere frame Retourne des bytes JPEG ou None si rien n'est disponible. """ # 1. Endpoints snapshot direct (differentes versions de photobooth-app) snapshot_paths = ( "/api/stream/snapshot", "/api/video/preview/snapshot", "/api/livestream/snapshot", "/api/cam/stream/snapshot", ) for path in snapshot_paths: 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: logger.debug("Snapshot OK via %s", path) return r.content except Exception as e: logger.debug("Snapshot %s: %s", path, e) # 2. Premiere frame d'un flux MJPEG (differentes versions) mjpeg_paths = ( "/api/aquisition/stream.mjpg", # photobooth-app v4+ (typo intentionnelle dans leur code) "/stream.mjpg", "/api/stream", "/api/video/stream", "/api/video/preview", "/api/livestream", "/livestream", ) for path in mjpeg_paths: 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: logger.debug("MJPEG frame OK via %s", path) return buf[start:end + 2] if len(buf) > 500_000: break except Exception as e: logger.debug("MJPEG %s: %s", path, e) logger.debug("Aucun endpoint liveview disponible sur %s", self._base) return None async def close(self): await self._client.aclose()