Spaces:

Surrendra
/

Multimedia_FAPI

Sleeping

Added a sample code from hugging face

d356a41 verified over 1 year ago

11.2 kB

# ########## Overview ################################################################
#  The application is divided into three main components:
#  Extracts metadata, transcripts, and thumbnails from YouTube videos.
#  Summarizes the extracted transcripts using the OpenAI API.
#  Builds a Streamlit web interface that ties everything together.
#
#Step 1: Extracting YouTube Data
#	The first step involves extracting the video ID, metadata (title and channel name), 
# 	transcript, and thumbnail image from a YouTube URL. 
#
#Step 2: Summarizing the Transcript
#	With the transcript in hand, the next step is to summarize it. This is done using 
# 	the OpenAI API in summarize_text.py. Before running this script, make sure to set 
# 	your OpenAI API key as an environment variable:
#
#Step 3: Building the Web Interface
#	Finally, the web interface is created using Streamlit. Streamlit is a Python library 
# 	that simplifies the process of creating web applications for data science and machine 
#   learning projects. This file combines the functionality of the other two scripts and 
# 	provides a user-friendly interface.
#Bringing It All Together
#	Extract Video ID and Metadata: When a user inputs a YouTube URL, the app extracts the 
# 	video ID and retrieves the title and channel name using extract_metadata(url).
#	Download Thumbnail: The app downloads the video’s thumbnail and displays it using 
# 	get_thumbnail_from_url(url).
#	Get and Summarize Transcript: The transcript is fetched using get_transcript_from_url(url) 
# 	and summarized using summarize_transcript(transcript, language). The summary is displayed 
# 	in the selected language.
#################################################################################################
# TO RUN, execute on the terminal:
#streamlit run C:\projects\ewp\ai-projects\ewp-sales-bots-repo\ewp-sales-bots-program\notebooks\summarize_mp4_video.py
#

import os
import subprocess
from moviepy.video.io.VideoFileClip import VideoFileClip
from moviepy.audio.io.AudioFileClip import AudioFileClip
import whisper
from openai import OpenAI
from fastapi import FastAPI
from transformers import pipeline

#import os
#os.system("apt-get update && apt-get install -y ffmpeg")

#result = subprocess.run(["ffmpeg", "-version"], capture_output=True, text=True)
#print(result.stdout)

#input_file = "input.mp4"
#output_file = "output.mp3"
#subprocess.run(["ffmpeg", "-i", input_file, output_file])

## create a new FASTAPI app instance
app=FastAPI()

# Initialize the text generation pipeline
#pipe = pipeline("text2text-generation", model="google/flan-t5-small")


@app.get("/")
def home():
    return {"message":"Hello World"}

# Define a function to handle the GET request at `/generate`


@app.get("/generate")
def generate(text:str):
    ## use the pipeline to generate text from given input text
    output=pipe(text)

    ## return the generate text in Json reposne
    return {"output":output[0]['generated_text']}

@app.get("/transcribe")
def get_transcript_mp4(mp4_path):
    #path_to_check = r"C:\Program Files\ffmpeg\bin"
    #if path_to_check in os.environ["PATH"]:
        #print(f"{path_to_check} is in the PATH")
    #else:
        #print(f"{path_to_check} is NOT in the PATH")
        #os.environ["PATH"] += os.pathsep + path_to_check
    ffmpeg_version = subprocess.run(["ffmpeg", "-version"], capture_output=True, text=True)
    print(ffmpeg_version.stdout)
    # Assuming the mp4 file has a transcript available
    # This is a placeholder function. You would need to implement actual transcript extraction from mp4.
    # For now, we will return a dummy transcript.
    if os.path.exists(mp4_path):
        print("File exists!: ", mp4_path)
        # construct mp3 path 
        # Replace the .mp4 extension with .mp3
        # Get the file name without extension
        base_name = os.path.splitext(mp4_path)[0]
        mp3_path = f"{base_name}.mp3"  # Append .mp3 extension
        #construct wav file path
        wav_path = f"{base_name}.wav"  # Append .mp3 extension
        # construct the file to store the transcript
        transcript_path = f"{base_name}_transcript.txt"
        video = VideoFileClip(mp4_path)
        video.audio.write_audiofile(mp3_path)
        print("Conversion to mp3 complete!")
        audio = AudioFileClip(mp3_path)
        audio.write_audiofile(wav_path)
        print("Conversion to wav complete!")
        # Initialize the Whisper model (small model is a good balance for speed and accuracy)
        print("Transcribing audio...")
        model = whisper.load_model("tiny.en")  # Options: "tiny", "base", "small", "medium", "large"
        # Transcribe the audio
            
        result = model.transcribe(wav_path)
        print("Transcription completed!")
        meet_data=result["text"]
        with open(transcript_path, "w") as file:
            file.write(meet_data)
            print("Txt written to file: ", transcript_path)
        return meet_data
    else:
        print("File NOT found! Check the path for: ", mp4_path)       
        return "Test summary"
    
