🔍 Code Extractor

Service class for managing TextSection entities, providing CRUD operations, versioning, chat functionality, and search capabilities.

Purpose

TextSectionService is a comprehensive service layer for managing text sections in a document management system. It handles creation, retrieval, updating, and deletion of text sections, maintains version history, manages chat configurations and messages, supports duplication and search operations, and integrates with a DatabaseManager for persistence. The service ensures proper state management, automatic versioning on content updates, and provides methods for reverting to previous versions.

Source Code

class TextSectionService:
    """Service for managing TextSections"""
    
    def __init__(self, db_manager: DatabaseManager):
        self.db = db_manager
    
    def create_text_section(
        self, 
        owner: str, 
        title: str, 
        section_type: SectionType = SectionType.TEXT,
        level: int = 1,
        initial_content: str = ""
    ) -> TextSection:
        """Create a new text section"""
        section_id = str(uuid.uuid4())
        version_id = str(uuid.uuid4())
        
        text_section = TextSection(
            id=section_id,
            owner=owner,
            title=title,
            section_type=section_type,
            level=level,
            current_content=initial_content,
            current_version_id=version_id
        )
        
        # Save the text section
        self.db.save_text_section(text_section)
        
        # Create initial version
        if initial_content:
            self.create_version(
                section_id, 
                initial_content, 
                owner, 
                "Initial version"
            )
        
        return text_section
    
    def get_text_section(self, section_id: str) -> Optional[TextSection]:
        """Get a text section by ID"""
        return self.db.get_text_section(section_id)
    
    def get_user_text_sections(self, owner: str) -> List[TextSection]:
        """Get all text sections for a user"""
        return self.db.get_user_text_sections(owner)
    
    def update_text_section_content(
        self, 
        section_id: str, 
        new_content: str, 
        author: str,
        change_summary: str = "",
        generated_by_ai: bool = False
    ) -> bool:
        """Update text section content with versioning"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        # Create new version
        version_id = self.create_version(
            section_id, 
            new_content, 
            author, 
            change_summary, 
            generated_by_ai
        )
        
        if version_id:
            # Update current content and version
            text_section.current_content = new_content
            text_section.current_version_id = version_id
            text_section.updated_at = datetime.now()
            
            return self.db.save_text_section(text_section)
        
        return False
    
    def update_text_section_title(self, section_id: str, new_title: str) -> bool:
        """Update text section title"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        text_section.title = new_title
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def create_version(
        self, 
        text_section_id: str, 
        content: str, 
        author: str,
        change_summary: str = "",
        generated_by_ai: bool = False
    ) -> Optional[str]:
        """Create a new version of text section content"""
        version_id = str(uuid.uuid4())
        
        version = TextSectionVersion(
            version_id=version_id,
            content=content,
            timestamp=datetime.now(),
            author=author,
            change_summary=change_summary,
            generated_by_ai=generated_by_ai
        )
        
        if self.db.save_text_section_version(version, text_section_id):
            return version_id
        return None
    
    def get_text_section_versions(self, text_section_id: str) -> List[TextSectionVersion]:
        """Get all versions for a text section"""
        return self.db.get_text_section_versions(text_section_id)
    
    def revert_to_version(self, text_section_id: str, version_id: str, author: str) -> bool:
        """Revert text section to a specific version"""
        versions = self.get_text_section_versions(text_section_id)
        target_version = next((v for v in versions if v.version_id == version_id), None)
        
        if target_version:
            return self.update_text_section_content(
                text_section_id,
                target_version.content,
                author,
                f"Reverted to version {version_id}",
                False
            )
        return False
    
    def update_chat_configuration(self, section_id: str, config: ChatConfiguration) -> bool:
        """Update chat configuration for a text section"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        text_section.chat_configuration = config
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def add_chat_message(
        self, 
        section_id: str, 
        role: str, 
        content: str, 
        references: List[Dict] = None
    ) -> bool:
        """Add a chat message to a text section"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        message = ChatMessage(
            id=str(uuid.uuid4()),
            role=role,
            content=content,
            timestamp=datetime.now(),
            references=references or []
        )
        
        text_section.chat_messages.append(message)
        
        # Update last references if it's an AI response
        if role == 'assistant' and references:
            text_section.last_references = references
        
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def clear_chat_history(self, section_id: str) -> bool:
        """Clear chat history for a text section"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        text_section.chat_messages = []
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def duplicate_text_section(self, section_id: str, new_owner: str, new_title: str = None) -> Optional[TextSection]:
        """Create a duplicate of an existing text section with new UUID"""
        original = self.get_text_section(section_id)
        if not original:
            return None
        
        # Create new text section with same content but new ID
        new_section = self.create_text_section(
            owner=new_owner,
            title=new_title or f"Copy of {original.title}",
            section_type=original.section_type,
            level=original.level,
            initial_content=original.current_content
        )
        
        # Copy chat configuration but not chat history
        new_section.chat_configuration = ChatConfiguration.from_dict(
            original.chat_configuration.to_dict()
        )
        
        # Copy tags and metadata
        new_section.tags = original.tags.copy()
        new_section.metadata = original.metadata.copy()
        
        self.db.save_text_section(new_section)
        
        return new_section
    
    def delete_text_section(self, section_id: str) -> bool:
        """Delete a text section and all its versions"""
        return self.db.delete_text_section(section_id)
    
    def update_chat_session_id(self, section_id: str, chat_session_id: str) -> bool:
        """Update the chat session ID for a text section"""
        try:
            print(f"🔄 DEBUG: Updating chat_session_id for section {section_id} to {chat_session_id}")
            
            # Get the section first
            section = self.get_text_section(section_id)
            if not section:
                print(f"❌ DEBUG: Section {section_id} not found")
                return False
            
            print(f"🔍 DEBUG: Section found, current chat_session_id: {section.chat_session_id}")
            
            # Update the chat session ID
            section.chat_session_id = chat_session_id
            print(f"🔄 DEBUG: Set new chat_session_id: {section.chat_session_id}")
            
            # Save using the database manager
            success = self.db.save_text_section(section)
            print(f"💾 DEBUG: Save result: {success}")
            
            # Verify the save by re-reading the section
            if success:
                verification_section = self.get_text_section(section_id)
                if verification_section:
                    print(f"✅ DEBUG: Verification - chat_session_id now: {verification_section.chat_session_id}")
                else:
                    print(f"⚠️ DEBUG: Could not verify save - section not found on re-read")
            
            return success
        except Exception as e:
            print(f"❌ DEBUG: Error updating chat session ID: {e}")
            return False
    
    def search_text_sections(
        self, 
        owner: str, 
        query: str = "", 
        section_type: SectionType = None,
        tags: List[str] = None
    ) -> List[TextSection]:
        """Search text sections by various criteria"""
        sections = self.get_user_text_sections(owner)
        
        if query:
            query_lower = query.lower()
            sections = [
                s for s in sections 
                if query_lower in s.title.lower() or query_lower in s.current_content.lower()
            ]
        
        if section_type:
            sections = [s for s in sections if s.section_type == section_type]
        
        if tags:
            sections = [
                s for s in sections 
                if any(tag in s.tags for tag in tags)
            ]
        
        return sections
    
    def get_user_text_sections_with_documents(self, owner: str) -> List[Dict]:
        """Get all text sections for a user with document information"""
        sections = self.get_user_text_sections(owner)
        result = []
        
        for section in sections:
            # Get documents that contain this section
            documents = self.db.get_documents_containing_section(section.id)
            
            section_dict = section.to_dict()
            section_dict['documents'] = [{'id': doc.id, 'title': doc.title} for doc in documents] if documents else []
            section_dict['document_count'] = len(documents) if documents else 0
            
            result.append(section_dict)
        
        return result
    
    def get_unique_user_text_sections(self, owner: str) -> List[Dict]:
        """Get unique text sections for a user (latest versions only) with document information, excluding headers"""
        # Get all sections grouped by title/content similarity to identify duplicates
        all_sections = self.get_user_text_sections_with_documents(owner)
        
        # Filter out header types - only show text/content sections
        filtered_sections = [
            section for section in all_sections 
            if section.get('section_type') in ['text', 'content']
        ]
        
        # Group by title and keep only the most recent
        unique_sections = {}
        for section in filtered_sections:
            title = section['title']
            created_at = datetime.fromisoformat(section['created_at'])
            
            if title not in unique_sections or created_at > datetime.fromisoformat(unique_sections[title]['created_at']):
                unique_sections[title] = section
        
        return list(unique_sections.values())

Parameters

Name	Type	Default	Kind
bases	-	-

Parameter Details

db_manager: An instance of DatabaseManager that handles all database operations for persisting and retrieving TextSection, TextSectionVersion, and related entities. This is the only required dependency for instantiation.

Return Value

Instantiation returns a TextSectionService object. Methods return various types: TextSection objects for creation/retrieval operations, bool for success/failure of update/delete operations, Optional[str] for version IDs, List[TextSection] for queries returning multiple sections, List[Dict] for sections with additional metadata, and Optional[TextSection] for operations that may not find the requested section.

Class Interface

Methods

Attributes

Name	Type	Description	Scope
db	DatabaseManager	Database manager instance used for all persistence operations throughout the service lifecycle	instance

Dependencies

• uuid
• typing
• datetime

Required Imports

import uuid
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime
from models import TextSection, DataSection, Document, DocumentSection, TextSectionVersion, ChatConfiguration, ChatMessage, ChatSession, SectionType, ContentStatus, DatabaseManager, DocumentVersion

Usage Example

from models import DatabaseManager, SectionType
from text_section_service import TextSectionService

# Initialize service with database manager
db_manager = DatabaseManager(connection_string='your_db_connection')
service = TextSectionService(db_manager)

# Create a new text section
section = service.create_text_section(
    owner='user123',
    title='Introduction',
    section_type=SectionType.TEXT,
    level=1,
    initial_content='This is the introduction.'
)

# Update content (creates new version automatically)
success = service.update_text_section_content(
    section_id=section.id,
    new_content='Updated introduction text.',
    author='user123',
    change_summary='Improved clarity'
)

# Get all versions
versions = service.get_text_section_versions(section.id)

# Add chat message
service.add_chat_message(
    section_id=section.id,
    role='user',
    content='Can you improve this section?'
)

# Search sections
results = service.search_text_sections(
    owner='user123',
    query='introduction',
    section_type=SectionType.TEXT
)

# Duplicate section
new_section = service.duplicate_text_section(
    section_id=section.id,
    new_owner='user456',
    new_title='Copy of Introduction'
)

# Delete section
service.delete_text_section(section.id)

Best Practices

• Always instantiate with a valid DatabaseManager that has an active database connection
• Content updates automatically create new versions - use update_text_section_content() rather than directly modifying TextSection objects
• Check return values for None/False to handle cases where sections don't exist or operations fail
• Use revert_to_version() to safely restore previous content while maintaining version history
• Chat messages are stored in the TextSection object - use add_chat_message() to maintain proper timestamps and structure
• When duplicating sections, chat history is intentionally not copied but configuration is preserved
• The service maintains referential integrity - deleting a section also deletes all its versions
• Search operations are performed in-memory after retrieval - for large datasets, consider database-level filtering
• Version IDs and section IDs are UUIDs generated automatically - don't attempt to create custom IDs
• The update_chat_session_id method includes extensive debug logging - monitor these logs in production
• Use get_unique_user_text_sections() to avoid showing duplicate sections to users, as it filters by title and excludes headers

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class TextSection 79.3% similar

  A dataclass representing a text section entity with versioning, chat interface, data analysis capabilities, and metadata management.

  From: /tf/active/vicechatdev/vice_ai/models.py

• class DocumentService 78.1% similar

  Service class for managing Document entities, including creation, retrieval, section management, versioning, and duplication operations.

  From: /tf/active/vicechatdev/vice_ai/services.py

• class DataSectionService 73.4% similar

  Service class for managing DataSection entities, providing CRUD operations and specialized update methods for analysis sessions, plots, and conclusions.

  From: /tf/active/vicechatdev/vice_ai/services.py

• class ChatSessionService 72.8% similar

  Service class for managing chat sessions, including creation, retrieval, message management, and configuration updates for document section-based conversations.

  From: /tf/active/vicechatdev/vice_ai/services.py

• class DocumentSection 64.5% similar

  A class representing a section within a complex document, supporting hierarchical structure with headers, text content, and references.

  From: /tf/active/vicechatdev/vice_ai/complex_app.py

class TextSectionService:
    """Service for managing TextSections"""
    
    def __init__(self, db_manager: DatabaseManager):
        self.db = db_manager
    
    def create_text_section(
        self, 
        owner: str, 
        title: str, 
        section_type: SectionType = SectionType.TEXT,
        level: int = 1,
        initial_content: str = ""
    ) -> TextSection:
        """Create a new text section"""
        section_id = str(uuid.uuid4())
        version_id = str(uuid.uuid4())
        
        text_section = TextSection(
            id=section_id,
            owner=owner,
            title=title,
            section_type=section_type,
            level=level,
            current_content=initial_content,
            current_version_id=version_id
        )
        
        # Save the text section
        self.db.save_text_section(text_section)
        
        # Create initial version
        if initial_content:
            self.create_version(
                section_id, 
                initial_content, 
                owner, 
                "Initial version"
            )
        
        return text_section
    
    def get_text_section(self, section_id: str) -> Optional[TextSection]:
        """Get a text section by ID"""
        return self.db.get_text_section(section_id)
    
    def get_user_text_sections(self, owner: str) -> List[TextSection]:
        """Get all text sections for a user"""
        return self.db.get_user_text_sections(owner)
    
    def update_text_section_content(
        self, 
        section_id: str, 
        new_content: str, 
        author: str,
        change_summary: str = "",
        generated_by_ai: bool = False
    ) -> bool:
        """Update text section content with versioning"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        # Create new version
        version_id = self.create_version(
            section_id, 
            new_content, 
            author, 
            change_summary, 
            generated_by_ai
        )
        
        if version_id:
            # Update current content and version
            text_section.current_content = new_content
            text_section.current_version_id = version_id
            text_section.updated_at = datetime.now()
            
            return self.db.save_text_section(text_section)
        
        return False
    
    def update_text_section_title(self, section_id: str, new_title: str) -> bool:
        """Update text section title"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        text_section.title = new_title
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def create_version(
        self, 
        text_section_id: str, 
        content: str, 
        author: str,
        change_summary: str = "",
        generated_by_ai: bool = False
    ) -> Optional[str]:
        """Create a new version of text section content"""
        version_id = str(uuid.uuid4())
        
        version = TextSectionVersion(
            version_id=version_id,
            content=content,
            timestamp=datetime.now(),
            author=author,
            change_summary=change_summary,
            generated_by_ai=generated_by_ai
        )
        
        if self.db.save_text_section_version(version, text_section_id):
            return version_id
        return None
    
    def get_text_section_versions(self, text_section_id: str) -> List[TextSectionVersion]:
        """Get all versions for a text section"""
        return self.db.get_text_section_versions(text_section_id)
    
    def revert_to_version(self, text_section_id: str, version_id: str, author: str) -> bool:
        """Revert text section to a specific version"""
        versions = self.get_text_section_versions(text_section_id)
        target_version = next((v for v in versions if v.version_id == version_id), None)
        
        if target_version:
            return self.update_text_section_content(
                text_section_id,
                target_version.content,
                author,
                f"Reverted to version {version_id}",
                False
            )
        return False
    
    def update_chat_configuration(self, section_id: str, config: ChatConfiguration) -> bool:
        """Update chat configuration for a text section"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        text_section.chat_configuration = config
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def add_chat_message(
        self, 
        section_id: str, 
        role: str, 
        content: str, 
        references: List[Dict] = None
    ) -> bool:
        """Add a chat message to a text section"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        message = ChatMessage(
            id=str(uuid.uuid4()),
            role=role,
            content=content,
            timestamp=datetime.now(),
            references=references or []
        )
        
        text_section.chat_messages.append(message)
        
        # Update last references if it's an AI response
        if role == 'assistant' and references:
            text_section.last_references = references
        
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def clear_chat_history(self, section_id: str) -> bool:
        """Clear chat history for a text section"""
        text_section = self.get_text_section(section_id)
        if not text_section:
            return False
        
        text_section.chat_messages = []
        text_section.updated_at = datetime.now()
        
        return self.db.save_text_section(text_section)
    
    def duplicate_text_section(self, section_id: str, new_owner: str, new_title: str = None) -> Optional[TextSection]:
        """Create a duplicate of an existing text section with new UUID"""
        original = self.get_text_section(section_id)
        if not original:
            return None
        
        # Create new text section with same content but new ID
        new_section = self.create_text_section(
            owner=new_owner,
            title=new_title or f"Copy of {original.title}",
            section_type=original.section_type,
            level=original.level,
            initial_content=original.current_content
        )
        
        # Copy chat configuration but not chat history
        new_section.chat_configuration = ChatConfiguration.from_dict(
            original.chat_configuration.to_dict()
        )
        
        # Copy tags and metadata
        new_section.tags = original.tags.copy()
        new_section.metadata = original.metadata.copy()
        
        self.db.save_text_section(new_section)
        
        return new_section
    
    def delete_text_section(self, section_id: str) -> bool:
        """Delete a text section and all its versions"""
        return self.db.delete_text_section(section_id)
    
    def update_chat_session_id(self, section_id: str, chat_session_id: str) -> bool:
        """Update the chat session ID for a text section"""
        try:
            print(f"🔄 DEBUG: Updating chat_session_id for section {section_id} to {chat_session_id}")
            
            # Get the section first
            section = self.get_text_section(section_id)
            if not section:
                print(f"❌ DEBUG: Section {section_id} not found")
                return False
            
            print(f"🔍 DEBUG: Section found, current chat_session_id: {section.chat_session_id}")
            
            # Update the chat session ID
            section.chat_session_id = chat_session_id
            print(f"🔄 DEBUG: Set new chat_session_id: {section.chat_session_id}")
            
            # Save using the database manager
            success = self.db.save_text_section(section)
            print(f"💾 DEBUG: Save result: {success}")
            
            # Verify the save by re-reading the section
            if success:
                verification_section = self.get_text_section(section_id)
                if verification_section:
                    print(f"✅ DEBUG: Verification - chat_session_id now: {verification_section.chat_session_id}")
                else:
                    print(f"⚠️ DEBUG: Could not verify save - section not found on re-read")
            
            return success
        except Exception as e:
            print(f"❌ DEBUG: Error updating chat session ID: {e}")
            return False
    
    def search_text_sections(
        self, 
        owner: str, 
        query: str = "", 
        section_type: SectionType = None,
        tags: List[str] = None
    ) -> List[TextSection]:
        """Search text sections by various criteria"""
        sections = self.get_user_text_sections(owner)
        
        if query:
            query_lower = query.lower()
            sections = [
                s for s in sections 
                if query_lower in s.title.lower() or query_lower in s.current_content.lower()
            ]
        
        if section_type:
            sections = [s for s in sections if s.section_type == section_type]
        
        if tags:
            sections = [
                s for s in sections 
                if any(tag in s.tags for tag in tags)
            ]
        
        return sections
    
    def get_user_text_sections_with_documents(self, owner: str) -> List[Dict]:
        """Get all text sections for a user with document information"""
        sections = self.get_user_text_sections(owner)
        result = []
        
        for section in sections:
            # Get documents that contain this section
            documents = self.db.get_documents_containing_section(section.id)
            
            section_dict = section.to_dict()
            section_dict['documents'] = [{'id': doc.id, 'title': doc.title} for doc in documents] if documents else []
            section_dict['document_count'] = len(documents) if documents else 0
            
            result.append(section_dict)
        
        return result
    
    def get_unique_user_text_sections(self, owner: str) -> List[Dict]:
        """Get unique text sections for a user (latest versions only) with document information, excluding headers"""
        # Get all sections grouped by title/content similarity to identify duplicates
        all_sections = self.get_user_text_sections_with_documents(owner)
        
        # Filter out header types - only show text/content sections
        filtered_sections = [
            section for section in all_sections 
            if section.get('section_type') in ['text', 'content']
        ]
        
        # Group by title and keep only the most recent
        unique_sections = {}
        for section in filtered_sections:
            title = section['title']
            created_at = datetime.fromisoformat(section['created_at'])
            
            if title not in unique_sections or created_at > datetime.fromisoformat(unique_sections[title]['created_at']):
                unique_sections[title] = section
        
        return list(unique_sections.values())