gradio_app.py

import gradio as gr
import json
import time
import logging
import re
from typing import Dict, Any, List, Tuple
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
from datetime import datetime
import os
import tempfile

# Hugging Face Transformers
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import gc

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class SyllabusFormatter:
    def __init__(self, model_name="microsoft/Phi-3-mini-4k-instruct"):
        """Initialize the formatter with Phi-3 model"""
        self.model_name = model_name
        self.tokenizer = None
        self.model = None
        self.pipe = None
        self.is_model_loaded = False
        self.processing_lock = threading.Lock()
        
    def load_model(self):
        """Load the Phi-3 model with optimizations"""
        if self.is_model_loaded:
            return True
            
        try:
            logger.info(f"Loading model: {self.model_name}")
            
            # Load tokenizer
            self.tokenizer = AutoTokenizer.from_pretrained(
                self.model_name,
                trust_remote_code=True
            )
            
            # Load model with optimizations
            self.model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32,
                device_map="auto" if torch.cuda.is_available() else None,
                trust_remote_code=True,
                low_cpu_mem_usage=True
            )
            
            # Create pipeline
            self.pipe = pipeline(
                "text-generation",
                model=self.model,
                tokenizer=self.tokenizer,
                device=0 if torch.cuda.is_available() else -1,
                torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32
            )
            
            self.is_model_loaded = True
            logger.info("Model loaded successfully!")
            return True
            
        except Exception as e:
            logger.error(f"Error loading model: {str(e)}")
            return False
    
    def create_formatting_prompt(self, unit_content: str, unit_name: str, subject_name: str = "") -> str:
        """Create a focused prompt for formatting syllabus content"""
        prompt = f"""<|system|>You are a professional academic syllabus formatter. Your job is to take poorly formatted syllabus content and make it beautifully organized and readable.



RULES:

1. PRESERVE every single word, topic, and concept from the original

2. NEVER add explanations, examples, or new content

3. ONLY restructure and format the existing text

4. Use clear headings, bullet points, and logical grouping

5. Separate different topics with proper spacing

6. Make it scannable and easy to read



FORMAT STYLE:

- Use main topic headings with proper capitalization

- Group related subtopics under main topics

- Use bullet points (•) for lists of concepts

- Use sub-bullets (◦) for details under main bullets

- Separate major sections with line breaks

- Keep technical terms exactly as written<|end|>



<|user|>Subject: {subject_name}

Unit: {unit_name}



Original content (poorly formatted):

{unit_content}



Task: Reformat this content to be beautifully organized and readable. Do NOT add any new information - only restructure what's already there.<|end|>



<|assistant|>"""
        return prompt
    
    def format_single_unit(self, unit_data: Tuple[str, str, str, str, str]) -> Tuple[str, str, str, str, str]:
        """Format a single unit's content"""
        branch, semester, subject, unit_name, unit_content = unit_data
        
        try:
            with self.processing_lock:
                # Create prompt
                prompt = self.create_formatting_prompt(unit_content, unit_name, subject)
                
                # Generate formatted content
                response = self.pipe(
                    prompt,
                    max_new_tokens=2048,
                    temperature=0.1,
                    do_sample=True,
                    top_p=0.9,
                    repetition_penalty=1.1,
                    pad_token_id=self.tokenizer.eos_token_id,
                    eos_token_id=self.tokenizer.eos_token_id
                )
                
                # Extract formatted content
                generated_text = response[0]['generated_text']
                assistant_start = generated_text.find("<|assistant|>")
                
                if assistant_start != -1:
                    formatted_content = generated_text[assistant_start + len("<|assistant|>"):].strip()
                else:
                    formatted_content = generated_text[len(prompt):].strip()
                
                # Clean up the content
                formatted_content = self.clean_generated_content(formatted_content)
                
                # Validate content
                if self.validate_formatted_content(unit_content, formatted_content):
                    return (branch, semester, subject, unit_name, formatted_content)
                else:
                    logger.warning(f"Validation failed for {subject} - {unit_name}")
                    return (branch, semester, subject, unit_name, unit_content)
                    
        except Exception as e:
            logger.error(f"Error formatting {subject} - {unit_name}: {str(e)}")
            return (branch, semester, subject, unit_name, unit_content)
    
    def clean_generated_content(self, content: str) -> str:
        """Clean up generated content"""
        # Remove special tokens
        content = re.sub(r'<\|.*?\|>', '', content)
        
        # Remove AI commentary
        lines = content.split('\n')
        cleaned_lines = []
        
        for line in lines:
            line = line.strip()
            if (line.startswith("Here") and ("formatted" in line.lower() or "organized" in line.lower())) or \
               line.startswith("I have") or line.startswith("The content has been") or \
               line.startswith("Note:") or line.startswith("This formatted version"):
                continue
            if line:
                cleaned_lines.append(line)
        
        content = '\n'.join(cleaned_lines)
        
        # Fix spacing
        content = re.sub(r'\n\s*\n\s*\n+', '\n\n', content)
        content = re.sub(r'\n([A-Z][^:\n]*:)\n', r'\n\n\1\n', content)
        
        return content.strip()
    
    def validate_formatted_content(self, original: str, formatted: str) -> bool:
        """Validate that formatted content preserves important information"""
        if len(formatted) < len(original) * 0.4:
            return False
        
        # Check for preservation of key terms
        original_words = set(re.findall(r'\b[A-Z][a-z]*(?:[A-Z][a-z]*)*\b', original))
        formatted_words = set(re.findall(r'\b[A-Z][a-z]*(?:[A-Z][a-z]*)*\b', formatted))
        
        missing_terms = original_words - formatted_words
        if len(missing_terms) > len(original_words) * 0.3:
            return False
        
        return True
    
    def extract_units_for_processing(self, syllabus_data: Dict[str, Any]) -> List[Tuple[str, str, str, str, str]]:
        """Extract all units for concurrent processing"""
        units = []
        
        for branch_name, branch_data in syllabus_data.get("syllabus", {}).items():
            if not isinstance(branch_data, dict):
                continue
                
            for sem_name, sem_data in branch_data.items():
                if not isinstance(sem_data, dict):
                    continue
                    
                for subject_name, subject_data in sem_data.items():
                    if not isinstance(subject_data, dict) or "content" not in subject_data:
                        continue
                        
                    content = subject_data["content"]
                    if not isinstance(content, dict):
                        continue
                    
                    for unit_name, unit_content in content.items():
                        if unit_name.startswith("Unit") and isinstance(unit_content, str):
                            units.append((branch_name, sem_name, subject_name, unit_name, unit_content))
        
        return units
    
    def format_syllabus_concurrent(self, syllabus_data: Dict[str, Any], progress_callback=None, max_workers=4) -> Dict[str, Any]:
        """Format syllabus using concurrent processing"""
        if not self.is_model_loaded:
            if not self.load_model():
                raise Exception("Failed to load model")
        
        # Extract units for processing
        units = self.extract_units_for_processing(syllabus_data)
        total_units = len(units)
        
        logger.info(f"Processing {total_units} units with {max_workers} workers")
        
        # Process units concurrently
        processed_units = {}
        completed_count = 0
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all tasks
            future_to_unit = {executor.submit(self.format_single_unit, unit): unit for unit in units}
            
            # Process completed tasks
            for future in as_completed(future_to_unit):
                try:
                    branch, semester, subject, unit_name, formatted_content = future.result()
                    
                    # Store the result
                    key = f"{branch}|{semester}|{subject}|{unit_name}"
                    processed_units[key] = formatted_content
                    
                    completed_count += 1
                    progress = (completed_count / total_units) * 100
                    
                    if progress_callback:
                        progress_callback(progress, f"Processed {subject} - {unit_name}")
                    
                    logger.info(f"Completed {completed_count}/{total_units} ({progress:.1f}%)")
                    
                except Exception as e:
                    logger.error(f"Error processing unit: {str(e)}")
        
        # Update the syllabus data with formatted content
        for branch_name, branch_data in syllabus_data.get("syllabus", {}).items():
            if not isinstance(branch_data, dict):
                continue
                
            for sem_name, sem_data in branch_data.items():
                if not isinstance(sem_data, dict):
                    continue
                    
                for subject_name, subject_data in sem_data.items():
                    if not isinstance(subject_data, dict) or "content" not in subject_data:
                        continue
                        
                    content = subject_data["content"]
                    if not isinstance(content, dict):
                        continue
                    
                    for unit_name in content.keys():
                        if unit_name.startswith("Unit"):
                            key = f"{branch_name}|{sem_name}|{subject_name}|{unit_name}"
                            if key in processed_units:
                                syllabus_data["syllabus"][branch_name][sem_name][subject_name]["content"][unit_name] = processed_units[key]
        
        # Add metadata
        if "metadata" not in syllabus_data:
            syllabus_data["metadata"] = {}
        
        syllabus_data["metadata"]["lastFormatted"] = datetime.now().isoformat()
        syllabus_data["metadata"]["formattingNote"] = "Content formatted using Phi-3 AI for enhanced readability"
        syllabus_data["metadata"]["originalContentPreserved"] = True
        syllabus_data["metadata"]["unitsProcessed"] = completed_count
        syllabus_data["metadata"]["formattingModel"] = self.model_name
        syllabus_data["metadata"]["version"] = "2.0"
        syllabus_data["metadata"]["processedConcurrently"] = True
        syllabus_data["metadata"]["maxWorkers"] = max_workers
        
        return syllabus_data

