""" AudioLib - Singleton class for managing audio library from Google Sheets music collections: https://drive.google.com/drive/folders/184VHlk1dPEAp5Xo_jJEe62h7yylxoBit """ import os import re 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 AudioLib: """ Singleton class that loads and manages audio library from Google Sheets. Handles background music selection and beat timing. Usage: audio_lib = get_audio_lib() music_url = audio_lib.select_background_music() beats = audio_lib.get_audio_beats(music_url) """ def __init__(self, gcloud_wrapper: Optional[GCloudWrapper] = None, initial_audio_index: int = 0): self._gcloud_wrapper = gcloud_wrapper or get_default_wrapper() self._audio_library: pd.DataFrame = self._load_from_gsheet() if len(self._audio_library) == 0: raise ValueError("Audio library is empty! Check AUDIO_LIBRARY_GSHEET_WORKSHEET env var and Google Sheet access.") self._current_audio_index = initial_audio_index % len(self._audio_library) logger.debug(f"✓ AudioLib initialized with {len(self._audio_library)} audio tracks, starting at index {self._current_audio_index}") @property def audio_library(self) -> pd.DataFrame: """Get the audio library DataFrame""" return self._audio_library @property def current_audio_index(self) -> int: """Get current audio index""" return self._current_audio_index @current_audio_index.setter def current_audio_index(self, value: int) -> None: """Set current audio index (wraps around)""" if len(self._audio_library) > 0: self._current_audio_index = value % len(self._audio_library) def _load_from_gsheet(self, account_id: str = "test_data") -> pd.DataFrame: """ Load audio library from Google Sheet. Args: account_id: Which account to use ('final_data' or 'test_data') """ try: worksheet_name = get_config_value("audio_library_gsheet_worksheet") if not worksheet_name: logger.error("AUDIO_LIBRARY_GSHEET_WORKSHEET env var not set!") return pd.DataFrame() logger.debug(f"Loading audio library using account: {account_id}") googleSheetReader = GoogleSheetReader( worksheet_name=worksheet_name, gcloud_wrapper=self._gcloud_wrapper, account_id=account_id, ) audio_df = googleSheetReader.get_filtered_dataframe() # Filter by beats timing if in beats_cut mode if get_config_value("setup_type") == "beats_cut": audio_df = clean_and_drop_empty(audio_df, "Beats Timing(SS:FF) AT 25FPS") return clean_and_drop_empty(audio_df, "AUDIO_LINK") 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 audio 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('audio_library_gsheet_worksheet')}'") else: raise ValueError(f"Failed to load audio library from Google Sheet: {error_msg}") return pd.DataFrame() def inc_audio_index(self) -> None: """Increment current audio index (wraps around)""" self._current_audio_index = (self._current_audio_index + 1) % len(self._audio_library) def select_background_music(self) -> str: """ Select background music SEQUENTIALLY (not random). Each call increments the index to ensure different music for each video. Returns: URL of the selected audio track """ if self._audio_library.empty: logger.error("❌ Audio library is empty") return "" selected = self._audio_library.iloc[self._current_audio_index]["AUDIO_LINK"] logger.debug( f"🎵 Selected background music #{self._current_audio_index + 1}/{len(self._audio_library)}: {selected}" ) # Increment index for next call (loop back to start if needed) self._current_audio_index = (self._current_audio_index + 1) % len(self._audio_library) return selected def get_audio_beats(self, audio_link: str) -> Optional[List[float]]: """ Load audio beats timing from audio_library and convert SS:FF (25 FPS) → seconds (float) Example: "01:12" → 1 + 12/25 = 1.48 Args: audio_link: URL of the audio track Returns: List of beat times in seconds, or None if not found """ try: if self._audio_library.empty: logger.error("Audio library is empty") return None # Find matching row row = self._audio_library.loc[ self._audio_library["AUDIO_LINK"] == audio_link ] if row.empty: logger.error(f"No audio entry found for: {audio_link}") return None beats_raw = row.iloc[0]["Beats Timing(SS:FF) AT 25FPS"] if pd.isna(beats_raw) or not str(beats_raw).strip(): logger.warning(f"No beat data for audio: {audio_link}") return None beats: List[float] = [] for token in str(beats_raw).split(","): token = token.strip() if ":" not in token: continue sec, frame = token.split(":", 1) beats.append( round(int(sec) + (int(frame) / 25.0), 2) ) return beats if beats else None except Exception as e: logger.error( f"Failed to compute audio beats map for {audio_link}: {e}" ) return None def reset_audio_index(self) -> None: """Reset audio index to start from beginning (useful for batch processing)""" self._current_audio_index = 0 logger.debug("🔄 Reset background music index to 0") def use_temp_audio_library(self) -> None: """ Temporary method to use specific audio tracks provided by the user. Overrides the loaded audio library. """ data = [ # { # "AUDIO_LINK": "https://storage.googleapis.com/somira/audio-track2.mp3", # "Beats Timing(SS:FF) AT 25FPS": "00:18,01:16,02:16,03:12,03:24,04:06,04:13,04:22,05:06,05:19,05:24,06:04,06:09,06:14,06:19,07:00,07:05,07:10,07:16,08:08,09:02,09:19,10:12,11:04,11:22,12:15" # }, { "AUDIO_LINK": "https://storage.googleapis.com/somira/audio-track3.mp3", "Beats Timing(SS:FF) AT 25FPS": "02:06,02:16,03:02,03:15,04:01,04:12,04:22,05:08,05:18,06:03,06:15,07:03,07:14,08:02,08:12,08:22,09:08,09:18,10:03" } ] self._audio_library = pd.DataFrame(data) self._current_audio_index = 0 logger.debug(f"✓ AudioLib switched to temporary library with {len(self._audio_library)} tracks") def __len__(self) -> int: return len(self._audio_library) # Module-level singleton instance _audio_lib: Optional[AudioLib] = None def get_audio_lib(initial_audio_index: int = 0) -> AudioLib: """ Get the singleton AudioLib instance. Args: initial_audio_index: Starting index for audio selection (only used on first call) Returns: AudioLib: The singleton instance """ global _audio_lib if _audio_lib is None: _audio_lib = AudioLib(initial_audio_index=initial_audio_index) return _audio_lib def reset_audio_lib() -> None: """Reset the singleton (useful for testing)""" global _audio_lib _audio_lib = None