Update app.py

app.py

@@ -1,129 +1,168 @@
 from flask import Flask, request, jsonify, render_template
-from datetime import datetime, timedelta
+from datetime import datetime
 from flask_cors import CORS
 from TTS.api import TTS
 import os
 import base64
-from helper import save_audio, generate_random_filename, save_to_dataset_repo, video_to_audio, validate_audio_file, ensure_wav_format
+from helper import (
+    save_audio,
+    generate_random_filename,
+    save_to_dataset_repo,
+    video_to_audio,
+    validate_audio_file,
+    ensure_wav_format,
+)
 import wave
-import requests
 
 app = Flask(__name__)
 CORS(app)
 os.environ["COQUI_TOS_AGREED"] = "1"
 
 device = "cpu"
-
 tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to(device)
-
 active_tasks = {}
 
+# Optional: set your max audio size (MB)
+MAX_AUDIO_SIZE_MB = 15
+
 
 @app.route("/")
 def greet_html():
     return render_template("home.html")
-    
+
+
 @app.route("/sign-in")
 def sign_in():
-  return render_template("sign_in.html")
+    return render_template("sign_in.html")
+
 
 @app.route("/user_dash")
 def user_dash():
-  user_id = request.args.get("user_id")
-  if user_id:
-    return render_template("u_dash.html", user_id=user_id)
-    
-@app.route("/generate_voice", methods=['POST'])
+    user_id = request.args.get("user_id")
+    if user_id:
+        return render_template("u_dash.html", user_id=user_id)
+    return jsonify({"error": "Missing user_id"}), 400
+
+
+@app.route("/generate_voice", methods=["POST"])
 def generate_voice():
-  try:
-    data = request.get_json()
-    if not data:
-      return jsonify({'error': 'No JSON body'}), 400
-    video = data.get('video')
-    text = data.get('text')
-    audio_base64 = data.get('audio')
-    task_id = data.get('task_id')
-    user_id = data.get('user_id')
-    if not user_id:
-      return jsonify({'error': 'you are required to signin before you could usw this Ai'})
-    if not text:
-      return jsonify({'error': 'please input a prompt'})
-    if task_id in active_tasks:
-      return jsonify({'error': f'There is already an active tasks for {task_id}'}), 409
-    active_tasks[task_id]={
-      "user_id": user_id,
-      "status": "Processing",
-      "created_at": datetime.now(),
-    }
-    process_vox(user_id, text, video, audio_base64, task_id)
-    return jsonify({'message': 'Processing started', 'task_id': task_id}), 202
-  except Exception as e:
-    return jsonify({'error': str(e)}), 500
-    
+    try:
+        data = request.get_json()
+        if not data:
+            return jsonify({"error": "No JSON body"}), 400
+
+        video = data.get("video")
+        text = data.get("text")
+        audio_base64 = data.get("audio")
+        task_id = data.get("task_id")
+        user_id = data.get("user_id")
+
+        if not user_id:
+            return jsonify({"error": "You must sign in before using this AI"}), 401
+        if not text:
+            return jsonify({"error": "Please input a prompt"}), 400
+        if task_id in active_tasks:
+            return jsonify({"error": f"There is already an active task for {task_id}"}), 409
+
+        active_tasks[task_id] = {
+            "user_id": user_id,
+            "status": "Processing",
+            "created_at": datetime.now(),
+        }
+
+        process_vox(user_id, text, video, audio_base64, task_id)
+        return jsonify({"message": "Processing started", "task_id": task_id}), 202
+
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+
 def process_vox(user_id, text, video, audio_base64, task_id):
