refactor: remove AssetSelector, use asset_manager singletons directly

- Remove AssetSelector wrapper class entirely
- automation.py now uses AudioLib and AssetProcessor directly
- Remove health_check methods (api_clients, automation, main)
- Remove video_library wrapper properties from asset_processor
- Simplify code by eliminating unnecessary abstraction layers

src/api_clients.py

@@ -745,55 +745,6 @@ class APIClients:
             logger.error(f"❌ Voice selection failed: {e}, using default")
             return await self._select_sequential_voice("female_young")
 
-
-
-    async def health_check(self) -> Dict[str, bool]:
-        """Check health of all API connections"""
-        logger.info("🏥 Running health check...")
-
-        health = {"gemini": False, "runwayml": False, "tts": False, "gcs": False}
-
-        try:
-            test_prompt = "Hello"
-            enhanced = await self.enhance_prompt(test_prompt)
-            if enhanced and len(enhanced) > 0:
-                health["gemini"] = True
-                logger.info("  ✅ Gemini API: Connected")
-            else:
-                logger.error("  ❌ Gemini API: No response")
-        except Exception as e:
-            logger.error(f"  ❌ Gemini API: {e}")
-
-        try:
-            from google.cloud.exceptions import NotFound
-
-            try:
-                self.gcs_bucket.exists()
-                health["gcs"] = True
-                logger.info("  ✅ Google Cloud Storage: Connected")
-            except NotFound:
-                logger.error("  ❌ Google Cloud Storage: Bucket not found")
-        except Exception as e:
-            logger.error(f"  ❌ Google Cloud Storage: {e}")
-
-        if self.runway_api_key and len(self.runway_api_key) > 10:
-            health["runwayml"] = True
-            logger.info("  ✅ RunwayML API (gen4_turbo): Configured")
-        else:
-            logger.error("  ❌ RunwayML API: Not configured")
-
-        if self.tts_client:
-            health["tts"] = True
-            logger.info("  ✅ TTS API: Configured")
-        else:
-            logger.error("  ❌ TTS API: Not configured")
-
-        all_healthy = all(health.values())
-        status = "✅ All systems operational!" if all_healthy else "⚠️ Some services have issues"
-        logger.info(f"\n{status}")
-
-        return health
-
     async def store_in_cache(self, file_path: str, method_type: str, file_ext: str = ".mp4") -> str:
         """Store file in Google Cloud Storage and return its public URL."""
         try:

src/asset_manager/asset_processor.py

@@ -29,11 +29,6 @@ class AssetProcessor:
         self.data_holder = data_holder
         self._video_lib = get_video_lib()
     
-    @property
-    def video_library(self) -> pd.DataFrame:
-        """Get video library from singleton"""
-        return self._video_lib.video_library
-    
     def _parse_duration(self, duration_str: str) -> int:
         """Parse duration from various string formats to integer seconds"""
         try:
@@ -100,8 +95,8 @@ Video Options: {video_context}
             selected = []
             for item in selection:
                 video_index = item["video_index"]
-                if video_index < len(self.video_library):
-                    video_row = self.video_library[self.video_library["Video URL (No Audio)"] == item["video_url"]]
+                if video_index < len(self._video_lib.video_library):
+                    video_row = self._video_lib.video_library[self._video_lib.video_library["Video URL (No Audio)"] == item["video_url"]]
                     video = video_row.iloc[0]
                     selected.append(
                         {
@@ -117,7 +112,7 @@ Video Options: {video_context}
                         }
                     )
                     if "alternate_video_index" in item:
-                        video_row = self.video_library[self.video_library["Video URL (No Audio)"] == item["alternate_video_url"]]
+                        video_row = self._video_lib.video_library[self._video_lib.video_library["Video URL (No Audio)"] == item["alternate_video_url"]]
                         video = video_row.iloc[0]
                         selected[-1]["alternate_url"] = video.get("Video URL (No Audio)", video.get("url", ""))
 