# Global formatter instance
formatter = SyllabusFormatter()

def format_syllabus_file(file_path, max_workers=4, progress=gr.Progress()):
    """Main function to format syllabus file"""
    try:
        # Load JSON file
        with open(file_path, 'r', encoding='utf-8') as f:
            syllabus_data = json.load(f)
        
        # Count units
        units = formatter.extract_units_for_processing(syllabus_data)
        total_units = len(units)
        
        progress(0, f"Found {total_units} units to process")
        
        # Progress callback
        def update_progress(percent, message):
            progress(percent/100, message)
        
        # Format the syllabus
        formatted_data = formatter.format_syllabus_concurrent(
            syllabus_data, 
            progress_callback=update_progress,
            max_workers=max_workers
        )
        
        # Save to temporary file
        with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f:
            json.dump(formatted_data, f, indent=2, ensure_ascii=False)
            temp_path = f.name
        
        progress(1.0, f"Completed! Processed {total_units} units")
        
        return temp_path, f"✅ Successfully formatted {total_units} units!"
        
    except Exception as e:
        error_msg = f"❌ Error: {str(e)}"
        logger.error(error_msg)
        return None, error_msg

def create_sample_json():
    """Create a sample JSON file for testing"""
    sample_data = {
        "metadata": {
            "totalFiles": 1,
            "generatedAt": datetime.now().isoformat(),
            "source": "Sample syllabus for testing",
            "description": "Sample syllabus content"
        },
        "syllabus": {
            "CSE": {
                "SEM1": {
                    "Mathematics": {
                        "extractedFrom": {
                            "path": "CSE > SEM1 > Mathematics",
                            "branch": "CSE",
                            "semester": "SEM1",
                            "subject": "Mathematics"
                        },
                        "content": {
                            "Unit I": "Differential Calculus: Limits, continuity, derivatives, applications of derivatives, maxima and minima, curve sketching, related rates, optimization problems, L'Hospital's rule, Taylor series, Partial derivatives, total differential, chain rule, implicit differentiation, Jacobians.",
                            "Unit II": "Integral Calculus: Integration techniques, definite integrals, applications of integrals, area under curves, volume of solids, arc length, surface area, Multiple integrals, double integrals, triple integrals, change of variables, applications in geometry and physics."
                        }
                    }
                }
            }
        }
    }
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f:
        json.dump(sample_data, f, indent=2, ensure_ascii=False)
        return f.name

