""" ContentStrategyLib - Singleton class for managing content strategies from Google Sheets """ import os import pandas as pd from typing import Optional, List from src.logger_config import logger from src.utils import clean_and_drop_empty from google_src.google_sheet import GoogleSheetReader from google_src import get_default_wrapper, GCloudWrapper from src.config import get_config_value class ContentStrategyLib: """ Singleton class that loads and manages content strategies from Google Sheets. Replaces the legacy CSV-based loading. Usage: lib = get_content_strategy_lib() df = lib.strategies # Auto-loads if empty """ def __init__(self, gcloud_wrapper: Optional[GCloudWrapper] = None): self._gcloud_wrapper = gcloud_wrapper or get_default_wrapper() self._strategies_df: pd.DataFrame = pd.DataFrame() self._worksheet_name = get_config_value("content_strategy_worksheet") if not self._worksheet_name: logger.warning("content_strategy_worksheet env var not set. Content strategies will be empty.") def load_strategies(self, account_id: str = "test_data") -> pd.DataFrame: """ Load content strategies from Google Sheet. Args: account_id: Which account to use ('final_data' or 'test_data') """ if not self._worksheet_name: return pd.DataFrame() try: logger.debug(f"Loading content strategies from Google Sheet: {self._worksheet_name} ({account_id})") googleSheetReader = GoogleSheetReader( worksheet_name=self._worksheet_name, gcloud_wrapper=self._gcloud_wrapper, account_id=account_id, ) df = googleSheetReader.get_filtered_dataframe() # Validate required columns similar to the old CSV method if get_config_value("ON_SCREEN_TEXT", "false").lower() == "true": required_cols = ["On-Screen Text"] else: required_cols = [ "Gemini Imagen4 Ultra Prompt (specific)", "TTS Script (AI Avatar)", "Runway Prompt Gen4 Turbo", "Captions", ] missing_cols = [col for col in required_cols if col not in df.columns] if missing_cols: logger.error(f"❌ Google Sheet missing columns: {', '.join(missing_cols)}") # We return empty or partial? Let's return what we have but warn self._strategies_df = df logger.debug(f"✓ Loaded {len(df)} content strategies") return df except Exception as e: error_msg = str(e) if str(e) else type(e).__name__ if "403" in error_msg or "permission" in error_msg.lower(): logger.error(f"❌ PERMISSION ERROR loading strategies: {error_msg}") elif "404" in error_msg or "not found" in error_msg.lower(): logger.error(f"❌ WORKSHEET NOT FOUND: '{self._worksheet_name}'") else: logger.error(f"Failed to load content strategies: {error_msg}") return pd.DataFrame() @property def strategies(self) -> pd.DataFrame: """Get the cached strategies DataFrame. Loads if empty.""" if self._strategies_df.empty: return self.load_strategies() return self._strategies_df def reload(self): """Force reload from Google Sheets""" return self.load_strategies() # Module-level singleton instance _content_strategy_lib: Optional[ContentStrategyLib] = None def get_content_strategy_lib() -> ContentStrategyLib: """ Get the singleton ContentStrategyLib instance. """ global _content_strategy_lib if _content_strategy_lib is None: _content_strategy_lib = ContentStrategyLib() return _content_strategy_lib