@@ -157,7 +152,7 @@ Video Options: {video_context}
                 f"{next((v.get('duration', 0) for v in self.data_holder.visual_assets['all_videos'] if v['url'] == row.get('Video URL (No Audio)')), 0)}s - "
                 f"Alignment: {row.get('Video Alignment with the TTS Script', row.get('alignment', ''))} - "
                 f"Usage Count: {self.data_holder.video_usage_count.get(row.get('Video URL (No Audio)'), 0)}"
-                for i, row in self.video_library.iterrows()
+                for i, row in self._video_lib.video_library.iterrows()
             ]
         )
 

src/asset_selector.py

@@ -1,79 +0,0 @@
-"""
-AssetSelector - Thin wrapper for backward compatibility
-Use asset_manager classes directly for new code.
-"""
-
-import pandas as pd
-from typing import List, Dict, Optional, Tuple
-from utils import logger
-from data_holder import DataHolder
-
-from asset_manager import get_video_lib, get_audio_lib, AssetProcessor
-
-
-class AssetSelector:
-    """
-    Wrapper class for backward compatibility.
-    New code should use asset_manager classes directly:
-    - get_video_lib() for video library
-    - get_audio_lib() for audio library  
-    - AssetProcessor(data_holder) for video selection
-    """
-    
-    def __init__(self, config: Dict, data_holder: DataHolder = None, gcloud_wrapper=None):
-        self.config = config
-        self.data_holder = data_holder
-        
-        # Use singletons from asset_manager
-        self._video_lib = get_video_lib()
-        initial_audio_index = config.get("current_audio_index", 0)
-        self._audio_lib = get_audio_lib(initial_audio_index)
-        self._audio_lib.current_audio_index = initial_audio_index
-        
-        # Processor for video selection (only create when data_holder available)
-        self._processor = AssetProcessor(data_holder) if data_holder else None
-
-    @property
-    def video_library(self) -> pd.DataFrame:
-        return self._video_lib.video_library
-    
-    @property
-    def audio_library(self) -> pd.DataFrame:
-        return self._audio_lib.audio_library
-    
-    @property
-    def current_audio_index(self) -> int:
-        return self._audio_lib.current_audio_index
-    
-    @current_audio_index.setter
-    def current_audio_index(self, value: int):
-        self._audio_lib.current_audio_index = value
-        self.config["current_audio_index"] = value
-
-    def inc_audio_index(self):
-        self._audio_lib.inc_audio_index()
-        self.config["current_audio_index"] = self._audio_lib.current_audio_index
-
-    def get_audio_beats(self, audio_link: str) -> Optional[List[float]]:
-        return self._audio_lib.get_audio_beats(audio_link)
-
-    async def select_videos(self, tts_script, timed_transcript, max_duration: int = 12) -> List[Dict]:
-        """Delegate to AssetProcessor"""
-        if not self._processor:
-            self._processor = AssetProcessor(self.data_holder)
-        return await self._processor.select_videos(tts_script, timed_transcript, max_duration)
-
-    def select_background_music(self) -> str:
-        selected = self._audio_lib.select_background_music()
-        self.config["current_audio_index"] = self._audio_lib.current_audio_index
-        return selected
-
-    def reset_audio_index(self):
-        self._audio_lib.reset_audio_index()
-        self.config["current_audio_index"] = 0
-
-    def select_random_videos(self, count: int) -> List[str]:
-        """Delegate to AssetProcessor"""
-        if not self._processor:
-            self._processor = AssetProcessor(self.data_holder)
-        return self._processor.select_random_videos(count)

src/automation.py

@@ -9,7 +9,6 @@ from typing import Dict, List, Optional, Any
 from pathlib import Path
 from api_clients import APIClients
 from video_renderer import VideoRenderer
-from asset_selector import AssetSelector
 from utils import logger
 import utils
 from moviepy.config import change_settings
