app.py

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

global notify
notify = ""
global temp_addr
temp_addr = "./yt_download" # + str(int(datetime.datetime.timestamp))
temp_audio_addr = ""

def clean_up():
    os.rmdir("./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")
    
    if(mp4_360p_files): mp4_360p_files.download(temp_addr)
    if(mp4_720p_files): mp4_720p_files.download(temp_addr)
    if(mp4_1080p_files): mp4_1080p_files.download(temp_addr)

    notify = "Video(s) fetched successfuly."
    print(notify)
    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. "
    print(notify)

    # For debugging use
    print(f_addr)

def extrator_pipeline(url):
    #clean_up()
    get_video(url)
    temp_audio_addr = extract_audio()
    return notify

demo = gr.Interface(
    fn=extrator_pipeline, 
    inputs="text", 
    outputs=gr.Audio(source = "upload"),
    examples=[
        [os.path.join(os.path.dirname(__file__),temp_audio_addr)]
    ]
)
demo.launch()