import pixeltable as pxt import os import openai import gradio as gr import getpass from pixeltable.iterators import FrameIterator from pixeltable.functions.video import extract_audio from pixeltable.functions.audio import get_metadata from pixeltable.functions import openai # Store OpenAI API Key if 'OPENAI_API_KEY' not in os.environ: os.environ['OPENAI_API_KEY'] = getpass.getpass('Enter your OpenAI API key:') MAX_VIDEO_SIZE_MB = 35 CONCURRENCY_LIMIT = 1 def process_and_generate_post(video_file, social_media_type, progress=gr.Progress()): progress(0, desc="Initializing...") # Create a Table, a View, and Computed Columns 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, fps=1) ) # 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 progress(0.1, desc="Creating UDFs...") # Custom User-Defined Function (UDF) for Generating Social Media Prompts @pxt.udf def prompt(A: str, B: str) -> list[dict]: system_msg = 'You are an expert in creating social media content and you generate effective post, based on user content. Respect the social media platform guidelines and constraints.' user_msg = f'A: "{A}" \n B: "{B}"' return [ {'role': 'system', 'content': system_msg}, {'role': 'user', 'content': user_msg} ] # Apply the UDF to create a new column t['message'] = prompt(t.sm_type, t.transcription_text) """## Generating Responses with OpenAI's GPT Model""" progress(0.2, desc="Calling LLMs") # # Generate responses using OpenAI's chat completion API t['response'] = openai.chat_completions(messages=t.message, model='gpt-4o-mini-2024-07-18', max_tokens=500) ## Extract the content of the response t['answer'] = t.response.choices[0].message.content 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 progress(0.4, desc="Inserting video...") # # Insert a video into the table. Pixeltable supports referencing external data sources like URLs t.insert([{ "video": video_file, "sm_type": social_media_type }]) progress(0.6, desc="Generating posts...") # Retrieve Social media posts social_media_post = t.select(t.answer).tail(1)['answer'][0] # Retrieve Audio audio = t.select(t.audio).tail(1)['audio'][0] # Retrieve thumbnails thumbnails = frames_view.select(frames_view.frame).tail(6)['frame'] progress(0.8, desc="Preparing results...") # Retrieve Pixeltable Table containing all videos and stored data df_output = t.select(t.transcription_text).tail(1)['transcription_text'][0] #Display content return social_media_post, thumbnails, df_output, audio 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.Monochrome()) as demo: gr.Markdown("""
Pixeltable is a declarative interface for working with text, images, embeddings, and even video, enabling you to store, transform, index, and iterate on data.
""" ) with gr.Row(): with gr.Column(): gr.Markdown("""