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("""