app.py

import sys
import os
import validators
import datatime
from moviepy.editor import *
import gradio as gr
from huggingface_hub import snapshot_download
import huggingdace_hub as hh
from pytube import YouTube

global notify
notify = ""
global temp_addr
temp_addr = "/yt_download/" + datatime.datetime.now()

def clean_up():
    hh.delete_folder("/yt_download")
    os.mkdir("/yt_download")

def convert_video_to_audio_ffmpeg(video_file, output_ext="mp3"):
    """Converts video to audio directly using `ffmpeg` command
    with the help of subprocess module"""
    filename, ext = os.path.splitext(video_file)
    subprocess.call(["ffmpeg", "-y", "-i", video_file, f"{filename}.{output_ext}"], 
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.STDOUT)
    return (filename + "." + ext)

def get_video(url):
    if(validators.url(url) == False):
        notify = "Invalid URL!"
        return False
    yt = YouTube(url)
    mp4_files = yt.streams.filter(file_extension="mp4")
    os.mkdir(temp_addr)
    print("Temp buffer folder created.")
    
    # Download
    mp4_360p_files = mp4_files.get_by_resolution("360p")
    mp4_720p_files = mp4_files.get_by_resolution("720p")
    mp4_1080p_files = mp4_files.get_by_resolution("1080p")
    
    mp4_360p_files.download(temp_addr)
    mp4_720p_files.download(temp_addr)
    mp4_1080p_files.download(temp_addr)

    notify = "Video(s) fetched successfuly."
    return True

def extract_audio():
    for file in os.listdir("temp_addr"):
        if file.endswith(".mp4"):
            f_addr = temp_addr + "/" + convert_video_to_audio_ffmpeg(file)
            print("Current audio address:" + f_addr)
            
    snapshot_download(repo_id="Campfireman/YouTubeAudioGrabNTake", allow_patterns="*.mp3")
    notify = "Sucess. Download request will be pulled up soon. "

def extrator_pipeline(url):
    clean_up()
    get_video(url)
    extract_audio()
    return notity

demo = gr.Interface(fn=extrator_pipeline, inputs="text", outputs="text")