Create app.py

app.py

@@ -0,0 +1,283 @@
+import os
+import uuid
+import tempfile
+import requests
+from pathlib import Path
+from dotenv import load_dotenv
+from gtts import gTTS
+from PyPDF2 import PdfReader
+from PIL import Image
+import gradio as gr
+from googletrans import Translator
+from sentence_transformers import SentenceTransformer, util
+
+# ------------------ Load API Keys ------------------
+load_dotenv()
+GROQ_API_KEY = os.getenv("GROQ_API_KEY", "").strip()
+OCR_SPACE_API_KEY = os.getenv("OCR_SPACE_API_KEY", "").strip()
+
+if not GROQ_API_KEY:
+    raise ValueError("❌ GROQ_API_KEY missing. Add it in Hugging Face Secrets.")
+if not OCR_SPACE_API_KEY:
+    raise ValueError("❌ OCR_SPACE_API_KEY missing. Add it in Hugging Face Secrets.")
+
+HEADERS = {"Authorization": f"Bearer {GROQ_API_KEY}"}
+
+# ------------------ Global States ------------------
+SESSION_HISTORY = {}
+PDF_CONTENT = {}       # session_id -> list of chunks
+PDF_EMBEDS = {}        # session_id -> list of embeddings
+IMAGE_TEXT = {}
+IMAGE_EMBEDS = {}
+CHUNK_SIZE = 1500  # Number of characters per chunk
+
+translator = Translator()
+embed_model = SentenceTransformer('all-MiniLM-L6-v2')
+
+# ------------------ Utility Functions ------------------
+def chunk_text(text, size=CHUNK_SIZE):
+    return [text[i:i+size] for i in range(0, len(text), size)]
+
+def synthesize_speech(text, lang="en"):
+    try:
+        tts = gTTS(text=text, lang=lang)
+        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3")
+        tts.save(temp_file.name)
+        return temp_file.name
+    except Exception as e:
+        print("TTS error:", e)
+        return None
+
+def select_relevant_chunk(question, chunks, chunk_embeds):
+    q_embed = embed_model.encode(question, convert_to_tensor=True)
+    scores = util.cos_sim(q_embed, chunk_embeds)[0]
+    top_idx = scores.argmax().item()
+    return chunks[top_idx]
+
+# ------------------ Voice Chat ------------------
+def transcribe_audio(audio_file):
+    try:
+        url = "https://api.groq.com/openai/v1/audio/transcriptions"
+        with open(audio_file, "rb") as f:
+            files = {"file": ("audio.wav", f, "audio/wav")}
+            data = {"model": "whisper-large-v3"}
+            resp = requests.post(url, headers=HEADERS, files=files, data=data)
+        resp.raise_for_status()
+        return resp.json().get("text", "")
+    except Exception as e:
+        return f"Error transcribing audio: {e}"
+
+def generate_response(session_id, user_text):
+    if session_id not in SESSION_HISTORY:
+        SESSION_HISTORY[session_id] = []
+    SESSION_HISTORY[session_id].append({"role": "user", "content": user_text})
+    messages = [{"role": "system", "content": "You are a helpful AI assistant."}] + SESSION_HISTORY[session_id]
+    body = {"model": "llama-3.1-8b-instant", "messages": messages}
+    try:
+        resp = requests.post("https://api.groq.com/openai/v1/chat/completions", headers=HEADERS, json=body)
+        resp.raise_for_status()
+        assistant_msg = resp.json()["choices"][0]["message"]["content"]
+        SESSION_HISTORY[session_id].append({"role": "assistant", "content": assistant_msg})
+        return assistant_msg
+    except Exception as e:
+        return f"Error generating response: {e}"
+
+def handle_voice(audio_file, session_id, tts_lang="en"):
+    if not audio_file:
+        return [], None
+    user_text = transcribe_audio(audio_file)
+
+    # Translate if needed
+    translated_text = user_text
+    if tts_lang != "en":
+        translated_text = translator.translate(user_text, src=tts_lang, dest="en").text
+
+    assistant_text = generate_response(session_id, translated_text)
+
+    # Translate back for TTS
+    tts_text = assistant_text
+    if tts_lang != "en":
+        tts_text = translator.translate(assistant_text, src="en", dest=tts_lang).text
+
+    audio_path = synthesize_speech(tts_text, lang=tts_lang)
+    return SESSION_HISTORY[session_id], audio_path
+
+def reset_voice():
+    new_id = str(uuid.uuid4())
+    SESSION_HISTORY[new_id] = []
+    return new_id, []
+
+# ------------------ PDF Handling ------------------
+def handle_pdf_upload(pdf_file, session_id):
+    if not pdf_file:
+        return "", "No file uploaded"
+    try:
+        reader = PdfReader(pdf_file.name)
+        text = ""
+        for page in reader.pages:
+            text += page.extract_text() or ""
+        if not text.strip():
+            return "", "No extractable content found in PDF."
+        chunks = chunk_text(text)
+        PDF_CONTENT[session_id] = chunks
+        PDF_EMBEDS[session_id] = embed_model.encode(chunks, convert_to_tensor=True)
+        return "PDF uploaded successfully!", ""
+    except Exception as e:
+        return "", f"Error processing PDF: {e}"
+
+def handle_pdf_question(question, session_id):
+    if session_id not in PDF_CONTENT:
+        return "Document not found. Please upload first."
+    chunk = select_relevant_chunk(question, PDF_CONTENT[session_id], PDF_EMBEDS[session_id])
+    messages = [
+        {"role": "system", "content": "You are a helpful assistant summarizing the PDF."},
+        {"role": "user", "content": f"PDF Content: {chunk} ... Question: {question}"}
+    ]
+    body = {"model": "llama-3.1-8b-instant", "messages": messages}
+    try:
+        resp = requests.post("https://api.groq.com/openai/v1/chat/completions", headers=HEADERS, json=body)
+        resp.raise_for_status()
+        return resp.json()["choices"][0]["message"]["content"]
+    except Exception as e:
+        return f"Error generating response: {e}"
+
+def handle_pdf_question_voice(audio_file, session_id, tts_lang="en"):
+    if not audio_file:
+        return "", None
+    question = transcribe_audio(audio_file)
+    
+    # Translate if needed
+    translated_question = question
+    if tts_lang != "en":
+        translated_question = translator.translate(question, src=tts_lang, dest="en").text
+
+    # Select relevant chunk
+    if session_id not in PDF_CONTENT:
+        answer = "No PDF uploaded. Please upload first."
+    else:
+        chunk = select_relevant_chunk(translated_question, PDF_CONTENT[session_id], PDF_EMBEDS[session_id])
+        messages = [
+            {"role": "system", "content": "You are a helpful assistant summarizing the PDF."},
+            {"role": "user", "content": f"PDF Content: {chunk} ... Question: {translated_question}"}
+        ]
+        body = {"model": "llama-3.1-8b-instant", "messages": messages}
+        try:
+            resp = requests.post("https://api.groq.com/openai/v1/chat/completions", headers=HEADERS, json=body)
+            resp.raise_for_status()
+            answer = resp.json()["choices"][0]["message"]["content"]
+        except Exception as e:
+            answer = f"Error generating response: {e}"
+
+    # Translate back for TTS
+    tts_text = answer
+    if tts_lang != "en":
+        tts_text = translator.translate(answer, src="en", dest=tts_lang).text
+    audio_path = synthesize_speech(tts_text, lang=tts_lang)
+    return answer, audio_path
+
+def download_pdf_summary(session_id):
+    if session_id not in SESSION_HISTORY:
+        return None
+    summary = "\n".join([msg["content"] for msg in SESSION_HISTORY[session_id] if msg["role"]=="assistant"])
+    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".txt")
+    with open(temp_file.name, "w", encoding="utf-8") as f:
+        f.write(summary)
+    return temp_file.name
+
+# ------------------ Image OCR via OCR.Space ------------------
+def handle_image_upload_ocr(image_file, session_id):
+    if not image_file:
+        return None, "No image uploaded"
+    try:
+        with open(image_file.name, "rb") as f:
+            response = requests.post(
+                'https://api.ocr.space/parse/image',
+                files={'file': f},
+                data={'apikey': OCR_SPACE_API_KEY, 'language': 'eng'}
+            )
+        result = response.json()
+        parsed_text = result['ParsedResults'][0]['ParsedText'] if result['ParsedResults'] else ""
+        if not parsed_text.strip():
+            return None, "No extractable text found in the image."
+        chunks = chunk_text(parsed_text)
+        IMAGE_TEXT[session_id] = chunks
+        IMAGE_EMBEDS[session_id] = embed_model.encode(chunks, convert_to_tensor=True)
+        return "Image uploaded successfully!", None
+    except Exception as e:
+        return None, f"Error reading image: {e}"
+
+def handle_image_question(question, session_id):
+    if session_id not in IMAGE_TEXT:
+        return "Image not found. Please upload first."
+    chunk = select_relevant_chunk(question, IMAGE_TEXT[session_id], IMAGE_EMBEDS[session_id])
+    messages = [
+        {"role": "system", "content": "You are a helpful assistant summarizing image text."},
+        {"role": "user", "content": f"Image Text: {chunk} ... Question: {question}"}
+    ]
+    body = {"model": "llama-3.1-8b-instant", "messages": messages}
+    try:
+        resp = requests.post("https://api.groq.com/openai/v1/chat/completions", headers=HEADERS, json=body)
+        resp.raise_for_status()
+        return resp.json()["choices"][0]["message"]["content"]
+    except Exception as e:
+        return f"Error generating response: {e}"
+
+# ------------------ Gradio UI ------------------
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    gr.Markdown("## 🛠 Multi-Mode AI Assistant (Voice, PDF, Image)")
+
+    session_voice = gr.State(str(uuid.uuid4()))
+    session_pdf = gr.State(str(uuid.uuid4()))
+    session_image = gr.State(str(uuid.uuid4()))
+
+    # --- Voice ---
+    with gr.Tab("🎤 Voice Chat"):
+        chat_voice = gr.Chatbot(type="messages", height=380)
+        with gr.Row():
+            mic = gr.Audio(type="filepath", label="Hold & speak")
+            tts_lang = gr.Dropdown(choices=["en", "ur"], value="en", label="Voice Language")
+            send_voice = gr.Button("Send")
+        audio_output = gr.Audio(label="Assistant Voice Output", type="filepath")
+        reset_v = gr.Button("♻ Reset Voice Chat")
+        send_voice.click(handle_voice, inputs=[mic, session_voice, tts_lang], outputs=[chat_voice, audio_output])
+        reset_v.click(reset_voice, outputs=[session_voice, chat_voice])
+
+    # --- PDF (Text) ---
+    with gr.Tab("📄 PDF Summarizer"):
+        pdf_output = gr.Textbox(label="Answer (Text Only)", lines=20, max_lines=40)
+        pdf_upload_btn = gr.File(label="Upload PDF", file_types=[".pdf"])
+        pdf_question = gr.Textbox(label="Ask a question about PDF", lines=2)
+        pdf_send_btn = gr.Button("Ask")
+        pdf_reset_btn = gr.Button("♻ Reset PDF")
+        pdf_download_btn = gr.Button("📥 Download Summary")
+        pdf_upload_btn.upload(handle_pdf_upload, inputs=[pdf_upload_btn, session_pdf], outputs=[pdf_output, pdf_output])
+        pdf_send_btn.click(handle_pdf_question, inputs=[pdf_question, session_pdf], outputs=[pdf_output])
+        pdf_reset_btn.click(lambda: (str(uuid.uuid4()), ""), outputs=[session_pdf, pdf_output])
+        pdf_download_btn.click(download_pdf_summary, inputs=[session_pdf], outputs=[pdf_output])
+
+    # --- PDF Voice Question ---
+    with gr.Tab("📄 PDF Voice Question"):
+        pdf_voice_chat = gr.Textbox(label="Assistant Answer", lines=10)
+        pdf_voice_audio = gr.Audio(label="Assistant Voice Output", type="filepath")
+        pdf_voice_input = gr.Audio(type="filepath", label="Hold & speak PDF question")
+        pdf_voice_lang = gr.Dropdown(choices=["en","ur"], value="en", label="Voice Language")
+        pdf_voice_btn = gr.Button("Ask via Voice")
+        pdf_voice_btn.click(
+            handle_pdf_question_voice, 
+            inputs=[pdf_voice_input, session_pdf, pdf_voice_lang], 
+            outputs=[pdf_voice_chat, pdf_voice_audio]
+        )
+
+    # --- Image ---
+    with gr.Tab("🖼 Image OCR"):
+        image_output = gr.Textbox(label="Answer (Text Only)", lines=20, max_lines=40)
+        image_upload_btn = gr.File(label="Upload Image", file_types=[".png", ".jpg", ".jpeg"])
+        image_question = gr.Textbox(label="Ask a question about Image", lines=2)
+        image_send_btn = gr.Button("Ask")
+        image_reset_btn = gr.Button("♻ Reset Image")
+        image_upload_btn.upload(handle_image_upload_ocr, inputs=[image_upload_btn, session_image], outputs=[image_output, image_output])
+        image_send_btn.click(handle_image_question, inputs=[image_question, session_image], outputs=[image_output])
+        image_reset_btn.click(lambda: (str(uuid.uuid4()), ""), outputs=[session_image, image_output])
+
+if __name__ == "__main__":
+    demo.launch()