Jules
4e431507fc
- WebSocket : reconnexion avec backoff exponentiel au lieu de reload page - Ecran pause avec tasse de cafe quand le backend est injoignable - Endpoint /api/health pour monitoring externe - Watchdog systemd (sd_notify READY=1 + WATCHDOG=1 toutes les 10s) - Service systemd durci (Type=notify, WatchdogSec=30, KillMode, limites) - WiFi watchdog : script + timer systemd, verifie la gateway toutes les 60s - Destinations : fallback url_tunnel (WireGuard) en priorite Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
221 lines
7.7 KiB
Python
221 lines
7.7 KiB
Python
import ftplib
|
|
import logging
|
|
import shutil
|
|
import threading
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import URLError
|
|
|
|
from backend.config import charger_config, mettre_a_jour_config
|
|
|
|
log = logging.getLogger("photobooth.destinations")
|
|
|
|
|
|
def distribuer_photo(chemin_photo: Path, imprimee: bool = False):
|
|
"""Copie la photo vers toutes les destinations activees."""
|
|
config = charger_config()
|
|
dest = config.get("destinations", {})
|
|
|
|
# Galerie live (booth) — toujours envoyer, imprimee ou non
|
|
booth = config.get("booth", {})
|
|
if booth.get("actif", False):
|
|
threading.Thread(
|
|
target=envoyer_booth,
|
|
args=(chemin_photo, booth),
|
|
daemon=True,
|
|
).start()
|
|
|
|
sous_dossier = None
|
|
if not dest.get("sauvegarder_tout", True) and not imprimee:
|
|
sous_dossier = "non_imprimees"
|
|
|
|
# Cle USB
|
|
if dest.get("cle_usb", False):
|
|
copier_usb(chemin_photo, dest.get("chemin_usb", "/media/usb"), sous_dossier)
|
|
|
|
# FTP
|
|
if dest.get("ftp", False):
|
|
envoyer_ftp(chemin_photo, dest, sous_dossier)
|
|
|
|
# Incrementer le compteur (photos imprimees seulement)
|
|
if imprimee:
|
|
compteur = config.get("compteur", {})
|
|
if compteur.get("actif", False):
|
|
compteur["photos_prises"] = compteur.get("photos_prises", 0) + 1
|
|
mettre_a_jour_config({"compteur": compteur})
|
|
|
|
|
|
def copier_usb(chemin_photo: Path, chemin_usb: str, sous_dossier: str | None = None):
|
|
"""Copie une photo sur la cle USB."""
|
|
dossier_usb = Path(chemin_usb)
|
|
if not dossier_usb.exists():
|
|
log.warning(f"Cle USB non trouvee : {chemin_usb}")
|
|
return False
|
|
|
|
dossier_dest = dossier_usb / "photobooth"
|
|
if sous_dossier:
|
|
dossier_dest = dossier_dest / sous_dossier
|
|
dossier_dest.mkdir(parents=True, exist_ok=True)
|
|
|
|
try:
|
|
shutil.copy2(chemin_photo, dossier_dest / chemin_photo.name)
|
|
log.info(f"Photo copiee sur USB : {chemin_photo.name} ({sous_dossier or 'root'})")
|
|
return True
|
|
except OSError as e:
|
|
log.error(f"Erreur copie USB : {e}")
|
|
return False
|
|
|
|
|
|
def envoyer_ftp(chemin_photo: Path, config_dest: dict, sous_dossier: str | None = None):
|
|
"""Envoie une photo par FTP."""
|
|
host = config_dest.get("ftp_host", "")
|
|
port = config_dest.get("ftp_port", 21)
|
|
user = config_dest.get("ftp_user", "")
|
|
password = config_dest.get("ftp_password", "")
|
|
chemin_distant = config_dest.get("ftp_chemin", "/photobooth")
|
|
if sous_dossier:
|
|
chemin_distant = f"{chemin_distant}/{sous_dossier}"
|
|
|
|
if not host:
|
|
log.warning("FTP non configure")
|
|
return False
|
|
|
|
try:
|
|
with ftplib.FTP() as ftp:
|
|
ftp.connect(host, port, timeout=10)
|
|
ftp.login(user, password)
|
|
# Creer le dossier si necessaire
|
|
try:
|
|
ftp.mkd(chemin_distant)
|
|
except ftplib.error_perm:
|
|
pass
|
|
ftp.cwd(chemin_distant)
|
|
with open(chemin_photo, "rb") as f:
|
|
ftp.storbinary(f"STOR {chemin_photo.name}", f)
|
|
log.info(f"Photo envoyee par FTP : {chemin_photo.name}")
|
|
return True
|
|
except (ftplib.all_errors, OSError) as e:
|
|
log.error(f"Erreur FTP : {e}")
|
|
return False
|
|
|
|
|
|
def _upload_booth(url: str, api_key: str, event_id: str, chemin_photo: Path) -> bool:
|
|
boundary = "----BoothUpload"
|
|
filename = chemin_photo.name
|
|
with open(chemin_photo, "rb") as f:
|
|
file_data = f.read()
|
|
body = (
|
|
f"--{boundary}\r\n"
|
|
f'Content-Disposition: form-data; name="photo"; filename="{filename}"\r\n'
|
|
f"Content-Type: image/jpeg\r\n\r\n"
|
|
).encode() + file_data + f"\r\n--{boundary}--\r\n".encode()
|
|
req = Request(f"{url}/admin/gallery/{event_id}/upload", data=body, method="POST")
|
|
req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}")
|
|
req.add_header("X-Api-Key", api_key)
|
|
with urlopen(req, timeout=10) as resp:
|
|
return resp.status == 200
|
|
|
|
|
|
def envoyer_booth(chemin_photo: Path, config_booth: dict):
|
|
"""Envoie une photo vers la galerie live booth."""
|
|
url = config_booth.get("url", "").rstrip("/")
|
|
url_tunnel = config_booth.get("url_tunnel", "").rstrip("/")
|
|
api_key = config_booth.get("api_key", "")
|
|
config = charger_config()
|
|
event_id = config.get("evenement", {}).get("event_id") or config_booth.get("event_id", "default")
|
|
|
|
if not url and not url_tunnel:
|
|
log.warning("Booth non configure (URL manquante)")
|
|
return False
|
|
|
|
for tentative_url in [u for u in (url_tunnel, url) if u]:
|
|
try:
|
|
if _upload_booth(tentative_url, api_key, event_id, chemin_photo):
|
|
log.info(f"Photo envoyee au booth via {tentative_url} : {chemin_photo.name}")
|
|
return True
|
|
log.error(f"Booth erreur HTTP via {tentative_url}")
|
|
except (URLError, OSError) as e:
|
|
log.warning(f"Echec envoi booth via {tentative_url} : {e}")
|
|
return False
|
|
|
|
|
|
def recuperer_booth_password() -> dict:
|
|
"""Recupere le mot de passe et l'info de la session booth."""
|
|
config = charger_config()
|
|
booth = config.get("booth", {})
|
|
url = booth.get("url", "").rstrip("/")
|
|
url_tunnel = booth.get("url_tunnel", "").rstrip("/")
|
|
api_key = booth.get("api_key", "")
|
|
event_id = config.get("evenement", {}).get("event_id") or booth.get("event_id", "default")
|
|
|
|
for tentative_url in [u for u in (url_tunnel, url) if u]:
|
|
try:
|
|
req = Request(f"{tentative_url}/admin/gallery/{event_id}/info")
|
|
req.add_header("X-Api-Key", api_key)
|
|
with urlopen(req, timeout=5) as resp:
|
|
import json
|
|
return json.loads(resp.read())
|
|
except (URLError, OSError) as e:
|
|
log.warning(f"Echec recuperation booth info via {tentative_url} : {e}")
|
|
return {"password": None}
|
|
|
|
|
|
def detecter_usb() -> list[str]:
|
|
"""Detecte les cles USB montees (cherche dans /media, /mnt et /media/<user>/)."""
|
|
usb_trouvees = []
|
|
candidats = []
|
|
|
|
for racine in [Path("/media"), Path("/mnt")]:
|
|
if not racine.exists():
|
|
continue
|
|
try:
|
|
for niveau1 in racine.iterdir():
|
|
if not niveau1.is_dir():
|
|
continue
|
|
if niveau1.is_mount():
|
|
candidats.append(niveau1)
|
|
else:
|
|
# /media/<user>/ → descendre encore (udiskie)
|
|
try:
|
|
for niveau2 in niveau1.iterdir():
|
|
if niveau2.is_dir() and niveau2.is_mount():
|
|
candidats.append(niveau2)
|
|
except PermissionError:
|
|
pass
|
|
except PermissionError:
|
|
pass
|
|
|
|
for chemin in candidats:
|
|
# Exclure les partitions système
|
|
try:
|
|
result = __import__("subprocess").run(
|
|
["findmnt", "-no", "FSTYPE", str(chemin)],
|
|
capture_output=True, text=True, timeout=3
|
|
)
|
|
fstype = result.stdout.strip()
|
|
if fstype in ("vfat", "exfat", "ntfs", "ext4", "ext3", "hfsplus"):
|
|
usb_trouvees.append(str(chemin))
|
|
except Exception:
|
|
usb_trouvees.append(str(chemin))
|
|
|
|
return usb_trouvees
|
|
|
|
|
|
def compteur_restant() -> dict:
|
|
"""Retourne l'etat du compteur."""
|
|
config = charger_config()
|
|
compteur = config.get("compteur", {})
|
|
limite = compteur.get("limite", 400)
|
|
prises = compteur.get("photos_prises", 0)
|
|
return {
|
|
"actif": compteur.get("actif", False),
|
|
"limite": limite,
|
|
"photos_prises": prises,
|
|
"restantes": max(0, limite - prises),
|
|
}
|
|
|
|
|
|
def reset_compteur():
|
|
"""Remet le compteur a zero."""
|
|
mettre_a_jour_config({"compteur": {"photos_prises": 0}})
|