refactor: simplify config to .env only, remove TOML

- Remove all config.toml files from setup/ folders
- Simplify config.py: remove TOML parsing, flatten logic
- All config now from environment variables only
- Keep setup/ folders for documentation purposes
- Remove unused functions (get_gcp_project_id, list_available_setups)

setup/beats_cut/config.toml

@@ -1,16 +0,0 @@
-# Beats Cut Random Videos Setup
-# Uses beat detection to sync video cuts with music
-
-[general]
-setup_type = "beats_cut"
-description = "Random videos with cuts synced to music beats"
-
-[video]
-beat_method = "downbeat"
-
-[gsheet]
-name = "Infloxa Data for Elvoro"
-id = "1djnE1u_QCveGlhjNnZRfXiY-3NLzO0V-04tZ9P2mVcs"
-video_library_worksheet = "Video Library"
-audio_library_worksheet = "Audio Library"
-logs_worksheet = "Infloxa Data for Elvoro LOGS"

setup/hard_cut/config.toml

@@ -1,16 +0,0 @@
-# Hard Cut Random Videos Setup
-# Uses fixed intervals instead of beat detection for video cuts
-
-[general]
-setup_type = "hard_cut"
-description = "Random videos with hard cuts at fixed intervals"
-
-[video]
-hard_cut_random_videos_interval = "0.5"
-
-[gsheet]
-name = "Infloxa Data for Elvoro"
-id = "1djnE1u_QCveGlhjNnZRfXiY-3NLzO0V-04tZ9P2mVcs"
-video_library_worksheet = "Video Library"
-audio_library_worksheet = "Audio Library"
-logs_worksheet = "Infloxa Data for Elvoro LOGS"

src/config.py

@@ -1,12 +1,11 @@
 """
 Unified Configuration Module
 
-Combines functionality from load_config.py and setup_config.py to provide
-a single source of truth for application configuration.
+Simple .env-based configuration. No TOML files needed.
+All configuration comes from environment variables.
 """
 
 import os
-import sys
 import json
 import logging
 from pathlib import Path
@@ -19,165 +18,11 @@ from google.auth import default
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 logger = logging.getLogger(__name__)
 
-# Use tomllib (Python 3.11+) or fall back to tomli
-if sys.version_info >= (3, 11):
-    import tomllib
-else:
-    try:
-        import tomli as tomllib
-    except ImportError:
-        tomllib = None
-
 # ------------------ Singleton & Cache ------------------
 
 _cached_config: Optional[Dict[str, Any]] = None
-_cached_setup_name: Optional[str] = None
 _config_initialized: bool = False
 
