Tools / src /asset_manager /text_overlay_lib.py
jebin2's picture
refactor: Centralize logger import to src.logger_config across various modules.
f20025d
"""
TextOverlayLib - Singleton class for managing text overlay content from Google Sheets
"""
import os
import pandas as pd
from typing import Optional
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 TextOverlayLib:
"""
Singleton class that loads and manages text overlay content from Google Sheets.
Uses round-robin auto-increment for text overlay selection.
Usage:
text_overlay_lib = get_text_overlay_lib()
text = text_overlay_lib.select_text_overlay()
"""
def __init__(self, gcloud_wrapper: Optional[GCloudWrapper] = None, initial_index: int = 0):
self._gcloud_wrapper = gcloud_wrapper or get_default_wrapper()
self._text_overlay_library: pd.DataFrame = self._load_from_gsheet()
if len(self._text_overlay_library) == 0:
raise ValueError("Text overlay library is empty! Check GSHEET_WORKSHEET_TEXT_OVERLAY env var and Google Sheet access.")
self._current_index = initial_index % len(self._text_overlay_library)
logger.debug(f"βœ“ TextOverlayLib initialized with {len(self._text_overlay_library)} text overlays, starting at index {self._current_index}")
@property
def text_overlay_library(self) -> pd.DataFrame:
"""Get the text overlay library DataFrame"""
return self._text_overlay_library
@property
def current_index(self) -> int:
"""Get current text overlay index"""
return self._current_index
@current_index.setter
def current_index(self, value: int) -> None:
"""Set current text overlay index (wraps around)"""
if len(self._text_overlay_library) > 0:
self._current_index = value % len(self._text_overlay_library)
def _load_from_gsheet(self, account_id: str = "test_data") -> pd.DataFrame:
"""
Load text overlay library from Google Sheet.
Args:
account_id: Which account to use ('final_data' or 'test_data')
"""
try:
worksheet_name = get_config_value("gsheet_worksheet_text_overlay")
if not worksheet_name:
logger.error("GSHEET_WORKSHEET_TEXT_OVERLAY env var not set!")
return pd.DataFrame()
logger.debug(f"Loading text overlay library using account: {account_id}")
googleSheetReader = GoogleSheetReader(
worksheet_name=worksheet_name,
gcloud_wrapper=self._gcloud_wrapper,
account_id=account_id,
)
text_df = googleSheetReader.get_filtered_dataframe()
column_name = get_config_value("gsheet_worksheet_text_overlay_column")
return clean_and_drop_empty(text_df, column_name)
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() or "forbidden" in error_msg.lower():
logger.error(f"❌ PERMISSION ERROR loading text overlay library: {error_msg}")
logger.error("Share the Google Sheet with the service account email as Editor!")
elif "404" in error_msg or "not found" in error_msg.lower():
logger.error(f"❌ WORKSHEET NOT FOUND: '{get_config_value('GSHEET_WORKSHEET_TEXT_OVERLAY')}'")
else:
logger.error(f"Failed to load text overlay library from Google Sheet: {error_msg}")
return pd.DataFrame()
def inc_index(self) -> None:
"""Increment current text overlay index (wraps around)"""
self._current_index = (self._current_index + 1) % len(self._text_overlay_library)
def select_text_overlay(self) -> str:
"""
Select text overlay SEQUENTIALLY (not random).
Each call increments the index to ensure different text for each video.
Returns:
Text overlay string
"""
if self._text_overlay_library.empty:
logger.error("❌ Text overlay library is empty")
return ""
column_name = get_config_value("gsheet_worksheet_text_overlay_column")
selected = self._text_overlay_library.iloc[self._current_index][column_name]
logger.debug(
f"πŸ“ Selected text overlay #{self._current_index + 1}/{len(self._text_overlay_library)}: {selected}"
)
# Increment index for next call (loop back to start if needed)
self._current_index = (self._current_index + 1) % len(self._text_overlay_library)
return selected
def reset_index(self) -> None:
"""Reset text overlay index to start from beginning (useful for batch processing)"""
self._current_index = 0
logger.debug("πŸ”„ Reset text overlay index to 0")
def __len__(self) -> int:
return len(self._text_overlay_library)
# Module-level singleton instance
_text_overlay_lib: Optional[TextOverlayLib] = None
def get_text_overlay_lib(initial_index: int = 0) -> TextOverlayLib:
"""
Get the singleton TextOverlayLib instance.
Args:
initial_index: Starting index for text overlay selection (only used on first call)
Returns:
TextOverlayLib: The singleton instance
"""
global _text_overlay_lib
if _text_overlay_lib is None:
_text_overlay_lib = TextOverlayLib(initial_index=initial_index)
return _text_overlay_lib
def reset_text_overlay_lib() -> None:
"""Reset the singleton (useful for testing)"""
global _text_overlay_lib
_text_overlay_lib = None