Create app.py

app.py

@@ -0,0 +1,239 @@
+# app.py
+import cv2
+import insightface
+import numpy as np
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+import base64
+import os
+import tempfile
+import uuid
+from werkzeug.utils import secure_filename
+from insightface.app import FaceAnalysis
+
+app = Flask(__name__)
+CORS(app)
+
+# Configuration
+app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB max file size
+app.config['UPLOAD_FOLDER'] = tempfile.gettempdir()
+ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'webp'}
+
+# Initialize face analysis globally
+face_app = FaceAnalysis(providers=['CPUExecutionProvider'])
+face_app.prepare(ctx_id=0, det_size=(640, 640))
+
+# Initialize face swapper
+swapper = insightface.model_zoo.get_model('inswapper_128.onnx')
+
+def allowed_file(filename):
+    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
+
+def decode_base64_image(base64_string):
+    """Decode base64 string to numpy image"""
+    if ',' in base64_string:
+        base64_string = base64_string.split(',')[1]
+    img_data = base64.b64decode(base64_string)
+    nparr = np.frombuffer(img_data, np.uint8)
+    return cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+
+def encode_image_to_base64(image):
+    """Encode numpy image to base64 string"""
+    _, buffer = cv2.imencode('.jpg', image)
+    return base64.b64encode(buffer).decode('utf-8')
+
+@app.route('/health', methods=['GET'])
+def health_check():
+    return jsonify({'status': 'healthy', 'message': 'Face Swap API is running'})
+
+@app.route('/swap', methods=['POST'])
+def swap_faces():
+    """
+    Swap faces between source and target images
+    Expected JSON payload:
+    {
+        "source_image": "base64_string or file_path",
+        "target_image": "base64_string or file_path",
+        "source_is_base64": true,  # optional, defaults to false
+        "target_is_base64": true   # optional, defaults to false
+    }
+    Or use multipart/form-data with files:
+    - source_file (image file)
+    - target_file (image file)
+    """
+    
+    try:
+        source_img = None
+        target_img = None
+        
+        # Handle JSON request
+        if request.is_json:
+            data = request.get_json()
+            
+            # Get source image
+            source_is_base64 = data.get('source_is_base64', False)
+            if source_is_base64:
+                source_img = decode_base64_image(data['source_image'])
+            else:
+                source_img = cv2.imread(data['source_image'])
+            
+            # Get target image
+            target_is_base64 = data.get('target_is_base64', False)
+            if target_is_base64:
+                target_img = decode_base64_image(data['target_image'])
+            else:
+                target_img = cv2.imread(data['target_image'])
+        
+        # Handle multipart form request
+        elif 'source_file' in request.files and 'target_file' in request.files:
+            source_file = request.files['source_file']
+            target_file = request.files['target_file']
+            
+            if source_file and allowed_file(source_file.filename):
+                source_filename = secure_filename(source_file.filename)
+                source_path = os.path.join(app.config['UPLOAD_FOLDER'], f"source_{uuid.uuid4()}_{source_filename}")
+                source_file.save(source_path)
+                source_img = cv2.imread(source_path)
+                os.remove(source_path)  # Clean up
+            else:
+                return jsonify({'error': 'Invalid source file type'}), 400
+            
+            if target_file and allowed_file(target_file.filename):
+                target_filename = secure_filename(target_file.filename)
+                target_path = os.path.join(app.config['UPLOAD_FOLDER'], f"target_{uuid.uuid4()}_{target_filename}")
+                target_file.save(target_path)
+                target_img = cv2.imread(target_path)
+                os.remove(target_path)  # Clean up
+            else:
+                return jsonify({'error': 'Invalid target file type'}), 400
+        
+        else:
+            return jsonify({'error': 'Invalid request. Provide source_image/target_image or source_file/target_file'}), 400
+        
+        # Validate images
+        if source_img is None:
+            return jsonify({'error': 'Could not read source image'}), 400
+        if target_img is None:
+            return jsonify({'error': 'Could not read target image'}), 400
+        
+        # Detect faces
+        source_faces = face_app.get(source_img)
+        target_faces = face_app.get(target_img)
+        
+        if len(source_faces) == 0:
+            return jsonify({'error': 'No face found in source image'}), 400
+        if len(target_faces) == 0:
+            return jsonify({'error': 'No face found in target image'}), 400
+        
+        # Perform face swap
+        swapped_image = swapper.get(target_img, target_faces[0], source_faces[0], paste_back=True)
+        
+        # Prepare response
+        return_type = request.args.get('return_type', 'base64')
+        
+        if return_type == 'file':
+            # Save and return file path
+            output_filename = f"swapped_{uuid.uuid4()}.jpg"
+            output_path = os.path.join(app.config['UPLOAD_FOLDER'], output_filename)
+            cv2.imwrite(output_path, swapped_image)
+            
+            return jsonify({
+                'success': True,
+                'message': 'Face swap completed successfully',
+                'output_path': output_path,
+                'filename': output_filename
+            })
+        else:
+            # Return base64 encoded image
+            encoded_image = encode_image_to_base64(swapped_image)
+            return jsonify({
+                'success': True,
+                'message': 'Face swap completed successfully',
+                'swapped_image': encoded_image
+            })
+    
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+@app.route('/swap/batch', methods=['POST'])
+def batch_swap_faces():
+    """
+    Batch face swap with multiple targets
+    Expected JSON payload:
+    {
+        "source_image": "base64_string or path",
+        "target_images": ["base64_string1", "base64_string2", ...],
+        "source_is_base64": true,
+        "targets_are_base64": true
+    }
+    """
+    try:
+        data = request.get_json()
+        
+        # Get source image
+        source_is_base64 = data.get('source_is_base64', False)
+        if source_is_base64:
+            source_img = decode_base64_image(data['source_image'])
+        else:
+            source_img = cv2.imread(data['source_image'])
+        
+        if source_img is None:
+            return jsonify({'error': 'Could not read source image'}), 400
+        
+        # Detect source face once
+        source_faces = face_app.get(source_img)
+        if len(source_faces) == 0:
+            return jsonify({'error': 'No face found in source image'}), 400
+        
+        source_face = source_faces[0]
+        
+        # Process all target images
+        results = []
+        target_images = data.get('target_images', [])
+        targets_are_base64 = data.get('targets_are_base64', False)
+        
+        for idx, target_img_data in enumerate(target_images):
+            try:
+                if targets_are_base64:
+                    target_img = decode_base64_image(target_img_data)
+                else:
+                    target_img = cv2.imread(target_img_data)
+                
+                if target_img is None:
+                    results.append({'index': idx, 'error': 'Could not read target image'})
+                    continue
+                
+                target_faces = face_app.get(target_img)
+                if len(target_faces) == 0:
+                    results.append({'index': idx, 'error': 'No face found in target image'})
+                    continue
+                
+                swapped_image = swapper.get(target_img, target_faces[0], source_face, paste_back=True)
+                encoded_image = encode_image_to_base64(swapped_image)
+                
+                results.append({
+                    'index': idx,
+                    'success': True,
+                    'swapped_image': encoded_image
+                })
+            except Exception as e:
+                results.append({'index': idx, 'error': str(e)})
+        
+        return jsonify({
+            'success': True,
+            'message': f'Processed {len(results)} images',
+            'results': results
+        })
+    
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+if __name__ == '__main__':
+    print("Starting Face Swap API Server...")
+    print("Make sure 'inswapper_128.onnx' is in the current directory")
+    print("API endpoints:")
+    print("  GET  /health - Health check")
+    print("  POST /swap - Single face swap")
+    print("  POST /swap/batch - Batch face swap")
+    print("\nStarting server on http://localhost:5000")
+    app.run(host='0.0.0.0', port=7860, debug=True)