@@ -27,18 +26,19 @@ import numpy as np
 from file_downloader import FileDownloader
 from data_holder import DataHolder
 import setup_config
-from asset_manager import get_asset_downloader
+from asset_manager import get_asset_downloader, get_audio_lib, AssetProcessor
 from file_downloader import FileDownloader
 
 class ContentAutomation:
-    def __init__(self, config: Dict[str, Any], data_holder: DataHolder = None, asset_selector: 'AssetSelector' = None, api_clients: 'APIClients' = None):
+    def __init__(self, config: Dict[str, Any], data_holder: DataHolder = None, api_clients: 'APIClients' = None):
         self.config = config
         self.data_holder = data_holder or DataHolder()
         # Reuse provided api_clients or create new one
         self.api_clients = api_clients or APIClients(config, self.data_holder)
         self.video_renderer = VideoRenderer(config, self.data_holder)
-        # Reuse provided asset_selector or create new one
-        self.asset_selector = asset_selector or AssetSelector(config, self.data_holder)
+        # Use asset_manager singletons directly
+        self._audio_lib = get_audio_lib(config.get("current_audio_index", 0))
+        self._asset_processor = AssetProcessor(self.data_holder)
         self.asset_downloader = get_asset_downloader()
         self.file_downloader = FileDownloader()
         self.pipeline_start_time = None
@@ -144,7 +144,7 @@ class ContentAutomation:
                 )
 
             logger.info("\n🎵 STEP 7: Background Music")
-            self.data_holder.visual_assets["background_music_url"] = self.asset_selector.select_background_music()
+            self.data_holder.visual_assets["background_music_url"] = self._audio_lib.select_background_music()
             local_path = self.file_downloader.safe_download(self.data_holder.visual_assets["background_music_url"])
             if local_path:
                 self.data_holder.visual_assets["background_music_local"] = str(local_path)
@@ -235,7 +235,7 @@ class ContentAutomation:
                     with AudioFileClip(self.data_holder.visual_assets["background_music_local"]) as audio_clip:
                         music_duration = audio_clip.duration - 0.5
 
