Upload 10 files

app.py

@@ -0,0 +1,86 @@
+# app.py
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+from config.database import db
+import numpy as np
+import tempfile
+import os
+from voice_embedder import get_embedding, cosine_match
+
+app = Flask(__name__)
+CORS(app)
+
+users_collection = db["users"]
+
+# ---------------- Upload PIN audio ----------------
+@app.route("/upload_pin", methods=["POST"])
+def upload_pin():
+    if "pin_audio" not in request.files:
+        return jsonify({"error": "No audio file found"}), 400
+
+    audio_file = request.files["pin_audio"]
+    temp_path = tempfile.mktemp(suffix=".wav")
+    audio_file.save(temp_path)
+
+    try:
+        emb = get_embedding(temp_path).tolist()
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+    finally:
+        os.remove(temp_path)
+
+    return jsonify({"message": "PIN processed successfully", "embedding": emb})
+
+# ---------------- Signup ----------------
+@app.route("/signup", methods=["POST"])
+def signup():
+    data = request.json
+    required = ("name", "email", "password", "embedding")
+    if not all(k in data for k in required):
+        return jsonify({"error": "Missing fields"}), 400
+
+    if users_collection.find_one({"email": data["email"]}):
+        return jsonify({"error": "Email already exists"}), 400
+
+    new_user = {
+        "name": data["name"],
+        "email": data["email"],
+        "password": data["password"],
+        "embedding": data["embedding"]
+    }
+    users_collection.insert_one(new_user)
+    return jsonify({"message": "Signup successful!"})
+
+# ---------------- Login ----------------
+@app.route("/login", methods=["POST"])
+def login():
+    if "pin_audio" not in request.files:
+        return jsonify({"error": "Audio file missing"}), 400
+
+    email = request.form.get("email")
+    user = users_collection.find_one({"email": email})
+    if not user:
+        return jsonify({"error": "User not found"}), 404
+
+    audio_file = request.files["pin_audio"]
+    temp_path = tempfile.mktemp(suffix=".wav")
+    audio_file.save(temp_path)
+
+    try:
+        new_emb = get_embedding(temp_path)
+        saved_emb = np.array(user["embedding"], dtype=np.float32)
+        score, status = cosine_match(saved_emb, new_emb)
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+    finally:
+        os.remove(temp_path)
+
+    return jsonify({"score": score, "status": status})
+
+# ---------------- Home ----------------
+@app.route("/")
+def home():
+    return jsonify({"message": "VoiceAuth Backend Running"})
+
+if __name__ == "__main__":
+    app.run(port=5000, debug=True)

config/database.py

@@ -0,0 +1,23 @@
+from pymongo import MongoClient
+import certifi
+
+MONGO_URI = "mongodb+srv://kataricoder_db_user:katari@cluster0.bfn51lm.mongodb.net/?retryWrites=true&w=majority"
+
+try:
+    client = MongoClient(
+        MONGO_URI,
+        tls=True,
+        tlsCAFile=certifi.where()   # Fix SSL handshake
+    )
+
+    db = client["mydatabase"]
+
+    client.admin.command("ping")
+
+    # FIXED: Correct PyMongo update query
+    db.users.update_many({}, {"$unset": {"voiceEmbedding": ""}})
+
+    print("🚀 MongoDB Connected Successfully!")
+
+except Exception as e:
+    print("❌ MongoDB Connection Failed:", e)

models/user_model.py

@@ -0,0 +1,10 @@
+# This file only defines your User schema structure
+
+def user_schema(name, email, password, embedding):
+    return {
+        "name": name,
+        "email": email,
+        "password": password,
+        "embedding": embedding  # vector (list of floats)
+    }
+

routes/user_routes.py

