""" Unified Configuration Module Simple .env-based configuration. No TOML files needed. All configuration comes from environment variables. """ import os import json import logging from pathlib import Path from typing import Dict, Any, Optional from dotenv import load_dotenv # Configure logging from src.logger_config import logger # ------------------ Singleton & Cache ------------------ _cached_config: Optional[Dict[str, Any]] = None _config_initialized: bool = False # ------------------ 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 environment variables. 1. Load .env 2. Resolve GCP Project & Secrets 3. Build config dict """ global _cached_config, _config_initialized # Return cache if valid if _config_initialized and not force_reload and _cached_config is not None: return _cached_config load_dotenv() setup_type = os.getenv("SETUP_TYPE") logger.info(f"āœ“ Loaded setup config: {setup_type}") # Build config from environment variables config = { # Core settings "setup_type": setup_type, "auth_method": "auto_detect", "github_token": os.getenv("GITHUB_TOKEN"), "ai_generation": _env_bool("AI_GENERATION"), "video_merge_type": os.getenv("VIDEO_MERGE_TYPE"), "video_merge_process": os.getenv("VIDEO_MERGE_PROCESS"), "job_index": _env_int("JOB_INDEX", 0), "total_jobs": _env_int("TOTAL_JOBS", 1), # 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"), "xai_api_key": os.getenv("XAI_API_KEY") or os.getenv("XAI_GROK_API_KEY"), "oneup_api_key": os.getenv("ONEUP_API_KEY"), "spark_key": os.getenv("SPARK_KEY"), "runway_2nd_api_key": os.getenv("RUNWAY_2ND_API_KEY"), "runway_3rd_api_key": os.getenv("RUNWAY_3RD_API_KEY"), "runway_4th_api_key": os.getenv("RUNWAY_4TH_API_KEY"), # GCS "gcs_bucket_name": os.getenv("GCS_BUCKET_NAME"), "gcs_bucket_folder_name": os.getenv("GCS_BUCKET_FOLDER_NAME", "video_rename"), "gcloud_final_data_credentials": os.getenv("GCLOUD_FINAL_DATA_CREDENTIALS"), "gcloud_test_data_credentials": os.getenv("GCLOUD_TEST_DATA_CREDENTIALS"), # Google Sheets "gsheet_id": os.getenv("GSHEET_ID"), "video_library_gsheet_worksheet": os.getenv("VIDEO_LIBRARY_GSHEET_WORKSHEET"), "audio_library_gsheet_worksheet": os.getenv("AUDIO_LIBRARY_GSHEET_WORKSHEET"), "logs_worksheet": f'{os.getenv("SETUP_TYPE")} LOGS', "content_strategy_worksheet": os.getenv("CONTENT_STRATEGY_FILE") or os.getenv("CONTENT_STRATEGY_GSHEET_WORKSHEET"), "gsheet_worksheet_text_overlay_column": os.getenv("GSHEET_WORKSHEET_TEXT_OVERLAY_COLUMN"), "product_video_library_gsheet_worksheet": os.getenv("PRODUCT_VIDEO_LIBRARY_GSHEET_WORKSHEET"), "product_name": os.getenv("PRODUCT_NAME"), "product_info": os.getenv("PRODUCT_INFO"), # Pipeline settings "caption_style": os.getenv("CAPTION_STYLE"), "beat_method": os.getenv("BEAT_METHOD"), "on_screen_text": os.getenv("ON_SCREEN_TEXT"), # May be string or bool in usage, keeping strict getenv for safety or _env_bool if confirmed boolean "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"), "use_grok_video": _env_bool("USE_GROK_VIDEO"), "video_generation_type": os.getenv("VIDEO_GENERATION_TYPE", "runway").lower(), "only_random_videos": _env_bool("ONLY_RANDOM_VIDEOS"), "generation_count": _env_int("GENERATION_COUNT", 100), "gsheet_worksheet_text_overlay": os.getenv("GSHEET_WORKSHEET_TEXT_OVERLAY"), "log_gsheet_id": os.getenv("LOG_GSHEET_ID", "1GARlzLgKB55aAA3EKgtpD7pK1eFG0FgUbW4eSmQy9C4"), "use_mock_gemini": _env_bool("TEST_AUTOMATION"), # Test settings "test_automation": _env_bool("TEST_AUTOMATION"), "test_data_directory": os.getenv("TEST_DATA_DIRECTORY"), "delete_all_a2e_videos": _env_bool("DELETE_ALL_A2E_VIDEOS"), "drive_upload_folder_id": os.getenv("DRIVE_UPLOAD_FOLDER_ID"), "drive_video_folder_id": os.getenv("DRIVE_VIDEO_FOLDER_ID"), # publisher settings "publisher_logs_worksheet": os.getenv("PUBLISHER_LOGS_WORKSHEET") or f'{os.getenv("GCS_BUCKET_NAME")} Publisher LOGS', "social_media_publisher_provider": os.getenv("SOCIAL_MEDIA_PUBLISHER_PROVIDER", "hybrid"), "social_media_publish_platforms": os.getenv("SOCIAL_MEDIA_PUBLISH_PLATFORMS"), "social_media_publisher_upload_limit_per_account": _env_int("SOCIAL_MEDIA_PUBLISHER_UPLOAD_LIMIT_PER_ACCOUNT", 1), "video_publish_public": _env_bool("VIDEO_PUBLISH_PUBLIC"), "gcs_bucket_name_public_for_social_publish": os.getenv("GCS_BUCKET_NAME_PUBLIC_FOR_SOCIAL_PUBLISH"), # TikTok "tiktok_access_token": os.getenv("TIKTOK_ACCESS_TOKEN"), "tiktok_client_key": os.getenv("TIKTOK_CLIENT_KEY"), "tiktok_client_secret": os.getenv("TIKTOK_CLIENT_SECRET"), # YouTube "youtube_refresh_token": os.getenv("YOUTUBE_REFRESH_TOKEN"), "youtube_client_secrets_json": os.getenv("YOUTUBE_CLIENT_SECRETS_JSON"), "video_category": os.getenv("VIDEO_CATEGORY", "22"), "scheduled_time": os.getenv("SCHEDULED_TIME"), "youtube_client_id": os.getenv("YOUTUBE_CLIENT_ID"), "youtube_client_secret": os.getenv("YOUTUBE_CLIENT_SECRET"), # Instagram "meta_app_id": os.getenv("META_APP_ID"), "meta_app_secret": os.getenv("META_APP_SECRET"), "instagram_app_id": os.getenv("INSTAGRAM_APP_ID"), "instagram_app_secret": os.getenv("INSTAGRAM_APP_SECRET"), # Threads (Separate App option) "threads_app_id": os.getenv("THREADS_APP_ID"), "threads_app_secret": os.getenv("THREADS_APP_SECRET"), # Facebook (can reuse Instagram/Meta app or have separate) "facebook_app_id": os.getenv("FACEBOOK_APP_ID"), "facebook_app_secret": os.getenv("FACEBOOK_APP_SECRET"), # Misc "encryption_key": os.getenv("ENCRYPTION_KEY"), "hf_token": os.getenv("HF_TOKEN"), "hf_space_url": os.getenv("HF_SPACE_URL"), # X (Twitter) "x_client_id": os.getenv("X_CLIENT_ID"), "x_client_secret": os.getenv("X_CLIENT_SECRET"), # ImageKit "imagekit_public_key": os.getenv("IMAGEKIT_PUBLIC_KEY"), "imagekit_private_key": os.getenv("IMAGEKIT_PRIVATE_KEY"), "imagekit_id": os.getenv("IMAGEKIT_ID"), "imagekit_url_endpoint": os.getenv("IMAGEKIT_URL_ENDPOINT"), } # On-screen CTA options config["on_screen_cta"] = [ "LINK IN BIO šŸ›ļø", "šŸ”— LINK IN ACCOUNT šŸ›ļø", "USE CODE: TIKTOK10 AT CHECKOUT", "šŸ”„ SELLING FAST ā€” LINK ON PROFILE", "LINK IN BIO šŸ›ļø", "šŸ›ļø SALE ENDING SOON - šŸ”— LINK IN BIO" ] # Initialize default empty collections config["video_usage_count"] = {} config["avatar_usage_count"] = {} config["visual_assets"] = {} _cached_config = config _config_initialized = True return config # ------------------ Public API ------------------ class ConfigProxy: """ Singleton proxy to access configuration. Lazily loads config on first access. """ def __init__(self): self._config = None def _ensure_loaded(self): if self._config is None: self._config = load_configuration() def get(self, key: str, default: Any = None) -> Any: self._ensure_loaded() # Direct match if key in self._config: return self._config[key] # Lowercase match if key.lower() in self._config: return self._config[key.lower()] return default def __getitem__(self, key: str) -> Any: self._ensure_loaded() if key in self._config: return self._config[key] if key.lower() in self._config: return self._config[key.lower()] raise KeyError(key) def __contains__(self, key: str) -> bool: self._ensure_loaded() return key in self._config def items(self): self._ensure_loaded() return self._config.items() def set(self, key: str, value: Any): """Set a configuration value.""" self._ensure_loaded() self._config[key] = value def __setitem__(self, key: str, value: Any): self._ensure_loaded() self._config[key] = value def reload(self): self._config = load_configuration(force_reload=True) # Global singleton config = ConfigProxy() def get_config_value(key: str, default: Any = None) -> Any: """Get a config value.""" value = config.get(key.lower(), default) return value if value is not None else default def set_config_value(key: str, value: Any): """Set a config value.""" config.set(key, value) # ------------------ Job Environment Configuration ------------------ def configure_job_environment(job_index: int): """ Configure environment variables for a parallel job index. Handles API key rotation for Google AI Studio. """ if job_index is None: return google_keys_env = [ "GOOGLE_AISTUDIO_2ND_API_KEY", "GOOGLE_AISTUDIO_3RD_API_KEY", "GOOGLE_AISTUDIO_IMAGEN_API_KEY", "GOOGLE_AISTUDIO_IMAGEN_API_KEY", "GOOGLE_AI_API_KEY", "GEMINI_API_KEY" ] os.environ["RUNWAYML_API_KEY"] = os.getenv("RUNWAYML_API_KEY", "") selected_key = google_keys_env[job_index % len(google_keys_env)] new_api_key = os.getenv(selected_key, "") if new_api_key: os.environ["GEMINI_API_KEY"] = new_api_key set_config_value("gemini_api_key", new_api_key) logger.debug(f"Using Google key: {selected_key}") else: logger.debug(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')}") required = ["gemini_api_key", "gcs_bucket_name"] 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}")