Config admin complète + IMAP polling

Config admin :
- Serveur upload : IP d'écoute, port, URL publique (QR code)
- Email IMAP : activer/désactiver, serveur, port, user, password,
  dossier, intervalle de polling
- Checkbox pour activer IMAP

Service IMAP :
- Thread daemon, polling configurable (défaut 30s)
- Connexion IMAP SSL, récupère mails UNSEEN
- Extrait les PJ photos (jpg/png/etc.)
- Sauvegarde dans data/photos/, notifie l'app via signal Qt
- Bouton Email passe au vert + Continuer quand photos reçues

main.py :
- Lit la config pour IP/port du serveur upload
- Démarre IMAP polling si activé dans la config

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-21 22:34:37 +01:00
parent 532f8d91c0
commit d8d7cbde7c
5 changed files with 273 additions and 61 deletions

View File

@@ -0,0 +1,119 @@
"""Service IMAP polling — consulte une boîte mail et récupère les pièces jointes photos."""
import imaplib
import email
import os
import time
import threading
from email.header import decode_header
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"}
class ImapSignals(QObject):
"""Signaux Qt pour communiquer du thread IMAP vers le thread principal."""
photos_received = pyqtSignal(list)
error = pyqtSignal(str)
class ImapWatcher:
"""Consulte une boîte IMAP en boucle et récupère les photos en PJ."""
def __init__(self):
self.signals = ImapSignals()
self._thread = None
self._running = False
self._seen_uids = set()
os.makedirs(UPLOAD_DIR, exist_ok=True)
def start(self, server, port, user, password, folder="INBOX", interval=30):
"""Démarre le polling IMAP dans un thread daemon."""
if not server or not user:
return
self._running = True
self._config = {
"server": server,
"port": port,
"user": user,
"password": password,
"folder": folder,
"interval": interval,
}
self._thread = threading.Thread(target=self._poll_loop, daemon=True)
self._thread.start()
def stop(self):
self._running = False
def _poll_loop(self):
"""Boucle de polling."""
while self._running:
try:
self._check_mail()
except Exception as e:
self.signals.error.emit(str(e))
time.sleep(self._config["interval"])
def _check_mail(self):
"""Connexion IMAP, récupère les nouveaux mails avec PJ photos."""
cfg = self._config
mail = imaplib.IMAP4_SSL(cfg["server"], cfg["port"])
mail.login(cfg["user"], cfg["password"])
mail.select(cfg["folder"])
# Chercher les mails non lus
status, data = mail.search(None, "UNSEEN")
if status != "OK" or not data[0]:
mail.logout()
return
saved_photos = []
for uid in data[0].split():
if uid in self._seen_uids:
continue
self._seen_uids.add(uid)
status, msg_data = mail.fetch(uid, "(RFC822)")
if status != "OK":
continue
msg = email.message_from_bytes(msg_data[0][1])
for part in msg.walk():
content_type = part.get_content_type()
filename = part.get_filename()
if filename:
# Décoder le nom de fichier
decoded_parts = decode_header(filename)
filename = ""
for content, encoding in decoded_parts:
if isinstance(content, bytes):
filename += content.decode(encoding or "utf-8", errors="replace")
else:
filename += content
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
if ext not in ALLOWED_EXTENSIONS:
continue
# Sauvegarder la pièce jointe
payload = part.get_payload(decode=True)
if not payload:
continue
safe_name = f"{int(time.time() * 1000)}_{filename}"
path = os.path.join(UPLOAD_DIR, safe_name)
with open(path, "wb") as f:
f.write(payload)
saved_photos.append(path)
mail.logout()
if saved_photos:
self.signals.photos_received.emit(saved_photos)