@app.get("/summarize")
def summarize_text(text, lang='en'):
    OpenAI.api_key = os.getenv("OPENAI_API_KEY")
    client = OpenAI(api_key=OpenAI.api_key)

    prompt = f"""
    The following text is in its original language. Provide the output in this language: {lang}. 
    Format the output as follows:
    
    Summary:
    short summary of the video
    
    Key Takeaways:
    succinct bullet point list of key takeaways
    
    Detailed important points:
    succinct bullet point list of feature details
    
    input text: {text}
    """
    
    response = client.chat.completions.create(
        messages=[{"role": "user", "content": prompt}],
        model="gpt-4o",
        temperature=0,
        #model="gpt-4-turbo", # better performance, slower inference
    )
    
    summary_text = response.to_dict()['choices'][0]['message']['content']
    return summary_text


import torch

import gradio as gr
import yt_dlp as youtube_dl
from transformers import pipeline
from transformers.pipelines.audio_utils import ffmpeg_read

import tempfile
import os

MODEL_NAME = "openai/whisper-large-v3"
BATCH_SIZE = 8
FILE_LIMIT_MB = 1000
YT_LENGTH_LIMIT_S = 3600  # limit to 1 hour YouTube files

device = 0 if torch.cuda.is_available() else "cpu"

pipe = pipeline(
    task="automatic-speech-recognition",
    model=MODEL_NAME,
    chunk_length_s=30,
    device=device,
)


def transcribe(inputs, task):
    if inputs is None:
        raise gr.Error("No audio file submitted! Please upload or record an audio file before submitting your request.")

    text = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True)["text"]
    return  text


def _return_yt_html_embed(yt_url):
    video_id = yt_url.split("?v=")[-1]
    HTML_str = (
        f'<center> <iframe width="500" height="320" src="https://www.youtube.com/embed/{video_id}"> </iframe>'
        " </center>"
    )
    return HTML_str

def download_yt_audio(yt_url, filename):
    info_loader = youtube_dl.YoutubeDL()
    
    try:
        info = info_loader.extract_info(yt_url, download=False)
    except youtube_dl.utils.DownloadError as err:
        raise gr.Error(str(err))
    
    file_length = info["duration_string"]
    file_h_m_s = file_length.split(":")
    file_h_m_s = [int(sub_length) for sub_length in file_h_m_s]
    
    if len(file_h_m_s) == 1:
        file_h_m_s.insert(0, 0)
    if len(file_h_m_s) == 2:
        file_h_m_s.insert(0, 0)
    file_length_s = file_h_m_s[0] * 3600 + file_h_m_s[1] * 60 + file_h_m_s[2]
    
    if file_length_s > YT_LENGTH_LIMIT_S:
        yt_length_limit_hms = time.strftime("%HH:%MM:%SS", time.gmtime(YT_LENGTH_LIMIT_S))
        file_length_hms = time.strftime("%HH:%MM:%SS", time.gmtime(file_length_s))
        raise gr.Error(f"Maximum YouTube length is {yt_length_limit_hms}, got {file_length_hms} YouTube video.")
    
    ydl_opts = {"outtmpl": filename, "format": "worstvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best"}
    
    with youtube_dl.YoutubeDL(ydl_opts) as ydl:
        try:
            ydl.download([yt_url])
        except youtube_dl.utils.ExtractorError as err:
            raise gr.Error(str(err))


def yt_transcribe(yt_url, task, max_filesize=75.0):
    html_embed_str = _return_yt_html_embed(yt_url)

    with tempfile.TemporaryDirectory() as tmpdirname:
        filepath = os.path.join(tmpdirname, "video.mp4")
        download_yt_audio(yt_url, filepath)
        with open(filepath, "rb") as f:
            inputs = f.read()

    inputs = ffmpeg_read(inputs, pipe.feature_extractor.sampling_rate)
    inputs = {"array": inputs, "sampling_rate": pipe.feature_extractor.sampling_rate}

    text = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True)["text"]

    return html_embed_str, text


