first push

bleep_that_sht/__init__.py

@@ -0,0 +1,9 @@
+import os
+
+base_dir = os.path.dirname(os.path.abspath(__file__))
+main_dir = os.path.dirname(base_dir)
+
+import whisper_timestamped as whisper
+
+model = whisper.load_model("tiny", device="cpu")
+model = whisper.load_model("base", device="cpu")

bleep_that_sht/audio_extractor.py

@@ -0,0 +1,13 @@
+from moviepy.editor import VideoFileClip
+
+
+def extract_audio(local_file_path: str, audio_filepath: str) -> None:
+    try:
+        video = VideoFileClip(local_file_path)
+        audio = video.audio
+        if audio is not None:
+            audio.write_audiofile(audio_filepath, verbose=False, logger=None)
+            audio.close()
+            video.close()
+    except Exception as e:
+        raise ValueError(f"error extracting audio from video {local_file_path}, exception: {e}")

bleep_that_sht/create.py

@@ -0,0 +1,89 @@
+from moviepy.editor import VideoFileClip, AudioFileClip, CompositeAudioClip
+from pydub import AudioSegment
+from bleep_that_sht import base_dir
+from bleep_that_sht.audio_extractor import extract_audio
+
+bleep_sound = AudioSegment.from_mp3(base_dir + "/bleep.mp3")
+bleep_first_sec = bleep_sound[1 * 1000 : 2 * 1000]
+
+
+# simple word cleaner - remove punctuation etc.,
+def word_cleaner(word: str) -> str:
+    return "".join(e for e in word if e.isalnum()).lower().strip()
+
+
+# collect all timestamped instances of bleep_word in transcript
+def query_transcript(bleep_words: list, timestamped_transcript: list) -> list:
+    transcript_words = sum(
+        [timestamped_transcript[i]["words"] for i in range(len(timestamped_transcript))],
+        [],
+    )
+    detected_bleep_words = []
+    for bleep_word in bleep_words:
+        detected_bleep_words += [v for v in transcript_words if word_cleaner(v["text"]) == word_cleaner(bleep_word)]
+    detected_bleep_words = sorted(detected_bleep_words, key=lambda d: d["start"])
+    return detected_bleep_words
+
+
+def bleep_replace(
+    og_video_path: str,
+    og_audio_path: str,
+    final_video_path: str,
+    final_audio_path: str,
+    bleep_words: list,
+    timestamped_transcript: dict,
+) -> None:
+
+    # # extract and save audio from original video
+    # extract_audio(local_file_path=og_video_path, audio_filepath=og_audio_path)
+
+    # input og audio file for splicing
+    test_sound = AudioSegment.from_mp3(og_audio_path)
+
+    # find bleep_words in timestamped transcript
+    bleep_word_instances = query_transcript(bleep_words, timestamped_transcript)
+
+    # start creation of test_sound_bleeped - by splicing in instance 0
+    test_clip = test_sound[:1]
+    test_sound_clips = [test_clip]
+
+    # loop over instances, thread in clips of bleep
+    prev_end_time = 1
+    for instance in bleep_word_instances:
+        # unpack bleep_word start / end times - converted to microseconds
+        start_time = int(instance["start"] * 1000) - 50
+        end_time = int(instance["end"] * 1000) + 50
+
+        # collect clip of test starting at previous end time, and leading to start_time of next bleep
+        test_clip = test_sound[prev_end_time:start_time]
+
+        # create bleep clip for this instance
+        bleep_clip = bleep_first_sec[: (end_time - start_time)]
+
+        # store test and bleep clips
+        test_sound_clips.append(test_clip)
+        test_sound_clips.append(bleep_clip)
+
+        # update prev_end_time
+        prev_end_time = end_time
+
+    # create final clip from test
+    test_clip = test_sound[prev_end_time:]
+    test_sound_clips.append(test_clip)
+
+    # save bleeped audio
+    bleeped_test_clip = sum(test_sound_clips)
+    bleeped_test_clip.export(final_audio_path, format="mp3")
+
+    # load in og video, overlay with bleeped audio
+    og_video = VideoFileClip(og_video_path)
+    bleep_audio = AudioFileClip(final_audio_path)
+    new_audioclip = CompositeAudioClip([bleep_audio])
+    og_video.audio = new_audioclip
+    og_video.write_videofile(
+        final_video_path,
+        codec="libx264",
+        audio_codec="aac",
+        temp_audiofile="temp-audio.m4a",
+        remove_temp=True,
+    )

bleep_that_sht/gradio_app_url_download.py

