Update app.py

app.py

@@ -596,6 +596,290 @@ def animate_chart_fallback(spec: ChartSpecification, df: pd.DataFrame, dur: floa
     except Exception:
         return str(out)  # Return path even if failed
 
+def safe_chart(desc: str, df: pd.DataFrame, dur: float, out: Path) -> str:
+    """
+    Enhanced safe chart generation with animation for video pipeline.
+    
+    This function integrates with the existing ChartGenerator system to create
+    animated charts that are suitable for video scenes. It provides multiple
+    fallback layers to ensure reliable chart generation.
+    
+    Args:
+        desc (str): Chart description/specification
+        df (pd.DataFrame): Source data
+        dur (float): Duration in seconds for animation
+        out (Path): Output video file path
+        
+    Returns:
+        str: Path to generated video file
+    """
+    try:
+        # Initialize the enhanced chart generator
+        llm = ChatGoogleGenerativeAI(
+            model="gemini-2.0-flash", 
+            google_api_key=API_KEY, 
+            temperature=0.1
+        )
+        chart_generator = create_chart_generator(llm, df)
+        
+        # Generate AI-driven chart specification
+        with st.spinner(f"Analyzing chart requirements: {desc}..."):
+            chart_spec = chart_generator.generate_chart_spec(desc)
+        
+        # Attempt enhanced animation with specification
+        try:
+            return animate_chart_with_spec(chart_spec, df, dur, out, fps=FPS)
+        except Exception as anim_error:
+            print(f"Enhanced animation failed: {anim_error}")
+            
+            # Fallback 1: Static chart with fade animation
+            try:
+                temp_png = Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.png"
+                if execute_chart_spec(chart_spec, df, temp_png):
+                    img = cv2.imread(str(temp_png))
+                    if img is not None:
+                        img = cv2.resize(img, (WIDTH, HEIGHT))
+                        return animate_image_fade(img, dur, out, fps=FPS)
+                    else:
+                        raise RuntimeError("Failed to load generated chart image")
+                else:
+                    raise RuntimeError("Chart specification execution failed")
+                    
+            except Exception as static_error:
+                print(f"Static chart generation failed: {static_error}")
+                
+                # Fallback 2: Quick chart generation
+                try:
+                    temp_png = Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.png"
+                    quick_chart(desc, df, temp_png)
+                    
+                    if temp_png.exists():
+                        img = cv2.imread(str(temp_png))
+                        if img is not None:
+                            img = cv2.resize(img, (WIDTH, HEIGHT))
+                            return animate_image_fade(img, dur, out, fps=FPS)
+                        else:
+                            raise RuntimeError("Failed to load quick chart image")
+                    else:
+                        raise RuntimeError("Quick chart generation failed")
+                        
+                except Exception as quick_error:
+                    print(f"Quick chart generation failed: {quick_error}")
+                    
+                    # Fallback 3: AI-generated image
+                    try:
+                        # Generate descriptive prompt for AI image generation
+                        img_prompt = f"Professional business chart showing {desc}. Clean, modern design with clear data visualization."
+                        img = generate_image_from_prompt(img_prompt)
+                        
+                        # Convert PIL to OpenCV format
+                        img_cv = cv2.cvtColor(
+                            np.array(img.resize((WIDTH, HEIGHT))), 
+                            cv2.COLOR_RGB2BGR
+                        )
+                        return animate_image_fade(img_cv, dur, out, fps=FPS)
+                        
+                    except Exception as ai_error:
+                        print(f"AI image generation failed: {ai_error}")
+                        
+                        # Fallback 4: Placeholder with text
+                        return create_placeholder_chart_video(desc, dur, out)
+                        
+    except Exception as e:
+        print(f"Safe chart generation completely failed: {e}")
+        # Ultimate fallback
+        return create_placeholder_chart_video(desc, dur, out)
+
+
+def create_placeholder_chart_video(desc: str, dur: float, out: Path) -> str:
+    """
+    Create a placeholder video with descriptive text when all chart generation fails.
+    
+    Args:
+        desc (str): Chart description
+        dur (float): Duration in seconds
+        out (Path): Output path
+        
+    Returns:
+        str: Path to generated placeholder video
+    """
+    try:
+        # Create a professional-looking placeholder
+        fig, ax = plt.subplots(figsize=(16, 9), dpi=100)
+        fig.patch.set_facecolor('#f8f9fa')
+        ax.set_facecolor('#ffffff')
+        
+        # Add title and description
+        ax.text(0.5, 0.65, "Data Visualization", 
+                ha='center', va='center', fontsize=24, fontweight='bold',
+                color='#2c3e50', transform=ax.transAxes)
+        
+        ax.text(0.5, 0.45, desc, 
+                ha='center', va='center', fontsize=16, 
+                color='#34495e', transform=ax.transAxes,
+                wrap=True, bbox=dict(boxstyle="round,pad=0.3", facecolor='#ecf0f1', alpha=0.8))
+        
+        ax.text(0.5, 0.25, "Chart generation in progress...", 
+                ha='center', va='center', fontsize=12, 
+                color='#7f8c8d', transform=ax.transAxes)
+        
+        # Add some decorative elements
+        ax.add_patch(plt.Rectangle((0.1, 0.1), 0.8, 0.8, 
+                                 fill=False, edgecolor='#3498db', linewidth=3,
+                                 transform=ax.transAxes))
+        
+        ax.set_xlim(0, 1)
+        ax.set_ylim(0, 1)
+        ax.axis('off')
+        
+        # Save as temporary image
+        temp_png = Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.png"
+        plt.savefig(temp_png, dpi=150, bbox_inches='tight', 
+                   facecolor='#f8f9fa', edgecolor='none')
+        plt.close()
+        
+        # Convert to video
+        img = cv2.imread(str(temp_png))
+        if img is not None:
+            img = cv2.resize(img, (WIDTH, HEIGHT))
+            return animate_image_fade(img, dur, out, fps=FPS)
+        else:
+            # Last resort: create solid color video
+            return create_solid_color_video(dur, out)
+            
+    except Exception as e:
+        print(f"Placeholder creation failed: {e}")
+        return create_solid_color_video(dur, out)
+
+
+def create_solid_color_video(dur: float, out: Path) -> str:
+    """
+    Create a simple solid color video as the ultimate fallback.
+    
+    Args:
+        dur (float): Duration in seconds
+        out (Path): Output path
+        
+    Returns:
+        str: Path to generated video
+    """
+    try:
+        # Create a simple colored frame
+        frame = np.full((HEIGHT, WIDTH, 3), [240, 240, 240], dtype=np.uint8)
+        
+        # Add simple text
+        cv2.putText(frame, "Data Visualization", 
+                   (WIDTH//2 - 200, HEIGHT//2 - 50), 
+                   cv2.FONT_HERSHEY_SIMPLEX, 2, (100, 100, 100), 3)
+        
+        cv2.putText(frame, "Loading...", 
+                   (WIDTH//2 - 80, HEIGHT//2 + 50), 
+                   cv2.FONT_HERSHEY_SIMPLEX, 1, (150, 150, 150), 2)
+        
+        # Write video
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        video_writer = cv2.VideoWriter(str(out), fourcc, FPS, (WIDTH, HEIGHT))
+        
+        total_frames = int(dur * FPS)
+        for i in range(total_frames):
+            video_writer.write(frame)
+        
+        video_writer.release()
+        return str(out)
+        
+    except Exception as e:
+        print(f"Solid color video creation failed: {e}")
+        # If even this fails, just return the output path
+        return str(out)
+
+
+def animate_image_fade(img: np.ndarray, dur: float, out: Path, fps: int = 24) -> str:
+    """
+    Create a fade-in animation for static images.
+    
+    Args:
+        img (np.ndarray): Input image in BGR format
+        dur (float): Duration in seconds
+        out (Path): Output video path
+        fps (int): Frames per second
+        
+    Returns:
+        str: Path to generated video
+    """
+    try:
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        video_writer = cv2.VideoWriter(str(out), fourcc, fps, (WIDTH, HEIGHT))
+        
+        total_frames = int(dur * fps)
+        fade_frames = min(int(fps * 0.5), total_frames // 3)  # 0.5 second fade or 1/3 of total
+        
+        for frame_idx in range(total_frames):
+            if frame_idx < fade_frames:
+                # Fade in
+                alpha = frame_idx / fade_frames
+                faded_img = cv2.addWeighted(img, alpha, np.zeros_like(img), 1 - alpha, 0)
+            else:
+                # Full opacity
+                faded_img = img
+            
+            video_writer.write(faded_img)
+        
+        video_writer.release()
+        return str(out)
+        
+    except Exception as e:
+        print(f"Image fade animation failed: {e}")
+        return str(out)
+
+
+def concat_media(file_paths: List[str], output_path: Path, media_type: str):
+    """
+    Concatenate multiple media files using FFmpeg.
+    
+    Args:
+        file_paths (List[str]): List of input file paths
+        output_path (Path): Output file path
+        media_type (str): Either 'video' or 'audio'
+    """
+    if not file_paths:
+        return
+        
+    try:
+        # Create temporary file list for FFmpeg
+        list_file = Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.txt"
+        
+        with open(list_file, 'w') as f:
+            for path in file_paths:
+                # Escape path for FFmpeg
+                escaped_path = str(path).replace('\\', '\\\\').replace("'", "\\'")
+                f.write(f"file '{escaped_path}'\n")
+        
+        # Build FFmpeg command
+        cmd = [
+            "ffmpeg", "-y", "-f", "concat", "-safe", "0",
+            "-i", str(list_file)
+        ]
+        
+        if media_type == "video":
+            cmd.extend(["-c", "copy"])
+        else:  # audio
+            cmd.extend(["-c:a", "aac"])
+            
+        cmd.append(str(output_path))
+        
+        # Execute FFmpeg command
+        subprocess.run(cmd, check=True, capture_output=True)
+        
+        # Clean up temporary file
+        list_file.unlink(missing_ok=True)
+        
+    except Exception as e:
+        print(f"Media concatenation failed: {e}")
+        # Create a fallback if concatenation fails
+        if file_paths:
+            # Just copy the first file as a fallback
+            subprocess.run(["cp", file_paths[0], str(output_path)], check=False)
+
 # ─── ENHANCED MAIN FUNCTIONS (DROP-IN REPLACEMENTS) ────────────────────────────
 
 def generate_report_bundle(buf: bytes, name: str, ctx: str, key: str):