Update services/video_analyzer.py

services/video_analyzer.py

@@ -1,151 +1,76 @@
+import json
 import os
 import tempfile
+import logging
 import pandas as pd
 import streamlit as st
+from typing import Dict, Any, List
+from google import genai
+import time
 
-from services.video_analyzer import analyze_video_only
-from components.render_analysis import render_analyzer_results
-from utils.video import get_video_thumbnail_base64
-from utils.dataframe import analysis_to_csv
-from database import insert_video_analysis, get_all_video_analyses
-
-def analyzer_page():
-    st.sidebar.header("Analyzer")
-    selected_tab = st.sidebar.radio("Select Mode", ["Video Analyser", "History"], index=0)
-
-    if selected_tab == "Video Analyser":
-        st.subheader(" Video Analyser")
-        uploaded_video = st.file_uploader(
-            "Upload Video",
-            type=['mp4', 'mov', 'avi', 'mkv'],
-            help="Upload a video for analysis"
-        )
-        analyse_button = st.button("Run Analysis", use_container_width=True)
-
-        if uploaded_video and analyse_button:
-            # Save to a temp file
-            suffix = os.path.splitext(uploaded_video.name)[1]
-            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
-                tmp.write(uploaded_video.read())
-                video_path = tmp.name
-
-            with st.spinner("Analyzing video..."):
-                result = analyze_video_only(video_path)
-
-    
-            st.session_state["video_name"] = uploaded_video.name
-            st.session_state["video_path"] = video_path
-            st.session_state["thumbnail"] = get_video_thumbnail_base64(video_path)
-            st.session_state["analysis"] = result
-
-        analysis = st.session_state.get("analysis")
-
-        if analysis is None:
-            return
-
-     
-        if not analysis:
-            st.error("Analysis returned no data.")
-            return
 
-        if "__error__" in analysis:
-            st.error(f"Analysis failed: {analysis['__error__']}")
-            return
-
-        
-        render_analyzer_results(analysis)
-
-        col1, col2 = st.columns(2)
+from config import configure_gemini
+from prompt.analyser_prompt import analyser_prompt
+from schema import AdAnalysis
+from utils.video import get_video_thumbnail_base64
+from dotenv import load_dotenv
 
-        # CSV export
-        with col1:
-            frames = []
-            if "storyboard" in analysis:
-                df_storyboard = pd.DataFrame(analysis["storyboard"])
-                df_storyboard["section"] = "Storyboard"
-                frames.append(df_storyboard)
+logger = logging.getLogger(__name__)
 
-            if "script" in analysis:
-                df_script = pd.DataFrame(analysis["script"])
-                df_script["section"] = "Script"
-                frames.append(df_script)
+GEMINI_API_KEY = os.getenv("GEMINI_KEY")
 
-            if "video_analysis" in analysis and "video_metrics" in analysis["video_analysis"]:
-                df_metrics = pd.DataFrame(analysis["video_analysis"]["video_metrics"])
-                df_metrics["section"] = "Metrics"
-                frames.append(df_metrics)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[logging.StreamHandler()]
+)
+logger = logging.getLogger(__name__)
 
-            if "timestamp_improvements" in analysis:
-                df_improvements = pd.DataFrame(analysis["timestamp_improvements"])
-                df_improvements["section"] = "Improvements"
-                frames.append(df_improvements)
 
-            if frames:
-                csv_content = pd.concat(frames, ignore_index=True).to_csv(index=False)
-                st.download_button(
-                    "Download CSV",
-                    data=csv_content,
-                    file_name=f"{st.session_state.get('video_name', 'analysis')}.csv",
-                    mime="text/csv",
-                    use_container_width=True,
-                )
-            else:
-                st.info("No tabular data available for CSV export.")
+def configure_gemini():
+    return genai.Client(api_key=GEMINI_API_KEY)
 
-        # Save to DB
-        with col2:
-            if st.button("Save to DB", use_container_width=True):
-                try:
-                    insert_video_analysis(
-                        video_name=st.session_state.get("video_name", "unknown"),
-                        response=analysis,
-                        thumbnail=st.session_state.get("thumbnail", "")
-                    )
-                    st.success("Analysis saved to database ")
-                except Exception as e:
-                    st.error(f"Failed to save analysis: {e}")
 
-    else:
-        # History tab
-        st.subheader(" History")
+def analyze_video_only(video_path: str) -> Dict[str, Any]:
+    client = configure_gemini()
+    try:
+        video_file = client.files.upload(file=video_path)
+        while getattr(video_file.state, "name", "") == "PROCESSING":
+            time.sleep(2)
+            video_file = client.files.get(name=video_file.name)
+        if getattr(video_file.state, "name", "") == "FAILED":
+            return {}
+        resp = client.models.generate_content(
+            model="gemini-2.0-flash",
+            contents=[analyser_prompt, video_file],
+            config={"response_mime_type": "application/json"}
+        )
+        raw = getattr(resp, "text", "") or ""
         try:
-            history_items = get_all_video_analyses(limit=20)
-        except Exception as e:
-            st.error(f"Failed to load history: {e}")
-            return
-
-        if not history_items:
-            st.info("No saved history available.")
-            return
-
-        video_titles = [
-            f"{item['video_name']} ({item['created_at'].strftime('%Y-%m-%d %H:%M')})"
-            for item in history_items
-        ]
-        selected = st.sidebar.radio("History Items", video_titles, index=0)
-        idx = video_titles.index(selected)
-        selected_data = history_items[idx]
-
-        st.subheader(f"Analysis for: {selected_data['video_name']}")
-        if selected_data.get("thumbnail"):
-            st.image("data:image/jpeg;base64," + selected_data["thumbnail"], width=150)
-
-        json_response = selected_data.get("response")
-        if not json_response:
-            st.info("No analysis payload found for this item.")
-            return
-
-        tabs = st.tabs(["Video Analysis"])
-        with tabs[0]:
-            render_analyzer_results(json_response)
+            model_obj = AdAnalysis.model_validate_json(raw)
+            return model_obj.model_dump()
+        except Exception:
             try:
-                csv_data = analysis_to_csv(json_response)
-                st.download_button(
-                    "Download CSV",
-                    data=csv_data,
-                    file_name=f"{selected_data['video_name']}_analysis.csv",
-                    mime="text/csv",
-                    use_container_width=True,
-                )
-            except Exception as e:
-                st.error(f"CSV export failed: {e}")
+                return json.loads(raw)
+            except Exception:
+                return {}
+    except Exception:
+        return {}
+
+
+def analyze_multiple_videos(video_files: List[st.runtime.uploaded_file_manager.UploadedFile]) -> List[Dict[str, Any]]:
+    results = []
+    for file in video_files:
+        with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(file.name)[1]) as tmp:
+            tmp.write(file.read())
+            video_path = tmp.name
+
+        analysis = analyze_video_only(video_path)
+        thumbnail_b64 = get_video_thumbnail_base64(video_path)
+
+        results.append({
+            "video_name": file.name,
+            "analysis": analysis,
+            "thumbnail": thumbnail_b64
+        })
+    return results