@@ -0,0 +1,187 @@
+from bleep_that_sht import main_dir
+from bleep_that_sht.transcribe import avaliable_models
+from bleep_that_sht.transcribe import transcribe
+from bleep_that_sht.audio_extractor import extract_audio
+from bleep_that_sht.create import bleep_replace
+from bleep_that_sht.yt_download import download_video
+import tempfile
+import uuid
+import os
+import io
+import gradio as gr
+
+
+HF_TOKEN = None
+
+try:
+    HF_TOKEN = os.environ.get("HF_TOKEN")
+except:
+    pass
+
+
+print("Setting up Gradio interface...")
+with gr.Blocks(theme=gr.themes.Soft(), title="🎬 Bleep That Sh*t 🙊") as demo:
+    with gr.Tabs():
+        with gr.TabItem("🎬 Bleep That Sh*t 🙊"):
+            with gr.Row():
+                with gr.Column(scale=4):
+                    url_input = gr.Textbox(
+                        value="https://www.youtube.com/shorts/43BhDHYBG0o",
+                        label="🔗 Paste YouTube / Shorts URL here",
+                        placeholder="e.g., https://www.youtube.com/watch?v=.",
+                        max_lines=1,
+                    )
+
+            with gr.Row():
+                with gr.Column(scale=8):
+                    bleep_words = gr.Textbox(
+                        placeholder="bleep keywords go here separated by commas",
+                        label="bleep-word list",
+                        value="treetz, ice, cream, chocolate, syrup, cookie, hooked, threats, treats, trees",
+                    )
+                with gr.Column(scale=3):
+                    model_selection = gr.Dropdown(choices=avaliable_models, value="base", label="whisper model (base only in HF space)", info="whisper model selection", interactive=False)
+                with gr.Column(scale=4):
+                    just_transcribe_button = gr.Button("Just Transcribe", variant="primary")
+                    transcribe_and_bleep_button = gr.Button("Transcribe & Bleep", variant="primary")
+
+            with tempfile.TemporaryDirectory() as tmpdirname:
+                with gr.Row():
+                    transcript_output = gr.Textbox(label="Video Transcript", placeholder="", max_lines=5, show_copy_button=True)
+
+                with gr.Row():
+                    og_video = gr.Video(
+                        visible=False,
+                        show_download_button=True,
+                        show_label=True,
+                        label="original video",
+                        format="mp4",
+                        width="50vw",
+                        height="50vw",
+                    )
+                                
+                    bleep_video = gr.Video(
+                        visible=False,
+                        show_download_button=True,
+                        show_label=True,
+                        label="bleeped video",
+                        format="mp4",
+                        width="50vw",
+                        height="50vw",
+                    )
+
+                @just_transcribe_button.click(inputs=[url_input, model_selection], outputs=[og_video, bleep_video, transcript_output])
+                def just_transcribe(url_input, model_selection):
+                    temporary_video_location = tmpdirname + "/original_" + str(uuid.uuid4()) + ".mp4"
+                    temporary_audio_location = temporary_video_location.replace("mp4", "mp3")
+
+                    download_video(url_input, temporary_video_location)
+                    filename = open(temporary_video_location, "rb")
+                    byte_file = io.BytesIO(filename.read())
+                    with open(temporary_video_location, "wb") as out:
+                        out.write(byte_file.read())
+                            
+                    new_og_video = gr.Video(
+                        value=temporary_video_location,
+                        visible=True,
+                        show_download_button=True,
+                        show_label=True,
+                        label="original video",
+                        format="mp4",
+                        width="50vw",
+                        height="50vw",
+                    )
+                    
+                    new_bleep_video = gr.Video(
+                        visible=False,
+                        show_download_button=True,
+                        show_label=True,
+                        label="bleeped video",
+                        format="mp4",
+                        width="50vw",
+                        height="50vw",
+                    )
+                    
+                    
+                    extract_audio(temporary_video_location, temporary_audio_location)
+                    transcript, timestamped_transcript = transcribe(local_file_path=temporary_audio_location, model=model_selection)
+
+                    return new_og_video, new_bleep_video, transcript
+                    
+                    
+                @transcribe_and_bleep_button.click(inputs=[url_input, model_selection, bleep_words], outputs=[og_video, bleep_video, transcript_output])
+                def transcribe_and_bleep(url_input, model_selection, bleep_words):
+                    if len(bleep_words) > 0:    
+                        temporary_video_location = tmpdirname + "/original_" + str(uuid.uuid4()) + ".mp4"
+                        temporary_audio_location = temporary_video_location.replace("mp4", "mp3")
+
+                        download_video(url_input, temporary_video_location)
+                        filename = open(temporary_video_location, "rb")
+                        byte_file = io.BytesIO(filename.read())
+                        with open(temporary_video_location, "wb") as out:
+                            out.write(byte_file.read())
+                                
+                        new_og_video = gr.Video(
+                            value=temporary_video_location,
+                            visible=True,
+                            show_download_button=True,
+                            show_label=True,
+                            label="original video",
+                            format="mp4",
+                            width="50vw",
+                            height="50vw",
+                        )
+                        
+                        extract_audio(temporary_video_location, temporary_audio_location)
+                        transcript, timestamped_transcript = transcribe(local_file_path=temporary_audio_location, model=model_selection)
+
+                        bleep_word_list = bleep_words.split(",")
+                        bleep_word_list = [v.strip() for v in bleep_word_list if len(v.strip()) > 0]
+                        bleep_video_output = temporary_video_location.replace("original", "bleep")
+                        bleep_audio_output = bleep_video_output.replace("mp4", "mp3")
+
+                        bleep_replace(
+                            temporary_video_location,
+                            temporary_audio_location,
+                            bleep_video_output,
+                            bleep_audio_output,
+                            bleep_word_list,
+                            timestamped_transcript,
+                        )
+
+                        new_bleep_video = gr.Video(
+                            value=bleep_video_output,
+                            visible=True,
+                            show_download_button=True,
+                            show_label=True,
+                            label="bleeped video",
+                            format="mp4",
+                            width="50vw",
+                            height="50vw",
+                        )
+            
+                        return new_og_video, new_bleep_video, transcript
+                    else:
+                        gr.Warning("bleep words empty!", duration=3)
+                        return None, None, None
+
+        with gr.TabItem("💡 About"):
+            with gr.Blocks() as about:
+                gr.Markdown(
+                    (
+                        "### Bleep out words of your choice from an input video.  \n"
+                        "How it works: \n\n"
+                        "1.  Provided a youtube / shorts url \n"
+                        "2.  Choose your your desired bleep keywords \n"
+                        "3.  (if running locally) Choose a model from the Whisper family to transcribe the audio (defaults to base only for HF space) \n"
+                        "4.  (optional) Press 'Just Transcribe' to examine / download just the transcription of the video (can help in choosing bleep words) \n"
+                        "5.  Press 'Transcribe and bleep' to transcribe and replace all instances of your keywords with *beep* sounds \n\n"
+                        "If you want to select your Whisper model / run longer videos pull and run the app locally. \n\n"
+                        "Notice: baseline (not fine tuned) Whisper models are used here - you may need to be creative to bleep out all the versions of an input word you want depending on its transcription. \n\n"
+                        "You do *not* need a GPU to run this locally.  Larger models take more time to process locally, but its doable. \n"
+                    )
+                )
+
+if __name__ == "__main__":
+    print("Launching Gradio interface...")
+    demo.launch()

