Added src folder

pages/reddit.py

@@ -0,0 +1,123 @@
+from pathlib import Path
+import random
+import streamlit as st
+
+import praw
+
+HOME = Path(__name__).parent.absolute()
+
+
+@st.cache_data
+def create_instance(*args, **kwargs):
+    reddit = praw.Reddit(
+        client_id=kwargs.get('client_id'),
+        client_secret=kwargs.get('client_secret'),
+        user_agent=kwargs.get('user_agent'),
+    )
+
+    subreddit = get_subreddit(reddit=reddit, subreddit=kwargs.get(
+        'subreddit'), nsfw=kwargs.get('nsfw'))
+    submission = get_random_submission(subreddit=subreddit)
+    st.session_state['submission'] = submission
+    return True
+
+
+def get_subreddit(*args, **kwargs):
+    reddit = kwargs.get('reddit')
+    subreddit = reddit.subreddit(kwargs.get('subreddit'))
+    nsfw = kwargs.get('nsfw')
+    try:
+        st.text(f"Subreddit: {subreddit.display_name}")
+    except Exception as exception:
+        st.exception(exception=exception)
+
+    if subreddit.over18 and not nsfw:
+        st.error(
+            body='subreddit has NSFW contents but you did not select to scrape them')
+    return subreddit
+
+
+def get_random_submission(*args, **kwargs):
+    subreddit = kwargs.get('subreddit')
+    submissions = [submission for submission in subreddit.hot(limit=10)]
+    return random.choice(submissions)
+
+
+# Streamlit Config
+st.set_page_config(
+    page_title="Whisper-TikTok",
+    page_icon="💬",
+    layout="wide",
+    initial_sidebar_state="expanded",
+    menu_items={
+        'Get Help': 'https://github.com/MatteoFasulo/Whisper-TikTok',
+        'Report a bug': "https://github.com/MatteoFasulo/Whisper-TikTok/issues",
+        'About':
+            """
+            # Whisper-TikTok
+            Whisper-TikTok is an innovative AI-powered tool that leverages the prowess of Edge TTS, OpenAI-Whisper, and FFMPEG to craft captivating TikTok videos also with a web application interface!
+
+            Mantainer: https://github.com/MatteoFasulo
+
+            If you find a bug or if you just have questions about the project feel free to reach me at https://github.com/MatteoFasulo/Whisper-TikTok
+            Any contribution to this project is welcome to improve the quality of work!
+            """
+    }
+)
+
+st.page_link("app.py", label="Home", icon="🏠")
+st.page_link("https://github.com/MatteoFasulo/Whisper-TikTok",
+             label="GitHub", icon="🌎")
+
+with st.sidebar:
+    with st.expander("ℹ️ How to use"):
+        st.write(
+            """
+            Before starting you will need to create a new [Reddit API App](https://www.reddit.com/prefs/apps) by selecting `script` (personal use).
+            Then, after putting the App name, http://localhost as `reddit uri` and `about url`, you have just to insert those values in this dashboard to use the Reddit API for scraping any subreddit.
+            """)
+    client_id = st.text_input(label='Reddit Client ID')
+    client_secret = st.text_input(
+        label='Reddit Client Secret', type='password')
+    user_agent = st.text_input(label='Reddit User Agent')
+
+
+st.title("🏆 Whisper-TikTok 🚀")
+st.subheader('Reddit section')
+st.write("""
+    This section allows you to generate videos from subreddits.""")
+
+st.divider()
+
+LEFT, RIGHT = st.columns(2)
+
+with LEFT:
+    num_videos = st.number_input(label='How many videos do you want to generate?',
+                                 min_value=1, max_value=10, value=1, step=1)
+
+    subreddit = st.text_input(
+        label='What Subreddit do you want to use', placeholder='AskReddit')
+
+    nsfw = st.checkbox(label='NSFW content?', value=False)
+
+    max_chars = st.slider(label='Maximum number of characters per line',
+                          min_value=10, max_value=50, value=38, step=1)
+
+    max_words = st.number_input(label='Maximum number of words per line', min_value=1,
+                                max_value=5, value=2, step=1)
+
+    result = st.button('Get subreddit')
+
+    with RIGHT:
+        if result:
+            create_instance(client_id=client_id, client_secret=client_secret,
+                            user_agent=user_agent, subreddit=subreddit, nsfw=nsfw)
+            submission = st.session_state['submission']
+            title = submission.title
+            submission.comment_sort = "new"
+            top_level_comments = list(submission.comments)
+            max_comments = 10
+            st.subheader(title)
+            for comment in top_level_comments[:max_comments]:
+                st.text(comment.body)
+                st.divider()

src/arg_parser.py

