#!/usr/bin/env python3 """ Validate mappings between Ferrero and Creative X systems Usage: python validate_mappings.py python validate_mappings.py --test-file /path/to/test_file.mp4 python validate_mappings.py --show-supported """ import argparse import sys from pathlib import Path from typing import Dict, Any # Add parent directory to path sys.path.insert(0, str(Path(__file__).parent.parent)) from config import load_config from core.filename_parser import FerreroFilenameParser from core.data_loader import DataLoader from core.mapping_resolver import MappingResolver from utils.file_handler import FileHandler def print_section(title: str): """Print formatted section header""" print("\n" + "=" * 70) print(f" {title}") print("=" * 70) def print_subsection(title: str): """Print formatted subsection header""" print(f"\n{title}") print("-" * 70) def show_supported_values(resolver: MappingResolver): """Display all supported brands and channels""" print_section("SUPPORTED VALUES IN CREATIVE X") # Brands print_subsection("Brands") brands = resolver.get_all_supported_brands() if brands: for code, name in brands.items(): print(f" {code:10s} → {name}") else: print(" (none configured)") print(f"\n Total: {len(brands)} brands") # Channels print_subsection("Social Media Channels") channels = resolver.get_all_supported_channels() # Group by channel channel_groups = {} for code, channel in channels.items(): if channel not in channel_groups: channel_groups[channel] = [] channel_groups[channel].append(code) for channel, codes in sorted(channel_groups.items()): print(f"\n {channel}:") for code in sorted(codes): mapping = resolver.get_creativex_channel_mapping(code) details = [] if mapping.get('publisher'): details.append(f"publisher={mapping['publisher']}") if mapping.get('placement'): details.append(f"placement={mapping['placement']}") if mapping.get('ad_format'): details.append(f"ad_format={mapping['ad_format']}") detail_str = f" ({', '.join(details)})" if details else "" print(f" {code:10s} → {mapping.get('ferrero_name', '')}{detail_str}") print(f"\n Total: {len(channels)} social media codes") def validate_filename(filename: str, parser: FerreroFilenameParser, resolver: MappingResolver) -> Dict[str, Any]: """ Validate a filename through complete mapping process Returns: dict: Validation results """ results = { 'filename': filename, 'parsed': False, 'parse_errors': [], 'mapped': False, 'mapping_errors': [], 'parsed_data': None, 'creativex_payload': None } # Step 1: Parse filename print_subsection(f"Parsing: {filename}") try: parsed_data = parser.parse(filename) results['parsed'] = True results['parsed_data'] = parsed_data print(f" ✓ Parsed successfully") print(f" Brand: {parsed_data.get('brand_code')} → {parsed_data.get('brand_name')}") print(f" Country: {parsed_data.get('country_code')} → {parsed_data.get('country_name')}") print(f" Language: {parsed_data.get('language_code')}") print(f" Channel: {parsed_data.get('social_media')} → {parsed_data.get('channel')}") print(f" Subject: {parsed_data.get('subject')}") print(f" Duration: {parsed_data.get('seconds')}") print(f" Ratio: {parsed_data.get('aspect_ratio')}") except Exception as e: results['parse_errors'].append(str(e)) print(f" ✗ Parse failed: {e}") return results # Step 2: Validate mappings print_subsection("Validating Creative X Mappings") is_valid, errors = resolver.validate_metadata_for_upload(parsed_data) if is_valid: print(f" ✓ All mappings valid") results['mapped'] = True else: results['mapping_errors'] = errors for error in errors: print(f" ✗ {error}") return results # Step 3: Build Creative X payload print_subsection("Creative X Payload") try: payload = resolver.build_creativex_payload(parsed_data) results['creativex_payload'] = payload print(f" brand_name: {payload.get('brand_name')}") print(f" channel: {payload.get('channel')}") if payload.get('publisher'): print(f" publisher: {payload.get('publisher')}") if payload.get('placement'): print(f" placement: {payload.get('placement')}") if payload.get('ad_format'): print(f" ad_format: {payload.get('ad_format')}") if payload.get('market_name'): print(f" market_name: {payload.get('market_name')}") if payload.get('language'): print(f" language: {payload.get('language')}") if payload.get('dimensions'): print(f" dimensions: {payload.get('dimensions')}") if payload.get('duration'): print(f" duration: {payload.get('duration')}") if payload.get('subject'): print(f" subject: {payload.get('subject')}") print(f"\n ✓ Ready for upload to Creative X") except Exception as e: results['mapping_errors'].append(str(e)) print(f" ✗ Payload build failed: {e}") return results def validate_test_file(file_path: str, parser: FerreroFilenameParser, resolver: MappingResolver): """Validate a test file""" print_section(f"VALIDATING TEST FILE") filename = FileHandler.get_filename(file_path) # Check file exists if not FileHandler.file_exists(file_path): print(f"\n✗ File not found: {file_path}") return False # Validate results = validate_filename(filename, parser, resolver) # Summary print_subsection("Summary") success = results['parsed'] and results['mapped'] if success: print(f" ✓ File is valid and ready for upload") print(f" ✓ Filename parsed successfully") print(f" ✓ All Creative X mappings valid") return True else: print(f" ✗ File validation failed") if not results['parsed']: print(f"\n Parse Errors:") for error in results['parse_errors']: print(f" - {error}") if not results['mapped'] and results['parsed']: print(f"\n Mapping Errors:") for error in results['mapping_errors']: print(f" - {error}") return False def validate_example_filenames(parser: FerreroFilenameParser, resolver: MappingResolver): """Validate example filenames""" print_section("VALIDATING EXAMPLE FILENAMES") examples = [ "1234567_NUT_MOMENT_OLV_6S_1x1_REF_GL_en_FBS_abcdef.mp4", "2345678_RAF_LUXURY_OLV_15S_16x9_MST_IT_it_IGF_xyz123.mp4", "3456789_NUT_BREAKFAST_OLV_10S_9x16_REF_DE_de_YTS_qwerty.mp4", ] results = [] for filename in examples: result = validate_filename(filename, parser, resolver) results.append(result) # Summary print_section("VALIDATION SUMMARY") total = len(results) passed = sum(1 for r in results if r['parsed'] and r['mapped']) failed = total - passed print(f"\n Total: {total}") print(f" Passed: {passed} ✓") print(f" Failed: {failed} ✗") if failed > 0: print("\n Failed files:") for result in results: if not (result['parsed'] and result['mapped']): print(f" - {result['filename']}") if result['parse_errors']: for error in result['parse_errors']: print(f" Parse: {error}") if result['mapping_errors']: for error in result['mapping_errors']: print(f" Mapping: {error}") return failed == 0 def main(): """CLI entry point""" parser = argparse.ArgumentParser( description='Validate mappings between Ferrero and Creative X' ) parser.add_argument('--test-file', help='Test a specific file') parser.add_argument('--show-supported', action='store_true', help='Show all supported brands and channels') parser.add_argument('--test-examples', action='store_true', help='Test example filenames') args = parser.parse_args() # Load configuration try: config = load_config() except Exception as e: print(f"Error loading configuration: {e}") sys.exit(1) # Initialize components try: data_loader = DataLoader(str(config.data_json_path)) mappings_path = config.project_root / 'mappings.json' mapping_resolver = MappingResolver(str(mappings_path)) filename_parser = FerreroFilenameParser(data_loader) except Exception as e: print(f"Error initializing components: {e}") sys.exit(1) print_section("CREATIVE X MAPPING VALIDATOR") print(f"\nConfiguration:") print(f" Data file: {config.data_json_path}") print(f" Mappings file: {mappings_path}") success = True # Show supported values if args.show_supported or (not args.test_file and not args.test_examples): show_supported_values(mapping_resolver) # Test specific file if args.test_file: file_success = validate_test_file(args.test_file, filename_parser, mapping_resolver) success = success and file_success # Test examples if args.test_examples: examples_success = validate_example_filenames(filename_parser, mapping_resolver) success = success and examples_success print() sys.exit(0 if success else 1) if __name__ == '__main__': main()