Simplified the app

Relied only on Pixeltable and removed loops

app.py

@@ -1,239 +1,127 @@
-# Import necessary libraries
-import gradio as gr
+
 import pixeltable as pxt
 import os
+import openai
+import gradio as gr
 import getpass
-from pixeltable.functions.video import extract_audio
-from pixeltable.functions import openai as pxop
 from pixeltable.iterators import FrameIterator
-import openai
-
-
-
-# Set up Pixeltable database and table
-db_directory = "video_db"
-table_name = "video_table"
-view_name = 'frames'
+from pixeltable.functions.video import extract_audio
+from pixeltable.functions.audio import get_metadata
+from pixeltable.functions import openai
+
+if 'OPENAI_API_KEY' not in os.environ:
+    os.environ['OPENAI_API_KEY'] = getpass.getpass('Enter your OpenAI API key:')
+
+pxt.drop_dir('directory', force=True)
+pxt.create_dir('directory')
+
+t = pxt.create_table(
+    'directory.video_table', {
+    "video": pxt.VideoType(nullable=True),
+    "sm_type": pxt.StringType(nullable=True),
+    }
+)
+
+frames_view = pxt.create_view(
+    "directory.frames",
+    t,
+    iterator=FrameIterator.create(video=t.video, num_frames=20)
+)
+
+# Create computed columns to store transformations and persist outputs
+t['audio'] = extract_audio(t.video, format='mp3')
+t['metadata'] = get_metadata(t.audio)
+t['transcription'] = openai.transcriptions(audio=t.audio, model='whisper-1')
+t['transcription_text'] = t.transcription.text
+
+@pxt.udf
+def prompt(A: str, B: str) -> list[dict]:
+    return [
+        {'role': 'system', 'content': 'You are an expert in creating social media content and you generate effective post, based on the video transcript and the type of social media asked for. Please respect the limitations in terms of characters and size of each social media platform'},
+        {'role': 'user', 'content': f'A: "{A}" \n B: "{B}"'}
+    ]
+
+t['message'] = prompt(t.sm_type, t.transcription_text)
+
+t['response'] = openai.chat_completions(messages=t.message, model='gpt-4o-mini-2024-07-18', max_tokens=500)
+t['answer'] = t.response.choices[0].message.content
 
-# Define constants
 MAX_VIDEO_SIZE_MB = 35
-GPT_MODEL = "gpt-4o-mini-2024-07-18"
-MAX_TOKENS = 500
-WHISPER_MODEL = "whisper-1"
-
-# Set OpenAI API key
-if "OPENAI_API_KEY" not in os.environ:
-    os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API Key:")
-
-# Clean up existing database and table if they exist
-pxt.drop_dir("video_db", force=True)
-if table_name in pxt.list_tables():
-    pxt.drop_table("video_db.video_table")
-
-# Create or use existing directory and table
-if db_directory not in pxt.list_dirs():
-    pxt.create_dir(db_directory)
-else:
-    print(f"Directory {db_directory} already exists. Using the existing directory.")
-
-if table_name not in pxt.list_tables():
-    t = pxt.create_table(
-        f"{db_directory}.{table_name}",
-        {
-            "video": pxt.VideoType(),
-            "video_filename": pxt.StringType(),
-            "sm_type": pxt.StringType(),
-            "sm_post": pxt.StringType(),
-        },
-    )
-else:
-    t = pxt.get_table(f"{db_directory}.{table_name}")
-    print(f"Table {table_name} already exists. Using the existing table.")
-
-# Function to generate social media post using OpenAI GPT-4 API
-def generate_social_media_post(transcript_text, social_media_type):
-    response = openai.chat.completions.create(
-        model=GPT_MODEL,
-        messages=[
-            {
-                "role": "system",
-                "content": f"You are an expert in creating social media content for {social_media_type}.",
-            },
-            {
-                "role": "user",
-                "content": f"Generate an effective and casual social media post based on this video transcript below. Make it a viral and suitable post for {social_media_type}. Transcript:\n{transcript_text}.",
-            },
-        ],
-        max_tokens=MAX_TOKENS,
-    )
-    return response.choices[0].message.content
-
-# Function to process the uploaded video and generate the post
-def process_and_generate_post(video_file, social_media_type):
-    if video_file:
-        try:
-            # Check video file size
-            video_size = os.path.getsize(video_file) / (1024 * 1024)  # Convert to MB
-            if video_size > MAX_VIDEO_SIZE_MB:
-                return f"The video file is larger than {MAX_VIDEO_SIZE_MB} MB. Please upload a smaller file."
-
-            video_filename = os.path.basename(video_file)
-            tr_audio_gen_flag = True
-            sm_gen_flag = True
-
-            # Check if video already exists in the table
-            video_df = t.where(t.video_filename == video_filename).tail(1)
-            if t.select().where(t.video_filename == video_filename).count() >= 1:
-                tr_audio_gen_flag = False
-
-            # Check if video and social media type combination exists
-            video_type_df = t.where(
-                (t.video_filename == video_filename) & (t.sm_type == social_media_type)
-            ).tail(1)
-            if video_type_df:
-                sm_gen_flag = False
-
-            # Insert video into PixelTable if it doesn't exist or if it's a new social media type
-            if (
-                (t.count() < 1)
-                or not (
-                    t.select().where(t.video_filename == video_filename).count() >= 1
-                )
-                or (video_df and not video_type_df)
-            ):
-                t.insert(
-                    [
-                        {
-                            "video": video_file,
-                            "video_filename": video_filename,
-                            "sm_type": social_media_type,
-                            "sm_post": "",
-                        }
-                    ]
-                )
 
