app.py

import pandas as pd
import gradio as gr
import logging
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from transformers import MarianMTModel, MarianTokenizer
from PIL import Image, ImageDraw, ImageFont
import textwrap
import os
from gtts import gTTS
import tempfile
from diffusers import DiffusionPipeline

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MultilingualGitaAnalyzer:
    def __init__(self, csv_path='/content/Bhagwad_Gita.csv'):
        """Initialize the Bhagavad Gita Analyzer with multilingual support."""
        logger.info("Initializing Multilingual Bhagavad Gita Analyzer")
        self.df = None
        self.vectorizer = None
        self.tfidf_matrix = None
        self.translation_model_to_en = None
        self.translation_tokenizer_to_en = None
        self.translation_models_from_en = {}
        self.translation_tokenizers_from_en = {}
        self.load_translation_models()
        self.load_dataset(csv_path)

    def load_translation_models(self):
        """Load translation models for multilingual support."""
        try:
            # Load MarianMT for translations (multi-language to English)
            logger.info("Loading MarianMT translation model for multi-language to English...")
            self.translation_tokenizer_to_en = MarianTokenizer.from_pretrained("Helsinki-NLP/opus-mt-mul-en")
            self.translation_model_to_en = MarianMTModel.from_pretrained("Helsinki-NLP/opus-mt-mul-en")
            logger.info("Translation model for multi-language to English loaded successfully.")

            # Load MarianMT for translations from English to other languages
            languages = ["hi", "es", "fr"]  # Hindi, Spanish, French
            for lang in languages:
                model_name = f"Helsinki-NLP/opus-mt-en-{lang}"
                logger.info(f"Loading MarianMT translation model for English to {lang}...")
                self.translation_tokenizers_from_en[lang] = MarianTokenizer.from_pretrained(model_name)
                self.translation_models_from_en[lang] = MarianMTModel.from_pretrained(model_name)
                logger.info(f"Translation model for English to {lang} loaded successfully.")
        except Exception as e:
            logger.error(f"Error loading translation models: {e}")
            self.translation_tokenizer_to_en = None
            self.translation_model_to_en = None
            self.translation_models_from_en = {}
            self.translation_tokenizers_from_en = {}

    def translate_to_english(self, text):
        """Translate text to English using MarianMT."""
        try:
            if not self.translation_model_to_en or not self.translation_tokenizer_to_en:
                return text  # Fallback: return original text
            inputs = self.translation_tokenizer_to_en(text, return_tensors="pt", truncation=True)
            outputs = self.translation_model_to_en.generate(**inputs)
            return self.translation_tokenizer_to_en.decode(outputs[0], skip_special_tokens=True)
        except Exception as e:
            logger.error(f"Error during translation to English: {e}")
            return text

    def translate_from_english(self, text, target_language="en"):
        """Translate from English to the selected target language."""
        try:
            if target_language == "en":
                return text  # No translation needed for English

            lang_code = {"Hindi": "hi", "Spanish": "es", "French": "fr"}.get(target_language, "en")
            if lang_code == "en":
                return text

            # If translation model is available, use it for translation
            tokenizer = self.translation_tokenizers_from_en.get(lang_code)
            model = self.translation_models_from_en.get(lang_code)
            if not tokenizer or not model:
                return text  # Fallback: return original text

            inputs = tokenizer(text, return_tensors="pt", truncation=True)
            outputs = model.generate(**inputs)
            translated_text = tokenizer.decode(outputs[0], skip_special_tokens=True)

            return translated_text
        except Exception as e:
            logger.error(f"Error during reverse translation: {e}")
            return text

    def load_dataset(self, csv_path):
        """Load and preprocess the Bhagavad Gita dataset."""
        try:
            self.df = pd.read_csv(csv_path)
            logger.info(f"Dataset loaded successfully with {len(self.df)} rows.")

            required_columns = {'ID', 'Chapter', 'Verse', 'EngMeaning'}
            missing_columns = required_columns - set(self.df.columns)
            if missing_columns:
                raise ValueError(f"Missing required columns: {missing_columns}")

            self.df['Chapter'] = self.df['Chapter'].fillna('').astype(str)
            self.df['Verse'] = self.df['Verse'].fillna('').astype(str)
            self.df['EngMeaning'] = self.df['EngMeaning'].fillna('').astype(str)

            self.vectorizer = TfidfVectorizer(stop_words='english')
            self.tfidf_matrix = self.vectorizer.fit_transform(self.df['EngMeaning'])
        except Exception as e:
            logger.error(f"Error loading dataset: {e}")
            self.df = None
            self.vectorizer = None
            self.tfidf_matrix = None

    def semantic_search(self, query, top_k=3):
        """Search for similar verses based on a query."""
        try:
            if self.df is None or self.vectorizer is None or self.tfidf_matrix is None:
                return ["Dataset not loaded or vectorizer not initialized."]

            query_vector = self.vectorizer.transform([query])
            cosine_similarities = cosine_similarity(query_vector, self.tfidf_matrix).flatten()
            top_indices = cosine_similarities.argsort()[-top_k:][::-1]

            results = []
            for idx in top_indices:
                verse = self.df.iloc[idx]
                chapter = verse['Chapter']
                verse_number = verse['Verse']
                eng_meaning = verse['EngMeaning']
                results.append(f"Chapter {chapter} - Verse {verse_number}: {eng_meaning}")
            return results
        except Exception as e:
            logger.error(f"Error in semantic search: {e}")
            return ["An error occurred during semantic search."]

    def generate_image(self, text):
        """Generate an image using the DiffusionPipeline."""
        try:
            pipe = DiffusionPipeline.from_pretrained("black-forest-labs/FLUX.1-dev")
            pipe.load_lora_weights("adirik/flux-cinestill")
            prompt = f"in the style of CNSTLL, {text}, night time, cinestill 800T"
            image = pipe(prompt).images[0]
            img_path = 'generated_image.png'
            image.save(img_path)
            return img_path
        except Exception as e:
            logger.error(f"Error generating image: {e}")
            return None

    def generate_audio(self, text):
        """Generate audio from the given text."""
        try:
            tts = gTTS(text=text, lang='en')
            audio_path = 'generated_audio.mp3'
            tts.save(audio_path)
            return audio_path
        except Exception as e:
            logger.error(f"Error generating audio: {e}")
            return None

    def gradio_interface(self):
        """Create a Gradio interface for the Bhagavad Gita Analyzer."""
        def process_query(question, preferred_language):
            try:
                # Translate the query to English if it's not in English
                if preferred_language != "English":
                    question_in_english = self.translate_to_english(question)
                    logger.info(f"Translated query to English: {question_in_english}")
                else:
                    question_in_english = question

                # Perform semantic search
                results = self.semantic_search(question_in_english)

                # Translate the results back to the preferred language if needed
                translated_results = [
                    self.translate_from_english(res, target_language=preferred_language)
                    for res in results
                ]

                # Provide an explanation in easy language
                explanation = "In simple terms, duty means doing what you are supposed to do. It's like your responsibilities or tasks that you need to complete. In today's world, it could be your job, taking care of your family, or helping others. It's about doing the right thing, even when it's hard."
                translated_explanation = self.translate_from_english(explanation, target_language=preferred_language)

                # Generate an image for the first result
                if translated_results:
                    image_path = self.generate_image(translated_results[0])
                else:
                    image_path = None

                # Generate audio for the first result
                if translated_results:
                    audio_path = self.generate_audio(translated_results[0])
                else:
                    audio_path = None

                return "\n\n".join(translated_results) + "\n\n" + translated_explanation, image_path, audio_path
            except Exception as e:
                logger.error(f"Error processing query: {e}")
                return "An unexpected error occurred.", None, None

        def daily_quote():
            # Placeholder for daily quote logic
            return "Daily Quote: 'The journey of a thousand miles begins with one step.' - Lao Tzu"

        def quiz():
            # Placeholder for quiz logic
            questions = [
                {
                    "question": "What is the meaning of duty?",
                    "options": ["A responsibility", "A hobby", "A choice", "A luxury"],
                    "answer": "A responsibility"
                },
                # Add more questions here
            ]
            return questions

        iface = gr.Blocks()

        with iface:
            gr.Markdown("# Multilingual Vedas Wisdom Finder")
            gr.Markdown("Ask questions in any language and get wisdom in your preferred language.")

            with gr.Row():
                with gr.Column():
                    question = gr.Textbox(label="Ask a Question")
                    preferred_language = gr.Dropdown(["English", "Hindi", "Spanish", "French"], label="Preferred Language")
                    submit_btn = gr.Button("Submit")

                with gr.Column():
                    output = gr.Textbox(label="Vedas Wisdom")
                    image_output = gr.Image(label="Relevant Image")
                    audio_output = gr.Audio(label="Relevant Audio")

            submit_btn.click(process_query, inputs=[question, preferred_language], outputs=[output, image_output, audio_output])

            gr.Markdown("## Daily Quote")
            daily_quote_output = gr.Textbox(label="Daily Quote", value=daily_quote())

            gr.Markdown("## Interactive Quiz")
            quiz_questions = quiz()
            for q in quiz_questions:
                gr.Markdown(f"**{q['question']}**")
                gr.Radio(q['options'], label="Select your answer")

            gr.Markdown("## Audio Verses")
            gr.Audio(label="Listen to a Verse", value="verse1.mp3")  # Ensure this file exists in the same directory

        return iface

def main():
    analyzer = MultilingualGitaAnalyzer('/content/Bhagwad_Gita.csv')
    interface = analyzer.gradio_interface()
    interface.launch(share=True)

if __name__ == "__main__":
    main()