""" Configuration Loader - Load YAML config with environment variable substitution Compatible with Python 3.6+ """ import os import re import yaml from dotenv import load_dotenv def load_config(config_path='config/config.yaml'): """ Load configuration from YAML file with environment variable substitution Supports: - ${VAR_NAME} - Required environment variable - ${VAR_NAME:-default} - Optional with default value """ # Load environment variables from .env file # Check for custom .env file path or environment-specific file dotenv_path = os.getenv('DOTENV_PATH') if dotenv_path: # Use explicitly specified .env file load_dotenv(dotenv_path, override=True) else: # Check for ENV variable to determine which .env file to load env_name = os.getenv('ENV', 'dev') if env_name == 'prod' or env_name == 'production': # Load .env-prod for production if os.path.exists('.env-prod'): load_dotenv('.env-prod', override=True) else: load_dotenv() # Fallback to default .env else: # Load default .env for dev/staging load_dotenv() # Read YAML file with open(config_path, 'r') as f: config_text = f.read() # Substitute environment variables config_text = substitute_env_vars(config_text) # Parse YAML config = yaml.safe_load(config_text) # Load environment-specific overrides if specified env = config.get('environment', 'staging') env_config_path = 'config/environments/{}.yaml'.format(env) if os.path.exists(env_config_path): with open(env_config_path, 'r') as f: env_config_text = f.read() env_config_text = substitute_env_vars(env_config_text) env_config = yaml.safe_load(env_config_text) # Merge environment-specific config config = deep_merge(config, env_config) return config def substitute_env_vars(text): """ Substitute ${VAR_NAME} and ${VAR_NAME:-default} patterns with environment variables """ def replacer(match): var_expr = match.group(1) # Check for default value syntax: VAR:-default if ':-' in var_expr: var_name, default = var_expr.split(':-', 1) return os.getenv(var_name, default) else: # Required variable value = os.getenv(var_expr) if value is None: raise ValueError("Required environment variable not set: {}".format(var_expr)) return value # Pattern: ${VAR_NAME} or ${VAR_NAME:-default} pattern = r'\$\{([^}]+)\}' return re.sub(pattern, replacer, text) def deep_merge(base, override): """ Deep merge two dictionaries """ result = base.copy() for key, value in override.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = deep_merge(result[key], value) else: result[key] = value return result def load_field_mappings(config): """ Load field mappings configuration Auto-detects environment (PPR or PROD) and loads appropriate file """ # Detect environment based on DAM URL dam_url = config.get('dam', {}).get('base_url', '') # Determine which field mappings file to use if 'ppr' in dam_url.lower(): # PPR environment mappings_file = 'config/field_mappings_ppr.yaml' env_name = 'PPR' else: # Production environment (default) mappings_file = 'config/field_mappings_prod.yaml' env_name = 'PROD' # Log which file is being loaded import logging logger = logging.getLogger('ConfigLoader') logger.info("Loading field mappings for {} environment: {}".format(env_name, mappings_file)) with open(mappings_file, 'r') as f: return yaml.safe_load(f) def load_country_code_mappings(): """ Load country code mappings: ISO 3166-1 Alpha-2 -> DAM Codes Returns: dict: ISO code -> DAM code mapping Example: {'BD': 'BG', 'DE': 'DE', 'IT': 'IT', ...} """ mapping_path = 'config/country_code_mappings.yaml' try: with open(mapping_path, 'r') as f: mappings = yaml.safe_load(f) return mappings if mappings else {} except Exception as e: # If file doesn't exist or fails to load, return empty dict # This allows system to work without mapping (uses ISO codes as-is) return {}