import asyncio import base64 import json import logging from pathlib import Path from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from backend.config import ( RACINE, DOSSIER_PHOTOS, DOSSIER_EXPORTS, DOSSIER_OVERLAYS, DOSSIER_FONDS, DOSSIER_ANIMATIONS, charger_config, sauvegarder_config, mettre_a_jour_config, ) from backend.camera import camera, GPHOTO2_DISPONIBLE try: import gphoto2 as gp except ImportError: gp = None from backend.gallery import lister_photos, compter_photos, supprimer_photo, vider_galerie from backend.effects import appliquer_filtre, appliquer_overlay, chroma_key, lister_overlays, lister_fonds, FILTRES from backend.collage import creer_strip, creer_collage, creer_impression_strip from backend.destinations import distribuer_photo, detecter_usb, compteur_restant, reset_compteur, recuperer_booth_password from backend.printer import lister_imprimantes, imprimer from backend.mailer import envoyer_photo from backend.qrcode_gen import generer_qr, qr_galerie logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s") log = logging.getLogger("photobooth") # Clients WebSocket connectes clients_ws: list[WebSocket] = [] capture_en_cours = False async def surveiller_dslr(): """Verifie periodiquement si un DSLR est branche et bascule dessus.""" while True: await asyncio.sleep(10) if capture_en_cours: continue try: if camera.mode != "gphoto2" and GPHOTO2_DISPONIBLE: dslrs = gp.Camera.autodetect() if len(dslrs) > 0: log.info(f"DSLR detecte : {dslrs[0][0]}, bascule automatique") camera.connecter(source="gphoto2") except Exception: pass @asynccontextmanager async def lifespan(app: FastAPI): """Demarrage et arret de l'application.""" log.info("Demarrage du photobooth") camera.connecter() task = asyncio.create_task(surveiller_dslr()) yield task.cancel() log.info("Arret du photobooth") camera.deconnecter() app = FastAPI(title="Photobooth", lifespan=lifespan) # Servir les fichiers statiques app.mount("/assets", StaticFiles(directory=str(RACINE / "frontend" / "assets")), name="assets") app.mount("/css", StaticFiles(directory=str(RACINE / "frontend" / "css")), name="css") app.mount("/js", StaticFiles(directory=str(RACINE / "frontend" / "js")), name="js") app.mount("/data", StaticFiles(directory=str(RACINE / "data")), name="data") # --- Pages --- @app.get("/") async def page_principale(): return FileResponse(str(RACINE / "frontend" / "index.html")) # --- API Config --- @app.get("/api/config") async def api_config(): return charger_config() @app.post("/api/config") async def api_config_update(modifications: dict): config = mettre_a_jour_config(modifications) await diffuser_ws({"type": "config_maj", "config": config}) return config # --- API Camera --- @app.post("/api/capturer") async def api_capturer(): global capture_en_cours # Verifier le compteur etat = compteur_restant() if etat["actif"] and etat["restantes"] <= 0: return JSONResponse({"erreur": "Limite de photos atteinte"}, status_code=403) capture_en_cours = True try: chemin = camera.capturer() finally: capture_en_cours = False if chemin is None: return JSONResponse({"erreur": "Echec capture"}, status_code=500) nom = chemin.name # Distribuer vers les destinations distribuer_photo(chemin, imprimee=False) await diffuser_ws({"type": "photo_capturee", "nom": nom}) return {"nom": nom, "chemin": f"/data/photos/{nom}"} @app.get("/api/preview") async def api_preview(): donnees = camera.preview() if donnees is None: return JSONResponse({"erreur": "Preview indisponible"}, status_code=500) b64 = base64.b64encode(donnees).decode("ascii") return {"image": f"data:image/jpeg;base64,{b64}"} @app.get("/api/camera/statut") async def api_camera_statut(): return { "connectee": camera.connectee, "mode": camera.mode, "appareils": camera.lister_appareils(), } @app.post("/api/camera/reconnecter") async def api_camera_reconnecter(body: dict = {}): camera.deconnecter() source = body.get("source") ok = camera.connecter(source=source) return {"connectee": ok, "mode": camera.mode} # --- API Effets --- @app.get("/api/filtres") async def api_filtres(): return FILTRES @app.post("/api/filtre") async def api_appliquer_filtre(donnees: dict): nom_photo = donnees.get("photo", "") filtre = donnees.get("filtre", "original") chemin = DOSSIER_PHOTOS / nom_photo if not chemin.exists(): return JSONResponse({"erreur": "Photo introuvable"}, status_code=404) if filtre == "original": return {"nom": nom_photo, "chemin": f"/data/photos/{nom_photo}"} chemin_export = appliquer_filtre(chemin, filtre) return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"} @app.get("/api/overlays") async def api_overlays(): return lister_overlays() @app.post("/api/overlay") async def api_appliquer_overlay(donnees: dict): nom_photo = donnees.get("photo", "") nom_overlay = donnees.get("overlay", "") chemin = DOSSIER_PHOTOS / nom_photo if not chemin.exists(): return JSONResponse({"erreur": "Photo introuvable"}, status_code=404) chemin_export = appliquer_overlay(chemin, nom_overlay) return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"} @app.post("/api/chroma") async def api_chroma_key(donnees: dict): nom_photo = donnees.get("photo", "") nom_fond = donnees.get("fond") chemin = DOSSIER_PHOTOS / nom_photo if not chemin.exists(): return JSONResponse({"erreur": "Photo introuvable"}, status_code=404) config = charger_config() conf_chroma = config.get("chroma_key", {}) chemin_export = chroma_key( chemin, nom_fond, couleur_cle=conf_chroma.get("couleur", "#00ff00"), tolerance=conf_chroma.get("tolerance", 40), ) return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"} @app.get("/api/fonds") async def api_fonds(): return lister_fonds() @app.delete("/api/fonds/{nom}") async def api_supprimer_fond(nom: str): chemin = DOSSIER_FONDS / nom if chemin.exists() and chemin.parent == DOSSIER_FONDS: chemin.unlink() return {"succes": True} return JSONResponse({"erreur": "Fichier introuvable"}, status_code=404) # --- API Collage --- @app.post("/api/strip") async def api_strip(donnees: dict): noms = donnees.get("photos", []) chemins = [DOSSIER_PHOTOS / n for n in noms] for c in chemins: if not c.exists(): return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404) chemin_strip = creer_strip(chemins) # Creer aussi la page d'impression (2 bandes sur 10x15 paysage) chemin_print = creer_impression_strip(chemin_strip) return { "nom": chemin_strip.name, "chemin": f"/data/exports/{chemin_strip.name}", "impression": chemin_print.name, "chemin_impression": f"/data/exports/{chemin_print.name}", } @app.post("/api/collage") async def api_collage(donnees: dict): noms = donnees.get("photos", []) colonnes = donnees.get("colonnes", 2) chemins = [DOSSIER_PHOTOS / n for n in noms] for c in chemins: if not c.exists(): return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404) chemin = creer_collage(chemins, colonnes=colonnes) return {"nom": chemin.name, "chemin": f"/data/exports/{chemin.name}"} # --- API Impression --- @app.get("/api/imprimantes") async def api_imprimantes(): return lister_imprimantes() @app.post("/api/imprimer") async def api_imprimer(donnees: dict): nom = donnees.get("photo", "") copies = donnees.get("copies", 1) # Limiter au max configure config = charger_config() copies_max = config.get("impression", {}).get("copies_max", 5) copies = max(1, min(copies, copies_max)) # Chercher dans exports d'abord, puis photos chemin = DOSSIER_EXPORTS / nom if not chemin.exists(): chemin = DOSSIER_PHOTOS / nom if not chemin.exists(): return JSONResponse({"erreur": "Photo introuvable"}, status_code=404) ok = imprimer(chemin, copies=copies) if ok: distribuer_photo(chemin, imprimee=True) return {"succes": ok} # --- API Booth (galerie live) --- @app.get("/api/booth/info") async def api_booth_info(): return recuperer_booth_password() # --- API Email --- @app.post("/api/email") async def api_email(donnees: dict): email_dest = donnees.get("email", "") nom = donnees.get("photo", "") if not email_dest: return JSONResponse({"erreur": "Email requis"}, status_code=400) chemin = DOSSIER_EXPORTS / nom if not chemin.exists(): chemin = DOSSIER_PHOTOS / nom if not chemin.exists(): return JSONResponse({"erreur": "Photo introuvable"}, status_code=404) ok = envoyer_photo(email_dest, chemin) return {"succes": ok} # --- API QR Code --- @app.get("/api/qr") async def api_qr(url: str = ""): """Genere un QR code pour une URL arbitraire.""" if not url: return JSONResponse({"erreur": "URL requise"}, status_code=400) import qrcode import io qr = qrcode.make(url, box_size=6, border=2) buf = io.BytesIO() qr.save(buf, format="PNG") buf.seek(0) from fastapi.responses import StreamingResponse return StreamingResponse(buf, media_type="image/png") @app.get("/api/qr/galerie") async def api_qr_galerie(): chemin = qr_galerie() if chemin is None: return JSONResponse({"erreur": "URL galerie non configuree"}, status_code=400) return {"chemin": f"/data/exports/{chemin.name}"} # --- API Galerie --- @app.get("/api/galerie") async def api_galerie(): return lister_photos("exports") @app.get("/api/galerie/compteur") async def api_compteur(): return compter_photos() @app.delete("/api/galerie/{nom}") async def api_supprimer_photo(nom: str): ok = supprimer_photo(nom) return {"succes": ok} @app.post("/api/galerie/vider") async def api_vider_galerie(): vider_galerie() return {"succes": True} # --- API Upload overlay/fond --- @app.post("/api/upload/overlay") async def api_upload_overlay(fichier: UploadFile = File(...)): chemin = DOSSIER_OVERLAYS / fichier.filename with open(chemin, "wb") as f: f.write(await fichier.read()) return {"nom": fichier.filename} @app.post("/api/upload/fond") async def api_upload_fond(fichier: UploadFile = File(...)): chemin = DOSSIER_FONDS / fichier.filename with open(chemin, "wb") as f: f.write(await fichier.read()) return {"nom": fichier.filename} # --- API Compteur --- @app.get("/api/compteur") async def api_compteur_etat(): return compteur_restant() @app.post("/api/compteur/reset") async def api_compteur_reset(): reset_compteur() return compteur_restant() # --- API Destinations --- @app.get("/api/usb/detecter") async def api_detecter_usb(): return detecter_usb() # --- API Cadres actifs --- @app.get("/api/cadres") async def api_cadres(): config = charger_config() tous = lister_overlays() actifs = config.get("cadres", {}).get("actifs", []) return {"tous": tous, "actifs": actifs} @app.post("/api/cadres") async def api_cadres_update(donnees: dict): actifs = donnees.get("actifs", []) mettre_a_jour_config({"cadres": {"actifs": actifs}}) return {"actifs": actifs} # --- API Animations compte a rebours --- ANIMATIONS_CAR = { "classique": "Classique (chiffres simples)", "explosion": "Explosion (chiffres qui eclatent)", "rebond": "Rebond (chiffres qui rebondissent)", "rotation": "Rotation 3D", "fondu": "Fondu enchaine", "emoji": "Emoji fun", "shake": "Tremblement", } @app.get("/api/animations") async def api_animations(): return ANIMATIONS_CAR @app.get("/api/animations/custom") async def api_animations_custom(): """Liste les animations personnalisees (GIF/video).""" extensions = {".gif", ".mp4", ".webm"} fichiers = [] if DOSSIER_ANIMATIONS.exists(): for f in sorted(DOSSIER_ANIMATIONS.iterdir()): if f.suffix.lower() in extensions: fichiers.append(f.name) config = charger_config() actives = config.get("animations_custom", {}).get("actives", []) return {"tous": fichiers, "actives": actives} @app.post("/api/animations/custom") async def api_animations_custom_update(donnees: dict): actives = donnees.get("actives", []) mettre_a_jour_config({"animations_custom": {"actives": actives}}) return {"actives": actives} @app.post("/api/upload/animation") async def api_upload_animation(fichier: UploadFile = File(...)): chemin = DOSSIER_ANIMATIONS / fichier.filename with open(chemin, "wb") as f: f.write(await fichier.read()) return {"nom": fichier.filename} @app.delete("/api/animations/custom/{nom}") async def api_supprimer_animation(nom: str): chemin = DOSSIER_ANIMATIONS / nom if chemin.exists() and chemin.parent == DOSSIER_ANIMATIONS: chemin.unlink() return {"succes": True} return JSONResponse({"erreur": "Fichier introuvable"}, status_code=404) # --- API Systeme --- @app.post("/api/systeme/redemarrer") async def api_redemarrer(): import subprocess log.warning("Redemarrage systeme demande") subprocess.Popen(["sudo", "reboot"]) return {"succes": True} @app.post("/api/systeme/eteindre") async def api_eteindre(): import subprocess log.warning("Extinction systeme demandee") subprocess.Popen(["sudo", "shutdown", "-h", "now"]) return {"succes": True} @app.post("/api/systeme/quitter-navigateur") async def api_quitter_navigateur(): import subprocess log.warning("Fermeture navigateur demandee") subprocess.Popen(["pkill", "-f", "firefox"]) subprocess.Popen(["pkill", "-f", "chromium"]) return {"succes": True} # --- WebSocket --- @app.websocket("/ws") async def websocket_endpoint(ws: WebSocket): await ws.accept() clients_ws.append(ws) log.info(f"Client WS connecte ({len(clients_ws)} clients)") try: while True: data = await ws.receive_text() msg = json.loads(data) await traiter_message_ws(msg, ws) except WebSocketDisconnect: clients_ws.remove(ws) log.info(f"Client WS deconnecte ({len(clients_ws)} clients)") async def traiter_message_ws(msg: dict, ws: WebSocket): """Traite les messages WebSocket entrants.""" type_msg = msg.get("type", "") if type_msg == "ping": await ws.send_json({"type": "pong"}) elif type_msg == "preview": donnees = camera.preview() if donnees: b64 = base64.b64encode(donnees).decode("ascii") await ws.send_json({"type": "preview", "image": f"data:image/jpeg;base64,{b64}"}) async def diffuser_ws(message: dict): """Envoie un message a tous les clients WebSocket.""" deconnectes = [] for ws in clients_ws: try: await ws.send_json(message) except Exception: deconnectes.append(ws) for ws in deconnectes: clients_ws.remove(ws) # Point d'entree if __name__ == "__main__": import uvicorn config = charger_config() conf_srv = config.get("serveur", {}) uvicorn.run( "backend.main:app", host=conf_srv.get("host", "0.0.0.0"), port=conf_srv.get("port", 8080), reload=False, log_level="info", )