#!/usr/bin/env python3 """ Universal YAML Validator for Activity Configurations This module provides comprehensive validation for activity YAML files, particularly battleship configurations and other interactive activities. It validates structure, syntax, Python code blocks, and logical consistency. """ import yaml import ast import re import sys import argparse from typing import Dict, List, Any, Optional, Tuple from pathlib import Path class ValidationError(Exception): """Custom exception for validation errors""" pass class ActivityYAMLValidator: """ Comprehensive validator for activity YAML configurations Validates: - YAML syntax and structure - Required fields and schema compliance - Python code blocks (processing_script, pre_script) - Logic flow and transitions - Battleship-specific rules - Token limits and AI prompt structures """ def __init__(self): self.errors = [] self.warnings = [] self.current_file = None def validate_file(self, file_path: str) -> Tuple[bool, List[str], List[str]]: """ Validate a YAML file and return results Returns: Tuple of (is_valid, errors, warnings) """ self.errors = [] self.warnings = [] self.current_file = file_path try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # Parse YAML try: data = yaml.safe_load(content) except yaml.YAMLError as e: self.errors.append(f"YAML syntax error: {e}") return False, self.errors, self.warnings # Validate structure self._validate_structure(data) # Validate sections if 'sections' in data: self._validate_sections(data['sections']) # Validate universal activity rules self._validate_activity_rules(data) # Validate Python code blocks self._validate_python_code(data) # Validate logic flow self._validate_logic_flow(data) return len(self.errors) == 0, self.errors, self.warnings except Exception as e: self.errors.append(f"Unexpected error: {e}") return False, self.errors, self.warnings def _validate_structure(self, data: Dict[str, Any]): """Validate basic YAML structure""" if not isinstance(data, dict): self.errors.append("Root level must be a dictionary") return # Check required top-level fields required_fields = ['sections'] for field in required_fields: if field not in data: self.errors.append(f"Missing required field: {field}") # Validate optional fields if 'default_max_attempts_per_step' in data: if not isinstance(data['default_max_attempts_per_step'], int) or data['default_max_attempts_per_step'] < 1: self.errors.append("default_max_attempts_per_step must be a positive integer") if 'tokens_for_ai_rubric' in data: if not isinstance(data['tokens_for_ai_rubric'], str): self.errors.append("tokens_for_ai_rubric must be a string") def _validate_sections(self, sections: List[Dict[str, Any]]): """Validate sections structure""" if not isinstance(sections, list): self.errors.append("sections must be a list") return if not sections: self.errors.append("At least one section is required") return section_ids = set() for i, section in enumerate(sections): if not isinstance(section, dict): self.errors.append(f"Section {i} must be a dictionary") continue # Validate section structure self._validate_section(section, i) # Check for duplicate section IDs if 'section_id' in section: if section['section_id'] in section_ids: self.errors.append(f"Duplicate section_id: {section['section_id']}") section_ids.add(section['section_id']) def _validate_section(self, section: Dict[str, Any], section_index: int): """Validate individual section""" required_fields = ['section_id', 'title', 'steps'] for field in required_fields: if field not in section: self.errors.append(f"Section {section_index}: Missing required field '{field}'") if 'steps' in section: self._validate_steps(section['steps'], section.get('section_id', f'section_{section_index}')) def _validate_steps(self, steps: List[Dict[str, Any]], section_id: str): """Validate steps within a section""" if not isinstance(steps, list): self.errors.append(f"Section {section_id}: steps must be a list") return if not steps: self.errors.append(f"Section {section_id}: At least one step is required") return step_ids = set() for i, step in enumerate(steps): if not isinstance(step, dict): self.errors.append(f"Section {section_id}, step {i}: Must be a dictionary") continue self._validate_step(step, section_id, i) # Check for duplicate step IDs if 'step_id' in step: if step['step_id'] in step_ids: self.errors.append(f"Section {section_id}: Duplicate step_id '{step['step_id']}'") step_ids.add(step['step_id']) def _validate_step(self, step: Dict[str, Any], section_id: str, step_index: int): """Validate individual step""" step_id = step.get('step_id', f'step_{step_index}') # Required fields required_fields = ['step_id', 'title'] for field in required_fields: if field not in step: self.errors.append(f"Section {section_id}, step {step_id}: Missing required field '{field}'") # Validate content_blocks or question has_content = 'content_blocks' in step has_question = 'question' in step if not has_content and not has_question: self.errors.append(f"Section {section_id}, step {step_id}: Must have either 'content_blocks' or 'question'") if has_content: self._validate_content_blocks(step['content_blocks'], section_id, step_id) if has_question: self._validate_question_step(step, section_id, step_id) def _validate_content_blocks(self, content_blocks: List[str], section_id: str, step_id: str): """Validate content blocks""" if not isinstance(content_blocks, list): self.errors.append(f"Section {section_id}, step {step_id}: content_blocks must be a list") return for i, block in enumerate(content_blocks): if not isinstance(block, str): self.errors.append(f"Section {section_id}, step {step_id}: content_blocks[{i}] must be a string") def _validate_question_step(self, step: Dict[str, Any], section_id: str, step_id: str): """Validate question-type step""" if 'question' in step and not isinstance(step['question'], str): self.errors.append(f"Section {section_id}, step {step_id}: 'question' must be a string") # Validate AI tokens if 'tokens_for_ai' in step: if not isinstance(step['tokens_for_ai'], str): self.errors.append(f"Section {section_id}, step {step_id}: 'tokens_for_ai' must be a string") if 'feedback_tokens_for_ai' in step: if not isinstance(step['feedback_tokens_for_ai'], str): self.errors.append(f"Section {section_id}, step {step_id}: 'feedback_tokens_for_ai' must be a string") # Validate buckets and transitions if 'buckets' in step: self._validate_buckets(step['buckets'], section_id, step_id) if 'transitions' in step: self._validate_transitions(step['transitions'], step.get('buckets', []), section_id, step_id) def _validate_buckets(self, buckets: List[str], section_id: str, step_id: str): """Validate buckets list""" if not isinstance(buckets, list): self.errors.append(f"Section {section_id}, step {step_id}: 'buckets' must be a list") return if not buckets: self.warnings.append(f"Section {section_id}, step {step_id}: Empty buckets list") return for i, bucket in enumerate(buckets): if not isinstance(bucket, str): self.errors.append(f"Section {section_id}, step {step_id}: buckets[{i}] must be a string") def _validate_transitions(self, transitions: Dict[str, Any], buckets: List[str], section_id: str, step_id: str): """Validate transitions dictionary""" if not isinstance(transitions, dict): self.errors.append(f"Section {section_id}, step {step_id}: 'transitions' must be a dictionary") return # Check that all buckets have corresponding transitions for bucket in buckets: if bucket not in transitions: self.errors.append(f"Section {section_id}, step {step_id}: Missing transition for bucket '{bucket}'") # Check for unused transitions for transition_key in transitions: if transition_key not in buckets: self.warnings.append(f"Section {section_id}, step {step_id}: Unused transition '{transition_key}'") # Validate each transition for bucket, transition in transitions.items(): self._validate_transition(transition, bucket, section_id, step_id) def _validate_transition(self, transition: Dict[str, Any], bucket: str, section_id: str, step_id: str): """Validate individual transition""" if not isinstance(transition, dict): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: Transition must be a dictionary") return # Validate next_section_and_step format if 'next_section_and_step' in transition: next_step = transition['next_section_and_step'] if not isinstance(next_step, str): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be a string") elif ':' not in next_step: self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be in format 'section_id:step_id'") # Validate metadata operations metadata_fields = ['metadata_add', 'metadata_tmp_add', 'metadata_remove', 'metadata_clear', 'metadata_feedback_filter'] for field in metadata_fields: if field in transition: if field == 'metadata_clear': if not isinstance(transition[field], bool): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be boolean") elif field == 'metadata_feedback_filter': if not isinstance(transition[field], list): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a list") else: for item in transition[field]: if not isinstance(item, str): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' items must be strings") elif field == 'metadata_remove': if isinstance(transition[field], str): # Single key to remove pass elif isinstance(transition[field], list): # List of keys to remove for item in transition[field]: if not isinstance(item, str): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' list items must be strings") else: self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a string or list of strings") else: if not isinstance(transition[field], dict): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a dictionary") # Validate other transition fields if 'run_processing_script' in transition: if not isinstance(transition['run_processing_script'], bool): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'run_processing_script' must be boolean") if 'ai_feedback' in transition: ai_feedback = transition['ai_feedback'] if not isinstance(ai_feedback, dict): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'ai_feedback' must be a dictionary") elif 'tokens_for_ai' in ai_feedback and not isinstance(ai_feedback['tokens_for_ai'], str): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: ai_feedback.tokens_for_ai must be a string") if 'content_blocks' in transition: if not isinstance(transition['content_blocks'], list): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'content_blocks' must be a list") else: for i, block in enumerate(transition['content_blocks']): if not isinstance(block, str): self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: content_blocks[{i}] must be a string") def _validate_python_code(self, data: Dict[str, Any]): """Validate Python code blocks in scripts""" def validate_code_block(code: str, location: str): if not code or not isinstance(code, str): return try: # Parse the code to check for syntax errors ast.parse(code) except SyntaxError as e: self.errors.append(f"{location}: Python syntax error - {e}") except Exception as e: self.errors.append(f"{location}: Python parsing error - {e}") # Check for common issues self._check_python_code_quality(code, location) # Recursively find and validate all Python code blocks self._find_and_validate_scripts(data, validate_code_block) def _find_and_validate_scripts(self, obj: Any, validator, path: str = "root"): """Recursively find and validate Python scripts""" if isinstance(obj, dict): for key, value in obj.items(): current_path = f"{path}.{key}" if key in ['processing_script', 'pre_script'] and isinstance(value, str): validator(value, current_path) else: self._find_and_validate_scripts(value, validator, current_path) elif isinstance(obj, list): for i, item in enumerate(obj): self._find_and_validate_scripts(item, validator, f"{path}[{i}]") def _check_python_code_quality(self, code: str, location: str): """Check Python code for common issues and best practices""" lines = code.split('\n') # Check for empty except blocks for i, line in enumerate(lines): stripped = line.strip() if stripped.startswith('except'): # Look for the next non-empty line next_line_idx = i + 1 while next_line_idx < len(lines) and not lines[next_line_idx].strip(): next_line_idx += 1 if next_line_idx < len(lines): next_line = lines[next_line_idx].strip() if next_line == 'pass': self.warnings.append(f"{location} line {i+1}: Empty except block with only 'pass'") # Check for potential security issues dangerous_patterns = [ ('exec(', "Use of exec() can be dangerous"), ('eval(', "Use of eval() can be dangerous"), ('__import__(', "Dynamic imports should be used carefully"), ] for pattern, message in dangerous_patterns: if pattern in code: self.warnings.append(f"{location}: {message}") # Check for proper indentation in else blocks for i, line in enumerate(lines): stripped = line.strip() if stripped == 'else:': # Check if the next non-empty line exists and is properly indented next_line_idx = i + 1 while next_line_idx < len(lines) and not lines[next_line_idx].strip(): next_line_idx += 1 if next_line_idx >= len(lines): self.errors.append(f"{location} line {i+1}: 'else:' block has no content") elif next_line_idx < len(lines): next_line = lines[next_line_idx] if not next_line.strip(): continue # Skip empty lines # Check if it's just a comment if next_line.strip().startswith('#') and next_line_idx + 1 < len(lines): following_line_idx = next_line_idx + 1 while following_line_idx < len(lines) and not lines[following_line_idx].strip(): following_line_idx += 1 if following_line_idx >= len(lines) or lines[following_line_idx].strip().startswith('#'): self.errors.append(f"{location} line {i+1}: 'else:' block contains only comments - add 'pass' statement") def _validate_activity_rules(self, data: Dict[str, Any]): """Validate universal activity rules""" if 'sections' not in data: return # Check that final steps don't have questions for section in data['sections']: if 'steps' not in section: continue steps = section['steps'] if not steps: continue # Find steps that don't have next transitions (terminal steps) terminal_steps = [] for step in steps: if 'transitions' not in step: terminal_steps.append(step) continue has_continuing_transition = False for transition in step['transitions'].values(): if 'next_section_and_step' in transition: has_continuing_transition = True break if not has_continuing_transition: terminal_steps.append(step) # Validate terminal steps for step in terminal_steps: step_id = step.get('step_id', 'unknown') section_id = section.get('section_id', 'unknown') if 'question' in step: self.errors.append(f"Section {section_id}, step {step_id}: Final/terminal steps cannot have questions") if 'buckets' in step and step['buckets']: self.errors.append(f"Section {section_id}, step {step_id}: Final/terminal steps should not have buckets") # Validate metadata_feedback_filter usage self._validate_metadata_filters(data) # Validate pre_script usage self._validate_pre_scripts(data) def _validate_metadata_filters(self, data: Dict[str, Any]): """Validate metadata_feedback_filter usage""" if 'sections' not in data: return for section in data['sections']: if 'steps' not in section: continue section_id = section.get('section_id', 'unknown') for step in section['steps']: step_id = step.get('step_id', 'unknown') if 'transitions' not in step: continue for bucket, transition in step['transitions'].items(): if 'metadata_feedback_filter' in transition: # Check if step has feedback_tokens_for_ai if 'feedback_tokens_for_ai' not in step: self.warnings.append(f"Section {section_id}, step {step_id}: metadata_feedback_filter used but no feedback_tokens_for_ai defined") def _validate_pre_scripts(self, data: Dict[str, Any]): """Validate pre_script usage""" if 'sections' not in data: return for section in data['sections']: if 'steps' not in section: continue section_id = section.get('section_id', 'unknown') for step in section['steps']: step_id = step.get('step_id', 'unknown') if 'pre_script' in step: # Check if step has a question (pre_script should be used with questions) if 'question' not in step: self.warnings.append(f"Section {section_id}, step {step_id}: pre_script typically used with question steps") # Validate pre_script is a string if not isinstance(step['pre_script'], str): self.errors.append(f"Section {section_id}, step {step_id}: pre_script must be a string") def _validate_logic_flow(self, data: Dict[str, Any]): """Validate logical flow and transitions between steps""" if 'sections' not in data: return # Build a map of all available steps all_steps = {} for section in data['sections']: section_id = section.get('section_id') if not section_id or 'steps' not in section: continue for step in section['steps']: step_id = step.get('step_id') if step_id: all_steps[f"{section_id}:{step_id}"] = step # Validate all transition targets for section in data['sections']: section_id = section.get('section_id') if not section_id or 'steps' not in section: continue for step in section['steps']: step_id = step.get('step_id') if not step_id or 'transitions' not in step: continue for bucket, transition in step['transitions'].items(): if 'next_section_and_step' in transition: target = transition['next_section_and_step'] if target not in all_steps: self.errors.append(f"Section {section_id}, step {step_id}: Invalid transition target '{target}'") def main(): """Command line interface for the validator""" parser = argparse.ArgumentParser(description='Validate activity YAML files') parser.add_argument('files', nargs='+', help='YAML files to validate') parser.add_argument('--strict', action='store_true', help='Treat warnings as errors') parser.add_argument('--quiet', action='store_true', help='Only show errors') args = parser.parse_args() validator = ActivityYAMLValidator() total_errors = 0 total_warnings = 0 for file_path in args.files: if not Path(file_path).exists(): print(f"āŒ File not found: {file_path}") total_errors += 1 continue if not args.quiet: print(f"\nšŸ“„ Validating: {file_path}") print("=" * 50) is_valid, errors, warnings = validator.validate_file(file_path) if errors: print(f"āŒ {len(errors)} error(s):") for error in errors: print(f" ā€¢ {error}") total_errors += len(errors) if warnings and not args.quiet: print(f"āš ļø {len(warnings)} warning(s):") for warning in warnings: print(f" ā€¢ {warning}") total_warnings += len(warnings) if is_valid and not warnings: print(f"āœ… {file_path} is valid!") elif is_valid: print(f"āœ… {file_path} is valid (with warnings)") else: print(f"āŒ {file_path} has errors") # Summary if not args.quiet: print(f"\nšŸ“Š Summary:") print(f" Files checked: {len(args.files)}") print(f" Errors: {total_errors}") print(f" Warnings: {total_warnings}") # Exit code exit_code = 0 if total_errors > 0: exit_code = 1 elif args.strict and total_warnings > 0: exit_code = 1 sys.exit(exit_code) if __name__ == '__main__': main()