Upload 3 files

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.10-slim
+
+# System dependencies
+RUN apt-get update && apt-get install -y \
+    ffmpeg \
+    libgl1-mesa-glx \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Copy all files
+COPY . .
+
+# Install Python dependencies
+RUN pip install --upgrade pip
+RUN pip install -r requirements.txt
+
+# Expose port
+EXPOSE 7860
+
+# Run your Flask app
+CMD ["python", "app.py"]

app.py

@@ -0,0 +1,128 @@
+import os
+os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"
+from flask import Flask, render_template, request, jsonify, Response, send_file
+import cv2
+import uuid
+import numpy as np
+from face_swap import face_swap
+from moviepy.editor import VideoFileClip
+
+app = Flask(__name__)
+app.config['UPLOAD_FOLDER'] = 'uploads'
+app.config['OUTPUT_FOLDER'] = 'outputs'
+
+os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
+os.makedirs(app.config['OUTPUT_FOLDER'], exist_ok=True)
+
+source_face = None
+anonymize = False
+save_faces = False
+webcam_active = False
+cap = None
+
+
+@app.route('/')
+def landing():
+    return render_template('landing.html')
+
+
+@app.route('/app')
+def main_app():
+    return render_template('app.html')
+
+
+@app.route('/upload-face', methods=['POST'])
+def upload_face():
+    global source_face
+    file = request.files['file']
+    path = os.path.join(app.config['UPLOAD_FOLDER'], f"face_{uuid.uuid4().hex}.jpg")
+    file.save(path)
+    source_face = cv2.imread(path)
+    return jsonify({'status': 'success'})
+
+
+@app.route('/upload-video', methods=['POST'])
+def upload_video():
+    global source_face
+    file = request.files['file']
+    path = os.path.join(app.config['UPLOAD_FOLDER'], f"video_{uuid.uuid4().hex}.mp4")
+    file.save(path)
+
+    cap = cv2.VideoCapture(path)
+    width, height = int(cap.get(3)), int(cap.get(4))
+    fps = cap.get(cv2.CAP_PROP_FPS) or 25.0
+
+    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+    output_path = os.path.join(app.config['OUTPUT_FOLDER'], f"output_{uuid.uuid4().hex}.mp4")
+    out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+        result = face_swap(source_face, frame)
+        out.write(result)
+
+    cap.release()
+    out.release()
+
+    try:
+        original = VideoFileClip(path)
+        edited = VideoFileClip(output_path)
+        final = edited.set_audio(original.audio)
+        final_path = output_path.replace(".mp4", "_with_audio.mp4")
+        final.write_videofile(final_path, codec='libx264', audio_codec='aac')
+        return send_file(final_path, as_attachment=True)
+    except Exception as e:
+        print("Audio merge error:", e)
+        return send_file(output_path, as_attachment=True)
+
+
+@app.route('/toggle-anonymize', methods=['POST'])
+def toggle_anonymize():
+    global anonymize
+    anonymize = not anonymize
+    return jsonify({'status': 'success', 'anonymize': anonymize})
+
+
+@app.route('/toggle-save-faces', methods=['POST'])
+def toggle_save_faces():
+    global save_faces
+    save_faces = not save_faces
+    return jsonify({'status': 'success', 'save_faces': save_faces})
+
+
+@app.route('/start-webcam')
+def start_webcam():
+    global webcam_active, cap
+    if not webcam_active:
+        cap = cv2.VideoCapture(0)
+        webcam_active = True
+    return Response(gen_frames(), mimetype='multipart/x-mixed-replace; boundary=frame')
+
+
+@app.route('/stop-webcam')
+def stop_webcam():
+    global webcam_active, cap
+    webcam_active = False
+    if cap:
+        cap.release()
+    return jsonify({'status': 'stopped'})
+
+
+def gen_frames():
+    global cap, webcam_active, source_face
+    while webcam_active and cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+        frame = face_swap(source_face, frame)
+        _, buffer = cv2.imencode('.jpg', frame)
+        frame_bytes = buffer.tobytes()
+        yield (b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n')
+    if cap:
+        cap.release()
+
+
+if __name__ == '__main__':
+    app.run(debug=True)

face_swap.py

@@ -0,0 +1,72 @@
+import os
+import cv2
+import numpy as np
+import warnings
+from datetime import datetime
+from insightface.app import FaceAnalysis
+from insightface.model_zoo import get_model
+
+warnings.filterwarnings("ignore", category=FutureWarning)
+
+# Try GPU, fallback to CPU
+try:
+    face_analyzer = FaceAnalysis(name='buffalo_l', providers=['CUDAExecutionProvider'])
+    face_analyzer.prepare(ctx_id=0)
+except Exception:
+    print("⚠️ CUDA not available for FaceAnalysis. Falling back to CPU.")
+    face_analyzer = FaceAnalysis(name='buffalo_l', providers=['CPUExecutionProvider'])
+    face_analyzer.prepare(ctx_id=0)
+
+try:
+    swapper = get_model('model/insightface/inswapper_128.onnx', providers=['CUDAExecutionProvider'])
+except Exception:
+    print("⚠️ CUDA not available for Swapper. Falling back to CPU.")
+    swapper = get_model('inswapper_128.onnx', providers=['CPUExecutionProvider'])
+
+def face_swap(source_img, target_img, anonymize=False, save_faces=False):
+    if source_img is None:
+        return target_img
+
+    source_img_rgb = cv2.cvtColor(source_img, cv2.COLOR_BGR2RGB)
+    target_img_rgb = cv2.cvtColor(target_img, cv2.COLOR_BGR2RGB)
+
+    # Resize source to target if needed
+    if source_img_rgb.shape != target_img_rgb.shape:
+        source_img_rgb = cv2.resize(source_img_rgb, (target_img_rgb.shape[1], target_img_rgb.shape[0]))
+
+    src_faces = face_analyzer.get(source_img_rgb)
+    tgt_faces = face_analyzer.get(target_img_rgb)
+
+    if len(tgt_faces) == 0 or len(src_faces) == 0:
+        return cv2.cvtColor(target_img_rgb, cv2.COLOR_RGB2BGR)
+
+    src_face = src_faces[0]
+
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    output_dir = f"saved_faces/{timestamp}"
+    if save_faces:
+        os.makedirs(output_dir, exist_ok=True)
+
+    for i, tgt_face in enumerate(tgt_faces):
+        x1, y1, x2, y2 = tgt_face.bbox.astype(int)
+
+        if anonymize:
+            target_img_rgb[y1:y2, x1:x2] = cv2.GaussianBlur(target_img_rgb[y1:y2, x1:x2], (99, 99), 30)
+        else:
+            try:
+                target_img_rgb = swapper.get(target_img_rgb, tgt_face, src_face, paste_back=True)
+            except Exception as e:
+                print(f"Swapper error on face {i}: {e}")
+
+        if save_faces:
+            face_crop = target_img_rgb[y1:y2, x1:x2]
+            face_bgr = cv2.cvtColor(face_crop, cv2.COLOR_RGB2BGR)
+            cv2.imwrite(f"{output_dir}/swapped_face_{i}.jpg", face_bgr)
+
+            side_by_side = np.hstack([
+                cv2.resize(cv2.cvtColor(source_img_rgb, cv2.COLOR_RGB2BGR), (200, 200)),
+                cv2.resize(face_bgr, (200, 200))
+            ])
+            cv2.imwrite(f"{output_dir}/source_target_face_{i}.jpg", side_by_side)
+
+    return cv2.cvtColor(target_img_rgb, cv2.COLOR_RGB2BGR)