src/asset_manager/audio_lib.py

"""
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