Update app.py

app.py

@@ -8,6 +8,13 @@ import tempfile
 from typing import List, Dict, Tuple, Optional
 from dataclasses import dataclass
 import subprocess
+from llama_index import VectorStoreIndex, SimpleDirectoryReader, StorageContext, load_index_from_storage
+from llama_index.vector_stores import SimpleVectorStore
+from docx import Document
+import fitz  # PyMuPDF
+import shutil
+import tempfile
+
 
 @dataclass
 class Question:
@@ -21,6 +28,98 @@ class QuizFeedback:
     selected: Optional[str]
     correct_answer: str
 
+class DocumentProcessor:
+    def __init__(self, storage_dir="./storage"):
+        self.storage_dir = storage_dir
+        self.temp_dir = tempfile.mkdtemp()
+        self.index = None
+        os.makedirs(storage_dir, exist_ok=True)
+    
+    def process_file(self, file_path: str) -> str:
+        """Process uploaded file and save to temp directory"""
+        file_ext = os.path.splitext(file_path)[1].lower()
+        
+        if file_ext == '.pdf':
+            return self._process_pdf(file_path)
+        elif file_ext in ['.doc', '.docx']:
+            return self._process_word(file_path)
+        else:
+            raise ValueError(f"Unsupported file type: {file_ext}")
+    
+    def _process_pdf(self, file_path: str) -> str:
+        """Extract text from PDF file"""
+        text = ""
+        try:
+            doc = fitz.open(file_path)
+            for page in doc:
+                text += page.get_text()
+            doc.close()
+            
+            # Save processed text
+            output_path = os.path.join(self.temp_dir, "processed.txt")
+            with open(output_path, "w", encoding="utf-8") as f:
+                f.write(text)
+            return output_path
+        except Exception as e:
+            raise Exception(f"Error processing PDF: {str(e)}")
+    
+    def _process_word(self, file_path: str) -> str:
+        """Extract text from Word file"""
+        try:
+            doc = Document(file_path)
+            text = "\n".join([paragraph.text for paragraph in doc.paragraphs])
+            
+            # Save processed text
+            output_path = os.path.join(self.temp_dir, "processed.txt")
+            with open(output_path, "w", encoding="utf-8") as f:
+                f.write(text)
+            return output_path
+        except Exception as e:
+            raise Exception(f"Error processing Word file: {str(e)}")
+    
+    def create_index(self) -> bool:
+        """Create vector store index from processed documents"""
+        try:
+            # Load documents from temp directory
+            documents = SimpleDirectoryReader(self.temp_dir).load_data()
+            
+            # Create and save index
+            self.index = VectorStoreIndex.from_documents(
+                documents,
+                storage_context=StorageContext.from_defaults(
+                    vector_store=SimpleVectorStore(),
+                    persist_dir=self.storage_dir
+                )
+            )
+            return True
+        except Exception as e:
+            print(f"Error creating index: {str(e)}")
+            return False
+    
+    def clear_index(self):
+        """Clear the vector store index and temporary files"""
+        try:
+            # Clear storage directory
+            if os.path.exists(self.storage_dir):
+                shutil.rmtree(self.storage_dir)
+                os.makedirs(self.storage_dir)
+            
+            # Clear temp directory
+            if os.path.exists(self.temp_dir):
+                shutil.rmtree(self.temp_dir)
+                self.temp_dir = tempfile.mkdtemp()
+            
+            self.index = None
+            return "Index cleared successfully"
+        except Exception as e:
+            return f"Error clearing index: {str(e)}"
+    
+    def cleanup(self):
+        """Clean up temporary directory"""
+        if os.path.exists(self.temp_dir):
+            shutil.rmtree(self.temp_dir)
+
+
 class QuizGenerator:
     def __init__(self, api_key: str):
         self.client = Groq(api_key=api_key)
@@ -421,6 +520,33 @@ class QuizApp:
         self.quiz_generator = QuizGenerator(api_key)
         self.certificate_generator = CertificateGenerator()
         self.current_questions: List[Question] = []
+        self.document_processor = DocumentProcessor()
+    
+    # Add this new method
+    def process_uploaded_file(self, file_path: str) -> str:
+        """Process uploaded file and return status message"""
+        try:
+            if not file_path:
+                return "⚠️ Please upload a file first."
+            
+            self.document_processor.process_file(file_path)
+            return "✅ File processed successfully! Click 'Create Index' to continue."
+        except Exception as e:
+            return f"❌ Error processing file: {str(e)}"
+    
+    # Add this new method
+    def create_index(self) -> str:
+        """Create vector store index and return status message"""
+        success = self.document_processor.create_index()
+        if success:
+            return "✅ Index created successfully!"
+        return "❌ Failed to create index. Please try again."
+    
+    # Add this new method
+    def clear_index(self) -> str:
+        """Clear vector store index and return status message"""
+        return self.document_processor.clear_index()
+
 
     def generate_questions(self, text: str, num_questions: int) -> Tuple[bool, List[Question]]:
         """
@@ -611,12 +737,27 @@ def create_quiz_interface():
                     name = gr.Textbox(label="Full Name", placeholder="Enter your full name")
                     email = gr.Textbox(label="Email", placeholder="Enter your email")
                 
+                # Add file upload section
+                with gr.Row():
+                    file_upload = gr.File(
+                        label="Upload PDF or Word Document",
+                        file_types=[".pdf", ".doc", ".docx"],
+                        type="filepath"
+                    )
+                    file_status = gr.Markdown("")
+                
+                with gr.Row():
+                    process_btn = gr.Button("Process File", variant="secondary", size="sm")
+                    index_btn = gr.Button("Create Index", variant="secondary", size="sm")
+                    clear_index_btn = gr.Button("Clear Index", variant="secondary", size="sm")
+                
+                # Or use text input
                 text_input = gr.Textbox(
-                    label="Learning Content",
+                    label="Or Enter Learning Content Directly",
                     placeholder="Enter the text content you want to be assessed on",
                     lines=10
                 )
-                
+        
                 num_questions = gr.Slider(
                     minimum=1,
                     maximum=20,
@@ -700,6 +841,18 @@ def create_quiz_interface():
 
 
         # Helper Functions
+
+        def process_file(file_path):
+            if not file_path:
+                return "⚠️ Please upload a file first."
+            return quiz_app.process_uploaded_file(file_path)
+        
+        def create_index():
+            return quiz_app.create_index()
+        
+        def clear_index():
+            return quiz_app.clear_index()
+        
         def on_generate_questions(text, num_questions):
             if not text.strip():
                 return [
@@ -944,7 +1097,21 @@ def create_quiz_interface():
             inputs=[score_display, name, course_name, company_logo, participant_photo],
             outputs=certificate_display
         )
+        process_btn.click(
+            fn=process_file,
+            inputs=[file_upload],
+            outputs=[file_status]
+        )
+        
+        index_btn.click(
+            fn=create_index,
+            outputs=[file_status]
+        )
         
+        clear_index_btn.click(
+            fn=clear_index,
+            outputs=[file_status]
+        )        
     return demo
 
 if __name__ == "__main__":