-                    beat_times = self.asset_selector.get_audio_beats(self.data_holder.visual_assets["background_music_url"])
+                    beat_times = self._audio_lib.get_audio_beats(self.data_holder.visual_assets["background_music_url"])
                     if beat_times:
                         beat_times = self.extend_beats_to_audio_end(
                             beat_times,
@@ -263,11 +263,9 @@ class ContentAutomation:
                 logger.info(f"Beat times: {beat_times}")
                 
                 num_videos_needed = len(beat_times) + 2
-
-            logger.info(f"video library size: {len(self.asset_selector.video_library)}")
             
             # Select enough videos
-            self.data_holder.visual_assets["selected_videos"] = self.asset_selector.select_random_videos(num_videos_needed)
+            self.data_holder.visual_assets["selected_videos"] = self._asset_processor.select_random_videos(num_videos_needed)
             logger.info(self.data_holder.visual_assets["selected_videos"])
 
             videos = self.data_holder.visual_assets["selected_videos"]
@@ -358,9 +356,9 @@ class ContentAutomation:
     async def _download_bg_music(self, try_next: bool = False):
         logger.info("\n🎵 STEP 1: Background Music")
         if try_next:
-            self.asset_selector.inc_audio_index()
+            self._audio_lib.inc_audio_index()
 
-        self.data_holder.visual_assets["background_music_url"] = self.asset_selector.select_background_music()
+        self.data_holder.visual_assets["background_music_url"] = self._audio_lib.select_background_music()
         local_path = self.file_downloader.safe_download(self.data_holder.visual_assets["background_music_url"])
         if local_path:
             self.data_holder.visual_assets["background_music_local"] = str(local_path)
@@ -408,7 +406,7 @@ class ContentAutomation:
         """Generate visual assets in parallel (hook video + library videos)"""
         tasks = {
             "hook_video": self._generate_hook_video(content_strategy),
-            "selected_videos": self.asset_selector.select_videos(
+            "selected_videos": self._asset_processor.select_videos(
                 tts_script=self.data_holder.tts_script,
                 timed_transcript=self.data_holder.visual_assets["timed_transcript"]),
             }
@@ -468,52 +466,6 @@ class ContentAutomation:
             traceback.print_exc()
             raise
 
-    async def health_check(self) -> Dict[str, bool]:
-        """Comprehensive health check of all components"""
-        logger.info("🏥 Running comprehensive health check...")
-
-        # Check API clients
-        api_health = await self.api_clients.health_check()
-
-        # Check asset selector
-        try:
-            asset_selector_healthy = len(self.asset_selector.video_library) > 0
-            if not asset_selector_healthy:
-                logger.warning("  ⚠️ Asset Selector: Video library is empty")
-        except Exception as e:
-            asset_selector_healthy = False
-            logger.error(f"  ❌ Asset Selector: {e}")
-
-        # Check video renderer
-        try:
-            video_renderer_healthy = self.video_renderer.temp_dir.exists()
-            if not video_renderer_healthy:
-                logger.warning("  ⚠️ Video Renderer: Temp directory issue")
-        except Exception as e:
-            video_renderer_healthy = False
-            logger.error(f"  ❌ Video Renderer: {e}")
-
-        # Combine all health statuses
-        health_status = {
-            **api_health,
-            "asset_selector": asset_selector_healthy,
-            "video_renderer": video_renderer_healthy,
-        }
-
-        operational_services = sum(health_status.values())
-        total_services = len(health_status)
-
-        print(f"\n📊 Health Summary: {operational_services}/{total_services} services operational")
-
-        if operational_services == total_services:
-            print("🎉 System is fully operational and ready for production!")
-        elif operational_services >= total_services - 2:
-            print("⚠️  System is mostly operational, but some features may be limited")
-        else:
-            print("❌ System has significant issues that need attention")
-
-        return health_status
-
     def extend_beats_to_audio_end(
         self,
         beats: List[float],
@@ -533,7 +485,6 @@ class ContentAutomation:
 
         return beats
 
-
     async def simple_demo(self):
         """Simple demo with proper audio handling"""
         logger.info("🎬 Starting Simple Demo with Audio Fix...")

src/main.py

@@ -154,33 +154,6 @@ async def run_pipeline(
 
     return result
 
-
-async def health_check_command(automation: ContentAutomation):
-    """Run health check on all services"""
-    try:
-        health_status = await automation.health_check()
-
-        print("\n" + "=" * 50)
-        print("🏥 SYSTEM HEALTH CHECK RESULTS")
-        print("=" * 50)
-
-        for service, status in health_status.items():
-            icon = "✅" if status else "❌"
-            print(f"{icon} {service.upper():<15} {'OPERATIONAL' if status else 'ISSUE DETECTED'}")
-
-        if all(health_status.values()):
-            print("\n🎉 All systems are ready for production!")
-            return 0
-        else:
-            print("\n⚠️  Some services need attention before running the pipeline.")
-            print("   Check the logs above for details.")
-            return 1
-
-    except Exception as e:
-        logger.error(f"Health check failed: {e}")
-        return 1
-
-
 async def test_command(automation: ContentAutomation):
     """Run simple demo test"""
     logger.info("\n🧪 Running Simple Demo Test...")

src/process_csv.py

@@ -113,7 +113,7 @@ async def process_row(row, config: dict):
     if SHARED_API_CLIENTS:
         SHARED_API_CLIENTS.data_holder = dataHolder
 
-    # AssetSelector uses singletons internally, no need to share
+    # ContentAutomation uses asset_manager singletons directly
     automation = ContentAutomation(
         config, dataHolder,
         api_clients=SHARED_API_CLIENTS