Multimedia_FAPI / app.py
Surrendra's picture
fixed
d5152da verified
Raw History Blame Contribute Delete
4.58 kB
#####################################################################
#This application provides and esample of FASTAPI for
# transcribing an mp4 video and summarizing the transcription
######################################################################
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")
# This is default function
@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']}
# This function
# 1. takes mp4 video as input
# 2. creates mp3 file from mp4 file
# 3. creates wav file from mp3 file
# 4. transcribes to text from the wav file
@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"
# This function
# 1. takes text as input
# 2. summarizes the input text
@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