Update app.py

app.py

@@ -1,277 +1,104 @@
-from flask import Flask, render_template, request, jsonify, send_from_directory
-from flask_cors import CORS
-from flask_limiter import Limiter
-from flask_limiter.util import get_remote_address
+from flask import Flask, render_template, request, jsonify
 from deepface import DeepFace
-from werkzeug.utils import secure_filename
 import os
 import tempfile
 import shutil
-import uuid
-import logging
-import time
-from datetime import datetime
-from functools import wraps
-import numpy as np
-import cv2
-from PIL import Image
-import io
-import threading
-import queue
-import hashlib
 
-# Configuration du logging
-logging.basicConfig(
-    filename='app.log',
-    level=logging.INFO,
-    format='%(asctime)s - %(levelname)s - %(message)s'
-)
-
-def timing_decorator(func):
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        start = time.time()
-        result = func(*args, **kwargs)
-        end = time.time()
-        logging.info(f'{func.__name__} took {end-start:.2f} seconds to execute')
-        return result
-    return wrapper
-
-class FaceAnalysisApp:
-    def __init__(self):
-        self.app = Flask(__name__, static_folder='static')
-        self.setup_app()
-        self.results_cache = {}
-        self.task_queue = queue.Queue()
-        self.setup_routes()
-        self.start_worker_thread()
-
-    def setup_app(self):
-        """Configure l'application Flask"""
-        # Configuration de base
-        self.app.config.update(
-            UPLOAD_FOLDER='static/uploads',
-            MAX_CONTENT_LENGTH=16 * 1024 * 1024,
-            ALLOWED_EXTENSIONS={'png', 'jpg', 'jpeg', 'gif'},
-            SECRET_KEY=os.urandom(24)
-        )
+app = Flask(__name__)
+
+# Configuration pour l'upload des images (facultatif, pour stocker les images temporairement)
+UPLOAD_FOLDER = 'uploads'
+app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
+ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg'}
+
+def allowed_file(filename):
+    return '.' in filename and \
+           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
+
+def process_image(image_path):
+    """
+    Traite une image avec DeepFace (détection, alignement, etc.).
+    Vous pouvez personnaliser cette fonction en fonction de vos besoins.
+    """
+    try:
+        # Exemple d'extraction des visages avec alignement et détection
+        faces = DeepFace.extract_faces(img_path=image_path, detector_backend='retinaface', align=True)
+        if len(faces) > 0 :
+            return faces[0]['facial_area']
+        else:
+             return None
+    
+    except Exception as e:
+        print(f"Erreur lors du traitement de l'image : {e}")
+        return None
+
+@app.route('/')
+def index():
+    return render_template('index.html')
+
+@app.route('/compare', methods=['POST'])
+def compare():
+    
+    # Gestion des fichiers uploadés
+    if 'file1' not in request.files or 'file2' not in request.files:
+        return jsonify({'error': 'Aucun fichier sélectionné'}), 400
+
+    file1 = request.files['file1']
+    file2 = request.files['file2']
+
+    if file1.filename == '' or file2.filename == '':
+        return jsonify({'error': 'Aucun fichier sélectionné'}), 400
+
+    if file1 and allowed_file(file1.filename) and file2 and allowed_file(file2.filename):
+        # Créer un dossier temporaire pour stocker les images
+        temp_dir = tempfile.mkdtemp(prefix="face_compare_", dir=app.config['UPLOAD_FOLDER'])
         