demo = gr.Blocks()

mf_transcribe = gr.Interface(
    fn=transcribe,
    inputs=[
        gr.inputs.Audio(source="microphone", type="filepath", optional=True),
        gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe"),
    ],
    outputs="text",
    layout="horizontal",
    theme="huggingface",
    title="Whisper Large V3: Transcribe Audio",
    description=(
        "Transcribe long-form microphone or audio inputs with the click of a button! Demo uses the OpenAI Whisper"
        f" checkpoint [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to transcribe audio files"
        " of arbitrary length."
    ),
    allow_flagging="never",
)

file_transcribe = gr.Interface(
    fn=transcribe,
    inputs=[
        gr.inputs.Audio(source="upload", type="filepath", optional=True, label="Audio file"),
        gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe"),
    ],
    outputs="text",
    layout="horizontal",
    theme="huggingface",
    title="Whisper Large V3: Transcribe Audio",
    description=(
        "Transcribe long-form microphone or audio inputs with the click of a button! Demo uses the OpenAI Whisper"
        f" checkpoint [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to transcribe audio files"
        " of arbitrary length."
    ),
    allow_flagging="never",
)

yt_transcribe = gr.Interface(
    fn=yt_transcribe,
    inputs=[
        gr.inputs.Textbox(lines=1, placeholder="Paste the URL to a YouTube video here", label="YouTube URL"),
        gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe")
    ],
    outputs=["html", "text"],
    layout="horizontal",
    theme="huggingface",
    title="Whisper Large V3: Transcribe YouTube",
    description=(
        "Transcribe long-form YouTube videos with the click of a button! Demo uses the OpenAI Whisper checkpoint"
        f" [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to transcribe video files of"
        " arbitrary length."
    ),
    allow_flagging="never",
)

with demo:
    gr.TabbedInterface([mf_transcribe, file_transcribe, yt_transcribe], ["Microphone", "Audio file", "YouTube"])

