Upload video + transcript to get a subtitled video

components/.env

@@ -0,0 +1,4 @@
+COGNITO_DOMAIN = "xxx"
+CLIENT_ID = "xxx"
+CLIENT_SECRET = "xxx"
+APP_URI = "xxx"

components/authenticate.py

@@ -0,0 +1,158 @@
+import os
+import streamlit as st
+from dotenv import load_dotenv
+import requests
+import base64
+import json
+
+load_dotenv()
+COGNITO_DOMAIN = os.environ.get("COGNITO_DOMAIN")
+CLIENT_ID = os.environ.get("CLIENT_ID")
+CLIENT_SECRET = os.environ.get("CLIENT_SECRET")
+APP_URI = os.environ.get("APP_URI")
+
+
+def init_state():
+    if "auth_code" not in st.session_state:
+        st.session_state["auth_code"] = ""
+    if "authenticated" not in st.session_state:
+        st.session_state["authenticated"] = False
+    if "user_cognito_groups" not in st.session_state:
+        st.session_state["user_cognito_groups"] = []
+
+# Get the authorization code after the user has logged in
+def get_auth_code():
+    auth_query_params = st.experimental_get_query_params()
+    try:
+        auth_code = dict(auth_query_params)["code"][0]
+    except (KeyError, TypeError):
+        auth_code = ""
+    return auth_code
+
+
+# Set the authorization code after the user has logged in
+def set_auth_code():
+    init_state()
+    auth_code = get_auth_code()
+    st.session_state["auth_code"] = auth_code
+
+
+# Get the access token from the authorization code
+def get_user_tokens(auth_code):
+    # Variables to make a post request
+    token_url = f"{COGNITO_DOMAIN}/oauth2/token"
+    client_secret_string = f"{CLIENT_ID}:{CLIENT_SECRET}"
+    client_secret_encoded = str(
+        base64.b64encode(client_secret_string.encode("utf-8")), "utf-8"
+    )
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Authorization": f"Basic {client_secret_encoded}",
+    }
+    body = {
+        "grant_type": "authorization_code",
+        "client_id": CLIENT_ID,
+        "code": auth_code,
+        "redirect_uri": APP_URI,
+    }
+
+    token_response = requests.post(token_url, headers=headers, data=body)
+    try:
+        access_token = token_response.json()["access_token"]
+        id_token = token_response.json()["id_token"]
+    except (KeyError, TypeError):
+        access_token = ""
+        id_token = ""
+
+    return access_token, id_token
+
+
+# Use access token to retrieve user info
+def get_user_info(access_token):
+    userinfo_url = f"{COGNITO_DOMAIN}/oauth2/userInfo"
+    headers = {
+        "Content-Type": "application/json;charset=UTF-8",
+        "Authorization": f"Bearer {access_token}",
+    }
+
+    userinfo_response = requests.get(userinfo_url, headers=headers)
+
+    return userinfo_response.json()
+
+
+# Decode access token to JWT to get user's cognito groups
+def pad_base64(data):
+    missing_padding = len(data) % 4
+    if missing_padding != 0:
+        data += "=" * (4 - missing_padding)
+    return data
+
+
+def get_user_cognito_groups(id_token):
+    user_cognito_groups = []
+    if id_token != "":
+        header, payload, signature = id_token.split(".")
+        printable_payload = base64.urlsafe_b64decode(pad_base64(payload))
+        payload_dict = json.loads(printable_payload)
+        try:
+            user_cognito_groups = list(dict(payload_dict)["cognito:groups"])
+        except (KeyError, TypeError):
+            pass
+    return user_cognito_groups
+
+
+# Set streamlit state variables
+def set_st_state_vars():
+    init_state()
+    auth_code = get_auth_code()
+    access_token, id_token = get_user_tokens(auth_code)
+    user_cognito_groups = get_user_cognito_groups(id_token)
+
+    if access_token != "":
+        st.session_state["auth_code"] = auth_code
+        st.session_state["authenticated"] = True
+        st.session_state["user_cognito_groups"] = user_cognito_groups
+
+
+# Login/ Logout HTML components
+login_link = f"{COGNITO_DOMAIN}/login?client_id={CLIENT_ID}&response_type=code&scope=email+openid&redirect_uri={APP_URI}"
+logout_link = f"{COGNITO_DOMAIN}/logout?client_id={CLIENT_ID}&logout_uri={APP_URI}"
+
+html_css_login = """
+<style>
+.button-login {
+  background-color: skyblue;
+  color: white !important;
+  padding: 1em 1.5em;
+  text-decoration: none;
+  text-transform: uppercase;
+}
+
+.button-login:hover {
+  background-color: #555;
+  text-decoration: none;
+}
+
+.button-login:active {
+  background-color: black;
+}
+
+</style>
+"""
+
+html_button_login = (
+    html_css_login
+    + f"<a href='{login_link}' class='button-login' target='_self'>Log In</a>"
+)
+html_button_logout = (
+    html_css_login
+    + f"<a href='{logout_link}' class='button-login' target='_self'>Log Out</a>"
+)
+
+
+def button_login():
+    return st.sidebar.markdown(f"{html_button_login}", unsafe_allow_html=True)
+
+
+def button_logout():
+    return st.sidebar.markdown(f"{html_button_logout}", unsafe_allow_html=True)

