feat: enhance video analysis pipeline with configuration loading and error handling

Author: smirk-dev

app/video_pipeline.py

@@ -1,27 +1,132 @@
-import time, threading
 
-def run_basic(video_path:str):
-    # Quick metadata
-    return {"duration": 123.0}
+import threading
+import yaml
+import traceback
+from src.processors.video_analyzer import VideoAnalyzer
+from src.processors.audio_analyzer import AudioAnalyzer
+from src.processors.scene_detector import SceneDetector
+from src.ai_models.emotion_detector import EmotionDetector
+from src.ai_models.intelligent_content_analyzer import IntelligentContentAnalyzer
+from src.suggestions.cut_suggester import CutSuggester
+import os
 
-def run_deep(video_path:str):
-    # Simulate heavy analysis
-    time.sleep(1.2)
-    return {
-        "emotions": {"happy":0.62,"neutral":0.25,"sad":0.13},
-        "objects": [{"label":"person","confidence":0.94}],
-        "cut_suggestions":[{"timestamp":28.4,"reason":"Scene shift","confidence":0.78}]
-    }
+def load_config():
+    with open(os.path.join(os.path.dirname(__file__), '../config.yaml'), 'r', encoding='utf-8') as f:
+        return yaml.safe_load(f)
+
+def run_basic(video_path:str, config):
+    # Quick metadata using VideoAnalyzer
+    try:
+        va = VideoAnalyzer(config)
+        meta = va.get_video_metadata(video_path)
+        return {"duration": meta.get("duration", 0), "status": "basic_ready"}
+    except Exception as e:
+        return {"duration": 0, "error": str(e), "status": "basic_failed"}
+
+def run_deep(video_path:str, config):
+    # Full pipeline: video, audio, scene, emotion, content, cut suggestions
+    try:
+        va = VideoAnalyzer(config)
+        aa = AudioAnalyzer(config)
+        sd = SceneDetector(config)
+        ed = EmotionDetector(config)
+        ca = IntelligentContentAnalyzer(config)
+        cs = CutSuggester(config)
+
+        # Video analysis
+        video_meta = va.get_video_metadata(video_path)
+        video_timeline = va.analyze_video_timeline(video_path)
+
+        # Audio extraction and analysis
+        audio_path = aa.extract_audio_from_video(video_path)
+        audio_features = aa.extract_audio_features(audio_path)
+        speech_emotions = aa.analyze_speech_emotion(audio_path)
+        speaker_changes = aa.detect_speaker_changes(audio_path)
+        energy_timeline = aa.analyze_audio_energy(audio_path)
+
+        # Scene detection
+        scenes = sd.detect_scenes(video_path, method='combined')
+
+        # Emotion detection (fuse text/audio/visual if available)
+        # For demo, use only speech_emotions and visual features
+        emotion_scores = {}
+        if speech_emotions:
+            # Use dominant emotion per chunk
+            for seg in speech_emotions:
+                for k, v in seg.get('all_emotions', {}).items():
+                    emotion_scores[k] = emotion_scores.get(k, 0) + v
+            # Normalize
+            total = sum(emotion_scores.values())
+            if total > 0:
+                emotion_scores = {k: v/total for k, v in emotion_scores.items()}
+        else:
+            emotion_scores = {"neutral": 1.0}
+
+        # Content analysis
+        content_analysis = ca.analyze_content(
+            video_path,
+            audio_or_text={"features": audio_features, "speech_emotions": speech_emotions, "energy_timeline": energy_timeline},
+            visual_or_metadata=video_timeline
+        )
+
+        # Cut suggestions
+        video_analysis = {
+            "scene_changes": scenes,
+            "timeline": video_timeline
+        }
+        audio_analysis = {
+            "features": audio_features,
+            "speech_emotions": speech_emotions,
+            "speaker_changes": speaker_changes,
+            "energy_timeline": energy_timeline
+        }
+        script_analysis = {}  # Placeholder for future script/subtitle analysis
+        cut_suggestions = cs.generate_suggestions(video_analysis, script_analysis, audio_analysis)
+
+        # Objects (from video timeline visual features)
+        objects = []
+        if video_timeline.get('visual_features'):
+            # For demo, treat high-confidence visual concepts as objects
+            for feats in video_timeline['visual_features']:
+                for k, v in feats.items():
+                    if v > 0.7:
+                        objects.append({"label": k, "confidence": v})
+        # Deduplicate
+        seen = set()
+        unique_objects = []
+        for o in objects:
+            if o['label'] not in seen:
+                unique_objects.append(o)
+                seen.add(o['label'])
+
+        # Clean up temp audio
+        try:
+            aa.cleanup_temp_files([audio_path])
+        except Exception:
+            pass
+
+        return {
+            "duration": video_meta.get("duration", 0),
+            "emotions": emotion_scores,
+            "objects": unique_objects,
+            "cut_suggestions": [dict(timestamp=s.timestamp, reason=s.reason, confidence=s.confidence) for s in cut_suggestions],
+            "scene_changes": scenes,
+            "content": content_analysis,
+            "status": "deep_ready"
+        }
+    except Exception as e:
+        return {"error": str(e), "traceback": traceback.format_exc(), "status": "deep_failed"}
 
 class VideoPipeline:
     def __init__(self, on_partial, on_complete):
         self.on_partial = on_partial
         self.on_complete = on_complete
+        self.config = load_config()
 
     def start(self, video_path:str):
         def worker():
-            basic = run_basic(video_path)
+            basic = run_basic(video_path, self.config)
             self.on_partial(basic)
-            deep = run_deep(video_path)
+            deep = run_deep(video_path, self.config)
             self.on_complete(deep)
         threading.Thread(target=worker, daemon=True).start()