# Gradio Interface
def create_interface():
    with gr.Blocks(
        title="Syllabus Formatter - AI-Powered JSON Syllabus Formatter",
        theme=gr.themes.Soft(
            primary_hue="blue",
            secondary_hue="purple",
            neutral_hue="gray"
        )
    ) as interface:
        
        gr.HTML("""

        <div style="text-align: center; padding: 20px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; border-radius: 10px; margin-bottom: 20px;">

            <h1 style="font-size: 2.5em; margin-bottom: 10px;">🎓 Syllabus Formatter</h1>

            <p style="font-size: 1.2em; opacity: 0.9;">AI-Powered JSON Syllabus Content Formatter using Phi-3</p>

            <p style="font-size: 1em; opacity: 0.8;">Upload your JSON syllabus file and get beautifully formatted content with concurrent processing for speed!</p>

        </div>

        """)
        
        with gr.Row():
            with gr.Column(scale=1):
                gr.HTML("""

                <div style="background: #f8f9fa; padding: 15px; border-radius: 8px; margin-bottom: 15px;">

                    <h3>📋 Instructions:</h3>

                    <ol>

                        <li>Upload your JSON syllabus file</li>

                        <li>Choose number of concurrent workers (1-8)</li>

                        <li>Click "Format Syllabus" to start processing</li>

                        <li>Download the formatted JSON file</li>

                    </ol>

                    <p><strong>Note:</strong> Only syllabus content will be formatted, metadata remains unchanged.</p>

                </div>

                """)
                
                file_input = gr.File(
                    label="📁 Upload JSON Syllabus File",
                    file_types=[".json"],
                    type="filepath"
                )
                
                workers_slider = gr.Slider(
                    minimum=1,
                    maximum=8,
                    value=4,
                    step=1,
                    label="🔄 Concurrent Workers",
                    info="More workers = faster processing (but more memory usage)"
                )
                
                format_btn = gr.Button(
                    "🚀 Format Syllabus",
                    variant="primary",
                    size="lg"
                )
                
                sample_btn = gr.Button(
                    "📝 Download Sample JSON",
                    variant="secondary"
                )
            
            with gr.Column(scale=1):
                status_output = gr.Textbox(
                    label="📊 Status",
                    lines=3,
                    interactive=False
                )
                
                download_output = gr.File(
                    label="📥 Download Formatted JSON",
                    visible=False
                )
                
                gr.HTML("""

                <div style="background: #e3f2fd; padding: 15px; border-radius: 8px; margin-top: 15px;">

                    <h3>✨ Features:</h3>

                    <ul>

                        <li>🤖 Powered by Microsoft Phi-3 AI model</li>

                        <li>⚡ Concurrent processing for speed</li>

                        <li>🔒 Preserves all original content</li>

                        <li>📊 Real-time progress tracking</li>

                        <li>🎯 Formats only syllabus content, not metadata</li>

                        <li>✅ Validation to ensure content integrity</li>

                    </ul>

                </div>

                """)
        
        # Event handlers
        def format_handler(file_path, max_workers):
            if file_path is None:
                return "❌ Please upload a JSON file first.", gr.update(visible=False)
            
            try:
                result_path, message = format_syllabus_file(file_path, int(max_workers))
                if result_path:
                    return message, gr.update(visible=True, value=result_path)
                else:
                    return message, gr.update(visible=False)
            except Exception as e:
                return f"❌ Error: {str(e)}", gr.update(visible=False)
        
        def sample_handler():
            sample_path = create_sample_json()
            return gr.update(visible=True, value=sample_path)
        
        format_btn.click(
            format_handler,
            inputs=[file_input, workers_slider],
            outputs=[status_output, download_output]
        )
        
        sample_btn.click(
            sample_handler,
            outputs=[gr.File(label="📥 Sample JSON File", visible=True)]
        )
        
        gr.HTML("""

        <div style="text-align: center; padding: 15px; margin-top: 20px; border-top: 1px solid #ddd;">

            <p style="color: #666;">

                Built with ❤️ using Hugging Face Spaces | 

                Powered by Microsoft Phi-3 | 

                Optimized for concurrent processing

            </p>

        </div>

        """)
    
    return interface

# Launch the app
if __name__ == "__main__":
    interface = create_interface()
    interface.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=True
    )