scripts/syllabus_formatter.py

#!/usr/bin/env python3
"""

Syllabus Formatter Script

This script downloads Phi-3 3B model and uses it to format syllabus content

to be more readable while preserving all content and structure.

"""

import json
import os
import sys
from pathlib import Path
import time
import logging
from typing import Dict, Any, List, Tuple
import re
import psutil  # For memory checks

# Imports for type hinting and core functionality
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
from transformers import BitsAndBytesConfig  # For 8-bit quantization
import requests

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('syllabus_formatter.log'),
        logging.StreamHandler()
    ]
)
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.processed_count = 0
        self.total_count = 0
        
    def setup_model(self):
        """Download and setup the Phi-3 model with CPU optimization"""
        logger.info(f"Setting up model: {self.model_name}")
        
        try:
            # Check available memory
            available_memory = psutil.virtual_memory().available / (1024 * 1024 * 1024)  # Convert to GB
            logger.info(f"Available system memory: {available_memory:.2f} GB")
            
            if available_memory < 4:  # We need at least 4GB free
                logger.warning("Low memory detected. Attempting to load with maximum optimization...")
            
            # Load tokenizer
            logger.info("Loading tokenizer...")
            self.tokenizer = AutoTokenizer.from_pretrained(
                self.model_name,
                trust_remote_code=True
            )
            
            # Load model with CPU optimizations
            logger.info("Loading model with CPU optimizations...")
            self.model = AutoModelForCausalLM.from_pretrained(
                self.model_name,
                torch_dtype=torch.float32,  # Use float32 for CPU
                device_map=None,  # Disable device mapping for CPU
                trust_remote_code=True,
                low_cpu_mem_usage=True
            )
            
            # Move model to CPU explicitly
            self.model = self.model.to('cpu')
            
            # Create pipeline with CPU settings
            logger.info("Creating CPU-optimized pipeline...")
            self.pipe = pipeline(
                "text-generation",
                model=self.model,
                tokenizer=self.tokenizer,
                device='cpu'  # Explicitly set to CPU
            )
            
            logger.info("Model setup complete with CPU optimizations!")
            return True
            
        except Exception as e:
            error_msg = str(e)
            if "paging file" in error_msg.lower():
                logger.error(
                    "Windows virtual memory (page file) is too small. Please:\n"
                    "1. Open System Properties > Advanced > Performance Settings > Advanced\n"
                    "2. Under Virtual Memory, click Change\n"
                    "3. Increase the page file size (recommended: 1.5x your RAM size)\n"
                    "4. Restart your computer"
                )
            else:
                logger.error(f"Error setting up model: {error_msg}")
            return False
    
    def create_formatting_prompt(self, unit_content: str, unit_name: str, subject_name: str = "") -> str:
        """Create a very clear, focused prompt for formatting syllabus content"""
        prompt = f"""<|system|>You are a professional academic syllabus formatter. Your ONLY job is to take badly 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. Make it professional and easy to scan.<|end|>



<|assistant|>"""
        return prompt
    
    def format_unit_content(self, unit_content: str, unit_name: str, subject_name: str = "") -> str:
        """Format a single unit\'s content using the AI model with focused prompting"""
        try:
            # Create a very clear, focused prompt
            prompt = self.create_formatting_prompt(unit_content, unit_name, subject_name)
            
            # Generate formatted content with specific parameters for better output
            response = self.pipe(
                prompt,
                max_new_tokens=2048,  # Increased for longer content
                temperature=0.1,      # Very low for consistent formatting
                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 the formatted content
            generated_text = response[0]['generated_text']
            
            # Find the assistant's response more reliably
            assistant_start = generated_text.find("<|assistant|>")
            if assistant_start != -1:
                formatted_content = generated_text[assistant_start + len("<|assistant|>"):].strip()
            else:
                # Fallback: try to find content after the prompt
                prompt_end = generated_text.find(prompt)
                if prompt_end != -1:
                    formatted_content = generated_text[prompt_end + len(prompt):].strip()
                else:
                    formatted_content = generated_text.strip()
            
            # Clean up the generated content
            formatted_content = self.clean_generated_content(formatted_content)
            
            # Validate the formatted content
            if not self.validate_formatted_content(unit_content, formatted_content, unit_name):
                logger.warning(f"Validation failed for {subject_name} - {unit_name}, using original")
                return unit_content
            
            logger.info(f"✓ Successfully formatted {subject_name} - {unit_name}")
            return formatted_content
            
        except Exception as e:
            logger.error(f"Error formatting {subject_name} - {unit_name}: {str(e)}")
            return unit_content  # Return original content if formatting fails
    
    def show_sample_comparison(self, original: str, formatted: str, subject: str, unit: str):
        """Show a before/after comparison for verification"""
        print("\n" + "="*80)
        print(f"📊 SAMPLE COMPARISON: {subject} - {unit}")
        print("="*80)
        print("🔴 BEFORE (Original):")
        print("-" * 40)
        print(original[:300] + "..." if len(original) > 300 else original)
        print("\n")
        print("🟢 AFTER (Formatted):")
        print("-" * 40)
        print(formatted[:300] + "..." if len(formatted) > 300 else formatted)
        print("="*80)
    
    def validate_formatted_content(self, original: str, formatted: str, unit_name: str) -> bool:
        """Validate that formatted content preserves all important information"""
        # Check length - formatted should not be drastically shorter
        if len(formatted) < len(original) * 0.4:
            logger.warning(f"Formatted content too short for {unit_name}")
            return False
        
        # Check for key technical terms preservation
        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))
        
        # Allow for some formatting differences but ensure major terms are preserved
        missing_important_terms = original_words - formatted_words
        if len(missing_important_terms) > len(original_words) * 0.3:
            logger.warning(f"Too many important terms missing in {unit_name}: {missing_important_terms}")
            return False
        
        return True
    
    def clean_generated_content(self, content: str) -> str:
        """Clean up generated content removing any artifacts and improving structure"""
        # Remove any remaining special tokens
        content = re.sub(r'<\|.*?\|>', '', content)
        
        # Remove any meta-commentary from the AI
        lines = content.split('\n')
        cleaned_lines = []
        
        for line in lines:
            line = line.strip()
            # Skip lines that look like AI commentary
            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:  # Only add non-empty lines
                cleaned_lines.append(line)
        
        content = '\n'.join(cleaned_lines)
        
        # Fix multiple consecutive newlines
        content = re.sub(r'\n\s*\n\s*\n+', '\n\n', content)
        
        # Ensure proper spacing around headers
        content = re.sub(r'\n([A-Z][^:\n]*:)\n', r'\n\n\1\n', content)
        
        return content.strip()
    
    def count_total_units(self, syllabus_data: Dict[str, Any]) -> int:
        """Count total number of units to process"""
        count = 0
        for branch_name, branch_data in syllabus_data.get("syllabus", {}).items():
            if isinstance(branch_data, dict):
                for sem_name, sem_data in branch_data.items():
                    if isinstance(sem_data, dict):
                        for subject_name, subject_data in sem_data.items():
                            if isinstance(subject_data, dict) and "content" in subject_data:
                                content = subject_data["content"]
                                if isinstance(content, dict):
                                    count += len([k for k in content.keys() if k.startswith("Unit")])
        return count
    
    def format_syllabus(self, input_file: str, output_file: str) -> bool:
        """Format the entire syllabus file"""
        try:
            # Load the syllabus file
            logger.info(f"Loading syllabus from: {input_file}")
            with open(input_file, 'r', encoding='utf-8') as f:
                syllabus_data = json.load(f)
            
            # Count total units
            self.total_count = self.count_total_units(syllabus_data)
            logger.info(f"Total units to process: {self.total_count}")
            
            # Process each branch
            for branch_name, branch_data in syllabus_data.get("syllabus", {}).items():
                if not isinstance(branch_data, dict):
                    continue
                    
                logger.info(f"Processing branch: {branch_name}")
                
                # Process each semester
                for sem_name, sem_data in branch_data.items():
                    if not isinstance(sem_data, dict):
                        continue
                        
                    logger.info(f"Processing {branch_name} - {sem_name}")
                    
                    # Process each subject
                    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
                        
                        logger.info(f"Processing {branch_name} - {sem_name} - {subject_name}")
                        
                        # Format each unit
                        for unit_name, unit_content in content.items():
                            if not unit_name.startswith("Unit") or not isinstance(unit_content, str):
                                continue
                            
                            self.processed_count += 1
                            progress = (self.processed_count / self.total_count) * 100
                            
                            logger.info(f"🔄 Processing {branch_name} > {sem_name} > {subject_name} > {unit_name} "
                                      f"({self.processed_count}/{self.total_count} - {progress:.1f}%)")
                            
                            # Show original content preview
                            preview = unit_content[:100].replace('\n', ' ') + "..." if len(unit_content) > 100 else unit_content
                            logger.info(f"📝 Original: {preview}")
                            
                            # Format the unit content with subject context
                            formatted_content = self.format_unit_content(
                                unit_content, 
                                unit_name, 
                                subject_name
                            )
                            
                            # Update the content
                            syllabus_data["syllabus"][branch_name][sem_name][subject_name]["content"][unit_name] = formatted_content
                            
                            # Show formatted content preview
                            formatted_preview = formatted_content[:100].replace('\n', ' ') + "..." if len(formatted_content) > 100 else formatted_content
                            logger.info(f"✨ Formatted: {formatted_preview}")
                            
                            # Add delay to prevent overwhelming the model
                            time.sleep(0.5)  # Increased delay for better processing
            
            # Add formatting metadata with detailed info
            if "metadata" not in syllabus_data:
                syllabus_data["metadata"] = {}
            
            syllabus_data["metadata"]["lastFormatted"] = time.strftime("%Y-%m-%dT%H:%M:%SZ")
            syllabus_data["metadata"]["formattingNote"] = "Content formatted using Phi-3 3B AI for enhanced readability and structure"
            syllabus_data["metadata"]["originalContentPreserved"] = True
            syllabus_data["metadata"]["unitsProcessed"] = self.processed_count
            syllabus_data["metadata"]["formattingModel"] = self.model_name
            syllabus_data["metadata"]["version"] = "2.0"
            
            # Save the formatted syllabus
            logger.info(f"Saving formatted syllabus to: {output_file}")
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(syllabus_data, f, indent=2, ensure_ascii=False)
            
            logger.info(f"Successfully formatted {self.processed_count} units!")
            return True
            
        except Exception as e:
            logger.error(f"Error formatting syllabus: {str(e)}")
            return False

def main():
    """Main function"""
    # Setup paths
    script_dir = Path(__file__).parent
    project_root = script_dir.parent
    syllabus_file = project_root / "public" / "Content-Meta" / "syllabus.json"
    output_file = project_root / "public" / "Content-Meta" / "syllabus_formatted.json"
    
    # Validate input file
    if not syllabus_file.exists():
        logger.error(f"Syllabus file not found: {syllabus_file}")
        return False
    
    # Create formatter
    formatter = SyllabusFormatter()
    
    # Setup model
    logger.info("Setting up Phi-3 model...")
    if not formatter.setup_model():
        logger.error("Failed to setup model")
        return False
    
    # Format syllabus
    logger.info("Starting syllabus formatting...")
    success = formatter.format_syllabus(str(syllabus_file), str(output_file))
    
    if success:
        logger.info(f"Formatting complete! Output saved to: {output_file}")
        logger.info("You can now review the formatted syllabus and replace the original if satisfied.")
    else:
        logger.error("Formatting failed!")
    
    return success

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)