deploy_tmp/block_validator.py

"""
Block Workspace Validator
Validates AI-generated workspace JSON against Blockly schema using Pydantic.
"""

from pydantic import BaseModel, Field, validator, ValidationError
from typing import List, Dict, Any, Optional, Union
from block_schema import BLOCKLY_SCHEMA, get_block_schema, get_allowed_children


class BlockSpec(BaseModel):
    """Specification for a single block"""
    id: str = Field(..., description="Unique block identifier")
    type: str = Field(..., description="Block type from schema")
    props: Dict[str, Any] = Field(default_factory=dict, description="Block properties")
    
    @validator('type')
    def validate_block_type(cls, v):
        """Ensure block type exists in schema"""
        if v not in BLOCKLY_SCHEMA['blocks']:
            available = list(BLOCKLY_SCHEMA['blocks'].keys())[:10]
            raise ValueError(
                f"Unknown block type: '{v}'. "
                f"Available types include: {', '.join(available)}... "
                f"(Total: {len(BLOCKLY_SCHEMA['blocks'])} types)"
            )
        return v
    
    @validator('id')
    def validate_id_format(cls, v):
        """Ensure ID follows naming convention"""
        if not v or not isinstance(v, str):
            raise ValueError("Block ID must be a non-empty string")
        if ' ' in v:
            raise ValueError(f"Block ID cannot contain spaces: '{v}'")
        return v


class ConnectionSpec(BaseModel):
    """Specification for a block connection"""
    from_path: str = Field(..., description="Source block + input (e.g., 'block1.CONTENT')")
    to_id: str = Field(..., description="Target block ID")
    
    @validator('from_path')
    def validate_from_path(cls, v):
        """Ensure from_path has valid format"""
        if '.' not in v:
            raise ValueError(
                f"Connection from_path must include input name: '{v}'. "
                f"Format: 'block_id.INPUT_NAME' (e.g., 'b1.CONTENT')"
            )
        return v


class WorkspaceSpec(BaseModel):
    """Specification for entire Blockly workspace"""
    blocks: List[BlockSpec] = Field(..., min_items=1, description="List of blocks")
    connections: List[List[str]] = Field(default_factory=list, description="Block connections")
    
    @validator('connections')
    def validate_connection_format(cls, v):
        """Ensure each connection is [from_path, to_id]"""
        for conn in v:
            if not isinstance(conn, list) or len(conn) != 2:
                raise ValueError(
                    f"Invalid connection format: {conn}. "
                    f"Expected: ['block_id.INPUT_NAME', 'target_id']"
                )
        return v


def validate_workspace(workspace: dict, max_blocks: int = 50) -> dict:
    """
    Validate workspace JSON against schema.
    
    Args:
        workspace: Dictionary with 'blocks' and 'connections'
        max_blocks: Maximum allowed blocks (safety limit)
    
    Returns:
        {
            "valid": bool,
            "errors": List[str] or None,
            "warnings": List[str] or None
        }
    """
    errors = []
    warnings = []
    
    try:
        # Parse with Pydantic (basic structure validation)
        spec = WorkspaceSpec(**workspace)
        
        # Block count limit
        if len(spec.blocks) > max_blocks:
            errors.append(
                f"Too many blocks: {len(spec.blocks)} exceeds limit of {max_blocks}. "
                f"Please simplify your request."
            )
        
        # Build block ID set for connection validation
        block_ids = {b.id for b in spec.blocks}
        block_map = {b.id: b for b in spec.blocks}
        
        # Validate each block
        for block in spec.blocks:
            block_schema = get_block_schema(block.type)
            if not block_schema:
                errors.append(f"Block '{block.id}': Type '{block.type}' not in schema")
                continue
            
            # Check required properties
            required_props = block_schema.get('required_props', [])
            for prop in required_props:
                if prop not in block.props:
                    errors.append(
                        f"Block '{block.id}' ({block.type}): Missing required property '{prop}'"
                    )
            
            # Validate property keys
            allowed_props = (
                block_schema.get('required_props', []) +
                block_schema.get('optional_props', [])
            )
            for prop in block.props.keys():
                if prop not in allowed_props:
                    warnings.append(
                        f"Block '{block.id}' ({block.type}): Unknown property '{prop}'. "
                        f"Allowed: {', '.join(allowed_props)}"
                    )
        
        # Validate connections
        for conn in spec.connections:
            from_path = conn[0]
            to_id = conn[1]
            
            # Parse from_path
            if '.' in from_path:
                from_id, input_name = from_path.split('.', 1)
            else:
                errors.append(
                    f"Connection '{from_path}' -> '{to_id}': "
                    f"from_path must include input name (e.g., 'block1.CONTENT')"
                )
                continue
            
            # Check block existence
            if from_id not in block_ids:
                errors.append(
                    f"Connection references non-existent source block: '{from_id}'"
                )
                continue
            
            if to_id not in block_ids:
                errors.append(
                    f"Connection references non-existent target block: '{to_id}'"
                )
                continue
            
            # Validate connection is allowed by schema
            from_block = block_map[from_id]
            to_block = block_map[to_id]
            
            allowed_children = get_allowed_children(from_block.type, input_name)
            
            # Check if connection is valid
            if allowed_children is not None:  # None means "*" (any block)
                if to_block.type not in allowed_children:
                    errors.append(
                        f"Invalid connection: '{from_path}' -> '{to_id}'. "
                        f"Input '{input_name}' of '{from_block.type}' does not accept '{to_block.type}'. "
                        f"Allowed: {', '.join(allowed_children)}"
                    )
        
        # Validate workspace structure
        # Check for html_document if present
        html_docs = [b for b in spec.blocks if b.type == 'html_document']
        if len(html_docs) > 1:
            errors.append(
                "Multiple 'html_document' blocks found. Only one root document is allowed."
            )
        
        return {
            "valid": len(errors) == 0,
            "errors": errors if errors else None,
            "warnings": warnings if warnings else None
        }
    
    except ValidationError as e:
        # Pydantic validation errors
        pydantic_errors = []
        for error in e.errors():
            field = ' -> '.join(str(loc) for loc in error['loc'])
            msg = error['msg']
            pydantic_errors.append(f"{field}: {msg}")
        
        return {
            "valid": False,
            "errors": pydantic_errors,
            "warnings": None
        }
    
    except Exception as e:
        return {
            "valid": False,
            "errors": [f"Validation error: {str(e)}"],
            "warnings": None
        }


def format_validation_errors(validation_result: dict) -> str:
    """
    Format validation errors for AI retry prompt.
    
    Returns a clear, actionable error message for the AI to correct.
    """
    if validation_result["valid"]:
        return "Validation passed."
    
    errors = validation_result.get("errors", [])
    warnings = validation_result.get("warnings", [])
    
    message = "VALIDATION FAILED. Please fix:\n\n"
    
    if errors:
        message += "ERRORS:\n"
        for i, error in enumerate(errors, 1):
            message += f"{i}. {error}\n"
    
    if warnings:
        message += "\nWARNINGS:\n"
        for i, warning in enumerate(warnings, 1):
            message += f"{i}. {warning}\n"
    
    message += "\nPlease output corrected JSON only."
    
    return message