Jules
4e267492c2
- Preview thread : poll uniquement quand ecran capture actif, keepalive 30s sinon - Sequence boot Canon : power cycle relais si absent USB au demarrage - Veille auto 30min : Canon coupe par relais apres inactivite, reveil au preview_start - Stats diagnostic /api/canon/stats : keepalive, preview, captures, connexions - Escalade reconnexion adoucie : pas d'USB reset les 2 premieres tentatives - Hard reset relais 10s (resistance decharge 990ohm sur dummy battery 8V) - Gestion fin evenement : date_fin + bouton terminer + guard 403 impression - Compteur base sur min(papier, ruban) avec cap evenement optionnel - Pre-eclairage countdown : projecteurs ON 3s avant capture pour mesure expo - Canon _configurer_init() : ISO 800, drivemode Single, viewfinder, EXIF preserve - Admin : detail evenement, compteur consommables, galerie impression Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WyN9xFp84P9VaWT3Qe2DXu
306 lines
11 KiB
Python
306 lines
11 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, copies: int = 1, format_papier: str | None = None):
|
|
"""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 (feuilles 10x15 consommees)
|
|
if imprimee:
|
|
compteur = config.get("compteur", {})
|
|
if compteur.get("actif", False):
|
|
compteur["photos_prises"] = compteur.get("photos_prises", 0) + copies
|
|
mettre_a_jour_config({"compteur": compteur})
|
|
maj_consommables(config, copies, format_papier)
|
|
|
|
|
|
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, basé sur les consommables restants."""
|
|
config = charger_config()
|
|
compteur = config.get("compteur", {})
|
|
conso = config.get("consommables", {})
|
|
prises = compteur.get("photos_prises", 0)
|
|
papier_rest = max(0, conso.get("papier_capacite", 400) - conso.get("papier_utilise", 0))
|
|
ruban_rest = max(0, round(conso.get("ruban_capacite", 400) - conso.get("ruban_utilise", 0), 1))
|
|
restantes = int(min(papier_rest, ruban_rest))
|
|
limite_event = compteur.get("limite", 0)
|
|
if compteur.get("actif") and limite_event > 0:
|
|
restantes = min(restantes, max(0, limite_event - prises))
|
|
papier_cap = conso.get("papier_capacite", 400)
|
|
ruban_cap = conso.get("ruban_capacite", 400)
|
|
capacite_conso = int(min(papier_cap, ruban_cap))
|
|
return {
|
|
"actif": compteur.get("actif", False),
|
|
"limite": limite_event,
|
|
"photos_prises": prises,
|
|
"restantes": restantes,
|
|
"capacite": capacite_conso,
|
|
"papier_restant": papier_rest,
|
|
"ruban_restant": int(ruban_rest),
|
|
}
|
|
|
|
|
|
def reset_compteur():
|
|
"""Remet le compteur a zero."""
|
|
mettre_a_jour_config({"compteur": {"photos_prises": 0}})
|
|
|
|
|
|
# --- Suivi consommables ---
|
|
|
|
RATIO_RUBAN_OPTIMAL = {
|
|
"10x15": 0.5,
|
|
"10x15-2up": 1.0,
|
|
"15x20": 1.0,
|
|
"15x20-2up": 1.0,
|
|
}
|
|
|
|
def maj_consommables(config: dict, copies: int, format_papier: str | None):
|
|
"""Met a jour le suivi consommables (papier, ruban, photos)."""
|
|
conso = config.get("consommables", {})
|
|
fmt = format_papier or config.get("impression", {}).get("format", "15x20")
|
|
rembobinage = config.get("impression", {}).get("rembobinage_ruban", False)
|
|
is_2up = "-2up" in fmt
|
|
|
|
feuilles = copies if not is_2up else (copies + 1) // 2
|
|
conso["papier_utilise"] = conso.get("papier_utilise", 0) + feuilles
|
|
conso["photos_imprimees"] = conso.get("photos_imprimees", 0) + copies
|
|
|
|
ratio_optimal = RATIO_RUBAN_OPTIMAL.get(fmt, 1.0)
|
|
if rembobinage:
|
|
ruban_consomme = feuilles * ratio_optimal
|
|
else:
|
|
ruban_consomme = feuilles * 1.0
|
|
|
|
conso["ruban_utilise"] = round(conso.get("ruban_utilise", 0) + ruban_consomme, 1)
|
|
|
|
# Poses perdues = ruban avance inutilement (difference entre consomme et optimal)
|
|
poses_perdues = feuilles * 1.0 - feuilles * ratio_optimal if not rembobinage else 0
|
|
conso["poses_perdues"] = round(conso.get("poses_perdues", 0) + poses_perdues, 1)
|
|
|
|
mettre_a_jour_config({"consommables": conso})
|
|
|
|
|
|
def consommables_etat() -> dict:
|
|
"""Retourne l'etat des consommables."""
|
|
config = charger_config()
|
|
conso = config.get("consommables", {})
|
|
papier_cap = conso.get("papier_capacite", 400)
|
|
ruban_cap = conso.get("ruban_capacite", 400)
|
|
papier_used = conso.get("papier_utilise", 0)
|
|
ruban_used = conso.get("ruban_utilise", 0)
|
|
photos = conso.get("photos_imprimees", 0)
|
|
perdues = conso.get("poses_perdues", 0)
|
|
return {
|
|
"papier_capacite": papier_cap,
|
|
"papier_utilise": papier_used,
|
|
"papier_restant": max(0, papier_cap - papier_used),
|
|
"ruban_capacite": ruban_cap,
|
|
"ruban_utilise": ruban_used,
|
|
"ruban_restant": max(0, round(ruban_cap - ruban_used, 1)),
|
|
"photos_imprimees": photos,
|
|
"poses_perdues": perdues,
|
|
}
|
|
|
|
|
|
def reset_consommables(quoi: str):
|
|
"""Reinitialise papier, ruban ou tout."""
|
|
config = charger_config()
|
|
conso = config.get("consommables", {})
|
|
if quoi in ("papier", "tout"):
|
|
conso["papier_utilise"] = 0
|
|
if quoi in ("ruban", "tout"):
|
|
conso["ruban_utilise"] = 0
|
|
conso["poses_perdues"] = 0
|
|
if quoi == "tout":
|
|
conso["photos_imprimees"] = 0
|
|
conso["poses_perdues"] = 0
|
|
mettre_a_jour_config({"consommables": conso})
|