@@ -0,0 +1,28 @@
+from flask import Blueprint, request, jsonify
+from config.database import db
+
+users_collection = db["users"]
+
+user_routes = Blueprint("user_routes", __name__)
+
+@user_routes.route("/signup", methods=["POST"])
+def signup():
+    data = request.json
+
+    required = ("name", "email", "password", "embedding")
+    if not all(k in data for k in required):
+        return jsonify({"error": "Missing fields"}), 400
+
+    if users_collection.find_one({"email": data["email"]}):
+        return jsonify({"error": "Email already exists"}), 400
+
+    new_user = {
+        "name": data["name"],
+        "email": data["email"],
+        "password": data["password"],
+        "embedding": data["embedding"],
+    }
+
+    users_collection.insert_one(new_user)
+
+    return jsonify({"message": "Signup successful"})

routes/voice_routes.py

@@ -0,0 +1,64 @@
+from flask import Blueprint, request, jsonify
+import numpy as np
+import tempfile
+import os
+
+from voice_embedder import get_embedding, cosine_match
+from config.database import db
+
+users_collection = db["users"]
+
+voice_routes = Blueprint("voice_routes", __name__)
+
+# --------------------------------------------------------
+# UPLOAD PIN AUDIO FOR SIGNUP
+# --------------------------------------------------------
+@voice_routes.route("/upload_pin", methods=["POST"])
+def upload_pin():
+    if "pin_audio" not in request.files:
+        return jsonify({"error": "No audio file found"}), 400
+
+    audio_file = request.files["pin_audio"]
+
+    temp_path = tempfile.mktemp(suffix=".wav")
+    audio_file.save(temp_path)
+
+    emb = get_embedding(temp_path).tolist()
+
+    os.remove(temp_path)
+
+    return jsonify({
+        "message": "PIN processed successfully",
+        "embedding": emb
+    })
+
+
+# --------------------------------------------------------
+# LOGIN WITH VOICE MATCH
+# --------------------------------------------------------
+@voice_routes.route("/login", methods=["POST"])
+def login():
+    if "pin_audio" not in request.files:
+        return jsonify({"error": "Audio file missing"}), 400
+
+    email = request.form.get("email")
+    user = users_collection.find_one({"email": email})
+
+    if not user:
+        return jsonify({"error": "User not found"}), 404
+
+    audio_file = request.files["pin_audio"]
+    temp_path = tempfile.mktemp(suffix=".wav")
+    audio_file.save(temp_path)
+
+    new_emb = get_embedding(temp_path)
+    saved_emb = np.array(user["embedding"], dtype=np.float32)
+
+    score, status = cosine_match(saved_emb, new_emb)
+
+    os.remove(temp_path)
+
+    return jsonify({
+        "score": score,
+        "status": status
+    })

voice_embedder.py

@@ -0,0 +1,55 @@
+import numpy as np
+from sklearn.metrics.pairwise import cosine_similarity
+import librosa
+import noisereduce as nr
+import torch
+from transformers import AutoModelForAudioXVector
+
+device = "cpu"
+
+# Load model globally once
+model = AutoModelForAudioXVector.from_pretrained(
+    "microsoft/wavlm-base-plus-sv"
+).to(device)
+
+
+# ------------------------------------------------------------
+# PREPROCESS AUDIO
+# ------------------------------------------------------------
+def preprocess(path):
+    y, sr = librosa.load(path, sr=16000)
+    y = nr.reduce_noise(y=y, sr=sr)
+    y, _ = librosa.effects.trim(y, top_db=25)
+    y = librosa.util.normalize(y)
+    return y
+
+
+# ------------------------------------------------------------
+# GET EMBEDDING
+# ------------------------------------------------------------
+def get_embedding(path):
+    y = preprocess(path)
+    audio = torch.tensor(y).float().unsqueeze(0).to(device)
+
+    with torch.no_grad():
+        outputs = model(audio)
+        emb = outputs.embeddings.cpu().numpy().squeeze()
+
+    emb = emb / np.linalg.norm(emb)
+    return emb.astype(np.float32)
+
+
+# ------------------------------------------------------------
+# COSINE MATCH
+# ------------------------------------------------------------
+def cosine_match(saved, new, threshold=0.75):
+    """Returns similarity score + authentication status"""
+
+    score = float(cosine_similarity([saved], [new])[0][0])
+
+    if score >= threshold:
+        return score, "Authenticated"
+    elif score >= 0.55:
+        return score, "Ask for PIN"
+    else:
+        return score, "Failed"