-# ------------------ Setup Config Logic ------------------
-
-def get_setup_dir() -> Path:
-    """Get the setup directory path."""
-    # Go from src/ to project root, then into setup/
-    project_root = Path(__file__).parent.parent
-    return project_root / "setup"
-
-
-def list_available_setups() -> list[str]:
-    """List all available setup configurations."""
-    setup_dir = get_setup_dir()
-    setups = []
-    
-    if setup_dir.exists():
-        for item in setup_dir.iterdir():
-            if item.is_dir() and (item / "config.toml").exists():
-                setups.append(item.name)
-    
-    return sorted(setups)
-
-
-def _flatten_config(config: Dict[str, Any], prefix: str = "") -> Dict[str, Any]:
-    """
-    Flatten nested TOML config into a flat dictionary.
-    """
-    result = {}
-    
-    for key, value in config.items():
-        full_key = f"{prefix}.{key}" if prefix else key
-        
-        if isinstance(value, dict):
-            # Recurse into nested dicts
-            nested = _flatten_config(value, full_key)
-            result.update(nested)
-        else:
-            # Add both the full nested key and the simple key
-            result[full_key] = value
-            # Also add simple key if it doesn't conflict
-            if key not in result:
-                result[key] = value
-    
-    return result
-
-
-def _apply_env_overrides(config: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Apply environment variable overrides to config values.
-    """
-    result = config.copy()
-    
-    # Map of config keys to env var names
-    env_mappings = {
-        "only_random_videos": "ONLY_RANDOM_VIDEOS",
-        "hard_cut_random_videos": "HARD_CUT_RANDOM_VIDEOS",
-        "use_veo": "USE_VEO",
-        "beat_method": "BEAT_METHOD",
-        "setup_type": "SETUP_TYPE",
-        "caption_style": "CAPTION_STYLE",
-        "content_strategy_file": "CONTENT_STRATEGY_FILE",
-        "on_screen_text": "ON_SCREEN_TEXT",
-        "is_onscreen_cta": "IS_ONSCREEN_CTA",
-        "is_a2e_lip_sync": "IS_A2E_LIP_SYNC",
-        "use_1x1_ratio": "USE_1X1_RATIO",
-        # GSheet mappings
-        "gsheet.name": "GSHEET_NAME",
-        "gsheet.id": "GSHEET_ID",
-        "video_library_worksheet": "VIDEO_LIBRARY_GSHEET_WORKSHEET",
-        "audio_library_worksheet": "AUDIO_LIBRARY_GSHEET_WORKSHEET",
-        "logs_worksheet": "GSHEET_WORKSHEET_LOGS",
-        "content_strategy_worksheet": "CONTENT_STRATEGY_GSHEET_WORKSHEET",
-        "test_automation": "TEST_AUTOMATION",
-        "test_data_directory": "TEST_DATA_DIRECTORY",
-        "delete_all_a2e_videos": "DELETE_ALL_A2E_VIDEOS",
-        "tiktok_access_token": "TIKTOK_ACCESS_TOKEN",
-        "tiktok_privacy_level": "TIKTOK_PRIVACY_LEVEL",
-        "youtube_refresh_token": "YOUTUBE_REFRESH_TOKEN",
-        "youtube_client_secrets_json": "YOUTUBE_CLIENT_SECRETS_JSON",
-        "youtube_client_secrets_base64": "YOUTUBE_CLIENT_SECRETS_BASE64",
-        "youtube_client_id": "YOUTUBE_CLIENT_ID",
-        "youtube_client_secret": "YOUTUBE_CLIENT_SECRET",
-        "video_category": "VIDEO_CATEGORY",
-        "video_privacy": "VIDEO_PRIVACY",
-        "scheduled_time": "SCHEDULED_TIME",
-        "instagram_access_token": "INSTAGRAM_ACCESS_TOKEN",
-        "instagram_account_id": "INSTAGRAM_ACCOUNT_ID",
-        "commit_progress": "COMMIT_PROGRESS",
-        "use_gemimi_video": "USE_GEMIMI_VIDEO",
-        "plain_video_count": "PlAIN_VIDEO_COUNT",
-    }
-    
-    for config_key, env_var in env_mappings.items():
-        env_value = os.getenv(env_var)
-        if env_value is not None:
-            # Convert to appropriate type based on existing config value
-            if config_key in result:
-                original_type = type(result[config_key])
-                if original_type == bool:
-                    result[config_key] = env_value.lower() in ("true", "1", "yes")
-                elif original_type == int:
-                    try:
-                        result[config_key] = int(env_value)
-                    except ValueError:
-                        pass
-                else:
-                    result[config_key] = env_value
-            else:
-                result[config_key] = env_value
-    
-    return result
-
-
-def _load_setup_config_data(setup_name: Optional[str] = None) -> Dict[str, Any]:
-    """Helper to load and process the TOML configuration file."""
-    if setup_name is None:
-        setup_name = os.getenv("SETUP_NAME")
-    
-    if not setup_name:
-        # If no setup name is provided or in env, return empty or raise?
-        # Preserving logic from setup_config.py which raised an error only if strictly required
-        # But here we want to handle the case where it might be missing gracefully if usage allows
-        return {}
-
-    setup_dir = get_setup_dir()
-    config_path = setup_dir / setup_name / "config.toml"
-    
-    if not config_path.exists():
-        # Fallback or error handled by caller/validator
-        return {}
-    
-    if tomllib is None:
-         logger.warning("TOML supported needed but not available. Install 'tomli' for Python < 3.11")
-         return {}
-
-    try:
-        with open(config_path, "rb") as f:
-            raw_config = tomllib.load(f)
-        
-        flat_config = _flatten_config(raw_config)
-        return _apply_env_overrides(flat_config)
-    except Exception as e:
-        logger.error(f"Error loading setup config: {e}")
-        return {}
-
 
 # ------------------ GCP Auth Configuration ------------------
 
@@ -195,7 +40,6 @@ def _resolve_gcp_project_id() -> tuple[Optional[str], Optional[str]]:
                       os.getenv("GOOGLE_APPLICATION_CREDENTIALS"))
 
     if gcp_creds_path:
-        # Set temp bucket env var side-effect (from original load_config)
         os.environ["MY_TEMP_GCS_BUCKET"] = os.getenv("MY_TEMP_GCS_BUCKET", "")
         
         try:
@@ -218,7 +62,6 @@ def _resolve_gcp_project_id() -> tuple[Optional[str], Optional[str]]:
     # 2. Workload Identity Federation / ADC
     if not gcp_project_id:
         try:
-            # This handles both WIF and ADC
             creds, project = default()
             if project:
                 gcp_project_id = project
@@ -263,58 +106,92 @@ def _resolve_gcp_project_id() -> tuple[Optional[str], Optional[str]]:
     return gcp_project_id, auth_method
 
 
+# ------------------ Helper to get bool from env ------------------
+
+def _env_bool(key: str, default: bool = False) -> bool:
+    """Get boolean from environment variable."""
+    return os.getenv(key, str(default)).lower() in ("true", "1", "yes")
+
+
+def _env_int(key: str, default: int = 0) -> int:
+    """Get integer from environment variable."""
+    try:
+        return int(os.getenv(key, str(default)))
+    except ValueError:
+        return default
+
+
 # ------------------ Main Load Function ------------------
 
 def load_configuration(force_reload: bool = False) -> Dict[str, Any]:
     """
-    Load configuration from all sources.
+    Load configuration from environment variables.
     
     1. Load .env
-    2. Load Setup Config (TOML)
-    3. Resolve GCP Project & Secrets
-    4. Merge & Validate
+    2. Resolve GCP Project & Secrets
+    3. Build config dict
     """
-    global _cached_config, _cached_setup_name, _config_initialized
+    global _cached_config, _config_initialized
 
-    setup_name = os.getenv("SETUP_NAME")
-    
     # Return cache if valid
-    if (_config_initialized and not force_reload and 
-        _cached_config is not None and 
-        _cached_setup_name == setup_name):
+    if _config_initialized and not force_reload and _cached_config is not None:
         return _cached_config
 
     load_dotenv()
     
-    # Load Setup Config (TOML)
-    # Note: We don't fail hard here if SETUP_NAME is missing, we just get empty setup config
-    # Validation happens later if critical keys are missing.
-    setup_config = _load_setup_config_data(setup_name)
-    if setup_config:
-         logger.info(f"✓ Loaded setup config: {setup_name}")
+    setup_type = os.getenv("SETUP_TYPE")
+    logger.info(f"✓ Loaded setup config: {setup_type}")
     
     # Resolve GCP Project
     gcp_project_id, auth_method = _resolve_gcp_project_id()
     if gcp_project_id:
         logger.info(f"✓ GCP Project ID: {gcp_project_id} ({auth_method})")
     
-    # Merge into final config
+    # Build config from environment variables
     config = {
-        **setup_config,
-        "gemini_api_key": os.getenv("GEMINI_API_KEY"),
-        "runwayml_api_key": os.getenv("RUNWAYML_API_KEY"),
-        "gcs_bucket_name": os.getenv("GCS_BUCKET_NAME"),
+        # Core settings
+        "setup_type": setup_type,
         "gcp_project_id": gcp_project_id,
         "auth_method": auth_method,
-        "setup_name": setup_name,
-        "setup_name": setup_name,
+        
+        # API Keys
+        "gemini_api_key": os.getenv("GEMINI_API_KEY"),
+        "runwayml_api_key": os.getenv("RUNWAYML_API_KEY"),
         "a2e_api_key": os.getenv("A2E_API_KEY"),
-        "test_automation": os.getenv("TEST_AUTOMATION", "false").lower() == "true",
+        
+        # GCS
+        "gcs_bucket_name": os.getenv("GCS_BUCKET_NAME"),
+        
+        # Google Sheets
+        "gsheet_name": os.getenv("GSHEET_NAME"),
+        "gsheet_id": os.getenv("GSHEET_ID"),
+        "video_library_worksheet": os.getenv("VIDEO_LIBRARY_GSHEET_WORKSHEET"),
+        "audio_library_worksheet": os.getenv("AUDIO_LIBRARY_GSHEET_WORKSHEET"),
+        "logs_worksheet": os.getenv("GSHEET_WORKSHEET_LOGS"),
+        "content_strategy_worksheet": os.getenv("CONTENT_STRATEGY_GSHEET_WORKSHEET"),
+        
+        # Pipeline settings
+        "caption_style": os.getenv("CAPTION_STYLE"),
+        "beat_method": os.getenv("BEAT_METHOD"),
+        "on_screen_text": _env_bool("ON_SCREEN_TEXT"),
+        "is_onscreen_cta": _env_bool("IS_ONSCREEN_CTA"),
+        "is_a2e_lip_sync": _env_bool("IS_A2E_LIP_SYNC"),
+        "use_1x1_ratio": _env_bool("USE_1X1_RATIO"),
+        "use_veo": _env_bool("USE_VEO"),
+        "use_gemimi_video": _env_bool("USE_GEMIMI_VIDEO"),
+        "only_random_videos": _env_bool("ONLY_RANDOM_VIDEOS"),
+        "plain_video_count": _env_int("PlAIN_VIDEO_COUNT", 100),
+        
+        # Test settings
+        "test_automation": _env_bool("TEST_AUTOMATION"),
         "test_data_directory": os.getenv("TEST_DATA_DIRECTORY"),
-        "delete_all_a2e_videos": os.getenv("DELETE_ALL_A2E_VIDEOS", "false").lower() == "true",
+        "delete_all_a2e_videos": _env_bool("DELETE_ALL_A2E_VIDEOS"),
+        
+        # TikTok
         "tiktok_access_token": os.getenv("TIKTOK_ACCESS_TOKEN"),
         "tiktok_privacy_level": os.getenv("TIKTOK_PRIVACY_LEVEL", "PUBLIC_TO_EVERYONE"),
-        "content_strategy_worksheet": os.getenv("CONTENT_STRATEGY_GSHEET_WORKSHEET"),
+        
+        # YouTube
         "youtube_refresh_token": os.getenv("YOUTUBE_REFRESH_TOKEN"),
         "youtube_client_secrets_json": os.getenv("YOUTUBE_CLIENT_SECRETS_JSON"),
         "youtube_client_secrets_base64": os.getenv("YOUTUBE_CLIENT_SECRETS_BASE64"),
@@ -323,13 +200,16 @@ def load_configuration(force_reload: bool = False) -> Dict[str, Any]:
         "video_category": os.getenv("VIDEO_CATEGORY", "22"),
         "video_privacy": os.getenv("VIDEO_PRIVACY", "private"),
         "scheduled_time": os.getenv("SCHEDULED_TIME"),
+        
+        # Instagram
         "instagram_access_token": os.getenv("INSTAGRAM_ACCESS_TOKEN"),
         "instagram_account_id": os.getenv("INSTAGRAM_ACCOUNT_ID"),
-        "commit_progress": os.getenv("COMMIT_PROGRESS", "false").lower() == "true",
-        "use_gemimi_video": os.getenv("USE_GEMIMI_VIDEO", "false").lower() == "true",
-        "plain_video_count": int(os.getenv("PlAIN_VIDEO_COUNT", 100)),
+        
+        # Misc
+        "commit_progress": _env_bool("COMMIT_PROGRESS"),
     }
 
+    # On-screen CTA options
     config["on_screen_cta"] = [
         "LINK IN BIO 🛍️",
         "🔗 LINK IN ACCOUNT 🛍️",
@@ -339,15 +219,12 @@ def load_configuration(force_reload: bool = False) -> Dict[str, Any]:
         "🛍️ SALE ENDING SOON - 🔗 LINK IN BIO"
     ]
 
-    # Initialize default empty collections for usage tracking
+    # Initialize default empty collections
     config["video_usage_count"] = {}
     config["avatar_usage_count"] = {}
     config["visual_assets"] = {}
 
-    # Required keys validation (soft validation - log error but don't crash module import)
-    # Crashes should handle at application start
     _cached_config = config
-    _cached_setup_name = setup_name
     _config_initialized = True
     
     return config
@@ -400,9 +277,7 @@ config = ConfigProxy()
 
 
 def get_config_value(key: str, default: Any = None) -> Any:
-    """
-    Get a config value. Preserves all types (dict, list, str, int, etc.).
-    """
+    """Get a config value."""
     value = config.get(key, default)
     return value if value is not None else default
 
@@ -410,36 +285,12 @@ def set_config_value(key: str, value: Any):
     """Set a config value."""
     config.set(key, value)
 
-def get_gcp_project_id() -> Optional[str]:
-    return config.get("gcp_project_id")
-
-# ------------------ CLI Test ------------------
-
-if __name__ == "__main__":
-    print("\n=== Unified Config Test ===\n")
-    try:
-        conf = load_configuration()
-        print("Configuration Loaded Successfully!")
-        print(f"Setup Name: {conf.get('setup_name')}")
-        print(f"Project ID: {conf.get('gcp_project_id')}")
-        
-        # Check required keys
-        required = ["gemini_api_key", "runwayml_api_key", "gcs_bucket_name", "gcp_project_id"]
-        missing = [k for k in required if not conf.get(k)]
-        
-        if missing:
-             print(f"\n[WARNING] Missing keys: {missing}")
-        else:
-             print("\nAll required keys present.")
-             
-    except Exception as e:
-        print(f"\n[ERROR] {e}")
-
+# ------------------ Job Environment Configuration ------------------
 
 def configure_job_environment(job_index: int):
     """
-    Configure environment variables and config settings specific to a parallel job index.
-    Primarily handles API key rotation for Google AI Studio.
+    Configure environment variables for a parallel job index.
+    Handles API key rotation for Google AI Studio.
     """
     if job_index is None:
         return
@@ -452,13 +303,10 @@ def configure_job_environment(job_index: int):
         "GOOGLE_AI_API_KEY"
     ]
     
-    # Ensure RUNWAYML_API_KEY is robust
     os.environ["RUNWAYML_API_KEY"] = os.getenv("RUNWAYML_API_KEY", "")
     
-    # Select key based on job index modulo
     selected_key = google_keys_env[job_index % len(google_keys_env)]
     
-    # Update both env and specific config
     new_api_key = os.getenv(selected_key, "")
     if new_api_key:
         os.environ["GEMINI_API_KEY"] = new_api_key
@@ -466,3 +314,25 @@ def configure_job_environment(job_index: int):
         print(f"Using Google key: {selected_key}")
     else:
         print(f"Warning: Selected key {selected_key} is empty or not set.")
+
+
+# ------------------ CLI Test ------------------
+
+if __name__ == "__main__":
+    print("\n=== Config Test ===\n")
+    try:
+        conf = load_configuration()
+        print("Configuration Loaded!")
+        print(f"Setup Type: {conf.get('setup_type')}")
+        print(f"Project ID: {conf.get('gcp_project_id')}")
+        
+        required = ["gemini_api_key", "gcs_bucket_name", "gcp_project_id"]
+        missing = [k for k in required if not conf.get(k)]
+        
+        if missing:
+             print(f"\n[WARNING] Missing: {missing}")
+        else:
+             print("\n✓ All required keys present.")
+             
+    except Exception as e:
+        print(f"\n[ERROR] {e}")