"""Serveur Flask d'upload photos — tourne en thread à côté de PyQt6."""
import os
import threading
import time
from flask import Flask, request, jsonify, render_template_string
from PyQt6.QtCore import QObject, pyqtSignal
UPLOAD_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "data", "photos")
ALLOWED_EXTENSIONS = {"jpg", "jpeg", "png", "bmp", "tiff", "tif", "webp", "heic"}
UPLOAD_PAGE = """
Photostation — Envoi de photos
Photostation
Sélectionnez vos photos à imprimer
📷
Appuyez ici pour choisir vos photos ou glissez-les
"""
class UploadSignals(QObject):
"""Signaux Qt pour communiquer du thread Flask vers le thread principal."""
photos_received = pyqtSignal(list) # liste de chemins
class UploadServer:
"""Serveur Flask d'upload, tourne dans un thread séparé."""
def __init__(self, host="0.0.0.0", port=80):
self.host = host
self.port = port
self.signals = UploadSignals()
self._callback = None
self.app = Flask(__name__)
self._setup_routes()
self._thread = None
os.makedirs(UPLOAD_DIR, exist_ok=True)
def _setup_routes(self):
@self.app.route("/")
@self.app.route("/upload")
def index():
return render_template_string(UPLOAD_PAGE)
@self.app.route("/upload", methods=["POST"])
def upload():
files = request.files.getlist("photos")
saved = []
for f in files:
if not f.filename:
continue
ext = f.filename.rsplit(".", 1)[-1].lower() if "." in f.filename else ""
if ext not in ALLOWED_EXTENSIONS:
continue
# Nom unique basé sur le timestamp
name = f"{int(time.time() * 1000)}_{f.filename}"
path = os.path.join(UPLOAD_DIR, name)
f.save(path)
saved.append(path)
if saved:
# QTimer.singleShot est thread-safe et exécute dans le thread principal
from PyQt6.QtCore import QTimer
from functools import partial
if self._callback:
QTimer.singleShot(0, partial(self._callback, saved))
return jsonify({"ok": True, "count": len(saved)})
def set_callback(self, callback):
"""Définit le callback appelé quand des photos sont uploadées."""
self._callback = callback
def start(self):
"""Démarre le serveur dans un thread daemon."""
self._thread = threading.Thread(
target=lambda: self.app.run(
host=self.host, port=self.port,
debug=False, use_reloader=False
),
daemon=True
)
self._thread.start()
def stop(self):
"""Le thread daemon s'arrête avec le processus principal."""
pass