File size: 7,882 Bytes
54639e8 f20025d 5f6fd2a 9d54dfd 54639e8 5f6fd2a 54639e8 12b7787 5f6fd2a 54639e8 ff14cb7 54639e8 6ffa615 782b3ed 54639e8 12b7787 54639e8 12b7787 54639e8 455e3bd 54639e8 455e3bd 54639e8 455e3bd 54639e8 ff14cb7 af2a248 ff60344 af2a248 ff60344 af2a248 ff14cb7 54639e8 ff14cb7 1d300b0 b8b8c8c 54639e8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 |
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,
}
|