Update video_gen.py

video_gen.py

@@ -0,0 +1,165 @@
+# -----------------------
+# Video Creation Functions
+# -----------------------
+import os
+import re
+import time
+import tempfile
+import requests
+import json
+import io
+import base64
+import numpy as np
+import cv2
+import logging
+import uuid
+import subprocess
+from pathlib import Path
+import urllib.parse
+import pandas as pd
+from PyPDF2 import PdfReader
+import plotly.graph_objects as go
+import matplotlib.pyplot as plt
+import matplotlib
+import matplotlib.pyplot as plt
+from io import BytesIO
+import dataframe_image as dfi
+from PIL import ImageFont, ImageDraw, Image
+import seaborn as sns
+
+
+
+def create_silent_video(images, durations, output_path, logo_path="sozo_logo2.png", font_path="lazy_dog.ttf"):
+    try:
+        height, width = 720, 1280
+        fps = 24
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        video = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+
+        if not video.isOpened():
+            st.error("Failed to create video file.")
+            return None
+
+        # Load font for text overlay
+        font_size = 45
+        font = ImageFont.truetype(font_path, font_size)
+
+        # Load logo for fallback and full-screen display at the end
+        logo = None
+        if logo_path:
+            logo = cv2.imread(logo_path)
+            if logo is not None:
+                logo = cv2.resize(logo, (width, height))  # Resize logo to full screen
+            else:
+                st.warning(f"Failed to load logo from {logo_path}. No fallback image will be used.")
+
+        for img, duration in zip(images, durations):
+            try:
+                img = img.convert("RGB")
+                img_resized = img.resize((width, height))
+                frame = np.array(img_resized)
+            except Exception as e:
+                print(f"Invalid image detected, replacing with logo: {e}")
+                if logo is not None:
+                    frame = logo  # Use the logo as a fallback
+                else:
+                    # If no logo is available, create a blank frame
+                    frame = np.zeros((height, width, 3), dtype=np.uint8)
+
+            # Convert to PIL for text drawing
+            pil_img = Image.fromarray(frame)
+            draw = ImageDraw.Draw(pil_img)
+
+            # Add "Sozo Dream Lab" text at bottom right
+            text1 = "Made With"
+            text2 = "Sozo Dream Lab"
+
+            # Calculate the height of the first text to adjust the second text's position
+            bbox = draw.textbbox((0, 0), text1, font=font)
+            text1_width = bbox[2] - bbox[0]
+            text1_height = bbox[3] - bbox[1]
+
+            text_position1 = (width - 270, height - 120)  # position for "Made with"
+            text_position2 = (width - 330, height - 120 + text1_height + 5) # position for "Sozo dream lab", +5 for a little gap.
+
+            draw.text(text_position1, text1, font=font, fill=(81, 34, 97, 255))  # RGB: Purple
+            draw.text(text_position2, text2, font=font, fill=(81, 34, 97, 255))  # RGB: Purple
+
+            # Convert back to OpenCV format
+            frame = np.array(pil_img)
+            frame_cv = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
+
+            # Write frame multiple times to match duration
+            for _ in range(int(duration * fps)):
+                video.write(frame_cv)
+
+        # Add full-screen logo frame at the end
+        if logo is not None:
+            for _ in range(int(3 * fps)):  # Display for 3 seconds
+                video.write(logo)
+
+        video.release()
+        return output_path
+
+    except Exception as e:
+        st.error(f"Error creating silent video: {e}")
+        return None
+
+
+def combine_video_audio(video_path, audio_files, output_path=None):
+    try:
+        if output_path is None:
+            output_path = f"final_video_{uuid.uuid4()}.mp4"
+        temp_audio_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3")
+        temp_audio_file.close()
+        if len(audio_files) > 1:
+            concat_list_path = tempfile.NamedTemporaryFile(delete=False, suffix=".txt")
+            with open(concat_list_path.name, 'w') as f:
+                for af in audio_files:
+                    f.write(f"file '{af}'\n")
+            concat_cmd = [
+                'ffmpeg', '-y', '-f', 'concat', '-safe', '0',
+                '-i', concat_list_path.name, '-c', 'copy', temp_audio_file.name
+            ]
+            subprocess.run(concat_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            os.unlink(concat_list_path.name)
+            combined_audio = temp_audio_file.name
+        else:
+            combined_audio = audio_files[0] if audio_files else None
+        if not combined_audio:
+            return video_path
+        combine_cmd = [
+            'ffmpeg', '-y', '-i', video_path, '-i', combined_audio,
+            '-map', '0:v', '-map', '1:a', '-c:v', 'libx264',
+            '-crf', '23', '-c:a', 'aac', '-shortest', output_path
+        ]
+        subprocess.run(combine_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        os.unlink(temp_audio_file.name)
+        return output_path
+    except Exception:
+        return video_path
+
+def create_video(images, audio_files, output_path=None):
+    try:
+        try:
+            subprocess.run(['ffmpeg', '-version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        except FileNotFoundError:
+            st.error("ffmpeg not installed.")
+            return None
+        if output_path is None:
+            output_path = f"output_video_{uuid.uuid4()}.mp4"
+        silent_video_path = f"silent_{uuid.uuid4()}.mp4"
+        durations = [get_audio_duration(af) if af else 5.0 for af in audio_files]
+        if len(durations) < len(images):
+            durations.extend([5.0]*(len(images)-len(durations)))
+        silent_video = create_silent_video(images, durations, silent_video_path)
+        if not silent_video:
+            return None
+        final_video = combine_video_audio(silent_video, audio_files, output_path)
+        try:
+            os.unlink(silent_video_path)
+        except Exception:
+            pass
+        return final_video
+    except Exception:
+        return None