app.py

# ========================================== #
# 0. HUGGING FACE ZERO-GPU INITIALIZATION    #
# ========================================== #
try:
    import spaces
except ImportError:
    class spaces:
        @staticmethod
        def GPU(fn):
            return fn

import os
import uuid
import whisper
import gradio as gr
import soundfile as sf
import numpy as np
import scipy.io.wavfile as wav
from groq import Groq
from f5_tts.api import F5TTS

# ========================================== #
# 1. System Setup & Configuration            #
# ========================================== #
GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "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")
voice_file = "my_voice_fixed.wav"

if not os.path.exists(voice_file):
    print(f"⚠️ Reference audio file '{voice_file}' not found! Creating safe baseline array profile...")
    sample_rate = 24000
    duration = 1.0
    audio_data = np.zeros(int(sample_rate * duration), dtype=np.int16)
    wav.write(voice_file, sample_rate, audio_data)

groq_client = Groq(api_key=GROQ_API_KEY) if GROQ_API_KEY else None
whisper_model = None
f5tts = None
auto_reference_text = None

# ========================================== #
# 2. Dynamic Text Response Generation        #
# ========================================== #
def get_ai_answer(question):
    global groq_client
    if not GROQ_API_KEY:
        return "Please add your active GROQ_API_KEY into Hugging Face Settings -> Secrets to chat."
    
    try:
        if groq_client is None:
            groq_client = Groq(api_key=GROQ_API_KEY)
            
        chat_completion = groq_client.chat.completions.create(
            messages=[
                {
                    "role": "system", 
                    "content": "You are an informative voice simulation assistant. Provide comprehensive, detailed, and clear explanations. Avoid single-sentence answers."
                },
                {"role": "user", "content": question}
            ],
            model="llama-3.3-70b-versatile",
            max_tokens=250, # INCREASED: Allows for longer, more detailed descriptions
            timeout=15
        )
        return chat_completion.choices[0].message.content.strip()
    except Exception as e:
        print(f"⚠️ Groq API issue detected: {e}")
        return f"The AI engine is currently offline. Details: {str(e)}"

# ========================================== #
# 3. Native Voice Synthesis Processing       #
# ========================================== #
def clone_and_speak(text):
    global f5tts, auto_reference_text
    try:
        if "offline" in text.lower() or "secrets" in text.lower():
            return None
            
        unique_filename = f"voice_output_{uuid.uuid4().hex[:8]}.wav"
        print(f"🔊 Rendering voice cloning track for text: '{text}'")
        
        if f5tts is None:
            print("Loading F5-TTS Core Neural Simulator...")
            f5tts = F5TTS()
            
        if auto_reference_text is None:
            auto_reference_text = "Hi, my name is Rabbani. I am software engineering in the Gamana solution."
            
        result = f5tts.infer(
            ref_file=voice_file,
            ref_text=auto_reference_text,
            gen_text=text
        )
        
        if isinstance(result, tuple):
            if len(result) == 2:
                audio_data, sample_rate = result[0], result[1]
            elif len(result) >= 3:
                audio_data, sample_rate = result[0], result[1]
            else:
                audio_data, sample_rate = result[0], 24000
        else:
            audio_data, sample_rate = result, 24000
            
        if hasattr(audio_data, "cpu"):
            audio_data = audio_data.cpu().numpy()
        elif hasattr(audio_data, "numpy"):
            audio_data = audio_data.numpy()
            
        audio_data = np.squeeze(audio_data)
        sf.write(unique_filename, audio_data, sample_rate, format='WAV')
        
        if os.path.exists(unique_filename) and os.path.getsize(unique_filename) > 0:
            print(f"🎉 Generated track successfully saved to: {unique_filename}")
            return unique_filename
        return None
    except Exception as e:
        print(f"❌ Core TTS System Exception triggered: {e}")
        return None

# ========================================== #
# 4. Request Router Pipeline                 #
# ========================================== #
@spaces.GPU
def chatbot_engine(text_input, audio_input):
    global whisper_model, auto_reference_text
    if whisper_model is None:
        print("Loading Lightweight Whisper Engine...")
        whisper_model = whisper.load_model("tiny")
        
    if auto_reference_text is None:
        try:
            ref_result = whisper_model.transcribe(voice_file, fp16=False)
            auto_reference_text = ref_result["text"].strip()
            if not auto_reference_text:
                auto_reference_text = "Hi, my name is Rabbani. I am software engineering in the Gamana solution."
        except Exception:
            auto_reference_text = "Hi, my name is Rabbani. I am software engineering in the Gamana solution."
            
    if audio_input is not None:
        user_transcription = whisper_model.transcribe(audio_input, fp16=False)
        question = user_transcription["text"]
    elif text_input and text_input.strip() != "":
        question = text_input
    else:
        return "System Notification: Please enter a text question or record your microphone.", None
        
    ai_text_response = get_ai_answer(question)
    generated_audio_track = clone_and_speak(ai_text_response)
    return ai_text_response, generated_audio_track, gr.update(value=""), gr.update(value=None)

# ========================================== #
# 5. Full-Output Single-Page UI Canvas       #
# ========================================== #
with gr.Blocks(title="Rabbani A I voice clone") as interface:
    gr.Markdown("# 🎙️ Rabbani A I voice clone")
    gr.Markdown("Type or speak your question below. The system processes the response dynamically and outputs both text and voice simulation results simultaneously.")
    
    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### 💬 Input Interaction Canvas")
            text_prompt_box = gr.Textbox(placeholder="Type your custom question query here...", label="Your Written Question")
            audio_prompt_box = gr.Audio(label="Or Record/Speak Your Question Directly", type="filepath")
            submit_btn = gr.Button("Send Question to AI Engine", variant="primary")
            clear_btn = gr.ClearButton(value="Clear Inputs")
            
        with gr.Column(scale=1):
            gr.Markdown("### 📊 Cloned Output Responses")
            text_output_box = gr.Textbox(label="AI Answer Text Output", interactive=False)
            audio_output_player = gr.Audio(
                label="Rendered Cloned Response Audio Player",
                type="filepath",
                interactive=False
            )
            
    submit_btn.click(
        fn=chatbot_engine,
        inputs=[text_prompt_box, audio_prompt_box],
        outputs=[text_output_box, audio_output_player, text_prompt_box, audio_prompt_box]
    )
    clear_btn.add([text_prompt_box, audio_prompt_box, text_output_box, audio_output_player])

interface.queue().launch()