""" Mapping resolver between Ferrero naming conventions and Creative X API """ import json from pathlib import Path from typing import Optional, Dict, Any class MappingResolver: """Resolves mappings between Ferrero codes and Creative X API values""" def __init__(self, mappings_json_path: str): """ Initialize mapping resolver Args: mappings_json_path: Path to mappings.json file Raises: FileNotFoundError: If mappings.json doesn't exist ValueError: If mappings.json has invalid structure """ self.mappings_path = Path(mappings_json_path) if not self.mappings_path.exists(): raise FileNotFoundError(f"mappings.json not found at: {mappings_json_path}") self.mappings = self._load_json() def _load_json(self) -> dict: """Load JSON data from file""" try: with open(self.mappings_path, 'r', encoding='utf-8') as f: return json.load(f) except json.JSONDecodeError as e: raise ValueError(f"Invalid JSON in mappings.json: {e}") def get_creativex_brand(self, ferrero_brand_code: str) -> Optional[Dict[str, Any]]: """ Map Ferrero brand code to Creative X brand information Args: ferrero_brand_code: Ferrero brand code (e.g., 'NUT', 'RAF') Returns: dict: { 'name': Creative X brand name, 'id': Creative X brand ID, 'ferrero_name': Original Ferrero brand name } None: If mapping doesn't exist """ brand_map = self.mappings.get('brand_mappings', {}).get(ferrero_brand_code.upper()) if not brand_map: return None return { 'name': brand_map.get('creativex_name'), 'id': brand_map.get('creativex_id'), 'ferrero_name': brand_map.get('ferrero_name') } def get_creativex_channel_mapping(self, ferrero_social_code: str) -> Optional[Dict[str, Any]]: """ Map Ferrero social media code to Creative X channel information Args: ferrero_social_code: Ferrero social code (e.g., 'FBS', 'IGF', 'YTB') Returns: dict: { 'channel': Creative X channel name (required), 'publisher': Creative X publisher (optional), 'placement': Creative X placement (optional), 'ad_format': Creative X ad format (optional), 'ferrero_name': Original Ferrero social media name } None: If mapping doesn't exist """ # Search through all channel mapping categories channel_categories = self.mappings.get('channel_mappings', {}) for category_name, category_data in channel_categories.items(): # Skip comment fields if category_name.startswith('_'): continue if ferrero_social_code.upper() in category_data: mapping = category_data[ferrero_social_code.upper()] return { 'channel': mapping.get('creativex_channel'), 'publisher': mapping.get('creativex_publisher'), 'placement': mapping.get('creativex_placement'), 'ad_format': mapping.get('creativex_ad_format'), 'ferrero_name': mapping.get('ferrero_name') } return None def get_creativex_market(self, ferrero_country_code: str, country_name: str = None) -> Dict[str, Any]: """ Map Ferrero country code to Creative X market information Args: ferrero_country_code: Ferrero country code (e.g., 'DE', 'IT', 'GL') country_name: Full country name from Ferrero data (optional) Returns: dict: { 'iso_code': ISO 3166-1 code (or None for 'Global'), 'name': Market name for Creative X API, 'direct_lookup': Boolean indicating if ISO code can be used directly } """ country_upper = ferrero_country_code.upper() # Check for special cases first special_cases = self.mappings.get('country_mappings', {}).get('SPECIAL_CASES', {}) if country_upper in special_cases: special = special_cases[country_upper] return { 'iso_code': special.get('creativex_iso_code'), 'name': special.get('creativex_name'), 'direct_lookup': False, 'note': special.get('_note') } # For standard ISO codes, use the country name from Ferrero data # Creative X API expects the full market name, not ISO code return { 'iso_code': country_upper, 'name': country_name, # Use full country name for API 'direct_lookup': True } def is_brand_supported(self, ferrero_brand_code: str) -> bool: """ Check if a Ferrero brand is supported in Creative X Args: ferrero_brand_code: Ferrero brand code Returns: bool: True if brand is mapped """ return ferrero_brand_code.upper() in self.mappings.get('brand_mappings', {}) def is_channel_supported(self, ferrero_social_code: str) -> bool: """ Check if a Ferrero social media code is supported in Creative X Args: ferrero_social_code: Ferrero social media code Returns: bool: True if channel is mapped """ return self.get_creativex_channel_mapping(ferrero_social_code) is not None def get_all_supported_brands(self) -> Dict[str, str]: """ Get all supported brand mappings Returns: dict: {ferrero_code: creativex_name} """ brand_mappings = self.mappings.get('brand_mappings', {}) return { code: data['creativex_name'] for code, data in brand_mappings.items() if not code.startswith('_') } def get_all_supported_channels(self) -> Dict[str, str]: """ Get all supported channel mappings Returns: dict: {ferrero_code: creativex_channel} """ result = {} channel_categories = self.mappings.get('channel_mappings', {}) for category_name, category_data in channel_categories.items(): if category_name.startswith('_'): continue for code, mapping in category_data.items(): result[code] = mapping.get('creativex_channel') return result def build_creativex_payload(self, parsed_metadata: Dict[str, Any]) -> Dict[str, Any]: """ Build Creative X preflight payload from parsed Ferrero metadata Args: parsed_metadata: Parsed metadata from filename parser Returns: dict: Creative X API payload structure Raises: ValueError: If required mappings are missing """ # Map brand brand_mapping = self.get_creativex_brand(parsed_metadata.get('brand_code', '')) if not brand_mapping: raise ValueError( f"Brand '{parsed_metadata.get('brand_code')}' is not supported in Creative X. " f"Supported brands: {list(self.get_all_supported_brands().keys())}" ) # Map channel channel_mapping = self.get_creativex_channel_mapping(parsed_metadata.get('social_media', '')) if not channel_mapping: raise ValueError( f"Social media code '{parsed_metadata.get('social_media')}' is not mapped to Creative X. " f"Check mappings.json for supported channels." ) # Map market - pass country name for proper lookup market_mapping = self.get_creativex_market( parsed_metadata.get('country_code', ''), parsed_metadata.get('country_name', '') ) # Build payload payload = { 'brand_name': brand_mapping['name'], 'channel': channel_mapping['channel'] } # Add optional channel fields if channel_mapping.get('publisher'): payload['publisher'] = channel_mapping['publisher'] if channel_mapping.get('placement'): payload['placement'] = channel_mapping['placement'] if channel_mapping.get('ad_format'): payload['ad_format'] = channel_mapping['ad_format'] # Add market name (always required by Creative X API) payload['market_name'] = market_mapping['name'] # Add other metadata if parsed_metadata.get('language_code'): payload['language'] = parsed_metadata['language_code'].lower() if parsed_metadata.get('aspect_ratio'): payload['dimensions'] = parsed_metadata['aspect_ratio'] if parsed_metadata.get('seconds'): # Remove 'S' suffix if present duration = parsed_metadata['seconds'].upper().replace('S', '') payload['duration'] = duration if parsed_metadata.get('subject'): payload['subject'] = parsed_metadata['subject'] if parsed_metadata.get('asset_type'): payload['asset_type'] = parsed_metadata['asset_type'] return payload def validate_metadata_for_upload(self, parsed_metadata: Dict[str, Any]) -> tuple[bool, list[str]]: """ Validate that parsed metadata can be mapped to Creative X Args: parsed_metadata: Parsed metadata from filename parser Returns: tuple: (is_valid: bool, errors: List[str]) """ errors = [] # Check brand brand_code = parsed_metadata.get('brand_code', '') if not self.is_brand_supported(brand_code): supported = ', '.join(self.get_all_supported_brands().keys()) errors.append( f"Brand code '{brand_code}' not supported in Creative X. " f"Supported: {supported}" ) # Check channel social_code = parsed_metadata.get('social_media', '') if not self.is_channel_supported(social_code): errors.append( f"Social media code '{social_code}' not mapped to Creative X channel. " f"Check mappings.json for available mappings." ) # Country codes are mostly standard, so we don't strictly validate # But we can warn about GL (Global) ambiguity country_code = parsed_metadata.get('country_code', '') if country_code.upper() == 'GL': market_mapping = self.get_creativex_market(country_code) if market_mapping.get('note'): # This is informational, not an error pass is_valid = len(errors) == 0 return is_valid, errors