-            # Extract audio and transcribe if needed
-            if tr_audio_gen_flag:
-                if not t.get_column(name="audio"):
-                    t["audio"] = extract_audio(t.video, format="mp3")
-                else:
-                    t.audio = extract_audio(t.video, format="mp3")
-
-                print("########### processing transcription #############")
-
-                if not t.get_column(name="transcription"):
-                    t["transcription"] = pxop.transcriptions(
-                        t.audio, model=WHISPER_MODEL
-                    )
-                else:
-                    t.transcription = pxop.transcriptions(t.audio, model=WHISPER_MODEL)
-
-            # Get the current video data
-            filtered_df = t.where(
-                (t.video_filename == video_filename) & (t.sm_type == social_media_type)
-            ).tail(1)
-
-            if len(filtered_df) == 0:
-                return "No matching video found in the table. Please ensure the video is uploaded correctly and try again."
-
-            cur_video_df = filtered_df[0]
-            plain_text = cur_video_df["transcription"]["text"]
-
-            # Generate or retrieve social media post
-            if (
-                t.select()
-                .where(
-                    (t.video_filename == video_filename)
-                    & (t.sm_type == social_media_type)
-                    & (t.sm_post != "")
-                )
-                .count()
-                >= 1
-            ):
-                print("retrieving existing social media post")
-                social_media_post = (
-                    t.select(t.sm_post)
-                    .where(
-                        (t.sm_type == social_media_type)
-                        & (t.video_filename == video_filename)
-                    )
-                    .collect()["sm_post"]
-                )
-            else:
-                print("generating new social media post")
-                social_media_post = generate_social_media_post(
-                    plain_text, social_media_type
-                )
-                if sm_gen_flag:
-                    cur_video_df.update({"sm_post": social_media_post})
-                    social_media_post = cur_video_df["sm_post"]
+def process_and_generate_post(video_file, social_media_type):
+    if not video_file:
+        return "Please upload a video file.", None
 
+    try:
+        # Check video file size
+        video_size = os.path.getsize(video_file) / (1024 * 1024)  # Convert to MB
+        if video_size > MAX_VIDEO_SIZE_MB:
+            return f"The video file is larger than {MAX_VIDEO_SIZE_MB} MB. Please upload a smaller file.", None
 
-            print("######## processing thumbnail ##########")
+        # Insert video in PixelTable
+        t.insert([{
+            "video": video_file,
+            "sm_type": social_media_type
+        }])
 
-            # images = []
-            # for frame in FrameIterator(video=video_file, num_frames=4):
-            #     images.append(frame['frame'])  # frame['frame'] is a PIL image
-            
-            if f"{db_directory}.{view_name}" in pxt.list_tables():
-                pxt.drop_table(f"{db_directory}.{view_name}")
+        # Retrieve Social media posts
+        social_media_post = t.select(t.answer).tail(1)['answer'][0]
 
-            frames_view = pxt.create_view(
-                f"{db_directory}.frames",
-                t,
-                iterator=FrameIterator.create(video=t.video, num_frames = 4)
-            )
+        # Retrieve thumbnails
+        frames = frames_view.select(frames_view.frame).tail(4)
+        thumbnails = [frame['frame'] for frame in frames]
 