demo.launch(enable_queue=True)

	# ########## Overview ################################################################
	# The application is divided into three main components:
	# Extracts metadata, transcripts, and thumbnails from YouTube videos.
	# Summarizes the extracted transcripts using the OpenAI API.
	# Builds a Streamlit web interface that ties everything together.
	#
	#Step 1: Extracting YouTube Data
	# The first step involves extracting the video ID, metadata (title and channel name),
	# transcript, and thumbnail image from a YouTube URL.
	#
	#Step 2: Summarizing the Transcript
	# With the transcript in hand, the next step is to summarize it. This is done using
	# the OpenAI API in summarize_text.py. Before running this script, make sure to set
	# your OpenAI API key as an environment variable:
	#
	#Step 3: Building the Web Interface
	# Finally, the web interface is created using Streamlit. Streamlit is a Python library
	# that simplifies the process of creating web applications for data science and machine
	# learning projects. This file combines the functionality of the other two scripts and
	# provides a user-friendly interface.
	#Bringing It All Together
	# Extract Video ID and Metadata: When a user inputs a YouTube URL, the app extracts the
	# video ID and retrieves the title and channel name using extract_metadata(url).
	# Download Thumbnail: The app downloads the video’s thumbnail and displays it using
	# get_thumbnail_from_url(url).
	# Get and Summarize Transcript: The transcript is fetched using get_transcript_from_url(url)
	# and summarized using summarize_transcript(transcript, language). The summary is displayed
	# in the selected language.
	#################################################################################################
	# TO RUN, execute on the terminal:
	#streamlit run C:\projects\ewp\ai-projects\ewp-sales-bots-repo\ewp-sales-bots-program\notebooks\summarize_mp4_video.py
	#

	import os
	import subprocess
	from moviepy.video.io.VideoFileClip import VideoFileClip
	from moviepy.audio.io.AudioFileClip import AudioFileClip
	import whisper
	from openai import OpenAI
	from fastapi import FastAPI
	from transformers import pipeline

	#import os
	#os.system("apt-get update && apt-get install -y ffmpeg")

	#result = subprocess.run(["ffmpeg", "-version"], capture_output=True, text=True)
	#print(result.stdout)

	#input_file = "input.mp4"
	#output_file = "output.mp3"
	#subprocess.run(["ffmpeg", "-i", input_file, output_file])

	## create a new FASTAPI app instance
	app=FastAPI()

	# Initialize the text generation pipeline
	#pipe = pipeline("text2text-generation", model="google/flan-t5-small")


	@app.get("/")
	def home():
	return {"message":"Hello World"}

	# Define a function to handle the GET request at `/generate`


	@app.get("/generate")
	def generate(text:str):
	## use the pipeline to generate text from given input text
	output=pipe(text)

	## return the generate text in Json reposne
	return {"output":output[0]['generated_text']}

	@app.get("/transcribe")
	def get_transcript_mp4(mp4_path):
	#path_to_check = r"C:\Program Files\ffmpeg\bin"
	#if path_to_check in os.environ["PATH"]:
	#print(f"{path_to_check} is in the PATH")
	#else:
	#print(f"{path_to_check} is NOT in the PATH")
	#os.environ["PATH"] += os.pathsep + path_to_check
	ffmpeg_version = subprocess.run(["ffmpeg", "-version"], capture_output=True, text=True)
	print(ffmpeg_version.stdout)
	# Assuming the mp4 file has a transcript available
	# This is a placeholder function. You would need to implement actual transcript extraction from mp4.
	# For now, we will return a dummy transcript.
	if os.path.exists(mp4_path):
	print("File exists!: ", mp4_path)
	# construct mp3 path
	# Replace the .mp4 extension with .mp3
	# Get the file name without extension
	base_name = os.path.splitext(mp4_path)[0]
	mp3_path = f"{base_name}.mp3" # Append .mp3 extension
	#construct wav file path
	wav_path = f"{base_name}.wav" # Append .mp3 extension
	# construct the file to store the transcript
	transcript_path = f"{base_name}_transcript.txt"
	video = VideoFileClip(mp4_path)
	video.audio.write_audiofile(mp3_path)
	print("Conversion to mp3 complete!")
	audio = AudioFileClip(mp3_path)
	audio.write_audiofile(wav_path)
	print("Conversion to wav complete!")
	# Initialize the Whisper model (small model is a good balance for speed and accuracy)
	print("Transcribing audio...")
	model = whisper.load_model("tiny.en") # Options: "tiny", "base", "small", "medium", "large"
	# Transcribe the audio

	result = model.transcribe(wav_path)
	print("Transcription completed!")
	meet_data=result["text"]
	with open(transcript_path, "w") as file:
	file.write(meet_data)
	print("Txt written to file: ", transcript_path)
	return meet_data
	else:
	print("File NOT found! Check the path for: ", mp4_path)
	return "Test summary"

	@app.get("/summarize")
	def summarize_text(text, lang='en'):
	OpenAI.api_key = os.getenv("OPENAI_API_KEY")
	client = OpenAI(api_key=OpenAI.api_key)

	prompt = f"""
	The following text is in its original language. Provide the output in this language: {lang}.
	Format the output as follows:

	Summary:
	short summary of the video

	Key Takeaways:
	succinct bullet point list of key takeaways

	Detailed important points:
	succinct bullet point list of feature details

	input text: {text}
	"""

	response = client.chat.completions.create(
	messages=[{"role": "user", "content": prompt}],
	model="gpt-4o",
	temperature=0,
	#model="gpt-4-turbo", # better performance, slower inference
	)

	summary_text = response.to_dict()['choices'][0]['message']['content']
	return summary_text


	import torch

	import gradio as gr
	import yt_dlp as youtube_dl
	from transformers import pipeline
	from transformers.pipelines.audio_utils import ffmpeg_read

	import tempfile
	import os

	MODEL_NAME = "openai/whisper-large-v3"
	BATCH_SIZE = 8
	FILE_LIMIT_MB = 1000
	YT_LENGTH_LIMIT_S = 3600 # limit to 1 hour YouTube files

	device = 0 if torch.cuda.is_available() else "cpu"

	pipe = pipeline(
	task="automatic-speech-recognition",
	model=MODEL_NAME,
	chunk_length_s=30,
	device=device,
	)


	def transcribe(inputs, task):
	if inputs is None:
	raise gr.Error("No audio file submitted! Please upload or record an audio file before submitting your request.")

	text = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True)["text"]
	return text


	def _return_yt_html_embed(yt_url):
	video_id = yt_url.split("?v=")[-1]
	HTML_str = (
	f'<center> <iframe width="500" height="320" src="https://www.youtube.com/embed/{video_id}"> </iframe>'
	" </center>"
	)
	return HTML_str

	def download_yt_audio(yt_url, filename):
	info_loader = youtube_dl.YoutubeDL()

	try:
	info = info_loader.extract_info(yt_url, download=False)
	except youtube_dl.utils.DownloadError as err:
	raise gr.Error(str(err))

	file_length = info["duration_string"]
	file_h_m_s = file_length.split(":")
	file_h_m_s = [int(sub_length) for sub_length in file_h_m_s]

	if len(file_h_m_s) == 1:
	file_h_m_s.insert(0, 0)
	if len(file_h_m_s) == 2:
	file_h_m_s.insert(0, 0)
	file_length_s = file_h_m_s[0] * 3600 + file_h_m_s[1] * 60 + file_h_m_s[2]

	if file_length_s > YT_LENGTH_LIMIT_S:
	yt_length_limit_hms = time.strftime("%HH:%MM:%SS", time.gmtime(YT_LENGTH_LIMIT_S))
	file_length_hms = time.strftime("%HH:%MM:%SS", time.gmtime(file_length_s))
	raise gr.Error(f"Maximum YouTube length is {yt_length_limit_hms}, got {file_length_hms} YouTube video.")

	ydl_opts = {"outtmpl": filename, "format": "worstvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best"}

	with youtube_dl.YoutubeDL(ydl_opts) as ydl:
	try:
	ydl.download([yt_url])
	except youtube_dl.utils.ExtractorError as err:
	raise gr.Error(str(err))


	def yt_transcribe(yt_url, task, max_filesize=75.0):
	html_embed_str = _return_yt_html_embed(yt_url)

	with tempfile.TemporaryDirectory() as tmpdirname:
	filepath = os.path.join(tmpdirname, "video.mp4")
	download_yt_audio(yt_url, filepath)
	with open(filepath, "rb") as f:
	inputs = f.read()

	inputs = ffmpeg_read(inputs, pipe.feature_extractor.sampling_rate)
	inputs = {"array": inputs, "sampling_rate": pipe.feature_extractor.sampling_rate}

	text = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True)["text"]

	return html_embed_str, text


	demo = gr.Blocks()

	mf_transcribe = gr.Interface(
	fn=transcribe,
	inputs=[
	gr.inputs.Audio(source="microphone", type="filepath", optional=True),
	gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe"),
	],
	outputs="text",
	layout="horizontal",
	theme="huggingface",
	title="Whisper Large V3: Transcribe Audio",
	description=(
	"Transcribe long-form microphone or audio inputs with the click of a button! Demo uses the OpenAI Whisper"
	f" checkpoint [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to transcribe audio files"
	" of arbitrary length."
	),
	allow_flagging="never",
	)

	file_transcribe = gr.Interface(
	fn=transcribe,
	inputs=[
	gr.inputs.Audio(source="upload", type="filepath", optional=True, label="Audio file"),
	gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe"),
	],
	outputs="text",
	layout="horizontal",
	theme="huggingface",
	title="Whisper Large V3: Transcribe Audio",
	description=(
	"Transcribe long-form microphone or audio inputs with the click of a button! Demo uses the OpenAI Whisper"
	f" checkpoint [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to transcribe audio files"
	" of arbitrary length."
	),
	allow_flagging="never",
	)

	yt_transcribe = gr.Interface(
	fn=yt_transcribe,
	inputs=[
	gr.inputs.Textbox(lines=1, placeholder="Paste the URL to a YouTube video here", label="YouTube URL"),
	gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe")
	],
	outputs=["html", "text"],
	layout="horizontal",
	theme="huggingface",
	title="Whisper Large V3: Transcribe YouTube",
	description=(
	"Transcribe long-form YouTube videos with the click of a button! Demo uses the OpenAI Whisper checkpoint"
	f" [{MODEL_NAME}](https://huggingface.co/{MODEL_NAME}) and 🤗 Transformers to transcribe video files of"
	" arbitrary length."
	),
	allow_flagging="never",
	)

	with demo:
	gr.TabbedInterface([mf_transcribe, file_transcribe, yt_transcribe], ["Microphone", "Audio file", "YouTube"])

	demo.launch(enable_queue=True)