bleep_that_sht/transcribe.py

@@ -0,0 +1,13 @@
+import whisper_timestamped as whisper
+from typing import Tuple
+
+avaliable_models = ["tiny", "base", "small", "medium", "large-v3"]
+
+
+def transcribe(local_file_path: str, model: str = "tiny", device: str = "cpu") -> Tuple[str, dict]:
+    assert model in avaliable_models, f"input model '{model}' not a member of available models = {avaliable_models}"
+    model = whisper.load_model(model, device="cpu")
+    process_output = whisper.transcribe(model, local_file_path, verbose=False)
+    transcript = process_output["text"]
+    timestamped_transcript = process_output["segments"]
+    return transcript, timestamped_transcript

bleep_that_sht/yt_download.py

@@ -0,0 +1,32 @@
+# from pytube import YouTube
+
+import yt_dlp
+import re
+
+
+def is_valid_youtube_url(url: str) -> bool:
+    if not isinstance(url, str):
+        return False
+    pattern = r"^https://www\.youtube\.com/watch\?v=[A-Za-z0-9_-]{11}$"  # youtube vido ids are always 11 chars long
+    if "shorts" in url:
+        pattern = r"^https://www\.youtube\.com/shorts/[A-Za-z0-9_-]{11}$"  # youtube vido ids are always 11 chars long
+    return re.match(pattern, url) is not None
+
+
+def download_video(url: str, savepath: str, my_proxies: dict = {}) -> None:
+    try:
+        print("Downloading video from youtube...")
+        if is_valid_youtube_url(url):
+            ydl_opts = {
+                'format': 'bestvideo[height<=720]+bestaudio/best', 
+                'merge_output_format': 'mp4',
+                'outtmpl': savepath,
+            }
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                ydl.download([url])
+
+            print("...done!")
+        else:
+            raise ValueError(f"invalid input url: {url}")
+    except Exception as e:
+        raise ValueError(f"yt_download failed with exception {e}")

requirements.gradio

@@ -0,0 +1,4 @@
+whisper-timestamped
+moviepy
+yt-dlp
+gradio