-            frames = frames_view.select(frames_view.frame).collect()
-            
-            return social_media_post, [frame['frame'] for frame in frames]
+        #Display content
+        return social_media_post, thumbnails
 
-        except Exception as e:
-            return f"An error occurred: {e}"
-    else:
-        return "Please upload a video file."
+    except Exception as e:
+        return f"An error occurred: {str(e)}", None
 
 # Gradio Interface
+import gradio as gr
+
 def gradio_interface():
     with gr.Blocks(theme=gr.themes.Glass()) as demo:
-        # Set up the UI components
-        gr.Markdown(
-            """<center><font size=12>Video to Social Media Post Generator</center>"""
-        )
-        gr.Markdown(
-            """<div align="center">
-<img src="https://raw.githubusercontent.com/pixeltable/pixeltable/main/docs/source/data/pixeltable-logo-large.png" alt="Pixeltable" width="20%" />
-"""
-        )
         gr.Markdown(
-            """<center><font size=6>Data Ops powered by <a href="https://github.com/pixeltable/pixeltable">Pixeltable</a></center>"""
+            """
+            <center>
+            <h1>Video to Social Media Post Generator</h1>
+            <img src="https://raw.githubusercontent.com/pixeltable/pixeltable/main/docs/source/data/pixeltable-logo-large.png" alt="Pixeltable" width="20%" />
+            <p>Pixeltable is a declarative interface for working with text, images, embeddings, and even video, enabling you to store, transform, index, and iterate on data.</p>
+            </center>
+            """
         )
-        gr.Markdown(
-            """<center>Pixeltable is a Python library providing a declarative interface for multimodal data (text, images, audio, video). It features built-in versioning, lineage tracking, and incremental updates, enabling users to store, transform, index, and iterate on data for their ML workflows. Data transformations, model inference, and custom logic are embedded as computed columns.
-</center>"""
-        )
-        video_input = gr.Video(label=f"Upload Video File (max {MAX_VIDEO_SIZE_MB}  MB):",
-                               include_audio = True,
-                               max_length= 300,
-                               height='400px',
-                               autoplay= True)
-        social_media_type = gr.Dropdown(
-            choices=["X (Twitter)", "Facebook", "LinkedIn"],
-            label="Select Social Media Platform:",
-            value="X (Twitter)",
-        )
-        generate_btn = gr.Button("Generate Post", interactive= True)
-
-        output = gr.Textbox(label="Generated Social Media Post", show_copy_button=True)
 
-        thumbnail = gr.Gallery(label = "Pick your favorite Post Thumbnail", show_download_button= True, show_fullscreen_button= True,
-                               height= '400px')
+        with gr.Row():
+            with gr.Column():
+                video_input = gr.Video(
+                    label=f"Upload Video File (max {MAX_VIDEO_SIZE_MB} MB):",
+                    include_audio=True,
+                    max_length=300,
+                    height='400px',
+                    autoplay=True
+                )
+                social_media_type = gr.Dropdown(
+                    choices=["X (Twitter)", "Facebook", "LinkedIn", "Instagram"],
+                    label="Select Social Media Platform:",
+                    value="X (Twitter)",
+                )
+                generate_btn = gr.Button("Generate Post")
+
+            with gr.Column():
+                output = gr.Textbox(label="Generated Social Media Post", show_copy_button=True)
+                thumbnail = gr.Gallery(
+                    label="Pick your favorite Post Thumbnail",
+                    show_download_button=True,
+                    show_fullscreen_button=True,
+                    height='400px'
+                )
 
-        examples = gr.Examples(
-            [["example1.mp4"], ["example2.mp4"],["example3.mp4"]], inputs=[video_input]
+        gr.Examples(
+            examples=[["example1.mp4"], ["example2.mp4"], ["example3.mp4"]],
+            inputs=[video_input]
         )
 
-        # Connect the generate button to the processing function
         generate_btn.click(
             fn=process_and_generate_post,
             inputs=[video_input, social_media_type],
@@ -243,4 +131,5 @@ def gradio_interface():
     return demo
 
 # Launch the Gradio interface
-gradio_interface().launch(show_api=False)
+if __name__ == "__main__":
+    gradio_interface().launch(show_api=False)