@@ -0,0 +1,98 @@
+import argparse
+import sys
+
+# voice_manager.py
+from src.voice_manager import VoicesManager
+
+import msg
+from utils import rgb_to_bgr
+
+
+async def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--model", default="small", help="Model to use",
+                        choices=["tiny", "base", "small", "medium", "large"], type=str)
+    parser.add_argument("--non_english", action='store_true',
+                        help="Don't use the english model.")
+    parser.add_argument("--url", metavar='U', default="https://www.youtube.com/watch?v=intRX7BRA90",
+                        help="Youtube URL to download as background video.", type=str)
+    parser.add_argument("--tts", default="en-US-ChristopherNeural",
+                        help="Voice to use for TTS", type=str)
+    parser.add_argument(
+        "--list-voices", help="Use `edge-tts --list-voices` to list all voices", action='help')
+    parser.add_argument("--random_voice", action='store_true',
+                        help="Random voice for TTS", default=False)
+    parser.add_argument("--gender", choices=["Male", "Female"],
+                        help="Gender of the random TTS voice", type=str)
+    parser.add_argument(
+        "--language", help="Language of the random TTS voice for example: en-US", type=str)
+    parser.add_argument("--sub_format",
+                        help="Subtitle format", choices=["u", "i", "b"], default="b", type=str)
+    parser.add_argument("--sub_position",
+                        help="Subtitle position", choices=[i for i in range(1, 10)], default=5, type=int)
+    parser.add_argument("--font", help="Subtitle font",
+                        default="Lexend Bold", type=str)
+    parser.add_argument("--font_color", help="Subtitle font color in hex format: FFF000",
+                        default="FFF000", type=str)
+    parser.add_argument(
+        "--font_size", help="Subtitle font size", default=21, type=int)
+    parser.add_argument('--max_characters', default=38,
+                        type=int, help='Max characters per line')
+    parser.add_argument('--max_words', default=2, type=int,
+                        help='Max words per segment')
+    parser.add_argument("--upload_tiktok", help="Upload to TikTok after creating the video",
+                        action='store_true', default=False)
+    parser.add_argument("-v", "--verbose", action='store_true',
+                        help="Verbose")
+    args = parser.parse_args()
+
+    if args.random_voice:  # Random voice
+        args.tts = None
+        if not args.gender:
+            print(
+                f"{msg.ERROR}When using --random_voice, please specify both --gender and --language arguments.")
+            sys.exit(1)
+
+        elif not args.language:
+            print(
+                f"{msg.ERROR}When using --random_voice, please specify both --gender and --language arguments.")
+            sys.exit(1)
+
+        elif args.gender and args.language:
+            # Check if voice is valid
+            voices_manager_obj = await VoicesManager().create()
+            voices = await VoicesManager().find(voices_manager_obj, args.gender, args.language)
+            args.tts = voices['Name']
+
+            # Check if language is english
+            if not str(args.language).startswith('en'):
+                args.non_english = True
+
+    else:
+        # Check if voice is valid
+        voices = await VoicesManager().create()
+        args.language = '-'.join(i for i in args.tts.split('-')[0:2])
+        voices = voices.find(Locale=args.language)
+        if len(voices) == 0:
+            # Voice not found
+            print(
+                f"{msg.ERROR}Specified TTS voice not found. Use `edge-tts --list-voices` to list all voices.")
+            sys.exit(1)
+
+    # Extract language from TTS voice
+    if args.tts:
+        lang_prefix = args.tts.split('-')[0]
+        if not lang_prefix.startswith('en'):
+            args.non_english = True
+
+    # Cast font color to lowercase
+    args.font_color = args.font_color.lower()
+
+    # Remove # from font color
+    if args.font_color.startswith('#'):
+        args.font_color = args.font_color[1:]
+
+    # Convert font color from RGB to BGR
+    args.font_color = rgb_to_bgr(args.font_color)
+
+    return args

src/logger.py

@@ -0,0 +1,38 @@
+import os
+import datetime
+import logging
+from pathlib import Path
+
+
+class KeepDir:
+    def __init__(self):
+        self.original_dir = os.getcwd()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        os.chdir(self.original_dir)
+
+    def chdir(self, path):
+        os.chdir(path)
+
+
+def setup_logger():
+    HOME = Path.cwd()
+    log_directory = HOME / 'log'
+    if not log_directory.exists():
+        log_directory.mkdir()
+
+    with KeepDir() as keep_dir:
+        keep_dir.chdir(log_directory)
+        log_filename = f'{datetime.date.today()}.log'
+        logging.basicConfig(
+            level=logging.INFO,
+            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+            handlers=[
+                logging.FileHandler(log_filename),
+            ]
+        )
+        logger = logging.getLogger(__name__)
+    return logger

src/subtitle_creator.py

