# encoding=utf-8 """ S3 Configuration and Service Module Handles S3 credentials and asset retrieval for the MOTM application. """ import os import json import boto3 from botocore.exceptions import ClientError, NoCredentialsError from datetime import datetime, timedelta from typing import Optional, Dict, Any class S3ConfigManager: """Manages S3 configuration settings.""" def __init__(self): self.config_file = 's3_config.json' self.config = self.load_config() def load_config(self) -> Dict[str, Any]: """Load S3 configuration from file.""" if os.path.exists(self.config_file): try: with open(self.config_file, 'r') as f: return json.load(f) except (json.JSONDecodeError, IOError): return self.get_default_config() return self.get_default_config() def get_default_config(self) -> Dict[str, Any]: """Get default S3 configuration.""" return { 'aws_access_key_id': '', 'aws_secret_access_key': '', 'aws_region': 'us-east-1', 'bucket_name': '', 'bucket_prefix': 'assets/', 'enable_s3': False, 'use_signed_urls': True, 'signed_url_expiry': 3600, # 1 hour in seconds 'fallback_to_static': True, 'storage_provider': 'aws', # 'aws' or 'minio' 'minio_endpoint': '', # MinIO endpoint URL 'minio_use_ssl': True # Whether to use SSL for MinIO } def save_config(self, config_data: Dict[str, Any]) -> bool: """Save S3 configuration to file.""" try: # Update environment variables os.environ['AWS_ACCESS_KEY_ID'] = config_data.get('aws_access_key_id', '') os.environ['AWS_SECRET_ACCESS_KEY'] = config_data.get('aws_secret_access_key', '') os.environ['AWS_DEFAULT_REGION'] = config_data.get('aws_region', 'us-east-1') with open(self.config_file, 'w') as f: json.dump(config_data, f, indent=2) self.config = config_data return True except IOError: return False def get_config_dict(self) -> Dict[str, Any]: """Get current configuration as dictionary.""" return self.config.copy() def test_connection(self, config_data: Optional[Dict[str, Any]] = None) -> tuple[bool, str]: """Test S3 connection with provided or current configuration.""" if config_data: test_config = config_data else: test_config = self.config if not test_config.get('enable_s3', False): return True, "S3 is disabled - using local static files" try: # Determine storage provider storage_provider = test_config.get('storage_provider', 'aws') if storage_provider == 'minio': # Create MinIO client minio_endpoint = test_config.get('minio_endpoint', '') if not minio_endpoint: return False, "MinIO endpoint is required when using MinIO" # Parse endpoint to remove protocol if present if minio_endpoint.startswith('http://'): endpoint_host = minio_endpoint[7:] use_ssl = False elif minio_endpoint.startswith('https://'): endpoint_host = minio_endpoint[8:] use_ssl = True else: endpoint_host = minio_endpoint use_ssl = test_config.get('minio_use_ssl', True) # Construct the full endpoint URL endpoint_url = f"{'https' if use_ssl else 'http'}://{endpoint_host}" s3_client = boto3.client( 's3', aws_access_key_id=test_config.get('aws_access_key_id', ''), aws_secret_access_key=test_config.get('aws_secret_access_key', ''), region_name=test_config.get('aws_region', 'us-east-1'), endpoint_url=endpoint_url, use_ssl=use_ssl, verify=True # Enable SSL certificate verification ) provider_name = "MinIO" else: # Create AWS S3 client s3_client = boto3.client( 's3', aws_access_key_id=test_config.get('aws_access_key_id', ''), aws_secret_access_key=test_config.get('aws_secret_access_key', ''), region_name=test_config.get('aws_region', 'us-east-1') ) provider_name = "AWS S3" # Test connection by listing bucket bucket_name = test_config.get('bucket_name', '') if not bucket_name: return False, "Bucket name is required when S3 is enabled" s3_client.head_bucket(Bucket=bucket_name) return True, f"Successfully connected to {provider_name} bucket: {bucket_name}" except ClientError as e: error_code = e.response['Error']['Code'] provider_name = "MinIO" if test_config.get('storage_provider', 'aws') == 'minio' else "AWS S3" if error_code == '404': return False, f"{provider_name} bucket '{bucket_name}' not found" elif error_code == '403': return False, f"Access denied to {provider_name} bucket. Check credentials and permissions" else: return False, f"{provider_name} error: {e.response['Error']['Message']}" except NoCredentialsError: return False, "AWS credentials not found" except Exception as e: provider_name = "MinIO" if test_config.get('storage_provider', 'aws') == 'minio' else "AWS S3" return False, f"{provider_name} connection error: {str(e)}" class S3AssetService: """Service for retrieving assets from S3.""" def __init__(self, config_manager: S3ConfigManager): self.config_manager = config_manager self._s3_client = None def _get_public_url(self, bucket_name: str, s3_key: str) -> str: """Generate public URL for S3 or MinIO.""" config = self.config_manager.config storage_provider = config.get('storage_provider', 'aws') if storage_provider == 'minio': minio_endpoint = config.get('minio_endpoint', '') use_ssl = config.get('minio_use_ssl', True) protocol = 'https' if use_ssl else 'http' # Parse endpoint to remove protocol if present if minio_endpoint.startswith('http://'): minio_endpoint = minio_endpoint[7:] elif minio_endpoint.startswith('https://'): minio_endpoint = minio_endpoint[8:] return f"{protocol}://{minio_endpoint}/{bucket_name}/{s3_key}" else: return f"https://{bucket_name}.s3.{config.get('aws_region', 'us-east-1')}.amazonaws.com/{s3_key}" @property def s3_client(self): """Lazy-loaded S3 client.""" if self._s3_client is None and self.config_manager.config.get('enable_s3', False): config = self.config_manager.config storage_provider = config.get('storage_provider', 'aws') if storage_provider == 'minio': # Create MinIO client minio_endpoint = config.get('minio_endpoint', '') use_ssl = config.get('minio_use_ssl', True) # Parse endpoint to remove protocol if present if minio_endpoint.startswith('http://'): endpoint_host = minio_endpoint[7:] use_ssl = False elif minio_endpoint.startswith('https://'): endpoint_host = minio_endpoint[8:] use_ssl = True else: endpoint_host = minio_endpoint # Construct the full endpoint URL endpoint_url = f"{'https' if use_ssl else 'http'}://{endpoint_host}" self._s3_client = boto3.client( 's3', aws_access_key_id=config.get('aws_access_key_id', ''), aws_secret_access_key=config.get('aws_secret_access_key', ''), region_name=config.get('aws_region', 'us-east-1'), endpoint_url=endpoint_url, use_ssl=use_ssl, verify=True # Enable SSL certificate verification ) else: # Create AWS S3 client self._s3_client = boto3.client( 's3', aws_access_key_id=config.get('aws_access_key_id', ''), aws_secret_access_key=config.get('aws_secret_access_key', ''), region_name=config.get('aws_region', 'us-east-1') ) return self._s3_client def get_logo_url(self, logo_path: str, club_name: str = '') -> str: """Get logo URL from S3 or fallback to static.""" config = self.config_manager.config if not config.get('enable_s3', False): return self._get_static_logo_url(logo_path) bucket_name = config.get('bucket_name', '') bucket_prefix = config.get('bucket_prefix', 'motm-assets/') if not bucket_name: return self._get_static_logo_url(logo_path) # Clean up the logo path if logo_path.startswith('/static/'): s3_key = bucket_prefix + logo_path.replace('/static/', '') elif logo_path.startswith('logos/'): s3_key = bucket_prefix + logo_path else: s3_key = bucket_prefix + 'logos/' + logo_path try: if config.get('use_signed_urls', True): # Generate signed URL expiry = config.get('signed_url_expiry', 3600) signed_url = self.s3_client.generate_presigned_url( 'get_object', Params={'Bucket': bucket_name, 'Key': s3_key}, ExpiresIn=expiry ) return signed_url else: # Return public URL return self._get_public_url(bucket_name, s3_key) except ClientError: # Fallback to static if S3 fails if config.get('fallback_to_static', True): return self._get_static_logo_url(logo_path) return '' def get_logo_url_public(self, logo_path: str, club_name: str = '') -> str: """Get logo URL from S3 using public URLs (no authentication).""" config = self.config_manager.config if not config.get('enable_s3', False): return self._get_static_logo_url(logo_path) bucket_name = config.get('bucket_name', '') bucket_prefix = config.get('bucket_prefix', 'assets/') if not bucket_name: return self._get_static_logo_url(logo_path) # Clean up the logo path if logo_path.startswith('/static/'): s3_key = bucket_prefix + logo_path.replace('/static/', '') elif logo_path.startswith('logos/'): s3_key = bucket_prefix + logo_path else: s3_key = bucket_prefix + 'logos/' + logo_path try: # Always return public URL (no authentication) return self._get_public_url(bucket_name, s3_key) except Exception: # Fallback to static if S3 fails if config.get('fallback_to_static', True): return self._get_static_logo_url(logo_path) return '' def get_asset_url(self, asset_path: str) -> str: """Get any asset URL from S3 or fallback to static.""" config = self.config_manager.config if not config.get('enable_s3', False): return f"/static/{asset_path}" bucket_name = config.get('bucket_name', '') bucket_prefix = config.get('bucket_prefix', 'motm-assets/') if not bucket_name: return f"/static/{asset_path}" # Clean up the asset path if asset_path.startswith('/static/'): s3_key = bucket_prefix + asset_path.replace('/static/', '') else: s3_key = bucket_prefix + asset_path try: if config.get('use_signed_urls', True): # Generate signed URL expiry = config.get('signed_url_expiry', 3600) signed_url = self.s3_client.generate_presigned_url( 'get_object', Params={'Bucket': bucket_name, 'Key': s3_key}, ExpiresIn=expiry ) return signed_url else: # Return public URL return self._get_public_url(bucket_name, s3_key) except ClientError: # Fallback to static if S3 fails if config.get('fallback_to_static', True): return f"/static/{asset_path}" return '' def get_asset_url_public(self, asset_path: str) -> str: """Get any asset URL from S3 using public URLs (no authentication).""" config = self.config_manager.config if not config.get('enable_s3', False): return f"/static/{asset_path}" bucket_name = config.get('bucket_name', '') bucket_prefix = config.get('bucket_prefix', 'assets/') if not bucket_name: return f"/static/{asset_path}" # Clean up the asset path if asset_path.startswith('/static/'): s3_key = bucket_prefix + asset_path.replace('/static/', '') else: s3_key = bucket_prefix + asset_path try: # Always return public URL (no authentication) return self._get_public_url(bucket_name, s3_key) except Exception: # Fallback to static if S3 fails if config.get('fallback_to_static', True): return f"/static/{asset_path}" return '' def upload_asset(self, file_path: str, s3_key: str) -> tuple[bool, str]: """Upload an asset to S3.""" config = self.config_manager.config if not config.get('enable_s3', False): return False, "S3 is disabled" bucket_name = config.get('bucket_name', '') bucket_prefix = config.get('bucket_prefix', 'motm-assets/') if not bucket_name: return False, "Bucket name not configured" full_s3_key = bucket_prefix + s3_key try: self.s3_client.upload_file(file_path, bucket_name, full_s3_key) return True, f"Successfully uploaded {s3_key} to S3" except ClientError as e: return False, f"Upload failed: {e.response['Error']['Message']}" except Exception as e: return False, f"Upload error: {str(e)}" def _get_static_logo_url(self, logo_path: str) -> str: """Get static logo URL as fallback.""" if logo_path.startswith('/static/'): return logo_path return f"/static/images/{logo_path}" # Global instances s3_config_manager = S3ConfigManager() s3_asset_service = S3AssetService(s3_config_manager)