-        # Initialisation CORS et Limiter
-        CORS(self.app)
-        self.limiter = Limiter(
-            self.app,
-            key_func=get_remote_address,
-            default_limits=["200 per day", "50 per hour"]
-        )
-
-    def start_worker_thread(self):
-        """Démarre le thread de traitement en arrière-plan"""
-        def worker():
-            while True:
-                task = self.task_queue.get()
-                if task is None:
-                    break
-                try:
-                    task()
-                except Exception as e:
-                    logging.error(f"Error in worker thread: {str(e)}")
-                finally:
-                    self.task_queue.task_done()
-
-        self.worker_thread = threading.Thread(target=worker, daemon=True)
-        self.worker_thread.start()
-
-    def validate_image(self, image_stream):
-        """Valide et optimise l'image"""
-        try:
-            img = Image.open(image_stream)
-            
-            # Vérification des dimensions
-            if img.size[0] > 2000 or img.size[1] > 2000:
-                img.thumbnail((2000, 2000), Image.LANCZOS)
-                
-            # Conversion en RGB si nécessaire
-            if img.mode not in ('RGB', 'L'):
-                img = img.convert('RGB')
-                
-            # Optimisation
-            output = io.BytesIO()
-            img.save(output, format='JPEG', quality=85, optimize=True)
-            output.seek(0)
-            
-            return output
-        except Exception as e:
-            logging.error(f"Image validation error: {str(e)}")
-            raise ValueError("Invalid image format")
-
-    def process_face_detection(self, image_path):
-        """Détecte les visages avec mise en cache"""
-        image_hash = hashlib.md5(open(image_path, 'rb').read()).hexdigest()
+        # Enregistre les fichiers dans le dossier temporaire
+        file1_path = os.path.join(temp_dir, file1.filename)
+        file2_path = os.path.join(temp_dir, file2.filename)
+        file1.save(file1_path)
+        file2.save(file2_path)
+
+        # Traitement des images (vous pouvez personnaliser cette partie)
+        processed_img1 = process_image(file1_path)
+        processed_img2 = process_image(file2_path)
         
-        if image_hash in self.results_cache:
-            return self.results_cache[image_hash]
-            
+        if processed_img1 is None or processed_img2 is None:
+          # Supprimer le dossier temporaire et son contenu
+          shutil.rmtree(temp_dir)
+          return jsonify({'error': 'Aucun visage détecté dans une ou plusieurs images'}), 400
+          
         try:
-            result = DeepFace.analyze(
-                img_path=image_path,
-                actions=['age', 'gender', 'race', 'emotion'],
-                enforce_detection=True
-            )
-            self.results_cache[image_hash] = result
-            return result
-        except Exception as e:
-            logging.error(f"Face detection error: {str(e)}")
-            raise
-
-    @timing_decorator
-    def verify_faces(self, image1_path, image2_path):
-        """Compare deux visages"""
-        try:
-            # Vérification des images
-            face1 = cv2.imread(image1_path)
-            face2 = cv2.imread(image2_path)
-            if face1 is None or face2 is None:
-                raise ValueError("Unable to read one or both images")
-
-            # Comparaison des visages
+            # Comparaison des visages avec DeepFace
             result = DeepFace.verify(
-                img1_path=image1_path,
-                img2_path=image2_path,
-                enforce_detection=True,
-                model_name="VGG-Face"
+                img1_path=file1_path,
+                img2_path=file2_path,
+                model_name="VGG-Face",
+                detector_backend="retinaface",
+                distance_metric="cosine",
+                enforce_detection=False
             )
-            
-            # Enrichissement des résultats
-            result.update({
-                'timestamp': datetime.now().isoformat(),
-                'confidence_score': 1 - result.get('distance', 0),
-                'processing_time': time.time()
-            })
-            
-            return result
-        except Exception as e:
-            logging.error(f"Face verification error: {str(e)}")
-            raise
 