@@ -0,0 +1,37 @@
+import os
+from pathlib import Path
+import torch
+
+
+def srt_create(whisper_model, path: str, series: str, part: int, text: str, filename: str, **kwargs) -> bool:
+    series = series.replace(' ', '_')
+
+    srt_path = f"{path}{os.sep}{series}{os.sep}"
+    srt_filename = f"{srt_path}{series}_{part}.srt"
+    ass_filename = f"{srt_path}{series}_{part}.ass"
+
+    absolute_srt_path = Path(srt_filename).absolute()
+    absolute_ass_path = Path(ass_filename).absolute()
+
+    word_dict = {
+        'Fontname': kwargs.get('font', 'Arial'),
+        'Alignment': kwargs.get('sub_position', 5),
+        'BorderStyle': '1',
+        'Outline': '1',
+        'Shadow': '2',
+        'Blur': '21',
+        'Fontsize': kwargs.get('font_size', 21),
+        'MarginL': '0',
+        'MarginR': '0',
+    }
+
+    transcribe = whisper_model.transcribe(
+        filename, regroup=True, fp16=torch.cuda.is_available())
+
+    transcribe.split_by_gap(0.5).split_by_length(kwargs.get(
+        'max_characters')).merge_by_gap(0.15, max_words=kwargs.get('max_words'))
+
+    transcribe.to_srt_vtt(str(absolute_srt_path), word_level=True)
+    transcribe.to_ass(str(absolute_ass_path), word_level=True,
+                      highlight_color=kwargs.get('font_color'), **word_dict)
+    return ass_filename

src/text_to_speech.py

@@ -0,0 +1,8 @@
+import edge_tts
+
+
+async def tts(final_text: str, voice: str = "en-US-ChristopherNeural", stdout: bool = False, outfile: str = "tts.mp3", args=None) -> bool:
+    communicate = edge_tts.Communicate(final_text, voice)
+    if not stdout:
+        await communicate.save(outfile)
+    return True

src/tiktok.py

@@ -0,0 +1,30 @@
+import logging
+import os
+
+from tiktok_uploader.upload import upload_video
+
+logger = logging.getLogger(__name__)
+
+
+def upload_tiktok(file, title: str, tags: list, headless: bool = False):
+    if not os.path.isfile('cookies.txt'):
+        logger.error('Cookie file not found')
+
+    else:
+        logger.info('Cookie file found')
+
+        if len(tags) > 0:
+            tags = ' '.join([f"#{tag}" for tag in tags])
+            description = f"{title} {tags}"
+        else:
+            description = title
+
+        try:
+            upload_video(file, description=description, cookies='cookies.txt',
+                         comment=True, stitch=False, duet=False, headless=headless)
+
+        except Exception as e:
+            logger.exception(e)
+            return False
+
+        return True

src/video_creator.py

@@ -0,0 +1,95 @@
+import json
+from pathlib import Path
+
+import stable_whisper as whisper
+from .logger import setup_logger
+from .subtitle_creator import srt_create
+from .text_to_speech import tts
+from .tiktok import upload_tiktok
+from .video_prepare import prepare_background
+from .video_downloader import download_video as youtube_download
+from utils import *
+
+HOME = Path.cwd()
+logger = setup_logger()
+media_folder = HOME / 'media'
+
+
+class VideoCreator:
+    def __init__(self, video, args):
+        self.args = args
+        self.video = video
+
+        self.series = video.get('series', '')
+        self.part = video.get('part', '')
+        self.text = video.get('text', '')
+        self.tags = video.get('tags', list())
+        self.outro = video.get('outro', '')
+        self.path = Path(media_folder).absolute()
+
+    def download_video(self, folder='background'):
+        youtube_download(url=self.args.url, folder=folder)
+        console.log(
+            f"{msg.OK}Video downloaded from {self.args.url} to {folder}")
+        logger.info(f"Video downloaded from {self.args.url} to {folder}")
+
+    def load_model(self):
+        model = self.args.model
+        if self.args.model != "large" and not self.args.non_english:
+            model = self.args.model + ".en"
+        whisper_model = whisper.load_model(model)
+
+        self.model = whisper_model
+        return whisper_model
+
+    def create_text(self):
+        req_text = f"{self.series} - Part {self.part}.\n{self.text}\n{self.outro}"
+        series = self.series.replace(' ', '_')
+        filename = f"{self.path}{os.sep}{series}{os.sep}{series}_{self.part}.mp3"
+
+        Path(f"{self.path}{os.sep}{series}").mkdir(parents=True, exist_ok=True)
+
+        self.req_text = req_text
+        self.mp3_file = filename
+        return req_text, filename
+
+    async def text_to_speech(self):
+        await tts(self.req_text, outfile=self.mp3_file, voice=self.args.tts, args=self.args)
+
+    def generate_transcription(self):
+        ass_filename = srt_create(self.model,
+                                  self.path, self.series, self.part, self.text, self.mp3_file, **vars(self.args))
+        ass_filename = Path(ass_filename).absolute()
+
+        self.ass_file = ass_filename
+        return ass_filename
+
+    def select_background(self):
+        try:
+            # Background video selected with WebUI
+            background_mp4 = self.args.mp4_background
+
+            with KeepDir() as keep_dir:
+                keep_dir.chdir("background")
+                background_mp4 = Path(background_mp4).absolute()
+        except AttributeError:
+            # CLI execution
+            background_mp4 = random_background()
+
+        background_mp4 = str(Path(background_mp4).absolute())
+
+        self.mp4_background = background_mp4
+        return background_mp4
+
+    def integrate_subtitles(self):
+        final_video = prepare_background(
+            self.mp4_background, filename_mp3=self.mp3_file, filename_srt=self.ass_file, verbose=self.args.verbose)
+        final_video = Path(final_video).absolute()
+
+        self.mp4_final_video = final_video
+        return final_video
+
+    def upload_to_tiktok(self):
+        uploaded = upload_tiktok(str(
+            self.mp4_final_video), title=f"{self.series} - {self.part}", tags=self.tags, headless=not self.args.verbose)
+        return uploaded

