content_state.py

# content_state.py

from typing import List, Set, Optional, Dict, Any
from datetime import datetime

class ContentState:
    """Tracks the state and progression of content generation"""
    def __init__(self):
        """Initialize content state tracking variables"""
        # Core content tracking
        self.current_summary: str = ""
        self.section_outlines: List[str] = []
        self.generated_sections: List[str] = []
        self.narrative_threads: List[str] = []
        self.key_points_covered: Set[str] = set()
        self.transition_points: List[str] = []
        self.manual_content: Optional[str] = None
        
        # Additional prompt and context management
        self.additional_prompts: Dict[str, str] = {}  # Store additional prompts per section
        self.context_history: List[Dict[str, Any]] = []  # Track generation context
        self.custom_instructions: Dict[str, str] = {}  # Store custom instructions per section
        
        # Generation state tracking
        self.total_words_generated: int = 0
        self.generation_timestamps: List[float] = []
        self.current_section_index: int = 0
        self.content_status: str = "not_started"  # not_started, in_progress, completed
        self.manual_edits: List[str] = []
        
    def set_manual_content(self, content: str) -> None:
        """Set pre-written content for this section"""
        self.manual_content = content
        self.manual_edits.append(content)
        self.content_status = "in_progress"
        
    def get_manual_content(self) -> Optional[str]:
        """Get pre-written content if it exists"""
        return self.manual_content
    
    def add_section(self, content: str) -> None:
        """Add a new generated section and update tracking"""
        self.generated_sections.append(content)
        self.total_words_generated += len(content.split())
        self.current_section_index += 1
        self.generation_timestamps.append(datetime.now().timestamp())
        
    def add_narrative_thread(self, thread: str) -> None:
        """Add a new narrative thread to track"""
        if thread not in self.narrative_threads:
            self.narrative_threads.append(thread)
        
    def add_key_point(self, point: str) -> None:
        """Add a key point to the covered points set"""
        self.key_points_covered.add(point)
        
    def add_transition(self, transition: str) -> None:
        """Add a transition point between sections"""
        self.transition_points.append(transition)
        
    def add_custom_prompt(self, section_id: str, prompt: str) -> None:
        """Add a custom prompt for a specific section"""
        self.additional_prompts[section_id] = prompt
        self._update_generation_status()
        
    def add_custom_instructions(self, section_id: str, instructions: str) -> None:
        """Add custom instructions for a specific section"""
        self.custom_instructions[section_id] = instructions
        self._update_generation_status()
        
    def get_section_context(self, section_id: str) -> Dict[str, Any]:
        """Get complete context for a section including custom prompts and instructions"""
        context = {
            'manual_content': self.manual_content,
            'additional_prompt': self.additional_prompts.get(section_id, ''),
            'custom_instructions': self.custom_instructions.get(section_id, ''),
            'current_summary': self.current_summary,
            'narrative_threads': self.narrative_threads,
            'key_points_covered': list(self.key_points_covered),
            'generation_status': self.content_status,
            'total_words': self.total_words_generated,
            'section_count': len(self.generated_sections)
        }
        
        if self.context_history:
            context['previous_context'] = self.context_history[-1]
        
        return context
        
    def update_context_history(self, section_id: str, generation_context: Dict[str, Any]) -> None:
        """Update context history with latest generation context"""
        self.context_history.append({
            'section_id': section_id,
            'context': generation_context,
            'timestamp': datetime.now().isoformat()
        })
        self._update_generation_status()
        
    def get_generation_progress(self) -> float:
        """Calculate generation progress as a percentage"""
        if not self.section_outlines:
            return 0.0
        return (self.current_section_index / len(self.section_outlines)) * 100
    
    def _update_generation_status(self) -> None:
        """Update the content generation status"""
        if self.total_words_generated == 0 and not self.manual_content:
            self.content_status = "not_started"
        elif self.get_generation_progress() >= 100:
            self.content_status = "completed"
        else:
            self.content_status = "in_progress"
    
    def reset_state(self) -> None:
        """Reset the content state to initial values"""
        self.__init__()
        
    def get_state_summary(self) -> Dict[str, Any]:
        """Get a summary of the current content state"""
        return {
            'status': self.content_status,
            'progress': self.get_generation_progress(),
            'total_words': self.total_words_generated,
            'sections_generated': len(self.generated_sections),
            'has_manual_content': bool(self.manual_content),
            'narrative_threads_count': len(self.narrative_threads),
            'key_points_count': len(self.key_points_covered),
            'custom_prompts_count': len(self.additional_prompts),
            'last_updated': self.generation_timestamps[-1] if self.generation_timestamps else None
        }