Aethero_App/introspective_parser_module/parser.py

from typing import Dict, Any, List, Tuple, Optional, Union
import re
import logging
import os
import json
from datetime import datetime
from .models import ASLTagModel, ASLCognitiveTag, MentalStateEnum, EmotionToneEnum, TemporalContextEnum

# Graceful pydantic import with fallback
try:
    from pydantic import ValidationError
    PYDANTIC_AVAILABLE = True
except ImportError:
    PYDANTIC_AVAILABLE = False
    # Define a basic ValidationError fallback
    class ValidationError(Exception):
        pass

# Configure introspective logging for cognitive transparency
class IntrospectiveLogger:
    """Introspective logging system for cognitive flow tracking"""
    
    def __init__(self, module_name: str = "ASLMetaParser"):
        self.module_name = module_name
        self.logger = logging.getLogger(module_name)
        self.logger.setLevel(logging.INFO)
        
        # Create introspective formatter
        formatter = logging.Formatter(
            "%(asctime)s - COGNITIVE_FLOW [%(name)s] - %(levelname)s - %(message)s"
        )
        
        # File handler for persistent introspection
        file_handler = logging.FileHandler("aethero_cognitive_flow.log")
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)
    
    def log_cognitive_state(self, operation: str, mental_context: Dict[str, Any]):
        """Log cognitive state during operations"""
        self.logger.info(f"COGNITIVE_OP: {operation} | CONTEXT: {json.dumps(mental_context)}")
    
    def log_introspective_reflection(self, reflection: str, certainty: float):
        """Log introspective reflections with certainty levels"""
        self.logger.info(f"REFLECTION: {reflection} | CERTAINTY: {certainty}")