src/video_downloader.py

@@ -0,0 +1,25 @@
+import subprocess
+from pathlib import Path
+
+import msg
+from utils import KeepDir
+
+HOME = Path.cwd()
+
+
+def download_video(url: str, folder: str = 'background'):
+    """
+    Downloads a video from the given URL and saves it to the specified folder.
+
+    Args:
+        url (str): The URL of the video to download.
+        folder (str, optional): The name of the folder to save the video in. Defaults to 'background'.
+    """
+    directory = HOME / folder
+    if not directory.exists():
+        directory.mkdir()
+
+    with KeepDir() as keep_dir:
+        keep_dir.chdir(folder)
+        subprocess.run(['yt-dlp', '-f bestvideo[ext=mp4]',
+                       '--restrict-filenames', url], check=True)

src/video_prepare.py

@@ -0,0 +1,62 @@
+import multiprocessing
+import os
+import subprocess
+import random
+
+from utils import *
+
+HOME = Path.cwd()
+
+
+def prepare_background(background_mp4: str, filename_mp3: str, filename_srt: str, verbose: bool = False) -> str:
+    video_info = get_info(background_mp4, kind='video')
+    video_duration = int(round(video_info.get('duration'), 0))
+
+    audio_info = get_info(filename_mp3, kind='audio')
+    audio_duration = int(round(audio_info.get('duration'), 0))
+
+    ss = random.randint(0, (video_duration-audio_duration))
+    audio_duration = convert_time(audio_duration)
+    if ss < 0:
+        ss = 0
+
+    srt_raw = filename_srt
+    srt_filename = filename_srt.name
+    srt_path = filename_srt.parent.absolute()
+
+    directory = HOME / 'output'
+    if not directory.exists():
+        directory.mkdir()
+
+    outfile = f"{HOME}{os.sep}output{os.sep}output_{ss}.mp4"
+
+    if verbose:
+        rich_print(
+            f"{filename_srt = }\n{background_mp4 = }\n{filename_mp3 = }\n", style='bold green')
+
+    args = [
+        "ffmpeg",
+        "-ss", str(ss),
+        "-t", str(audio_duration),
+        "-i", background_mp4,
+        "-i", filename_mp3,
+        "-map", "0:v",
+        "-map", "1:a",
+        "-vf", f"crop=ih/16*9:ih, scale=w=1080:h=1920:flags=lanczos, gblur=sigma=2, ass='{srt_raw.absolute()}'",
+        "-c:v", "libx264",
+        "-crf", "23",
+        "-c:a", "aac",
+        "-ac", "2",
+        "-b:a", "192K",
+        f"{outfile}",
+        "-y",
+        "-threads", f"{multiprocessing.cpu_count()}"]
+
+    if verbose:
+        rich_print('[i] FFMPEG Command:\n'+' '.join(args)+'\n', style='yellow')
+
+    with KeepDir() as keep_dir:
+        keep_dir.chdir(srt_path)
+        subprocess.run(args, check=True)
+
+    return outfile

src/voice_manager.py

@@ -0,0 +1,17 @@
+import sys
+
+import edge_tts
+
+
+class VoicesManager:
+    @staticmethod
+    async def create():
+        return await edge_tts.VoicesManager.create()
+
+    @staticmethod
+    def find(voices, Gender, Locale):
+        voices = voices.find(Gender=Gender, Locale=Locale)
+        if len(voices) == 0:
+            print(f"Specified TTS language not found. Make sure you are using the correct format. For example: en-US")
+            sys.exit(1)
+        return voices['Name']