-    def setup_routes(self):
-        """Configure les routes de l'application"""
-        
-        @self.app.route('/')
-        def index():
-            return render_template('index.html')
-
-        @self.app.route('/verify', methods=['POST'])
-        @self.limiter.limit("10 per minute")
-        def verify_faces_endpoint():
-            try:
-                # Vérification des fichiers
-                if 'image1' not in request.files or 'image2' not in request.files:
-                    return jsonify({'error': 'Two images are required'}), 400
-
-                image1 = request.files['image1']
-                image2 = request.files['image2']
-
-                # Validation des images
-                try:
-                    image1_stream = self.validate_image(image1)
-                    image2_stream = self.validate_image(image2)
-                except ValueError as e:
-                    return jsonify({'error': str(e)}), 400
-
-                # Traitement des images
-                with tempfile.TemporaryDirectory() as temp_dir:
-                    # Sauvegarde temporaire
-                    paths = []
-                    for img, stream in [(image1, image1_stream), (image2, image2_stream)]:
-                        path = os.path.join(temp_dir, secure_filename(img.filename))
-                        with open(path, 'wb') as f:
-                            f.write(stream.getvalue())
-                        paths.append(path)
-
-                    # Vérification des visages
-                    result = self.verify_faces(paths[0], paths[1])
-
-                    # Sauvegarde des résultats positifs
-                    if result['verified']:
-                        permanent_dir = os.path.join(self.app.static_folder, 'verified_faces')
-                        os.makedirs(permanent_dir, exist_ok=True)
-                        
-                        saved_paths = []
-                        for i, path in enumerate(paths, 1):
-                            name = f"face{i}_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}.jpg"
-                            dest = os.path.join(permanent_dir, name)
-                            shutil.copy2(path, dest)
-                            saved_paths.append(f'/static/verified_faces/{name}')
-                        
-                        result['image1_url'] = saved_paths[0]
-                        result['image2_url'] = saved_paths[1]
-
-                return jsonify(result)
+            # Formater la réponse
+            response_data = {
+                'verified': result['verified'],
+                'similarity': round((1 - result['distance']) * 100, 1) if result['verified'] else round((1 - result['distance']) * 100, 1)
+            }
 
-            except Exception as e:
-                logging.error(f"Verification endpoint error: {str(e)}")
-                return jsonify({'error': 'An internal error occurred'}), 500
+            # Supprimer le dossier temporaire et son contenu
+            shutil.rmtree(temp_dir)
+            return jsonify(response_data)
 
-        @self.app.route('/analyze', methods=['POST'])
-        @self.limiter.limit("20 per minute")
-        def analyze_face_endpoint():
-            try:
-                if 'image' not in request.files:
-                    return jsonify({'error': 'No image provided'}), 400
-
-                image = request.files['image']
-                
-                # Validation de l'image
-                try:
-                    image_stream = self.validate_image(image)
-                except ValueError as e:
-                    return jsonify({'error': str(e)}), 400
-
-                # File d'attente pour les résultats
-                result_queue = queue.Queue()
-                
-                def process_task():
-                    try:
-                        with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as temp_file:
-                            temp_file.write(image_stream.getvalue())
-                            result = self.process_face_detection(temp_file.name)
-                            result_queue.put(('success', result))
-                    except Exception as e:
-                        result_queue.put(('error', str(e)))
-                    finally:
-                        try:
-                            os.unlink(temp_file.name)
-                        except:
-                            pass
-
-                # Ajout de la tâche à la file d'attente
-                self.task_queue.put(process_task)
-                
-                # Attente du résultat
-                try:
-                    status, result = result_queue.get(timeout=30)
-                    if status == 'error':
-                        return jsonify({'error': result}), 500
-                    return jsonify(result)
-                except queue.Empty:
-                    return jsonify({'error': 'Processing timeout'}), 408
-
-            except Exception as e:
-                logging.error(f"Analysis endpoint error: {str(e)}")
-                return jsonify({'error': 'An internal error occurred'}), 500
-
-        @self.app.errorhandler(413)
-        def request_entity_too_large(error):
-            return jsonify({'error': 'File too large'}), 413
-
-        @self.app.errorhandler(429)
-        def ratelimit_handler(e):
-            return jsonify({'error': 'Rate limit exceeded'}), 429
+        except Exception as e:
+            # Supprimer le dossier temporaire et son contenu
+            shutil.rmtree(temp_dir)
+            print(f"Erreur lors de la comparaison des visages : {e}")
+            return jsonify({'error': str(e)}), 500
 
-    def run(self, host='0.0.0.0', port=5000, debug=False):
-        """Démarre l'application Flask"""
-        self.app.run(host=host, port=port, debug=debug)
+    else:
+        return jsonify({'error': 'Type de fichier non autorisé'}), 400
 
 if __name__ == '__main__':
-    app = FaceAnalysisApp()
-    app.run(debug=True)
+    # Créer le dossier d'upload si nécessaire
+    os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
+    app.run(debug=True) # Mettre debug=False en production