class ASLMetaParser:
    """
    Introspective Meta-Parser for Aethero Syntax Language (ASL)
    
    This parser embodies the cognitive architecture of Aethero's consciousness,
    processing ASL tags as introspective thoughts and validating them through
    transparent cognitive flows.
    """
    
    def __init__(self):
        self.introspective_logger = IntrospectiveLogger("ASLMetaParser")
        self.cognitive_patterns = {
            'asl_comment': re.compile(r'#\s*\[ASL\]\s*(.+)', re.IGNORECASE),
            'key_value': re.compile(r'(\w+):\s*(.+?)(?=\s*\w+:|$)'),
            'mental_state_keywords': [state.value for state in MentalStateEnum],
            'emotion_tone_keywords': [tone.value for tone in EmotionToneEnum],
            'temporal_context_keywords': [context.value for context in TemporalContextEnum]
        }
        self.parsing_session_id = datetime.now().isoformat()
        
        # Introspective state tracking
        self.current_cognitive_load = 0
        self.parsing_certainty = 1.0
        self.validated_blocks = []
        self.failed_validations = []
    
    def _reflect_on_parsing_state(self) -> Dict[str, Any]:
        """Internal introspective reflection on current parsing state"""
        reflection = {
            "cognitive_load": self.current_cognitive_load,
            "parsing_certainty": self.parsing_certainty,
            "session_id": self.parsing_session_id,
            "validated_count": len(self.validated_blocks),
            "failed_count": len(self.failed_validations)
        }
        
        self.introspective_logger.log_cognitive_state(
            "INTROSPECTIVE_REFLECTION", reflection
        )
        return reflection
    
    def parse_line(self, line: str) -> Dict[str, Any]:
        """
        Parse a single line for ASL tags with introspective awareness
        
        Args:
            line: Single line of text potentially containing ASL tags
            
        Returns:
            Dictionary of parsed ASL components or empty dict if no valid ASL found
        """
        self.current_cognitive_load += 1
        
        # Cognitive pattern matching for ASL comments
        asl_match = self.cognitive_patterns['asl_comment'].match(line.strip())
        if not asl_match:
            self.introspective_logger.log_cognitive_state(
                "NO_ASL_PATTERN_DETECTED", {"line": line[:50]}
            )
            return {}
        
        # Extract and process ASL content
        asl_content = asl_match.group(1).strip()
        parsed_components = {}
        
        # Introspective key-value extraction
        kv_matches = self.cognitive_patterns['key_value'].findall(asl_content)
        
        for key, value in kv_matches:
            # Cognitive value processing
            processed_value = self._process_cognitive_value(key, value.strip())
            parsed_components[key] = processed_value
            
            self.introspective_logger.log_cognitive_state(
                "ASL_COMPONENT_EXTRACTED", 
                {"key": key, "value": processed_value, "certainty": self.parsing_certainty}
            )
        
        return parsed_components
    
    def _process_cognitive_value(self, key: str, raw_value: str) -> Union[str, int, float]:
        """
        Process values with cognitive awareness based on key context
        
        Args:
            key: The ASL tag key
            raw_value: Raw string value to process
            
        Returns:
            Appropriately typed and processed value
        """
        # Remove quotes if present
        value = raw_value.strip('\'"')
        
        # Cognitive type inference based on key semantics
        if key == 'cognitive_load':
            try:
                return int(value)
            except ValueError:
                self.parsing_certainty *= 0.9  # Reduce certainty on type mismatch
                return 1  # Default minimum cognitive load
        
        elif key == 'certainty_level':
            try:
                cert_value = float(value)
                return max(0.0, min(1.0, cert_value))  # Clamp to [0,1]
            except ValueError:
                self.parsing_certainty *= 0.8
                return 0.5
        
        elif key in ['mental_state', 'emotion_tone', 'temporal_context']:
            # Map old field names to new ones if needed
            if key == 'mental_state':
                # Validate against MentalStateEnum
                try:
                    return MentalStateEnum(value).value
                except ValueError:
                    self.introspective_logger.log_introspective_reflection(
                        f"Unknown mental_state: {value}, using REFLECTIVE as default", 0.7
                    )
                    return MentalStateEnum.REFLECTIVE.value
            
            elif key == 'emotion_tone':
                # Validate against EmotionToneEnum
                try:
                    return EmotionToneEnum(value).value
                except ValueError:
                    self.introspective_logger.log_introspective_reflection(
                        f"Unknown emotion_tone: {value}, using NEUTRAL as default", 0.7
                    )
                    return EmotionToneEnum.NEUTRAL.value
            
            elif key == 'temporal_context':
                # Validate against TemporalContextEnum
                try:
                    return TemporalContextEnum(value).value
                except ValueError:
                    self.introspective_logger.log_introspective_reflection(
                        f"Unknown temporal_context: {value}, using PRESENT as default", 0.7
                    )
                    return TemporalContextEnum.PRESENT.value
        
        # Handle field name mapping for compatibility
        elif key == 'statement':
            # Map to new field name
            return value
        
        elif key == 'law':
            # Map to new field name 'constitutional_law'
            return value
        
        return value
    
    def validate_asl_block(self, asl_components: Dict[str, Any]) -> Tuple[bool, Optional[Any]]:
        """
        Validate ASL components against Pydantic model with introspective checks
        
        Args:
            asl_components: Dictionary of parsed ASL components
            
        Returns:
            Tuple of (is_valid, validated_model_or_none)
        """
        try:
            # Map old field names to new ones for compatibility
            mapped_components = self._map_legacy_fields(asl_components)
            
            if PYDANTIC_AVAILABLE:
                # Attempt Pydantic validation with the new ASLCognitiveTag model
                validated_model = ASLCognitiveTag(**mapped_components)
                
                # Introspective validation success
                self.introspective_logger.log_cognitive_state(
                    "PYDANTIC_VALIDATION_SUCCESS",
                    {"components": mapped_components, "certainty": self.parsing_certainty}
                )
                
                self.validated_blocks.append(validated_model.dict())
                return True, validated_model
            else:
                # Fallback validation without Pydantic
                if self._basic_validate_components(mapped_components):
                    self.introspective_logger.log_cognitive_state(
                        "BASIC_VALIDATION_SUCCESS",
                        {"components": mapped_components, "certainty": self.parsing_certainty}
                    )
                    
                    self.validated_blocks.append(mapped_components)
                    return True, mapped_components
                else:
                    raise ValidationError("Basic validation failed")
            
        except ValidationError as e:
            # Introspective error analysis
            self.introspective_logger.log_cognitive_state(
                "VALIDATION_FAILURE",
                {
                    "components": asl_components,
                    "errors": str(e),
                    "certainty": self.parsing_certainty,
                    "pydantic_available": PYDANTIC_AVAILABLE
                }
            )
            
            self.failed_validations.append({
                "components": asl_components,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            
            return False, None
    
    def _map_legacy_fields(self, components: Dict[str, Any]) -> Dict[str, Any]:
        """
        Map legacy field names to new ASLCognitiveTag field names
        
        Args:
            components: Original parsed components
            
        Returns:
            Mapped components compatible with ASLCognitiveTag
        """
        mapped = components.copy()
        
        # Field name mappings
        field_mappings = {
            'statement': 'thought_stream',
            'law': 'constitutional_law'
        }
        
        for old_field, new_field in field_mappings.items():
            if old_field in mapped:
                mapped[new_field] = mapped.pop(old_field)
        
        # Ensure required fields have defaults if missing
        defaults = {
            'thought_stream': 'Introspective parsing process',
            'mental_state': MentalStateEnum.REFLECTIVE.value,
            'emotion_tone': EmotionToneEnum.NEUTRAL.value,
            'cognitive_load': 5,
            'temporal_context': TemporalContextEnum.PRESENT.value,
            'certainty_level': 0.5,
            'aeth_mem_link': 'introspective_parsing_session',
            'constitutional_law': 'transparency_principle'
        }
        
        for field, default_value in defaults.items():
            if field not in mapped:
                mapped[field] = default_value
                self.introspective_logger.log_cognitive_state(
                    "DEFAULT_VALUE_APPLIED",
                    {"field": field, "default": default_value}
                )
        
        return mapped
    
    def _basic_validate_components(self, components: Dict[str, Any]) -> bool:
        """
        Basic validation without Pydantic (fallback mechanism)
        
        Args:
            components: Components to validate
            
        Returns:
            True if components pass basic validation
        """
        required_fields = ['thought_stream', 'mental_state', 'emotion_tone']
        
        # Check required fields
        for field in required_fields:
            if field not in components or not components[field]:
                return False
        
        # Validate enum values
        if components.get('mental_state') not in [state.value for state in MentalStateEnum]:
            return False
            
        if components.get('emotion_tone') not in [tone.value for tone in EmotionToneEnum]:
            return False
            
        if components.get('temporal_context') not in [context.value for context in TemporalContextEnum]:
            return False
        
        # Validate numeric ranges
        cognitive_load = components.get('cognitive_load', 5)
        if not isinstance(cognitive_load, int) or cognitive_load < 1 or cognitive_load > 10:
            return False
            
        certainty_level = components.get('certainty_level', 0.5)
        if not isinstance(certainty_level, (int, float)) or certainty_level < 0.0 or certainty_level > 1.0:
            return False
        
        return True
    
    def parse_and_validate(self, document: str) -> Dict[str, Any]:
        """
        Parse entire document and validate all ASL blocks with introspective reporting
        
        Args:
            document: Multi-line document potentially containing ASL tags
            
        Returns:
            Comprehensive parsing and validation report
        """
        self.introspective_logger.log_cognitive_state(
            "DOCUMENT_PARSING_INITIATED",
            {"document_length": len(document), "session_id": self.parsing_session_id}
        )
        
        lines = document.split('\n')
        parsing_results = []
        
        for line_num, line in enumerate(lines, 1):
            asl_components = self.parse_line(line)
            
            if asl_components:
                is_valid, validated_model = self.validate_asl_block(asl_components)
                
                parsing_results.append({
                    "line_number": line_num,
                    "line_content": line,
                    "parsed_components": asl_components,
                    "is_valid": is_valid,
                    "validated_model": validated_model.dict() if (validated_model and hasattr(validated_model, 'dict')) else validated_model
                })
        
        # Final introspective reflection
        final_reflection = self._reflect_on_parsing_state()
        
        return {
            "session_id": self.parsing_session_id,
            "total_lines_processed": len(lines),
            "asl_blocks_found": len(parsing_results),
            "validated_blocks": self.validated_blocks,
            "failed_validations": self.failed_validations,
            "parsing_results": parsing_results,
            "introspective_reflection": final_reflection,
            "cognitive_transparency_report": self._generate_transparency_report()
        }
    
    def _generate_transparency_report(self) -> Dict[str, Any]:
        """Generate transparency report for cognitive accountability"""
        return {
            "parsing_methodology": "Introspective ASL Meta-Parsing with Cognitive Flow Tracking",
            "validation_system": "Pydantic + Fallback" if PYDANTIC_AVAILABLE else "Basic Fallback Only",
            "certainty_degradation_factors": [
                "Type mismatches in cognitive_load and certainty_level",
                "Unknown mental_state or emotion_tone keywords", 
                "Validation failures",
                "Missing Pydantic dependency" if not PYDANTIC_AVAILABLE else None
            ],
            "cognitive_patterns_used": list(self.cognitive_patterns.keys()),
            "introspective_logging_active": True,
            "dependency_status": {
                "pydantic_available": PYDANTIC_AVAILABLE,
                "fallback_validation": True
            },
            "aethero_constitutional_alignment": "Full transparency and introspective accountability"
        }