72e9b54ff6
- Ferrero filename parser with NEW format support - CreativeX API client with retry logic - State persistence with resume capability - Upload and status checking scripts - Comprehensive documentation - Virtual environment support
106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""
|
|
Configuration management for CreativeX API Integration
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
class Config:
|
|
"""Application configuration loaded from environment variables"""
|
|
|
|
def __init__(self):
|
|
"""Initialize configuration from environment variables"""
|
|
# Load .env file
|
|
load_dotenv()
|
|
|
|
# API Configuration
|
|
self.api_base_url = os.getenv(
|
|
'CREATIVEX_API_BASE_URL',
|
|
'https://staging-api.creativex.com/api/v3'
|
|
)
|
|
self.access_token = os.getenv('CREATIVEX_ACCESS_TOKEN')
|
|
|
|
if not self.access_token:
|
|
raise ValueError("CREATIVEX_ACCESS_TOKEN not set in environment or .env file")
|
|
|
|
# Paths
|
|
self.project_root = Path(__file__).parent.absolute()
|
|
self.data_json_path = Path(os.getenv(
|
|
'DATA_JSON_PATH',
|
|
str(self.project_root / 'data.json')
|
|
))
|
|
|
|
if not self.data_json_path.exists():
|
|
raise FileNotFoundError(
|
|
f"data.json not found at {self.data_json_path}. "
|
|
"Please copy from Ferrero naming convention project."
|
|
)
|
|
|
|
self.state_file = self.project_root / 'data' / 'uploads_state.json'
|
|
self.results_dir = self.project_root / 'data' / 'results'
|
|
self.logs_dir = self.project_root / 'data' / 'logs'
|
|
|
|
# API Settings
|
|
self.api_timeout = int(os.getenv('API_TIMEOUT', '30'))
|
|
self.api_max_retries = int(os.getenv('API_MAX_RETRIES', '3'))
|
|
|
|
# Upload Settings
|
|
self.max_file_size_mb = int(os.getenv('MAX_FILE_SIZE_MB', '500'))
|
|
self.chunk_size_mb = int(os.getenv('CHUNK_SIZE_MB', '5'))
|
|
|
|
# Polling Settings
|
|
self.default_poll_interval = int(os.getenv('POLL_INTERVAL_MINUTES', '30'))
|
|
self.max_wait_hours = int(os.getenv('MAX_WAIT_HOURS', '48'))
|
|
|
|
# Logging
|
|
self.log_level = os.getenv('LOG_LEVEL', 'INFO')
|
|
|
|
# Create required directories
|
|
self._create_directories()
|
|
|
|
def _create_directories(self):
|
|
"""Ensure all required directories exist"""
|
|
for dir_path in [self.results_dir, self.logs_dir, self.state_file.parent]:
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
def to_dict(self) -> dict:
|
|
"""
|
|
Return configuration as dictionary (excluding sensitive data)
|
|
|
|
Returns:
|
|
dict: Configuration values
|
|
"""
|
|
return {
|
|
'api_base_url': self.api_base_url,
|
|
'data_json_path': str(self.data_json_path),
|
|
'state_file': str(self.state_file),
|
|
'results_dir': str(self.results_dir),
|
|
'logs_dir': str(self.logs_dir),
|
|
'api_timeout': self.api_timeout,
|
|
'api_max_retries': self.api_max_retries,
|
|
'max_file_size_mb': self.max_file_size_mb,
|
|
'default_poll_interval': self.default_poll_interval,
|
|
'max_wait_hours': self.max_wait_hours,
|
|
'log_level': self.log_level,
|
|
}
|
|
|
|
def __repr__(self) -> str:
|
|
"""String representation of configuration"""
|
|
config_dict = self.to_dict()
|
|
return f"Config({', '.join(f'{k}={v}' for k, v in config_dict.items())})"
|
|
|
|
|
|
def load_config() -> Config:
|
|
"""
|
|
Load and return configuration
|
|
|
|
Returns:
|
|
Config: Initialized configuration object
|
|
|
|
Raises:
|
|
ValueError: If required configuration is missing
|
|
FileNotFoundError: If required files don't exist
|
|
"""
|
|
return Config()
|