pages/02_📼_Upload_Video_File.py

@@ -9,6 +9,8 @@ from io import StringIO
 import numpy as np
 import pathlib
 import os
+import components.authenticate as authenticate
+import torch
 
 st.set_page_config(page_title="Auto Subtitled Video Generator", page_icon=":movie_camera:", layout="wide")
 
@@ -36,7 +38,7 @@ current_size = "None"
 col1, col2 = st.columns([1, 3])
 with col1:
     lottie = load_lottieurl("https://assets1.lottiefiles.com/packages/lf20_HjK9Ol.json")
-    st_lottie(lottie, speed=1, height=250, width=250)
+    st_lottie(lottie)
 
 with col2:
     st.write("""
@@ -49,8 +51,10 @@ with col2:
 
 @st.cache(allow_output_mutation=True)
 def change_model(current_size, size):
+    torch.cuda.is_available()
+    DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
     if current_size != size:
-        loaded_model = whisper.load_model(size)
+        loaded_model = whisper.load_model(size, device=DEVICE)
         return loaded_model
     else:
         raise Exception("Model size is the same as the current size.")
@@ -98,7 +102,7 @@ def getSubs(segments: Iterator[dict], format: str, maxLineWidth: int) -> str:
 def generate_subtitled_video(video, audio, transcript):
     video_file = ffmpeg.input(video)
     audio_file = ffmpeg.input(audio)
-    ffmpeg.concat(video_file.filter("subtitles", transcript), audio_file, v=1, a=1).output("final.mp4").run(quiet=True, overwrite_output=True)
+    ffmpeg.concat(video_file.filter("subtitles", transcript), audio_file, v=1, a=1).output("final.mp4").global_args('-report').run(quiet=True, overwrite_output=True)
     video_with_subs = open("final.mp4", "rb")
     return video_with_subs
 
@@ -108,7 +112,7 @@ def main():
     loaded_model = change_model(current_size, size)
     st.write(f"Model is {'multilingual' if loaded_model.is_multilingual else 'English-only'} "
         f"and has {sum(np.prod(p.shape) for p in loaded_model.parameters()):,} parameters.")
-    input_file = st.file_uploader("File", type=["mp4", "avi", "mov", "mkv"])
+    input_file = st.file_uploader("Upload Video File", type=["mp4", "avi", "mov", "mkv"])
     # get the name of the input_file
     if input_file is not None:
         filename = input_file.name[:-4]
@@ -226,5 +230,10 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
-    st.markdown("###### Made with :heart: by [@BatuhanYılmaz](https://twitter.com/batuhan3326) [![this is an image link](https://i.imgur.com/thJhzOO.png)](https://www.buymeacoffee.com/batuhanylmz)")
+    authenticate.set_st_state_vars()
+    if st.session_state["authenticated"]:
+        main()
+        authenticate.button_logout()
+    else:
+        st.info("Please log in or sign up to use the app.")
+        authenticate.button_login()

pages/03_📝_Upload_Video_File_and_Transcript.py

@@ -0,0 +1,137 @@
+import streamlit as st
+from streamlit_lottie import st_lottie
+from utils import write_vtt, write_srt
+import ffmpeg
+import requests
+from typing import Iterator
+from io import StringIO
+import numpy as np
+import pathlib
+import os
+import components.authenticate as authenticate
+
+
+st.set_page_config(page_title="Auto Subtitled Video Generator", page_icon=":movie_camera:", layout="wide")
+
+# Define a function that we can use to load lottie files from a link.
+@st.cache(allow_output_mutation=True)
+def load_lottieurl(url: str):
+    r = requests.get(url)
+    if r.status_code != 200:
+        return None
+    return r.json()
+
+
+APP_DIR = pathlib.Path(__file__).parent.absolute()
+
+LOCAL_DIR = APP_DIR / "local_transcript"
+LOCAL_DIR.mkdir(exist_ok=True)
+save_dir = LOCAL_DIR / "output"
+save_dir.mkdir(exist_ok=True)
+
+
+col1, col2 = st.columns([1, 3])
+with col1:
+    lottie = load_lottieurl("https://assets6.lottiefiles.com/packages/lf20_cjnxwrkt.json")
+    st_lottie(lottie)
+
+with col2:
+    st.write("""
+    ## Auto Subtitled Video Generator 
+    ##### ➠ Upload a video file and a transcript as .srt or .vtt file and get a video with subtitles.
+    ##### ➠ Processing time will increase as the video length increases. """)
+
+
+def getSubs(segments: Iterator[dict], format: str, maxLineWidth: int) -> str:
+    segmentStream = StringIO()
+
+    if format == 'vtt':
+        write_vtt(segments, file=segmentStream, maxLineWidth=maxLineWidth)
+    elif format == 'srt':
+        write_srt(segments, file=segmentStream, maxLineWidth=maxLineWidth)
+    else:
+        raise Exception("Unknown format " + format)
+
+    segmentStream.seek(0)
+    return segmentStream.read()
+
+
+def split_video_audio(uploaded_file):
+    with open(f"{save_dir}/input.mp4", "wb") as f:
+            f.write(uploaded_file.read())
+    audio = ffmpeg.input(f"{save_dir}/input.mp4")
+    audio = ffmpeg.output(audio, f"{save_dir}/output.wav", acodec="pcm_s16le", ac=1, ar="16k")
+    ffmpeg.run(audio, overwrite_output=True)
+
+
+def main():
+    uploaded_video = st.file_uploader("Upload Video File", type=["mp4", "avi", "mov", "mkv"])
+    # get the name of the input_file
+    if uploaded_video is not None:
+        filename = uploaded_video.name[:-4]
+    else:
+        filename = None
+    transcript_file = st.file_uploader("Upload Transcript File", type=["srt", "vtt"])
+    if transcript_file is not None:
+        transcript_name = transcript_file.name
+    else:
+        transcript_name = None
+    if uploaded_video is not None and transcript_file is not None:
+        if transcript_name[-3:] == "vtt":
+            with open("uploaded_transcript.vtt", "wb") as f:
+                f.writelines(transcript_file)
+                f.close()
+            with open(os.path.join(os.getcwd(), "uploaded_transcript.vtt"), "rb") as f:
+                vtt_file = f.read()
+            if st.button("Generate Video with Subtitles"):
+                with st.spinner("Generating Subtitled Video"):
+                    split_video_audio(uploaded_video)
+                    video_file = ffmpeg.input(f"{save_dir}/input.mp4")
+                    audio_file = ffmpeg.input(f"{save_dir}/output.wav")
+                    ffmpeg.concat(video_file.filter("subtitles", "uploaded_transcript.vtt"), audio_file, v=1, a=1).output("final.mp4").global_args('-report').run(quiet=True, overwrite_output=True)
+                    video_with_subs = open("final.mp4", "rb")
+                col3, col4 = st.columns(2)
+                with col3:
+                    st.video(uploaded_video)
+                with col4:
+                    st.video(video_with_subs)
+                st.download_button(label="Download Video with Subtitles",
+                                    data=video_with_subs,
+                                    file_name=f"{filename}_with_subs.mp4")
+
+        elif transcript_name[-3:] == "srt":
+            with open("uploaded_transcript.srt", "wb") as f:
+                f.writelines(transcript_file)
+                f.close()
+            with open(os.path.join(os.getcwd(), "uploaded_transcript.srt"), "rb") as f:
+                srt_file = f.read()
+            if st.button("Generate Video with Subtitles"):
+                with st.spinner("Generating Subtitled Video"):
+                    split_video_audio(uploaded_video)
+                    video_file = ffmpeg.input(f"{save_dir}/input.mp4")
+                    audio_file = ffmpeg.input(f"{save_dir}/output.wav")
+                    ffmpeg.concat(video_file.filter("subtitles",  "uploaded_transcript.srt"), audio_file, v=1, a=1).output("final.mp4").run(quiet=True, overwrite_output=True)
+                    video_with_subs = open("final.mp4", "rb")
+                col3, col4 = st.columns(2)
+                with col3:
+                    st.video(uploaded_video)
+                with col4:
+                    st.video(video_with_subs)
+                st.download_button(label="Download Video with Subtitles",
+                                    data=video_with_subs,
+                                    file_name=f"{filename}_with_subs.mp4")
+        else:
+            st.error("Please upload a .srt or .vtt file")
+    else:
+        st.info("Please upload a video file and a transcript file")
+
+
+if __name__ == "__main__":
+    authenticate.set_st_state_vars()
+    if st.session_state["authenticated"]:
+        main()
+        authenticate.button_logout()
+    else:
+        st.info("Please log in or sign up to use the app.")
+        authenticate.button_login()
+        

pages/{03_🔊_Upload_Audio_File.py → 04_🔊_Upload_Audio_File.py}

@@ -9,6 +9,8 @@ from io import StringIO
 import numpy as np
 import pathlib
 import os
+import components.authenticate as authenticate
+import torch
 
 st.set_page_config(page_title="Auto Transcriber", page_icon="🔊", layout="wide")
 
@@ -32,7 +34,7 @@ save_dir.mkdir(exist_ok=True)
 col1, col2 = st.columns([1, 3])
 with col1:
     lottie = load_lottieurl("https://assets1.lottiefiles.com/packages/lf20_1xbk4d2v.json")
-    st_lottie(lottie, speed=1, height=250, width=250)
+    st_lottie(lottie)
 
 with col2:
     st.write("""
@@ -48,8 +50,10 @@ current_size = "None"
 
 @st.cache(allow_output_mutation=True)
 def change_model(current_size, size):
+    torch.cuda.is_available()
+    DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
     if current_size != size:
-        loaded_model = whisper.load_model(size)
+        loaded_model = whisper.load_model(size, device=DEVICE)
         return loaded_model
     else:
         raise Exception("Model size is the same as the current size.")
@@ -98,7 +102,7 @@ def main():
     loaded_model = change_model(current_size, size)
     st.write(f"Model is {'multilingual' if loaded_model.is_multilingual else 'English-only'} "
         f"and has {sum(np.prod(p.shape) for p in loaded_model.parameters()):,} parameters.")
-    input_file = st.file_uploader("Upload an audio file", type=["mp3", "wav", "m4a"])
+    input_file = st.file_uploader("Upload Audio File", type=["mp3", "wav", "m4a"])
     if input_file is not None:
         filename = input_file.name[:-4]
     else:
@@ -201,5 +205,10 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
-    st.markdown("###### Made with :heart: by [@BatuhanYılmaz](https://twitter.com/batuhan3326) [![this is an image link](https://i.imgur.com/thJhzOO.png)](https://www.buymeacoffee.com/batuhanylmz)")
+    authenticate.set_st_state_vars()
+    if st.session_state["authenticated"]:
+        main()
+        authenticate.button_logout()
+    else:
+        st.info("Please log in or sign up to use the app.")
+        authenticate.button_login()