message_adapter.py

from typing import List, Optional, Dict, Any, Union
from models import Message
import re


class MessageAdapter:
    """Converts between OpenAI message format and Claude Code prompts."""
    
    @staticmethod
    def has_structured_format(content: str) -> bool:
        """
        Detect if content has structured format (XML, JSON, etc).
        Used to determine if content should be preserved as-is.
        """
        if not content or len(content) < 10:
            return False
            
        # Check for XML-like patterns (opening and closing tags)
        import re
        xml_pattern = r'<([a-zA-Z_][\w\-\.]*)(\s[^>]*)?>.*?</\1>'
        if re.search(xml_pattern, content, re.DOTALL):
            return True
            
        # Check for JSON-like patterns
        content_stripped = content.strip()
        if (content_stripped.startswith('{') and content_stripped.endswith('}')) or \
           (content_stripped.startswith('[') and content_stripped.endswith(']')):
            try:
                import json
                json.loads(content_stripped)
                return True
            except:
                pass
                
        # Check for structured format indicators
        structured_indicators = [
            '```',  # Code blocks
            '<?xml',  # XML declaration
            '<!DOCTYPE',  # HTML/XML doctype
        ]
        
        for indicator in structured_indicators:
            if indicator in content:
                return True
                
        return False
    
    @staticmethod
    def messages_to_prompt(messages: List[Message]) -> tuple[str, Optional[str]]:
        """
        Convert OpenAI messages to Claude Code prompt format.
        Returns (prompt, system_prompt)
        """
        system_prompt = None
        conversation_parts = []
        
        for message in messages:
            if message.role == "system":
                # Use the last system message as the system prompt
                system_prompt = message.content
            elif message.role == "user":
                conversation_parts.append(f"Human: {message.content}")
            elif message.role == "assistant":
                conversation_parts.append(f"Assistant: {message.content}")
        
        # Join conversation parts
        prompt = "\n\n".join(conversation_parts)
        
        # If the last message wasn't from the user, add a prompt for assistant
        if messages and messages[-1].role != "user":
            prompt += "\n\nHuman: Please continue."
            
        return prompt, system_prompt
    
    @staticmethod
    def messages_to_prompt_with_images(
        messages: List[Message], 
        image_mappings: Dict[str, str]
    ) -> tuple[str, Optional[str], List[str]]:
        """
        Convert OpenAI messages to Claude Code prompt format with image references.
        
        Args:
            messages: List of Message objects
            image_mappings: Dictionary mapping image URLs to local file paths
            
        Returns:
            Tuple of (prompt, system_prompt, list_of_image_paths)
        """
        system_prompt = None
        conversation_parts = []
        referenced_images = []
        
        for message in messages:
            if message.role == "system":
                # Use the last system message as the system prompt
                system_prompt = message.content
            else:
                # Process the content to include image references
                content = MessageAdapter._process_content_with_images(
                    message.content, 
                    image_mappings, 
                    referenced_images
                )
                
                if message.role == "user":
                    conversation_parts.append(f"Human: {content}")
                elif message.role == "assistant":
                    conversation_parts.append(f"Assistant: {content}")
        
        # Join conversation parts
        prompt = "\n\n".join(conversation_parts)
        
        # If the last message wasn't from the user, add a prompt for assistant
        if messages and messages[-1].role != "user":
            prompt += "\n\nHuman: Please continue."
        
        # Get unique image paths
        unique_image_paths = list(set(referenced_images))
        
        return prompt, system_prompt, unique_image_paths
    
    @staticmethod
    def _process_content_with_images(
        content: Union[str, List[Any]], 
        image_mappings: Dict[str, str],
        referenced_images: List[str]
    ) -> str:
        """
        Process message content to include image file references.
        
        Args:
            content: Message content (string or list of content parts)
            image_mappings: URL to file path mappings
            referenced_images: List to append referenced image paths to
            
        Returns:
            Processed content string with image references
        """
        if isinstance(content, str):
            return content
        
        # Process array content
        text_parts = []
        
        for part in content:
            if isinstance(part, dict):
                if part.get('type') == 'text':
                    text_parts.append(part.get('text', ''))
                elif part.get('type') == 'image_url':
                    # Get the image URL
                    image_url = part.get('image_url', {}).get('url', '')
                    
                    # Look up the local file path
                    if image_url in image_mappings:
                        file_path = image_mappings[image_url]
                        # Add reference to the image in the text
                        text_parts.append(f"[Image: {file_path}]")
                        referenced_images.append(file_path)
                    else:
                        # Image wasn't processed (maybe failed to download)
                        text_parts.append("[Image: Failed to process]")
        
        # Join text parts with spaces
        return " ".join(text_parts) if text_parts else ""
    
    @staticmethod
    def filter_content(content: str) -> str:
        """
        Filter content for unsupported features and tool usage.
        Remove thinking blocks, tool calls, and image references.
        """
        if not content:
            return content
            
        # Remove thinking blocks (common when tools are disabled but Claude tries to think)
        thinking_pattern = r'<thinking>.*?</thinking>'
        content = re.sub(thinking_pattern, '', content, flags=re.DOTALL)
        
        # Extract content from attempt_completion blocks (these contain the actual user response)
        attempt_completion_pattern = r'<attempt_completion>(.*?)</attempt_completion>'
        attempt_matches = re.findall(attempt_completion_pattern, content, flags=re.DOTALL)
        if attempt_matches:
            # Use the content from the attempt_completion block
            extracted_content = attempt_matches[0].strip()
            
            # If there's a <result> tag inside, extract from that
            result_pattern = r'<result>(.*?)</result>'
            result_matches = re.findall(result_pattern, extracted_content, flags=re.DOTALL)
            if result_matches:
                extracted_content = result_matches[0].strip()
            
            if extracted_content:
                content = extracted_content
        else:
            # Remove other tool usage blocks (when tools are disabled but Claude tries to use them)
            tool_patterns = [
                r'<read_file>.*?</read_file>',
                r'<write_file>.*?</write_file>',
                r'<bash>.*?</bash>',
                r'<search_files>.*?</search_files>',
                r'<str_replace_editor>.*?</str_replace_editor>',
                r'<args>.*?</args>',
                r'<ask_followup_question>.*?</ask_followup_question>',
                r'<attempt_completion>.*?</attempt_completion>',
                r'<question>.*?</question>',
                r'<follow_up>.*?</follow_up>',
                r'<suggest>.*?</suggest>',
            ]
            
            for pattern in tool_patterns:
                content = re.sub(pattern, '', content, flags=re.DOTALL)
        
        # No longer filter out images - they are now supported!
        # Images are processed separately by the ImageHandler
        
        # Clean up extra whitespace and newlines
        content = re.sub(r'\n\s*\n\s*\n', '\n\n', content)  # Multiple newlines to double
        content = content.strip()
        
        # If content is now empty or only whitespace, provide a fallback
        if not content or content.isspace():
            return "I understand you're testing the system. How can I help you today?"
            
        return content
    
    @staticmethod
    def format_claude_response(content: str, model: str, finish_reason: str = "stop") -> Dict[str, Any]:
        """Format Claude response for OpenAI compatibility."""
        return {
            "role": "assistant",
            "content": content,
            "finish_reason": finish_reason,
            "model": model
        }
    
    @staticmethod
    def estimate_tokens(text: str) -> int:
        """
        Rough estimation of token count.
        OpenAI's rule of thumb: ~4 characters per token for English text.
        """
        return len(text) // 4
    
    @staticmethod
    def validate_xml_tool_response(content: str) -> bool:
        """Check if content contains valid XML tool tags."""
        import re
        
        # Check for ANY XML tool-like tags (not just specific ones)
        # This regex looks for opening XML tags that look like tools
        # (lowercase words, possibly with underscores)
        tool_pattern = r'<([a-z][a-z_]*[a-z]|[a-z])>'
        
        # Also check for common tool tags from configuration
        from xml_tools_config import get_known_xml_tools
        
        content_lower = content.lower()
        known_tools = get_known_xml_tools()
        
        # Check if any known tools are present (wrap tool names in angle brackets)
        has_known_tool = any(f"<{tool}>" in content_lower for tool in known_tools)
        
        # Or check if there's any XML-like tool structure
        has_xml_structure = bool(re.search(tool_pattern, content))
        
        return has_known_tool or has_xml_structure