371 lines
15 KiB
Python
371 lines
15 KiB
Python
"""
|
|
GigLez Environment Configuration
|
|
|
|
Handles environment-based settings for development and production modes
|
|
"""
|
|
|
|
import os
|
|
from enum import Enum
|
|
from typing import Optional, List
|
|
from pathlib import Path
|
|
from loguru import logger
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
# =============================================================================
|
|
# LOAD ENVIRONMENT
|
|
# =============================================================================
|
|
|
|
# Determine which .env file to load
|
|
MODE = os.getenv("GIGLEZ_MODE", "development")
|
|
|
|
if MODE == "production":
|
|
env_file = ".env.production"
|
|
elif MODE == "development":
|
|
env_file = ".env.development"
|
|
else:
|
|
env_file = ".env"
|
|
|
|
# Load environment variables
|
|
env_path = Path(__file__).parent.parent / env_file
|
|
if env_path.exists():
|
|
load_dotenv(env_path)
|
|
logger.info(f"Loaded environment from: {env_file}")
|
|
else:
|
|
logger.warning(f"Environment file not found: {env_file}, using defaults")
|
|
|
|
|
|
# =============================================================================
|
|
# ENUMS
|
|
# =============================================================================
|
|
|
|
class DeploymentMode(str, Enum):
|
|
"""Deployment mode enumeration"""
|
|
DEVELOPMENT = "development"
|
|
PRODUCTION = "production"
|
|
|
|
|
|
class StorageType(str, Enum):
|
|
"""Storage backend types"""
|
|
FILESYSTEM = "filesystem"
|
|
S3 = "s3"
|
|
|
|
|
|
class BackgroundMode(str, Enum):
|
|
"""Background task processing mode"""
|
|
SYNC = "sync" # Synchronous (development)
|
|
CELERY = "celery" # Celery async (production)
|
|
|
|
|
|
class LogFormat(str, Enum):
|
|
"""Log output format"""
|
|
TEXT = "text"
|
|
JSON = "json"
|
|
|
|
|
|
# =============================================================================
|
|
# SETTINGS CLASS
|
|
# =============================================================================
|
|
|
|
class Settings:
|
|
"""
|
|
Application settings with environment-based configuration
|
|
|
|
Supports both development (Termux) and production (Server) modes
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""Initialize settings from environment variables"""
|
|
|
|
# =================================================================
|
|
# DEPLOYMENT MODE
|
|
# =================================================================
|
|
self.mode = DeploymentMode(os.getenv("GIGLEZ_MODE", "development"))
|
|
|
|
# =================================================================
|
|
# DATABASE
|
|
# =================================================================
|
|
self.db_host = os.getenv("GIGLEZ_DB_HOST", "localhost")
|
|
self.db_port = int(os.getenv("GIGLEZ_DB_PORT", "5432"))
|
|
self.db_name = os.getenv("GIGLEZ_DB_NAME", "giglez")
|
|
self.db_user = os.getenv("GIGLEZ_DB_USER", "giglez_user")
|
|
self.db_password = os.getenv("GIGLEZ_DB_PASSWORD", "giglez_secure_password_2026")
|
|
self.db_pool_size = int(os.getenv("GIGLEZ_DB_POOL_SIZE", "10"))
|
|
self.db_max_overflow = int(os.getenv("GIGLEZ_DB_MAX_OVERFLOW", "20"))
|
|
self.db_echo = os.getenv("GIGLEZ_DB_ECHO", "false").lower() == "true"
|
|
|
|
# =================================================================
|
|
# API
|
|
# =================================================================
|
|
self.api_host = os.getenv("GIGLEZ_API_HOST", "0.0.0.0")
|
|
self.api_port = int(os.getenv("GIGLEZ_API_PORT", "8000"))
|
|
self.api_workers = int(os.getenv("GIGLEZ_API_WORKERS", "4"))
|
|
|
|
# CORS
|
|
cors_origins = os.getenv("GIGLEZ_API_CORS_ORIGINS", "*")
|
|
self.cors_origins = [o.strip() for o in cors_origins.split(",")] if cors_origins != "*" else ["*"]
|
|
|
|
# Authentication
|
|
self.require_auth = os.getenv("GIGLEZ_REQUIRE_AUTH", "true").lower() == "true"
|
|
self.jwt_secret_key = os.getenv("GIGLEZ_JWT_SECRET_KEY", "dev-secret-key")
|
|
self.jwt_algorithm = os.getenv("GIGLEZ_JWT_ALGORITHM", "HS256")
|
|
self.jwt_expire_minutes = int(os.getenv("GIGLEZ_JWT_EXPIRE_MINUTES", "1440"))
|
|
|
|
# =================================================================
|
|
# FILE STORAGE
|
|
# =================================================================
|
|
self.storage_type = StorageType(os.getenv("GIGLEZ_STORAGE_TYPE", "filesystem"))
|
|
self.storage_path = os.getenv("GIGLEZ_STORAGE_PATH", "./storage")
|
|
self.storage_bucket = os.getenv("GIGLEZ_STORAGE_BUCKET", "giglez-captures")
|
|
self.s3_endpoint = os.getenv("GIGLEZ_S3_ENDPOINT", "https://s3.amazonaws.com")
|
|
|
|
# AWS credentials (for S3)
|
|
self.aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID")
|
|
self.aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
|
|
self.aws_region = os.getenv("AWS_REGION", "us-east-1")
|
|
|
|
# Upload limits
|
|
self.max_upload_size = int(os.getenv("GIGLEZ_MAX_UPLOAD_SIZE", "10485760")) # 10MB
|
|
self.max_batch_files = int(os.getenv("GIGLEZ_MAX_BATCH_FILES", "100"))
|
|
|
|
# =================================================================
|
|
# HARDWARE (Termux specific)
|
|
# =================================================================
|
|
self.enable_hardware = os.getenv("GIGLEZ_ENABLE_HARDWARE", "false").lower() == "true"
|
|
self.tembed_port = os.getenv("GIGLEZ_TEMBED_PORT", "/dev/ttyUSB0")
|
|
self.tembed_baud = int(os.getenv("GIGLEZ_TEMBED_BAUD", "115200"))
|
|
self.gps_provider = os.getenv("GIGLEZ_GPS_PROVIDER", "termux-api")
|
|
|
|
# =================================================================
|
|
# PERFORMANCE
|
|
# =================================================================
|
|
self.batch_size = int(os.getenv("GIGLEZ_BATCH_SIZE", "512"))
|
|
self.cache_size = int(os.getenv("GIGLEZ_CACHE_SIZE", "256"))
|
|
self.redis_url = os.getenv("GIGLEZ_REDIS_URL", "")
|
|
self.background_mode = BackgroundMode(os.getenv("GIGLEZ_BACKGROUND_MODE", "sync"))
|
|
|
|
# Celery (if background_mode == celery)
|
|
self.celery_broker = os.getenv("GIGLEZ_CELERY_BROKER", "redis://localhost:6379/1")
|
|
self.celery_result_backend = os.getenv("GIGLEZ_CELERY_RESULT_BACKEND", "redis://localhost:6379/2")
|
|
|
|
# =================================================================
|
|
# LOGGING
|
|
# =================================================================
|
|
self.log_level = os.getenv("GIGLEZ_LOG_LEVEL", "INFO")
|
|
self.log_file = os.getenv("GIGLEZ_LOG_FILE", "")
|
|
self.log_colorize = os.getenv("GIGLEZ_LOG_COLORIZE", "true").lower() == "true"
|
|
self.log_format = LogFormat(os.getenv("GIGLEZ_LOG_FORMAT", "text"))
|
|
|
|
# =================================================================
|
|
# GPS VALIDATION
|
|
# =================================================================
|
|
self.gps_max_accuracy = float(os.getenv("GIGLEZ_GPS_MAX_ACCURACY", "50.0"))
|
|
self.gps_min_accuracy = float(os.getenv("GIGLEZ_GPS_MIN_ACCURACY", "10.0"))
|
|
|
|
# =================================================================
|
|
# OFFLINE SUPPORT (Development)
|
|
# =================================================================
|
|
self.enable_offline_queue = os.getenv("GIGLEZ_ENABLE_OFFLINE_QUEUE", "false").lower() == "true"
|
|
self.queue_path = os.getenv("GIGLEZ_QUEUE_PATH", "./queue")
|
|
self.auto_upload = os.getenv("GIGLEZ_AUTO_UPLOAD", "true").lower() == "true"
|
|
|
|
# =================================================================
|
|
# SECURITY (Production)
|
|
# =================================================================
|
|
self.rate_limit_enabled = os.getenv("GIGLEZ_RATE_LIMIT_ENABLED", "false").lower() == "true"
|
|
self.rate_limit_per_minute = int(os.getenv("GIGLEZ_RATE_LIMIT_PER_MINUTE", "60"))
|
|
self.rate_limit_per_hour = int(os.getenv("GIGLEZ_RATE_LIMIT_PER_HOUR", "1000"))
|
|
self.force_https = os.getenv("GIGLEZ_FORCE_HTTPS", "false").lower() == "true"
|
|
|
|
# =================================================================
|
|
# FEATURES
|
|
# =================================================================
|
|
self.debug_endpoints = os.getenv("GIGLEZ_DEBUG_ENDPOINTS", "false").lower() == "true"
|
|
self.auto_reload = os.getenv("GIGLEZ_AUTO_RELOAD", "false").lower() == "true"
|
|
self.debug_errors = os.getenv("GIGLEZ_DEBUG_ERRORS", "false").lower() == "true"
|
|
self.enable_metrics = os.getenv("GIGLEZ_ENABLE_METRICS", "false").lower() == "true"
|
|
|
|
# =================================================================
|
|
# SIGNATURE DATABASES
|
|
# =================================================================
|
|
self.signature_auto_update = os.getenv("GIGLEZ_SIGNATURE_AUTO_UPDATE", "false").lower() == "true"
|
|
self.signature_update_cron = os.getenv("GIGLEZ_SIGNATURE_UPDATE_CRON", "0 2 * * *")
|
|
|
|
# =====================================================================
|
|
# COMPUTED PROPERTIES
|
|
# =====================================================================
|
|
|
|
@property
|
|
def is_production(self) -> bool:
|
|
"""Check if running in production mode"""
|
|
return self.mode == DeploymentMode.PRODUCTION
|
|
|
|
@property
|
|
def is_development(self) -> bool:
|
|
"""Check if running in development mode"""
|
|
return self.mode == DeploymentMode.DEVELOPMENT
|
|
|
|
@property
|
|
def database_url(self) -> str:
|
|
"""Get database connection URL"""
|
|
return f"postgresql://{self.db_user}:{self.db_password}@{self.db_host}:{self.db_port}/{self.db_name}"
|
|
|
|
@property
|
|
def database_url_safe(self) -> str:
|
|
"""Get database URL without password (for logging)"""
|
|
return f"postgresql://{self.db_user}:****@{self.db_host}:{self.db_port}/{self.db_name}"
|
|
|
|
@property
|
|
def use_redis(self) -> bool:
|
|
"""Check if Redis is configured"""
|
|
return bool(self.redis_url)
|
|
|
|
@property
|
|
def use_celery(self) -> bool:
|
|
"""Check if Celery background tasks are enabled"""
|
|
return self.background_mode == BackgroundMode.CELERY
|
|
|
|
@property
|
|
def use_s3_storage(self) -> bool:
|
|
"""Check if S3 storage is configured"""
|
|
return self.storage_type == StorageType.S3
|
|
|
|
@property
|
|
def use_local_storage(self) -> bool:
|
|
"""Check if local filesystem storage is configured"""
|
|
return self.storage_type == StorageType.FILESYSTEM
|
|
|
|
# =====================================================================
|
|
# VALIDATION
|
|
# =====================================================================
|
|
|
|
def validate(self) -> List[str]:
|
|
"""
|
|
Validate configuration
|
|
|
|
Returns:
|
|
List of validation errors (empty if valid)
|
|
"""
|
|
errors = []
|
|
|
|
# Production checks
|
|
if self.is_production:
|
|
if self.jwt_secret_key == "dev-secret-key":
|
|
errors.append("JWT secret key must be changed in production")
|
|
|
|
if not self.require_auth:
|
|
errors.append("Authentication must be enabled in production")
|
|
|
|
if self.debug_endpoints:
|
|
errors.append("Debug endpoints should be disabled in production")
|
|
|
|
if self.use_s3_storage and not (self.aws_access_key_id and self.aws_secret_access_key):
|
|
errors.append("AWS credentials required for S3 storage")
|
|
|
|
# Development checks
|
|
if self.is_development:
|
|
if self.enable_hardware and not os.path.exists(self.tembed_port):
|
|
logger.warning(f"T-Embed port not found: {self.tembed_port}")
|
|
|
|
# Storage checks
|
|
if self.use_local_storage:
|
|
storage_path = Path(self.storage_path)
|
|
if not storage_path.exists():
|
|
try:
|
|
storage_path.mkdir(parents=True, exist_ok=True)
|
|
logger.info(f"Created storage directory: {storage_path}")
|
|
except Exception as e:
|
|
errors.append(f"Cannot create storage directory: {e}")
|
|
|
|
# Queue checks
|
|
if self.enable_offline_queue:
|
|
queue_path = Path(self.queue_path)
|
|
if not queue_path.exists():
|
|
try:
|
|
queue_path.mkdir(parents=True, exist_ok=True)
|
|
logger.info(f"Created queue directory: {queue_path}")
|
|
except Exception as e:
|
|
errors.append(f"Cannot create queue directory: {e}")
|
|
|
|
return errors
|
|
|
|
# =====================================================================
|
|
# DISPLAY
|
|
# =====================================================================
|
|
|
|
def display(self):
|
|
"""Display current configuration"""
|
|
logger.info("=" * 60)
|
|
logger.info("GigLez Configuration")
|
|
logger.info("=" * 60)
|
|
logger.info(f"Mode: {self.mode.value}")
|
|
logger.info(f"Database: {self.database_url_safe}")
|
|
logger.info(f"API: {self.api_host}:{self.api_port}")
|
|
logger.info(f"Storage: {self.storage_type.value}")
|
|
logger.info(f"Hardware: {'Enabled' if self.enable_hardware else 'Disabled'}")
|
|
logger.info(f"Authentication: {'Required' if self.require_auth else 'Optional'}")
|
|
logger.info(f"Background Tasks: {self.background_mode.value}")
|
|
logger.info(f"Redis: {'Enabled' if self.use_redis else 'Disabled'}")
|
|
logger.info(f"Log Level: {self.log_level}")
|
|
logger.info("=" * 60)
|
|
|
|
# Validation
|
|
errors = self.validate()
|
|
if errors:
|
|
logger.error("Configuration errors:")
|
|
for error in errors:
|
|
logger.error(f" - {error}")
|
|
else:
|
|
logger.success("Configuration validated successfully")
|
|
|
|
|
|
# =============================================================================
|
|
# GLOBAL SETTINGS INSTANCE
|
|
# =============================================================================
|
|
|
|
settings = Settings()
|
|
|
|
|
|
# =============================================================================
|
|
# HELPER FUNCTIONS
|
|
# =============================================================================
|
|
|
|
def get_settings() -> Settings:
|
|
"""Get global settings instance"""
|
|
return settings
|
|
|
|
|
|
def reload_settings():
|
|
"""Reload settings from environment"""
|
|
global settings
|
|
settings = Settings()
|
|
return settings
|
|
|
|
|
|
# =============================================================================
|
|
# TESTING
|
|
# =============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
# Display configuration
|
|
settings.display()
|
|
|
|
# Test validation
|
|
print("\nValidation Results:")
|
|
errors = settings.validate()
|
|
if errors:
|
|
print("Errors found:")
|
|
for error in errors:
|
|
print(f" ❌ {error}")
|
|
else:
|
|
print(" ✅ Configuration is valid")
|
|
|
|
# Test properties
|
|
print("\nComputed Properties:")
|
|
print(f" is_production: {settings.is_production}")
|
|
print(f" is_development: {settings.is_development}")
|
|
print(f" use_redis: {settings.use_redis}")
|
|
print(f" use_celery: {settings.use_celery}")
|
|
print(f" use_s3_storage: {settings.use_s3_storage}")
|
|
print(f" use_local_storage: {settings.use_local_storage}")
|