-    temp_audio_path=None
-  try:
-    if audio_base64:
-      if audio_base64.startswith('data:audio/'):
-        audio_base64=audio_base64.split(',', 1)[1]
-        temp_audio_path = f'/tmp/temp_ref_{task_id}.wav'
-        with open(temp_audio_path, 'wb') as f:
-          f.write(base64.b64decode(audio_base64))
-    elif video:
-      temp_audio_path = video_to_audio(video, output_path=temp_audio_path)
-    temp_audio_path = ensure_wav_format(temp_audio_path)
-  
-    valid, msg = validate_audio_file(temp_audio_path, MAX_AUDIO_SIZE_MB)
-    if not valid:
-      raise Exception(f"Invalid audio file: {msg}")
-    result_file= clone(text, temp_audio_path)
-    out_dir = "user_audios"
-    
-    os.makedirs(out_dir, exist_ok=True)
-    file_name = generate_random_filename("mp3")
-    file_path = os.path.join(out_dir, file_name)
-
-    with open(result_file, 'rb') as src, open(file_path, 'wb') as dst:
-      dst.write(src.read())
-    
-    with wave.open(file_path, 'rb') as wf:
-      dura = wf.getnframes() / float(wf.getframerate())
-      duration=f"{dura:.2f}"
-      title=text[:20]
-    # Upload + save metadata
-    audio_url = save_to_dataset_repo(file_path, f"user/data/audios/{file_name}", file_name)
-    active_tasks[task_id].update({'status': 'completed', 'audio_url': audio_url, 'completion_time': datetime.now()})
-    save_audio(user_id, audio_url, title or "Audio", text, duration)
-  except Exception as e:
-    active_tasks[task_id] = {'status': 'failed', 'error': str(e), 'completion_time': datetime.now()}
-  finally:
-    if os.path.exists(temp_audio_path):
-      os.remove(temp_audio_path)
-  
+    temp_audio_path = None
+    try:
+        # Handle audio or video input
+        if audio_base64:
+            if audio_base64.startswith("data:audio/"):
+                audio_base64 = audio_base64.split(",", 1)[1]
+            temp_audio_path = f"/tmp/temp_ref_{task_id}.wav"
+            with open(temp_audio_path, "wb") as f:
+                f.write(base64.b64decode(audio_base64))
+        elif video:
+            temp_audio_path = video_to_audio(video, output_path=temp_audio_path)
+
+        temp_audio_path = ensure_wav_format(temp_audio_path)
+
+        valid, msg = validate_audio_file(temp_audio_path, MAX_AUDIO_SIZE_MB)
+        if not valid:
+            raise Exception(f"Invalid audio file: {msg}")
+
+        result_file = clone(text, temp_audio_path)
+        out_dir = "user_audios"
+        os.makedirs(out_dir, exist_ok=True)
+
+        file_name = generate_random_filename("mp3")
+        file_path = os.path.join(out_dir, file_name)
+
+        with open(result_file, "rb") as src, open(file_path, "wb") as dst:
+            dst.write(src.read())
+
+        with wave.open(file_path, "rb") as wf:
+            dura = wf.getnframes() / float(wf.getframerate())
+            duration = f"{dura:.2f}"
+            title = text[:20]
+
+        # Upload + save metadata
+        audio_url = save_to_dataset_repo(file_path, f"user/data/audios/{file_name}", file_name)
+        active_tasks[task_id].update(
+            {
+                "status": "completed",
+                "audio_url": audio_url,
+                "completion_time": datetime.now(),
+            }
+        )
+        save_audio(user_id, audio_url, title or "Audio", text, duration)
+
+    except Exception as e:
+        active_tasks[task_id] = {
+            "status": "failed",
+            "error": str(e),
+            "completion_time": datetime.now(),
+        }
+
+    finally:
+        if temp_audio_path and os.path.exists(temp_audio_path):
+            os.remove(temp_audio_path)
+
+
 def clone(text, audio):
-  tts.tts_to_file(text=text, speaker_wav=audio, language="en", file_path="./output.wav")
-  return "./output.wav"
-    
-@app.route('/task_status')
+    tts.tts_to_file(text=text, speaker_wav=audio, language="en", file_path="./output.wav")
+    return "./output.wav"
+
+
+@app.route("/task_status")
 def task_status():
     task_id = request.args.get("task_id")
     if not task_id:
-        return jsonify({'error': 'task_id parameter is required'}), 400
-        
+        return jsonify({"error": "task_id parameter is required"}), 400
+
     if task_id not in active_tasks:
-        return jsonify({'status': 'not found'}), 404
+        return jsonify({"status": "not found"}), 404
 
     task = active_tasks[task_id]
     response_data = {
-        'status': task['status'],
-        'start_time': task.get('created_at').isoformat() if task.get('created_at') else None
+        "status": task["status"],
+        "start_time": task.get("created_at").isoformat() if task.get("created_at") else None,
     }
 
-    if task['status'] == 'completed':
-        response_data['audio_url'] = task.get('audio_url')
-        response_data['completion_time'] = task.get('completion_time').isoformat() if task.get('completion_time') else None
-    elif task['status'] == 'failed':
-        response_data['error'] = task.get('error')
-        response_data['completion_time'] = task.get('completion_time').isoformat() if task.get('completion_time') else None
+    if task["status"] == "completed":
+        response_data["audio_url"] = task.get("audio_url")
+        response_data["completion_time"] = (
+            task.get("completion_time").isoformat() if task.get("completion_time") else None
+        )
+    elif task["status"] == "failed":
+        response_data["error"] = task.get("error")
+        response_data["completion_time"] = (
+            task.get("completion_time").isoformat() if task.get("completion_time") else None
+        )
 
     return jsonify(response_data)