433 lines
15 KiB
Python
433 lines
15 KiB
Python
import ftplib
|
|
import json
|
|
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, RACINE
|
|
|
|
log = logging.getLogger("photobooth.destinations")
|
|
|
|
FICHIER_BOOTH_SPOOL = RACINE / "data" / "booth_spool.json"
|
|
_booth_spool_lock = threading.Lock()
|
|
|
|
|
|
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)
|
|
|
|
# Notifier booth que la photo a ete imprimee
|
|
if imprimee and booth.get("actif", False):
|
|
threading.Thread(
|
|
target=notifier_booth_impression,
|
|
args=(chemin_photo.name, copies, booth, config),
|
|
daemon=True,
|
|
).start()
|
|
|
|
# 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. En cas d'echec, met en spool."""
|
|
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}")
|
|
|
|
_ajouter_booth_spool(str(chemin_photo))
|
|
return False
|
|
|
|
|
|
def notifier_booth_impression(filename: str, copies: int, config_booth: dict, config: dict):
|
|
"""Notifie booth.copydev.fr qu'une photo a ete imprimee."""
|
|
url = config_booth.get("url", "").rstrip("/")
|
|
url_tunnel = config_booth.get("url_tunnel", "").rstrip("/")
|
|
api_key = config_booth.get("api_key", "")
|
|
event_id = config.get("evenement", {}).get("event_id") or config_booth.get("event_id", "default")
|
|
|
|
body = json.dumps({"copies": copies}).encode()
|
|
for tentative_url in [u for u in (url_tunnel, url) if u]:
|
|
try:
|
|
req = Request(
|
|
f"{tentative_url}/admin/gallery/{event_id}/photo/{filename}/imprimee",
|
|
data=body, method="POST",
|
|
)
|
|
req.add_header("Content-Type", "application/json")
|
|
req.add_header("X-Api-Key", api_key)
|
|
with urlopen(req, timeout=10) as resp:
|
|
if resp.status == 200:
|
|
log.info(f"Booth notifie impression: {filename} x{copies}")
|
|
return
|
|
except (URLError, OSError) as e:
|
|
log.warning(f"Echec notification impression booth: {e}")
|
|
|
|
|
|
# --- Spool galerie booth ---
|
|
|
|
def _ajouter_booth_spool(chemin: str):
|
|
with _booth_spool_lock:
|
|
spool = _charger_booth_spool()
|
|
if chemin not in spool:
|
|
spool.append(chemin)
|
|
_sauver_booth_spool(spool)
|
|
log.info(f"Photo mise en spool galerie ({len(spool)} en attente)")
|
|
|
|
|
|
def _charger_booth_spool() -> list:
|
|
if not FICHIER_BOOTH_SPOOL.exists():
|
|
return []
|
|
try:
|
|
with open(FICHIER_BOOTH_SPOOL, "r") as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, OSError):
|
|
return []
|
|
|
|
|
|
def _sauver_booth_spool(spool: list):
|
|
try:
|
|
with open(FICHIER_BOOTH_SPOOL, "w") as f:
|
|
json.dump(spool, f)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def taille_booth_spool() -> int:
|
|
with _booth_spool_lock:
|
|
return len(_charger_booth_spool())
|
|
|
|
|
|
def traiter_booth_spool() -> int:
|
|
"""Retente l'envoi des photos en spool. Retourne le nombre envoyees."""
|
|
with _booth_spool_lock:
|
|
spool = _charger_booth_spool()
|
|
if not spool:
|
|
return 0
|
|
|
|
config = charger_config()
|
|
booth = config.get("booth", {})
|
|
if not booth.get("actif", False):
|
|
return 0
|
|
|
|
envoyes = 0
|
|
restants = []
|
|
for chemin_str in spool:
|
|
chemin = Path(chemin_str)
|
|
if not chemin.exists():
|
|
log.warning(f"Spool booth: photo introuvable {chemin}, ignoree")
|
|
continue
|
|
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")
|
|
ok = 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):
|
|
log.info(f"Spool booth: {chemin.name} envoyee via {tentative_url}")
|
|
ok = True
|
|
break
|
|
except (URLError, OSError):
|
|
pass
|
|
if ok:
|
|
envoyes += 1
|
|
else:
|
|
restants.append(chemin_str)
|
|
break
|
|
|
|
if envoyes > 0 or len(restants) < len(spool):
|
|
idx = envoyes + len(restants)
|
|
restants.extend(spool[idx:])
|
|
with _booth_spool_lock:
|
|
_sauver_booth_spool(restants)
|
|
log.info(f"Spool booth: {envoyes} envoyee(s), {len(restants)} en attente")
|
|
|
|
return envoyes
|
|
|
|
|
|
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", 320) - conso.get("papier_utilise", 0))
|
|
ruban_rest = max(0, round(conso.get("ruban_capacite", 320) - 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", 320)
|
|
ruban_cap = conso.get("ruban_capacite", 320)
|
|
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": 2.0,
|
|
}
|
|
# Poids papier : combien d'unites 10x15 une feuille consomme
|
|
POIDS_PAPIER = {
|
|
"10x15": 1, "10x15-2up": 1,
|
|
"15x20": 2, "15x20-2up": 2,
|
|
"15x15": 2, "15x15-2up": 2,
|
|
}
|
|
|
|
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
|
|
poids = POIDS_PAPIER.get(fmt, 1)
|
|
conso["papier_utilise"] = conso.get("papier_utilise", 0) + feuilles * poids
|
|
conso["photos_imprimees"] = conso.get("photos_imprimees", 0) + copies
|
|
|
|
ratio_optimal = RATIO_RUBAN_OPTIMAL.get(fmt, 1.0)
|
|
ruban_par_feuille = float(poids)
|
|
if rembobinage:
|
|
ruban_consomme = feuilles * ratio_optimal
|
|
else:
|
|
ruban_consomme = feuilles * ruban_par_feuille
|
|
|
|
conso["ruban_utilise"] = round(conso.get("ruban_utilise", 0) + ruban_consomme, 1)
|
|
|
|
poses_perdues = feuilles * ruban_par_feuille - 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", 320)
|
|
ruban_cap = conso.get("ruban_capacite", 320)
|
|
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})
|