|
|
""" |
|
|
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}" |
|
|
) |
|
|
|
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
|
|
|
_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 |
|
|
|