Delete sozo_gen.py

sozo_gen.py

@@ -1,872 +0,0 @@
-# sozo_gen.py
-
-import os
-import re
-import json
-import logging
-import uuid
-import io
-from pathlib import Path
-import pandas as pd
-import numpy as np
-import matplotlib
-matplotlib.use("Agg")
-import matplotlib.pyplot as plt
-from matplotlib.animation import FuncAnimation, FFMpegWriter
-import seaborn as sns
-from scipy import stats
-from PIL import Image, ImageDraw, ImageFont
-import cv2
-import inspect
-import tempfile
-import subprocess
-from typing import Dict, List, Tuple, Any
-from langchain_google_genai import ChatGoogleGenerativeAI
-from google import genai
-import requests
-# In sozo_gen.py, near the other google imports
-from google.genai import types as genai_types
-import math # Add this import at the top of your sozo_gen.py file
-import shutil
-# --- Configuration ---
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - [%(funcName)s] - %(message)s')
-FPS, WIDTH, HEIGHT = 24, 1280, 720
-MAX_CHARTS, VIDEO_SCENES = 5, 5
-MAX_CONTEXT_TOKENS = 750000
-
-# --- API Initialization ---
-API_KEY = os.getenv("GOOGLE_API_KEY")
-if not API_KEY:
-    raise ValueError("GOOGLE_API_KEY environment variable not set.")
-
-PEXELS_API_KEY = os.getenv("PEXELS_API_KEY")
-
-# --- Helper Functions ---
-def load_dataframe_safely(buf, name: str):
-    ext = Path(name).suffix.lower()
-    df = (pd.read_excel if ext in (".xlsx", ".xls") else pd.read_csv)(buf)
-    df.columns = df.columns.astype(str).str.strip()
-    df = df.dropna(how="all")
-    if df.empty or len(df.columns) == 0: raise ValueError("No usable data found")
-    return df
-
-def deepgram_tts(txt: str, voice_model: str):
-    DG_KEY = os.getenv("DEEPGRAM_API_KEY")
-    if not DG_KEY or not txt: return None
-    txt = re.sub(r"[^\w\s.,!?;:-]", "", txt)
-    try:
-        r = requests.post("https://api.deepgram.com/v1/speak", params={"model": voice_model}, headers={"Authorization": f"Token {DG_KEY}", "Content-Type": "application/json"}, json={"text": txt}, timeout=30)
-        r.raise_for_status()
-        return r.content
-    except Exception as e:
-        logging.error(f"Deepgram TTS failed: {e}")
-        return None
-
-def generate_silence_mp3(duration: float, out: Path):
-    subprocess.run([ "ffmpeg", "-y", "-f", "lavfi", "-i", "anullsrc=r=44100:cl=mono", "-t", f"{duration:.3f}", "-q:a", "9", str(out)], check=True, capture_output=True)
-
-def audio_duration(path: str) -> float:
-    try:
-        res = subprocess.run([ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", path], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
-        return float(res.stdout.strip())
-    except Exception: return 5.0
-
-TAG_RE = re.compile( r'[<[]\s*generate_?chart\s*[:=]?\s*[\"\'“”]?(?P<d>[^>\"\'”\]]+?)[\"\'“”]?\s*[>\]]', re.I, )
-TAG_RE_PEXELS = re.compile( r'[<[]\s*generate_?stock_?video\s*[:=]?\s*[\"\'“”]?(?P<d>[^>\"\'”\]]+?)[\"\'“”]?\s*[>\]]', re.I, )
-extract_chart_tags = lambda t: list( dict.fromkeys(m.group("d").strip() for m in TAG_RE.finditer(t or "")) )
-extract_pexels_tags = lambda t: list( dict.fromkeys(m.group("d").strip() for m in TAG_RE_PEXELS.finditer(t or "")) )
-
-re_scene = re.compile(r"^\s*scene\s*\d+[:.\- ]*", re.I | re.M)
-def clean_narration(txt: str) -> str:
-    txt = TAG_RE.sub("", txt); txt = TAG_RE_PEXELS.sub("", txt); txt = re_scene.sub("", txt)
-    phrases_to_remove = [r"chart tag", r"chart_tag", r"narration", r"stock video tag"]
-    for phrase in phrases_to_remove: txt = re.sub(phrase, "", txt, flags=re.IGNORECASE)
-    txt = re.sub(r"\s*\([^)]*\)", "", txt); txt = re.sub(r"[\*#_]", "", txt)
-    return re.sub(r"\s{2,}", " ", txt).strip()
-
-def placeholder_img() -> Image.Image: return Image.new("RGB", (WIDTH, HEIGHT), (230, 230, 230))
-
-def _sanitize_for_json(data):
-    """Recursively sanitizes a dict/list for JSON compliance."""
-    if isinstance(data, dict):
-        return {k: _sanitize_for_json(v) for k, v in data.items()}
-    if isinstance(data, list):
-        return [_sanitize_for_json(i) for i in data]
-    if isinstance(data, float) and (math.isnan(data) or math.isinf(data)):
-        return None
-    return data
-    
-def detect_dataset_domain(df: pd.DataFrame) -> str:
-    """Analyzes column names to detect the dataset's primary domain."""
-    domain_keywords = {
-        "health insurance": ["charges", "bmi", "smoker", "beneficiary"],
-        "finance": ["revenue", "profit", "cost", "budget", "expense", "stock"],
-        "marketing": ["campaign", "conversion", "click", "customer", "segment"],
-        "survey": ["satisfaction", "rating", "feedback", "opinion", "score"],
-        "food": ["nutrition", "calories", "ingredients", "restaurant"]
-    }
-    columns_lower = [col.lower() for col in df.columns]
-    for domain, keywords in domain_keywords.items():
-        if any(keyword in col for col in columns_lower for keyword in keywords):
-            logging.info(f"Dataset domain detected: {domain}")
-            return domain
-    logging.info("No specific dataset domain detected, using generic terms.")
-    return "data"
-
-# NEW: Keyword extraction for better Pexels searches
-def extract_keywords_for_query(text: str, llm) -> str:
-    prompt = f"""
-    Extract a maximum of 3 key nouns or verbs from the following text to use as a search query for a stock video. 
-    Focus on concrete actions and subjects.
-    Example: 'Our analysis shows a significant growth in quarterly revenue and strong partnerships.' -> 'data analysis growth'
-    Output only the search query keywords, separated by spaces.
-
-    Text: "{text}"
-    """
-    try:
-        response = llm.invoke(prompt).content.strip()
-        return response if response else text
-    except Exception as e:
-        logging.error(f"Keyword extraction failed: {e}. Using original text.")
-        return text
-
-# UPDATED: Pexels search now loops short videos
-def search_and_download_pexels_video(query: str, duration: float, out_path: Path) -> str:
-    if not PEXELS_API_KEY:
-        logging.warning("PEXELS_API_KEY not set. Cannot fetch stock video.")
-        return None
-    try:
-        headers = {"Authorization": PEXELS_API_KEY}
-        params = {"query": query, "per_page": 10, "orientation": "landscape"}
-        response = requests.get("https://api.pexels.com/videos/search", headers=headers, params=params, timeout=20)
-        response.raise_for_status()
-        videos = response.json().get('videos', [])
-        if not videos:
-            logging.warning(f"No Pexels videos found for query: '{query}'")
-            return None
-
-        video_to_download = None
-        for video in videos:
-            for f in video.get('video_files', []):
-                if f.get('quality') == 'hd' and f.get('width') >= 1280:
-                    video_to_download = f['link']
-                    break
-            if video_to_download:
-                break
-        
-        if not video_to_download:
-            logging.warning(f"No suitable HD video file found for query: '{query}'")
-            return None
-
-        with requests.get(video_to_download, stream=True, timeout=60) as r:
-            r.raise_for_status()
-            with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_dl_file:
-                for chunk in r.iter_content(chunk_size=8192):
-                    temp_dl_file.write(chunk)
-                temp_dl_path = Path(temp_dl_file.name)
-
-        # UPDATED: Added -stream_loop -1 to loop short videos
-        cmd = [
-            "ffmpeg", "-y", 
-            "-stream_loop", "-1", # Loop the input video
-            "-i", str(temp_dl_path),
-            "-vf", f"scale={WIDTH}:{HEIGHT}:force_original_aspect_ratio=decrease,pad={WIDTH}:{HEIGHT}:(ow-iw)/2:(oh-ih)/2,setsar=1",
-            "-t", f"{duration:.3f}", # Cut the looped video to the exact duration
-            "-c:v", "libx264", "-pix_fmt", "yuv420p", "-an",
-            str(out_path)
-        ]
-        subprocess.run(cmd, check=True, capture_output=True)
-        temp_dl_path.unlink()
-        return str(out_path)
-
-    except Exception as e:
-        logging.error(f"Pexels video processing failed for query '{query}': {e}")
-        if 'temp_dl_path' in locals() and temp_dl_path.exists():
-            temp_dl_path.unlink()
-        return None
-
-class ChartSpecification:
-    def __init__(self, chart_type: str, title: str, x_col: str, y_col: str = None, size_col: str = None, agg_method: str = None, filter_condition: str = None, top_n: int = None, color_scheme: str = "professional"):
-        self.chart_type = chart_type; self.title = title; self.x_col = x_col; self.y_col = y_col; self.size_col = size_col
-        self.agg_method = agg_method or "sum"; self.filter_condition = filter_condition; self.top_n = top_n; self.color_scheme = color_scheme
-
-class ChartGenerator:
-    def __init__(self, llm, df: pd.DataFrame):
-        self.llm = llm; self.df = df
-
-    def generate_chart_spec(self, description: str, context: Dict) -> ChartSpecification:
-        spec_prompt = f"""
-        You are a data visualization expert. Based on the dataset context and chart description, generate a precise chart specification.
-        **Dataset Context:** {json.dumps(context, indent=2)}
-        **Chart Request:** {description}
-        **Return a JSON specification with these exact fields:**
-        {{
-            "chart_type": "bar|pie|line|scatter|hist|heatmap|area|bubble", 
-            "title": "Professional chart title", 
-            "x_col": "column_name_for_x_axis_or_null_for_heatmap",
-            "y_col": "column_name_for_y_axis_or_null", 
-            "size_col": "column_name_for_bubble_size_or_null",
-            "agg_method": "sum|mean|count|max|min|null", 
-            "top_n": "number_for_top_n_filtering_or_null"
-        }}
-        Return only the JSON specification, no additional text.
-        """
-        try:
-            response = self.llm.invoke(spec_prompt).content.strip()
-            if response.startswith("```json"): response = response[7:-3]
-            elif response.startswith("```"): response = response[3:-3]
-            spec_dict = json.loads(response)
-            valid_keys = [p.name for p in inspect.signature(ChartSpecification).parameters.values() if p.name not in ['reasoning', 'filter_condition', 'color_scheme']]
-            filtered_dict = {k: v for k, v in spec_dict.items() if k in valid_keys}
-            return ChartSpecification(**filtered_dict)
-        except Exception as e:
-            logging.error(f"Spec generation failed: {e}. Using fallback.")
-            numeric_cols = context.get('schema', {}).get('numeric_columns', list(self.df.select_dtypes(include=['number']).columns))
-            categorical_cols = context.get('schema', {}).get('categorical_columns', list(self.df.select_dtypes(exclude=['number']).columns))
-            ctype = "bar"
-            for t in ["pie", "line", "scatter", "hist", "heatmap", "area", "bubble"]:
-                if t in description.lower(): ctype = t
-            x = categorical_cols[0] if categorical_cols else self.df.columns[0]
-            y = numeric_cols[0] if numeric_cols and len(self.df.columns) > 1 else (self.df.columns[1] if len(self.df.columns) > 1 else None)
-            return ChartSpecification(ctype, description, x, y)
-
-def execute_chart_spec(spec: ChartSpecification, df: pd.DataFrame, output_path: Path) -> bool:
-    try:
-        plot_data = prepare_plot_data(spec, df)
-        fig, ax = plt.subplots(figsize=(12, 8)); plt.style.use('default')
-        if spec.chart_type == "bar": ax.bar(plot_data.index.astype(str), plot_data.values, color='#2E86AB', alpha=0.8); ax.set_xlabel(spec.x_col); ax.set_ylabel(spec.y_col); ax.tick_params(axis='x', rotation=45)
-        elif spec.chart_type == "pie": ax.pie(plot_data.values, labels=plot_data.index, autopct='%1.1f%%', startangle=90); ax.axis('equal')
-        elif spec.chart_type == "line": ax.plot(plot_data.index, plot_data.values, marker='o', linewidth=2, color='#A23B72'); ax.set_xlabel(spec.x_col); ax.set_ylabel(spec.y_col); ax.grid(True, alpha=0.3)
-        elif spec.chart_type == "scatter": ax.scatter(plot_data.iloc[:, 0], plot_data.iloc[:, 1], alpha=0.6, color='#F18F01'); ax.set_xlabel(spec.x_col); ax.set_ylabel(spec.y_col); ax.grid(True, alpha=0.3)
-        elif spec.chart_type == "hist": ax.hist(plot_data.values, bins=20, color='#C73E1D', alpha=0.7, edgecolor='black'); ax.set_xlabel(spec.x_col); ax.set_ylabel('Frequency'); ax.grid(True, alpha=0.3)
-        elif spec.chart_type == "area": ax.fill_between(plot_data.index, plot_data.values, color="#4E79A7", alpha=0.4); ax.plot(plot_data.index, plot_data.values, color="#4E79A7", alpha=0.8); ax.set_xlabel(spec.x_col); ax.set_ylabel(spec.y_col); ax.grid(True, alpha=0.3)
-        elif spec.chart_type == "heatmap": sns.heatmap(plot_data, annot=True, cmap="viridis", ax=ax); plt.xticks(rotation=45, ha="right"); plt.yticks(rotation=0)
-        elif spec.chart_type == "bubble": 
-            sizes = (plot_data[spec.size_col] - plot_data[spec.size_col].min() + 1) / (plot_data[spec.size_col].max() - plot_data[spec.size_col].min() + 1) * 2000 + 50
-            ax.scatter(plot_data[spec.x_col], plot_data[spec.y_col], s=sizes, alpha=0.6, color='#59A14F'); ax.set_xlabel(spec.x_col); ax.set_ylabel(spec.y_col); ax.grid(True, alpha=0.3)
-        
-        ax.set_title(spec.title, fontsize=14, fontweight='bold', pad=20); plt.tight_layout()
-        plt.savefig(output_path, dpi=300, bbox_inches='tight', facecolor='white'); plt.close()
-        return True
-    except Exception as e: logging.error(f"Static chart generation failed for '{spec.title}': {e}"); return False
-
-def prepare_plot_data(spec: ChartSpecification, df: pd.DataFrame):
-    if spec.chart_type not in ["heatmap"]:
-        if spec.x_col not in df.columns or (spec.y_col and spec.y_col not in df.columns): raise ValueError(f"Invalid columns in chart spec: {spec.x_col}, {spec.y_col}")
-
-    if spec.chart_type in ["bar", "pie"]:
-        if not spec.y_col: return df[spec.x_col].value_counts().nlargest(spec.top_n or 10)
-        grouped = df.groupby(spec.x_col)[spec.y_col].agg(spec.agg_method or 'sum')
-        return grouped.nlargest(spec.top_n or 10)
-    elif spec.chart_type in ["line", "area"]: return df.set_index(spec.x_col)[spec.y_col].sort_index()
-    elif spec.chart_type == "scatter": return df[[spec.x_col, spec.y_col]].dropna()
-    elif spec.chart_type == "bubble":
-        if not spec.size_col or spec.size_col not in df.columns: raise ValueError("Bubble chart requires a valid size_col.")
-        return df[[spec.x_col, spec.y_col, spec.size_col]].dropna()
-    elif spec.chart_type == "hist": return df[spec.x_col].dropna()
-    elif spec.chart_type == "heatmap":
-        numeric_cols = df.select_dtypes(include=np.number).columns
-        if not numeric_cols.any(): raise ValueError("Heatmap requires numeric columns.")
-        return df[numeric_cols].corr()
-    return df[spec.x_col]
-
-# UPDATED: animate_chart now uses blit=False for accurate timing
-
-def animate_chart(spec: ChartSpecification, df: pd.DataFrame, dur: float, out: Path, fps: int = FPS) -> str:
-    plot_data = prepare_plot_data(spec, df)
-    frames = math.ceil(dur * fps)
-    fig, ax = plt.subplots(figsize=(WIDTH / 100, HEIGHT / 100), dpi=100)
-    plt.tight_layout(pad=3.0)
-    ctype = spec.chart_type
-
-    init_func, update_func = None, None
-
-    if ctype == "line":
-        plot_data = plot_data.sort_index()
-        x_full, y_full = plot_data.index, plot_data.values
-        
-        ax.set_xlim(x_full.min(), x_full.max())
-        ax.set_ylim(y_full.min() * 0.9, y_full.max() * 1.1)
-        ax.set_title(spec.title); ax.grid(alpha=.3); ax.set_xlabel(spec.x_col); ax.set_ylabel(spec.y_col)
-        
-        line, = ax.plot([], [], lw=2, color='#A23B72')
-        markers, = ax.plot([], [], 'o', color='#A23B72', markersize=5)
-
-        def init():
-            line.set_data([], [])
-            markers.set_data([], [])
-            return line, markers
-        def update(i):
-            k = max(2, int(len(x_full) * (i / (frames - 1))))
-            line.set_data(x_full[:k], y_full[:k])
-            markers.set_data(x_full[:k], y_full[:k])
-            return line, markers
-        init_func, update_func = init, update
-        
-        anim = FuncAnimation(fig, update, init_func=init, frames=frames, blit=True, interval=1000 / fps)
-        anim.save(str(out), writer=FFMpegWriter(fps=fps), dpi=144)
-        plt.close(fig)
-        return str(out)
-
-    # Fallback to the slightly slower but reliable blit=False for other types
-    # This ensures stability across all chart types while the line chart is optimized
-    if ctype == "bar":
-        bars = ax.bar(plot_data.index.astype(str), np.zeros_like(plot_data.values, dtype=float), color="#1f77b4")
-        ax.set_ylim(0, plot_data.max() * 1.1 if not pd.isna(plot_data.max()) and plot_data.max() > 0 else 1)
-        ax.set_title(spec.title); plt.xticks(rotation=45, ha="right")
-        def init(): return bars
-        def update(i):
-            for b, h in zip(bars, plot_data.values): b.set_height(h * (i / (frames - 1)))
-            return bars
-        init_func, update_func = init, update
-    elif ctype == "scatter":
-        x_full, y_full = plot_data.iloc[:, 0], plot_data.iloc[:, 1]
-        slope, intercept, _, _, _ = stats.linregress(x_full, y_full)
-        reg_line_x = np.array([x_full.min(), x_full.max()])
-        reg_line_y = slope * reg_line_x + intercept
-        scat = ax.scatter([], [], alpha=0.7, color='#F18F01')
-        line, = ax.plot([], [], 'r--', lw=2)
-        ax.set_xlim(x_full.min(), x_full.max()); ax.set_ylim(y_full.min(), y_full.max())
-        ax.set_title(spec.title); ax.grid(alpha=.3); ax.set_xlabel(spec.x_col); ax.set_ylabel(spec.y_col)
-        def init():
-            scat.set_offsets(np.empty((0, 2))); line.set_data([], [])
-            return []
-        def update(i):
-            point_frames = int(frames * 0.7)
-            if i <= point_frames:
-                k = max(1, int(len(x_full) * (i / point_frames)))
-                scat.set_offsets(plot_data.iloc[:k].values)
-            else:
-                line_frame = i - point_frames; line_total_frames = frames - point_frames
-                current_x = reg_line_x[0] + (reg_line_x[1] - reg_line_x[0]) * (line_frame / line_total_frames)
-                line.set_data([reg_line_x[0], current_x], [reg_line_y[0], slope * current_x + intercept])
-            return []
-        init_func, update_func = init, update
-    elif ctype == "pie":
-        wedges, _, _ = ax.pie(plot_data, labels=plot_data.index, startangle=90, autopct='%1.1f%%')
-        ax.set_title(spec.title); ax.axis('equal')
-        def init(): [w.set_alpha(0) for w in wedges]; return []
-        def update(i): [w.set_alpha(i / (frames - 1)) for w in wedges]; return []
-        init_func, update_func = init, update
-    elif ctype == "hist":
-        _, _, patches = ax.hist(plot_data, bins=20, alpha=0)
-        ax.set_title(spec.title); ax.set_xlabel(spec.x_col); ax.set_ylabel("Frequency")
-        def init(): [p.set_alpha(0) for p in patches]; return []
-        def update(i): [p.set_alpha((i / (frames - 1)) * 0.7) for p in patches]; return []
-        init_func, update_func = init, update
-    elif ctype == "heatmap":
-        sns.heatmap(plot_data, annot=True, cmap="viridis", ax=ax, alpha=0)
-        ax.set_title(spec.title)
-        def init(): ax.collections[0].set_alpha(0); return []
-        def update(i): ax.collections[0].set_alpha(i / (frames - 1)); return []
-        init_func, update_func = init, update
-    else:
-        ax.text(0.5, 0.5, f"'{ctype}' animation not implemented", ha='center', va='center')
-        def init(): return []
-        def update(i): return []
-        init_func, update_func = init, update
-
-    anim = FuncAnimation(fig, update_func, init_func=init_func, frames=frames, blit=False, interval=1000 / fps)
-    anim.save(str(out), writer=FFMpegWriter(fps=fps), dpi=144)
-    plt.close(fig)
-    return str(out)
-
-def safe_chart(desc: str, df: pd.DataFrame, dur: float, out: Path, context: Dict) -> str:
-    try:
-        llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", google_api_key=API_KEY, temperature=0.1)
-        chart_generator = ChartGenerator(llm, df)
-        chart_spec = chart_generator.generate_chart_spec(desc, context)
-        return animate_chart(chart_spec, df, dur, out)
-    except Exception as e:
-        logging.error(f"Chart animation failed for '{desc}': {e}. Raising exception to trigger fallback.")
-        raise e # Raise exception to be caught by the video generator's fallback logic
-
-def concat_media(file_paths: List[str], output_path: Path):
-    valid_paths = [p for p in file_paths if Path(p).exists() and Path(p).stat().st_size > 100]
-    if not valid_paths: raise ValueError("No valid media files to concatenate.")
-    if len(valid_paths) == 1: import shutil; shutil.copy2(valid_paths[0], str(output_path)); return
-    list_file = output_path.with_suffix(".txt")
-    with open(list_file, 'w') as f:
-        for path in valid_paths: f.write(f"file '{Path(path).resolve()}'\n")
-    cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(list_file), "-c", "copy", str(output_path)]
-    try:
-        subprocess.run(cmd, check=True, capture_output=True, text=True)
-    finally:
-        list_file.unlink(missing_ok=True)
-
-def sanitize_for_firebase_key(text: str) -> str:
-    forbidden_chars = ['.', '$', '#', '[', ']', '/']
-    for char in forbidden_chars:
-        text = text.replace(char, '_')
-    return text
-
-def analyze_data_intelligence(df: pd.DataFrame) -> Dict:
-    numeric_cols = df.select_dtypes(include=['number']).columns.tolist()
-    categorical_cols = df.select_dtypes(exclude=['number']).columns.tolist()
-    is_timeseries = any('date' in col.lower() or 'time' in col.lower() for col in df.columns)
-    opportunities = []
-    if is_timeseries: opportunities.append("temporal trends")
-    if len(numeric_cols) > 1: opportunities.append("correlations between metrics")
-    if len(categorical_cols) > 0 and len(numeric_cols) > 0: opportunities.append("segmentation by category")
-    if df.isnull().sum().sum() > 0: opportunities.append("impact of missing data")
-    return {
-        "insight_opportunities": opportunities,
-        "is_timeseries": is_timeseries,
-        "has_correlations": len(numeric_cols) > 1,
-        "has_segments": len(categorical_cols) > 0 and len(numeric_cols) > 0
-    }
-
-def generate_visualization_strategy(intelligence: Dict) -> str:
-    strategy = "Vary your visualizations to keep the report engaging. "
-    if intelligence["is_timeseries"]: strategy += "Use 'line' or 'area' charts to explore temporal trends. "
-    if intelligence["has_correlations"]: strategy += "Use 'scatter' or 'heatmap' charts to reveal correlations. "
-    if intelligence["has_segments"]: strategy += "Use 'bar' or 'pie' charts to compare segments. "
-    return strategy
-
-def get_augmented_context(df: pd.DataFrame, user_ctx: str) -> Dict:
-    """Creates a detailed, JSON-safe summary of the dataframe for the AI."""
-    numeric_cols = df.select_dtypes(include=['number']).columns.tolist()
-    categorical_cols = df.select_dtypes(exclude=['number']).columns.tolist()
-    
-    context = {
-        "user_context": user_ctx,
-        "dataset_shape": {"rows": df.shape[0], "columns": df.shape[1]},
-        "schema": {"numeric_columns": numeric_cols, "categorical_columns": categorical_cols},
-        "data_previews": {}
-    }
-    
-    for col in categorical_cols[:5]:
-        unique_vals = df[col].unique()
-        context["data_previews"][col] = {
-            "count": len(unique_vals),
-            "values": unique_vals[:5].tolist()
-        }
-        
-    for col in numeric_cols[:5]:
-        context["data_previews"][col] = {
-            "mean": df[col].mean(),
-            "min": df[col].min(),
-            "max": df[col].max()
-        }
-    
-    # Sanitize the entire structure before returning
-    return _sanitize_for_json(json.loads(json.dumps(context, default=str)))
-
-def generate_report_draft(buf, name: str, ctx: str, uid: str, project_id: str, bucket):
-    logging.info(f"Generating guided storyteller report draft for project {project_id}")
-    df = load_dataframe_safely(buf, name)
-    llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", google_api_key=API_KEY, temperature=0.3)
-    
-    data_context_str, context_for_charts = "", {}
-    try:
-        df_json = df.to_json(orient='records')
-        estimated_tokens = len(df_json) / 4
-        if estimated_tokens < MAX_CONTEXT_TOKENS:
-            logging.info(f"Using full JSON context for report generation.")
-            data_context_str = f"Here is the full dataset in JSON format:\n{df_json}"
-            context_for_charts = get_augmented_context(df, ctx)
-        else:
-            raise ValueError("Dataset too large for full context.")
-    except Exception as e:
-        logging.warning(f"Falling back to augmented summary context for report generation: {e}")
-        augmented_context = get_augmented_context(df, ctx)
-        data_context_str = f"The full dataset is too large to display. Here is a detailed summary:\n{json.dumps(augmented_context, indent=2)}"
-        context_for_charts = augmented_context
-
-    md = ""
-    try:
-        # --- Pass 1: The "Visualization Strategist" ---
-        strategist_prompt = f"""
-        You are a data visualization expert. Your task is to create a diverse palette of unique and impactful charts for a data storyteller.
-        Based on the provided data context, identify the 4-5 most distinct and insightful stories that can be visualized.
-        
-        **Data Context:**
-        {data_context_str}
-
-        **Your Goal:**
-        Your primary goal is to select a **diverse palette of chart types**. A high-quality response will use a mix of different charts from the available list to create a visually engaging and comprehensive report. **Do not use the same chart type more than twice.**
-
-        **Strategic Hints:**
-        - Consider a `histogram` to show the distribution of a key variable (like age or bmi).
-        - Consider a `pie chart` for a clear part-to-whole relationship (e.g., smoker vs. non-smoker proportions).
-        - Consider a `heatmap` if the dataset has multiple numeric columns and you believe the overall pattern of their correlations is a key insight in itself.
-
-        **Output Format:**
-        Return ONLY a valid JSON array of strings. Each string must be a unique chart description tag.
-        
-        Example:
-        ["bar | Average Charges by Smoker Status", "scatter | Charges vs. BMI", "hist | Distribution of Beneficiary Ages", "pie | Regional Proportions"]
-        """
-        logging.info("Executing Visualization Strategist Pass...")
-        strategist_response = llm.invoke(strategist_prompt).content.strip()
-        if strategist_response.startswith("```json"):
-            strategist_response = strategist_response[7:-3]
-        chart_palette = json.loads(strategist_response)
-        logging.info(f"Strategist Pass successful. Palette has {len(chart_palette)} unique charts.")
-
-        # --- Pass 2: The "Master Storyteller" ---
-        storyteller_prompt = f"""
-        You are an elite data storyteller and business intelligence expert. Your mission is to write a comprehensive, flowing narrative that analyzes the entire dataset provided. Your goal is to create a captivating story that **drives action**.
-
-        **Data Context:**
-        {data_context_str}
-
-        **Narrative Construction Guidelines:**
-        1.  **Use Compelling Headers:** Structure your report with multiple sections using Markdown headings (`##` or `###`). Do not write one long block of text. Create curiosity with your headers (e.g., 'The Smoking Premium: A Costly Habit', 'Geographic Hotspots: Where Charges Are Highest').
-        2.  **Weave a Story:** Don't just describe the charts one by one. Connect the findings together. For example, how does 'age' relate to 'smoker status' and how do they both impact 'charges'?
-        3.  **Drive to Action:** Conclude your report with a dedicated section titled `## Actionable Recommendations`. Based on your analysis, provide specific, data-driven suggestions that a business leader could implement.
-
-        **Your Toolbox (Most Important):**
-        To support your story with visuals, you have been provided with a pre-approved 'palette' of unique charts. As you write your narrative, you **must** integrate each of these chart tags, one time, at the most logical point in the story.
-        - You **must** use every chart tag from the provided palette exactly once.
-        - Do **not** repeat chart tags.
-        - Do **not** invent new chart tags.
-        - Insert the tags in the format `<generate_chart: "the_description">`.
-
-        **Chart Palette:**
-        {json.dumps(chart_palette, indent=2)}
-
-        Now, write the complete, comprehensive Markdown report.
-        """
-        logging.info("Executing Master Storyteller Pass...")
-        md = llm.invoke(storyteller_prompt).content.strip()
-        logging.info("Master Storyteller Pass successful.")
-
-    except Exception as e:
-        logging.error(f"Guided Storyteller system failed: {e}. Reverting to single-pass fallback.")
-        fallback_prompt = f"""
-        You are an elite data storyteller and business intelligence expert. Your mission is to uncover the compelling, hidden narrative in this dataset and present it as a captivating story in Markdown format that drives action.
-        **Data Context:** {data_context_str}
-        **Your Grounding Rules (Most Important):**
-        1.  **Strict Accuracy:** Your entire analysis and narrative **must strictly** use the column names provided in the 'Data Context' section.
-        2.  **Chart Support:** Wherever a key finding is made, you **must** support it with a chart tag: `<generate_chart: "chart_type | a specific, compelling description">`.
-        3.  **Chart Accuracy:** The column names used in your chart descriptions **must** also be an exact match from the provided data context.
-        Now, begin your report. Let the data's story unfold naturally.
-        """
-        md = llm.invoke(fallback_prompt).content.strip()
-
-    chart_descs = extract_chart_tags(md)[:MAX_CHARTS]
-    chart_urls = {}
-    chart_generator = ChartGenerator(llm, df)
-
-    for desc in chart_descs:
-        safe_desc = sanitize_for_firebase_key(desc)
-        md = md.replace(f'<generate_chart: "{desc}">', f'<generate_chart: "{safe_desc}">')
-        md = md.replace(f'<generate_chart: {desc}>', f'<generate_chart: "{safe_desc}">')
-
-        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
-            img_path = Path(temp_file.name)
-            try:
-                chart_spec = chart_generator.generate_chart_spec(desc, context_for_charts)
-                if execute_chart_spec(chart_spec, df, img_path):
-                    blob_name = f"sozo_projects/{uid}/{project_id}/charts/{uuid.uuid4().hex}.png"
-                    blob = bucket.blob(blob_name)
-                    blob.upload_from_filename(str(img_path))
-                    chart_urls[safe_desc] = blob.public_url
-            finally:
-                if os.path.exists(img_path):
-                    os.unlink(img_path)
-
-    return {"raw_md": md, "chartUrls": chart_urls, "data_context": context_for_charts}
-
-def generate_single_chart(df: pd.DataFrame, description: str, uid: str, project_id: str, bucket):
-    logging.info(f"Generating single chart '{description}' for project {project_id}")
-    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", google_api_key=API_KEY, temperature=0.1)
-    chart_generator = ChartGenerator(llm, df)
-    context = get_augmented_context(df, "")
-    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
-        img_path = Path(temp_file.name)
-        try:
-            chart_spec = chart_generator.generate_chart_spec(description, context)
-            if execute_chart_spec(chart_spec, df, img_path):
-                blob_name = f"sozo_projects/{uid}/{project_id}/charts/{uuid.uuid4().hex}.png"
-                blob = bucket.blob(blob_name)
-                blob.upload_from_filename(str(img_path))
-                logging.info(f"Uploaded single chart to {blob.public_url}")
-                return blob.public_url
-        finally:
-            if os.path.exists(img_path):
-                os.unlink(img_path)
-    return None
-
-
-def generate_video_from_project(df: pd.DataFrame, raw_md: str, data_context: Dict, uid: str, project_id: str, voice_model: str, bucket):
-    logging.info(f"Generating video for project {project_id} with voice {voice_model}")
-    llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", google_api_key=API_KEY, temperature=0.2)
-    
-    domain = detect_dataset_domain(df)
-
-    story_prompt = f"""
-    Based on the following report, create a script for a {VIDEO_SCENES}-scene video.
-    1. The first scene MUST be an "Introduction". It must contain narration and a stock video tag like: <generate_stock_video: "search query">.
-    2. The last scene MUST be a "Conclusion". It must also contain narration and a stock video tag.
-    3. The middle scenes should each contain narration and one chart tag from the report.
-    4. Separate each scene with '[SCENE_BREAK]'.
-    Report: {raw_md}
-    Only output the script, no extra text.
-    """
-    script = llm.invoke(story_prompt).content.strip()
-    scenes = [s.strip() for s in script.split("[SCENE_BREAK]") if s.strip()]
-    video_parts, audio_parts, temps = [], [], []
-    total_audio_duration = 0.0
-    conclusion_video_path = None
-
-    for i, sc in enumerate(scenes):
-        mp4 = Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.mp4"
-        narrative = clean_narration(sc)
-        if not narrative:
-            logging.warning(f"Scene {i+1} has no narration, skipping.")
-            continue
-
-        audio_bytes = deepgram_tts(narrative, voice_model)
-        mp3 = Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.mp3"
-        audio_dur = 5.0
-        if audio_bytes:
-            mp3.write_bytes(audio_bytes)
-            audio_dur = audio_duration(str(mp3))
-            if audio_dur <= 0.1: audio_dur = 5.0
-        else:
-            generate_silence_mp3(audio_dur, mp3)
-        
-        audio_parts.append(str(mp3)); temps.append(mp3)
-        total_audio_duration += audio_dur
-        
-        video_dur = audio_dur + 1.5 
-        
-        try:
-            # --- Primary Visual Generation ---
-            chart_descs = extract_chart_tags(sc)
-            pexels_descs = extract_pexels_tags(sc)
-            is_conclusion_scene = any(k in narrative.lower() for k in ["conclusion", "summary", "in closing", "final thoughts"])
-
-            if pexels_descs:
-                logging.info(f"Scene {i+1}: Processing Pexels scene.")
-                base_keywords = extract_keywords_for_query(narrative, llm)
-                final_query = f"{base_keywords} {domain}"
-                video_path = search_and_download_pexels_video(final_query, video_dur, mp4)
-                if not video_path: raise ValueError("Pexels search returned no results for chained query.")
-                video_parts.append(video_path)
-                if is_conclusion_scene:
-                    conclusion_video_path = video_path
-            elif chart_descs:
-                logging.info(f"Scene {i+1}: Primary attempt with animated chart.")
-                if not chart_descs: raise ValueError("AI script failed to provide a chart tag for this scene.")
-                safe_chart(chart_descs[0], df, video_dur, mp4, data_context)
-                video_parts.append(str(mp4))
-            else:
-                raise ValueError("No visual tag found in scene script.")
-        except Exception as e:
-            logging.warning(f"Scene {i+1}: Primary visual failed ({e}). Triggering Fallback Tier 1.")
-            # --- Fallback Tier 1: Context-Aware Pexels Replacement ---
-            try:
-                fallback_keywords = extract_keywords_for_query(narrative, llm)
-                final_fallback_query = f"{fallback_keywords} {domain}"
-                logging.info(f"Fallback Tier 1: Searching Pexels with query: '{final_fallback_query}'")
-                
-                video_path = search_and_download_pexels_video(final_fallback_query, video_dur, mp4)
-                if not video_path: raise ValueError("Fallback Pexels search returned no results.")
-                
-                video_parts.append(video_path)
-                logging.info(f"Scene {i+1}: Successfully recovered with a relevant Pexels video.")
-            except Exception as fallback_e:
-                # --- Fallback Tier 2: Looping Conclusion Failsafe ---
-                logging.error(f"Scene {i+1}: Fallback Tier 1 also failed ({fallback_e}). Marking for final failsafe.")
-                video_parts.append("FALLBACK_NEEDED")
-
-        temps.append(mp4)
-
-    if not conclusion_video_path:
-        logging.warning("No conclusion video was generated; creating a generic one for fallbacks.")
-        fallback_mp4 = Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.mp4"
-        conclusion_video_path = search_and_download_pexels_video(f"data visualization abstract {domain}", 5.0, fallback_mp4)
-        if conclusion_video_path: temps.append(fallback_mp4)
-
-    final_video_parts = []
-    for part in video_parts:
-        if part == "FALLBACK_NEEDED":
-            if conclusion_video_path:
-                fallback_copy_path = Path(tempfile.gettempdir()) / f"fallback_{uuid.uuid4().hex}.mp4"
-                shutil.copy(conclusion_video_path, fallback_copy_path)
-                temps.append(fallback_copy_path)
-                final_video_parts.append(str(fallback_copy_path))
-                logging.info(f"Applying unique copy of conclusion video as fallback for a failed scene.")
-            else:
-                logging.error("Cannot apply fallback; no conclusion video available. A scene will be missing.")
-        else:
-            final_video_parts.append(part)
-
-    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_vid, \
-            tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as temp_aud, \
-            tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as final_vid, \
-            tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as branded_vid:
-        
-        silent_vid_path = Path(temp_vid.name)
-        audio_mix_path = Path(temp_aud.name)
-        final_vid_path = Path(final_vid.name)
-        branded_vid_path = Path(branded_vid.name)
-
-        concat_media(final_video_parts, silent_vid_path)
-        concat_media(audio_parts, audio_mix_path)
-        
-        cmd = [
-            "ffmpeg", "-y", "-i", str(silent_vid_path), "-i", str(audio_mix_path),
-            "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac",
-            "-map", "0:v:0", "-map", "1:a:0",
-            "-t", f"{total_audio_duration:.3f}",
-            str(final_vid_path)
-        ]
-        subprocess.run(cmd, check=True, capture_output=True)
-        
-        upload_path = final_vid_path
-        logo_path = Path("sozob.png")
-
-        if logo_path.exists():
-            logging.info("Logo 'sozob.png' found. Adding full-screen end-card.")
-            duration_for_filter = total_audio_duration
-            
-            filter_complex = f"[1:v]scale={WIDTH}:{HEIGHT}[logo];[0:v][logo]overlay=0:0:enable='gte(t,{duration_for_filter - 2})'"
-
-            logo_cmd = [
-                "ffmpeg", "-y",
-                "-i", str(final_vid_path),
-                "-i", str(logo_path),
-                "-filter_complex", filter_complex,
-                "-map", "0:a",
-                "-c:a", "copy",
-                "-c:v", "libx264", "-pix_fmt", "yuv420p",
-                str(branded_vid_path)
-            ]
-            try:
-                subprocess.run(logo_cmd, check=True, capture_output=True)
-                upload_path = branded_vid_path
-            except subprocess.CalledProcessError as e:
-                logging.error(f"Failed to add logo end-card. Uploading unbranded video. Error: {e.stderr.decode()}")
-        else:
-            logging.warning("Logo 'sozob.png' not found in root directory. Skipping end-card.")
-
-        blob_name = f"sozo_projects/{uid}/{project_id}/video.mp4"
-        blob = bucket.blob(blob_name)
-        blob.upload_from_filename(str(upload_path))
-        logging.info(f"Uploaded video to {blob.public_url}")
-        
-        for p in temps + [silent_vid_path, audio_mix_path, final_vid_path, branded_vid_path]:
-            if os.path.exists(p): os.unlink(p)
-            
-        return blob.public_url
-    return None
-
-# In sozo_gen.py, add these new functions at the end of the file
-
-def generate_image_with_gemini(prompt: str) -> Image.Image:
-    """Generates an image using the specified Gemini model and client configuration."""
-    logging.info(f"Generating Gemini image with prompt: '{prompt}'")
-    try:
-        # Use the genai.Client as per the correct implementation
-        client = genai.Client(api_key=API_KEY)
-        full_prompt = f"A professional, 3d digital art style illustration for a business presentation: {prompt}"
-        
-        response = client.models.generate_content(
-            model="gemini-2.0-flash-exp",
-            contents=full_prompt,
-            config=genai_types.GenerateContentConfig(
-                response_modalities=["Text", "Image"]
-            ),
-        )
-        
-        # Find the image part in the response
-        img_part = next((part for part in response.candidates[0].content.parts if part.content_type == "Image"), None)
-        
-        if img_part:
-            # The content is already bytes, so we can open it directly
-            return Image.open(io.BytesIO(img_part.content)).convert("RGB")
-        else:
-            logging.error("Gemini response did not contain an image.")
-            return None
-    except Exception as e:
-        logging.error(f"Gemini image generation failed: {e}")
-        return None
-
-def generate_slides_from_report(raw_md: str, chart_urls: dict, uid: str, project_id: str, bucket, llm):
-    """
-    Uses an AI planner to convert a report into a 10-slide presentation deck.
-    """
-    logging.info(f"Generating slides for project {project_id}")
-
-    planner_prompt = f"""
-    You are an expert presentation designer. Your task is to convert the following data analysis report into a concise and visually engaging 10-slide deck.
-
-    **Full Report Content:**
-    ---
-    {raw_md}
-    ---
-
-    **Instructions:**
-    1.  Read the entire report to understand the core narrative and key findings.
-    2.  Create a plan for exactly 10 slides.
-    3.  For each slide, define a `title` and short `content` (2-3 bullet points or a brief paragraph).
-    4.  For the visual on each slide, you must decide between two types:
-        - If a report section is supported by an existing chart (indicated by a `<generate_chart:...>` tag), you **must** use it. Set `visual_type: "existing_chart"` and `visual_ref: "the exact chart description from the tag"`.
-        - For key points without a chart (like introductions, conclusions, or text-only insights), you **must** request a new image. Set `visual_type: "new_image"` and `visual_ref: "a concise, descriptive prompt for an AI to generate a 3D digital art style illustration"`.
-    5.  You must request exactly 3-4 new images to balance the presentation.
-
-    **Output Format:**
-    Return ONLY a valid JSON array of 10 slide objects. Do not include any other text or markdown formatting.
-
-    Example:
-    [
-      {{ "slide_number": 1, "title": "Introduction", "content": "...", "visual_type": "new_image", "visual_ref": "A 3D illustration of a rising stock chart" }},
-      {{ "slide_number": 2, "title": "Sales by Region", "content": "...", "visual_type": "existing_chart", "visual_ref": "bar | Sales by Region" }},
-      ...
-    ]
-    """
-
-    try:
-        plan_response = llm.invoke(planner_prompt).content.strip()
-        if plan_response.startswith("```json"):
-            plan_response = plan_response[7:-3]
-        slide_plan = json.loads(plan_response)
-    except Exception as e:
-        logging.error(f"Failed to generate or parse slide plan: {e}")
-        return None
-
-    final_slides = []
-    for slide in slide_plan:
-        try:
-            image_url = None
-            visual_type = slide.get("visual_type")
-            visual_ref = slide.get("visual_ref")
-
-            if visual_type == "existing_chart":
-                sanitized_ref = sanitize_for_firebase_key(visual_ref)
-                image_url = chart_urls.get(sanitized_ref)
-                if not image_url:
-                    logging.warning(f"Could not find existing chart for ref: '{visual_ref}' (sanitized: '{sanitized_ref}')")
-            
-            elif visual_type == "new_image":
-                img = generate_image_with_gemini(visual_ref)
-                if img:
-                    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
-                        img_path = Path(temp_file.name)
-                        img.save(img_path, format="PNG")
-                        
-                        blob_name = f"sozo_projects/{uid}/{project_id}/slides/slide_{uuid.uuid4().hex}.png"
-                        blob = bucket.blob(blob_name)
-                        blob.upload_from_filename(str(img_path))
-                        image_url = blob.public_url
-                        logging.info(f"Uploaded new slide image to {image_url}")
-                        os.unlink(img_path)
-
-            if not image_url:
-                logging.warning(f"Visual generation failed for slide {slide.get('slide_number')}. Skipping visual for this slide.")
-
-            final_slides.append({
-                "slide_number": slide.get("slide_number"),
-                "title": slide.get("title"),
-                "content": slide.get("content"),
-                "image_url": image_url or ""
-            })
-        except Exception as slide_e:
-            logging.error(f"Failed to process slide {slide.get('slide_number')}: {slide_e}")
-            continue
-            
-    return final_slides