#!/usr/bin/env python3 """ Validate demo JSON files to ensure all dates use BASE_TS markers. This script enforces the new architecture requirement that all temporal data in demo fixtures must use BASE_TS markers for deterministic sessions. """ import json import sys from pathlib import Path from typing import Dict, List, Tuple, Set # Date/time fields that should use BASE_TS markers or be null DATE_TIME_FIELDS = { # Common fields 'created_at', 'updated_at', # Procurement 'order_date', 'required_delivery_date', 'estimated_delivery_date', 'expected_delivery_date', 'sent_to_supplier_at', 'supplier_confirmation_date', 'approval_deadline', # Production 'planned_start_time', 'planned_end_time', 'actual_start_time', 'actual_end_time', 'completed_at', 'install_date', 'last_maintenance_date', 'next_maintenance_date', # Inventory 'received_date', 'expiration_date', 'best_before_date', 'original_expiration_date', 'transformation_date', 'final_expiration_date', # Orders 'order_date', 'delivery_date', 'promised_date', 'last_order_date', # Forecasting 'forecast_date', 'prediction_date', # Schedules 'schedule_date', 'shift_start', 'shift_end', 'finalized_at', # Quality 'check_time', # Generic 'date', 'start_time', 'end_time' } class ValidationError: """Represents a validation error""" def __init__(self, file_path: Path, entity_type: str, entity_index: int, field_name: str, value: any, message: str): self.file_path = file_path self.entity_type = entity_type self.entity_index = entity_index self.field_name = field_name self.value = value self.message = message def __str__(self): return ( f"❌ {self.file_path.name} » {self.entity_type}[{self.entity_index}] » " f"{self.field_name}: {self.message}\n" f" Value: {self.value}" ) def validate_date_value(value: any, field_name: str) -> Tuple[bool, str]: """ Validate a single date field value. Returns: (is_valid, error_message) """ # Null values are allowed if value is None: return True, "" # BASE_TS markers are the expected format if isinstance(value, str) and value.startswith("BASE_TS"): # Validate BASE_TS marker format if value == "BASE_TS": return True, "" # Should be "BASE_TS + ..." or "BASE_TS - ..." parts = value.split() if len(parts) < 3: return False, f"Invalid BASE_TS marker format (expected 'BASE_TS +/- ')" if parts[1] not in ['+', '-']: return False, f"Invalid BASE_TS operator (expected + or -)" # Extract offset parts (starting from index 2) offset_parts = ' '.join(parts[2:]) # Validate offset components (must contain d, h, or m) if not any(c in offset_parts for c in ['d', 'h', 'm']): return False, f"BASE_TS offset must contain at least one of: d (days), h (hours), m (minutes)" return True, "" # ISO 8601 timestamps are NOT allowed (should use BASE_TS) if isinstance(value, str) and ('T' in value or 'Z' in value): return False, "Found ISO 8601 timestamp - should use BASE_TS marker instead" # offset_days dictionaries are NOT allowed (legacy format) if isinstance(value, dict) and 'offset_days' in value: return False, "Found offset_days dictionary - should use BASE_TS marker instead" # Unknown format return False, f"Unknown date format (type: {type(value).__name__})" def validate_entity(entity: Dict, entity_type: str, entity_index: int, file_path: Path) -> List[ValidationError]: """ Validate all date fields in a single entity. Returns: List of validation errors """ errors = [] # Check for legacy offset_days fields for key in entity.keys(): if key.endswith('_offset_days'): base_field = key.replace('_offset_days', '') errors.append(ValidationError( file_path, entity_type, entity_index, key, entity[key], f"Legacy offset_days field found - migrate to BASE_TS marker in '{base_field}' field" )) # Validate date/time fields for field_name, value in entity.items(): if field_name in DATE_TIME_FIELDS: is_valid, error_msg = validate_date_value(value, field_name) if not is_valid: errors.append(ValidationError( file_path, entity_type, entity_index, field_name, value, error_msg )) return errors def validate_json_file(file_path: Path) -> List[ValidationError]: """ Validate all entities in a JSON file. Returns: List of validation errors """ try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) except json.JSONDecodeError as e: return [ValidationError( file_path, "FILE", 0, "JSON", None, f"Invalid JSON: {e}" )] except Exception as e: return [ValidationError( file_path, "FILE", 0, "READ", None, f"Failed to read file: {e}" )] errors = [] # Validate each entity type for entity_type, entities in data.items(): if isinstance(entities, list): for i, entity in enumerate(entities): if isinstance(entity, dict): errors.extend( validate_entity(entity, entity_type, i, file_path) ) return errors def main(): """Main validation function""" # Find all JSON files in demo fixtures root_dir = Path(__file__).parent.parent fixtures_dir = root_dir / "shared" / "demo" / "fixtures" if not fixtures_dir.exists(): print(f"❌ Fixtures directory not found: {fixtures_dir}") return 1 # Find all JSON files json_files = sorted(fixtures_dir.rglob("*.json")) if not json_files: print(f"❌ No JSON files found in {fixtures_dir}") return 1 print(f"🔍 Validating {len(json_files)} JSON files...\n") # Validate each file all_errors = [] files_with_errors = 0 for json_file in json_files: errors = validate_json_file(json_file) if errors: files_with_errors += 1 all_errors.extend(errors) # Print file header relative_path = json_file.relative_to(fixtures_dir) print(f"\n📄 {relative_path}") print(f" Found {len(errors)} error(s):") # Print each error for error in errors: print(f" {error}") # Print summary print("\n" + "=" * 80) if all_errors: print(f"\n❌ VALIDATION FAILED") print(f" Total errors: {len(all_errors)}") print(f" Files with errors: {files_with_errors}/{len(json_files)}") print(f"\n💡 Fix these errors by:") print(f" 1. Replacing ISO timestamps with BASE_TS markers") print(f" 2. Removing *_offset_days fields") print(f" 3. Using format: 'BASE_TS +/- ' where offset uses d/h/m") print(f" Examples: 'BASE_TS', 'BASE_TS + 2d', 'BASE_TS - 4h', 'BASE_TS + 1h30m'") return 1 else: print(f"\n✅ ALL VALIDATIONS PASSED") print(f" Files validated: {len(json_files)}") print(f" All date fields use BASE_TS markers correctly") return 0 if __name__ == "__main__": sys.exit(main())