src/pipelines/base.py

import time
import asyncio
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

from src.logger_config import logger
import src.utils as utils
from src.config import get_config_value, set_config_value
from google_src.gcs_utils import upload_file_to_gcs
from src.video_renderer import VideoRenderer
from google_src.tts import GoogleTTS
from google_src.stt import GoogleSTT
from file_downloader import get_file_downloader
from src.asset_manager import get_asset_downloader, get_audio_lib, AssetProcessor, get_text_overlay_lib
from video_editor import loudness_normalize, remove_green_bg
from video_editor.onscreen_cta import add_cta
from google_src.ai_studio_sdk import generate

class ContentAutomationBase(ABC):
    def __init__(self):
        # Initialize Google Services directly
        self.tts = GoogleTTS()
        self.stt = GoogleSTT()
        
        self.video_renderer = VideoRenderer()
        # Use asset_manager singletons directly
        self._audio_lib = get_audio_lib(get_config_value("current_audio_index", 0))
        # self._audio_lib.use_temp_audio_library()
        if get_config_value("on_screen_text", False):
            self._text_overlay_lib = get_text_overlay_lib(get_config_value("current_audio_index", 0))
        else:
            self._text_overlay_lib = None
        self._asset_processor = AssetProcessor()
        self.asset_downloader = get_asset_downloader()
        self.file_downloader = get_file_downloader()  # Use singleton
        self.pipeline_start_time = None

    async def workflow_pre_config(self):
        logger.info("\n🎭 STEP 1: Pre Config")
        self.pipeline_start_time = time.time()

        logger.info("\n STEP 2: Load video assets")
        videos = await self.asset_downloader.download_all_videos()
        
        visual_assets = get_config_value("visual_assets", {})
        visual_assets["all_videos"] = videos
        set_config_value("visual_assets", visual_assets)

        logger.info("\n🎵 STEP 3: Background Music")
        await self._download_bg_music(try_next=False)

    async def _download_bg_music(self, try_next: bool = False):
        if try_next:
            self._audio_lib.inc_audio_index()

        visual_assets = get_config_value("visual_assets", {})
        visual_assets["background_music_url"] = self._audio_lib.select_background_music()
        local_path = await self.file_downloader.safe_download(visual_assets["background_music_url"])
        if local_path:
            visual_assets["background_music_local"] = str(local_path)
        set_config_value("visual_assets", visual_assets)

    async def video_features(self, video_path: str):
        if get_config_value("use_1x1_ratio", False):
            video_path = utils.ratio_1x1_to9x16(video_path)

        if get_config_value("is_a2e_lip_sync", False):
            video_path = remove_green_bg.process_video_with_ffmpeg(get_config_value("visual_assets")["a2e_video_local_path"], video_path)

        if get_config_value("on_screen_text", False):
            visual_assets = get_config_value("visual_assets", {})
            text_overlay = self._text_overlay_lib.select_text_overlay()
            visual_assets["timed_transcript"] = [
                    {
                        "start_time": 0,
                        "end_time": 15,
                        "word": text_overlay
                    }
                ]
            set_config_value("visual_assets", visual_assets)
            video_path = await self.video_renderer._add_timed_subtitles(video_path, group_all=True, position=250)
            set_config_value("video_no_audio_path_with_caption", video_path)

        elif get_config_value("is_onscreen_cta", False):
            on_screen_cta = get_config_value("on_screen_cta")[get_config_value("current_audio_index", 0) % len(get_config_value("on_screen_cta"))]
            cap_parma = self.video_renderer._get_caption_style()[0]()
            padding = cap_parma.get("safe_zone_padding", 20)
            # bottom_safe_y = 250
            video_path = add_cta(
                video_path,
                on_screen_cta,
                padding=padding
                # bottom_safe_y=bottom_safe_y
            )

        return video_path

    @abstractmethod
    async def generate_content(self) -> str:
        """
        Abstract method to generate content. 
        Must return the path to the video file before audio merging.
        """
        pass

    def generate_broll_filename(self) -> str | None:
        """
        Generate a safe filename based on the provided prompt using Gemini.
        Returns a string suitable for use as a filename (lowercase, underscores, no special chars).
        """
        try:
            import random
            seed = random.randint(1000, 9999)
            
            user_prompt = f"""Generate ONE unique luxury b-roll video title (2-3 words max).

    TONE: Modern, aspirational, Instagram-worthy. Think luxury influencer content.

    EXAMPLES of the vibe (DO NOT COPY THESE):
    "Golden Hour Aesthetic", "Rich Life Moments", "Boss Life Vibes", "Money Moves", "Champagne Dreams", "VIP Treatment", "Designer Everything", "Empire Building"

    WORD COMBINATIONS - Mix and match:

    Part 1 (Descriptors):
    Golden, Luxury, Rich, Elite, Opulent, Wealthy, Premium, High End, Lavish, Classy, Million Dollar, Success, Boss, First Class, Diamond, Gold, Platinum, VIP, Exclusive, Private, Penthouse, Yacht, Designer, Five Star, Empire, Sophisticated, Expensive, Refined, Upper Class, Affluent

    Part 2 (Vibes/Things):
    Hour, Lifestyle, Life, Vibes, Moments, Routine, Living, Aesthetic, Mindset, Dreams, Energy, Moves, Standard, Treatment, Access, Views, Club, Everything, Quality, Taste, Style, Circle, Society, Story

    Patterns:
    - [Luxury word] + [Thing]: "Diamond Lifestyle", "Platinum Energy"
    - [Adjective] + [Vibe word]: "Exclusive Vibes", "Premium Moments"  
    - [Noun] + [Action/State]: "Empire Mindset", "Yacht Dreams"

    CRITICAL: Be creative. Mix words in fresh ways. Avoid repeating common combos.
    Variation seed: {seed}

    Output ONLY the title (2-3 words). No punctuation, no explanation."""

            return generate(user_prompt)
            
        except Exception as e:
            logger.error("Failed to generate b-roll filename: %s", e)
            return None

    async def run_pipeline(self):
        await self.workflow_pre_config()
        
        # Generate the video components
        final_video_path = await self.generate_content()

        # Apply common video features
        final_video_path = await self.video_features(final_video_path)

        # STEP 7: Add audio to video
        logger.info("\n🔊 STEP 8: Add Audio to Video")
        final_video_path = await self.video_renderer.add_audio_to_video(final_video_path)

        final_video_path = loudness_normalize.normalize_loudness(final_video_path)

        # STEP 8: Upload to cloud storage
        final_url = None
        gcs_filename = None
        # Generate UUID-based filename managed by utility
        logger.info("\n☁️ STEP 9: Cloud Storage Upload")

        file_name = self.generate_broll_filename()
        # Final Result
        upload_result = upload_file_to_gcs(
            final_video_path,
            destination_blob_name=file_name
        )
        final_url = upload_result["url"]
        gcs_filename = upload_result["gcs_filename"]

        # Pipeline completion
        if self.pipeline_start_time:
            elapsed_time = time.time() - self.pipeline_start_time
        else:
            elapsed_time = 0
            
        logger.info(f"\n✅ Enhanced pipeline completed in {elapsed_time:.2f}s")

        return {
            "success": True,
            "final_url": final_url,
            "gcs_filename": gcs_filename,
            "local_path": final_video_path,
        }