Add stable-ts to requirements.txt and update video.json structure #16
Conversation
Split main.py into multiple files
WalkthroughWalkthroughThe project underwent a significant refactoring for better modularity and maintainability. Key improvements include the introduction of asynchronous file operations, modular argument parsing, enhanced video creation and processing, and direct TikTok uploading capabilities. The addition of new dependencies supports these features, streamlining the video production pipeline from content generation to publication. Changes
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? TipsChat with CodeRabbit Bot (
|
MatteoFasulo
deleted the
15-webui-gradio-issue-all-the-project-is-on-one-file-mainpy
branch
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review Status
Actionable comments generated: 13
Configuration used: CodeRabbit UI
Files ignored due to filter (1)
- video.json
Files selected for processing (12)
- main.py (4 hunks)
- requirements.txt (1 hunks)
- src/arg_parser.py (1 hunks)
- src/logger.py (1 hunks)
- src/subtitle_creator.py (1 hunks)
- src/text_to_speech.py (1 hunks)
- src/tiktok.py (1 hunks)
- src/video_creator.py (1 hunks)
- src/video_downloader.py (1 hunks)
- src/video_prepare.py (1 hunks)
- src/voice_manager.py (1 hunks)
- utils.py (1 hunks)
Files skipped from review due to trivial changes (1)
- requirements.txt
Additional comments: 1
utils.py (1)
- 101-115: The
convert_timefunction converts time in seconds to a formatted string. The function is correct, but it's not clear if it's used in the context of the application. Ensure that this utility function is necessary and used within the codebase.
| 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 |
There was a problem hiding this comment.
The tts function correctly uses the edge_tts package to perform text-to-speech operations. The function signature is clear, and the use of type hints improves readability. The function always returns True, which might not be informative if the operation fails. Consider returning a more descriptive result or raising an exception on failure.
| 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'] |
There was a problem hiding this comment.
The find method in the VoicesManager class exits the program if no voices are found. This could be problematic for a library function, as it takes control away from the calling code. Consider raising an exception instead of exiting directly to allow the caller to handle the situation appropriately.
| 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]+bestaudio[ext=m4a]', | ||
| '--restrict-filenames', url], check=True) | ||
| print(f"{msg.OK}Background video downloaded successfully") |
There was a problem hiding this comment.
The download_video function uses subprocess.run to invoke yt-dlp, which is a blocking call. Since the rest of the codebase is moving towards asynchronous operations, consider using an asynchronous library or running this in a separate thread or process to avoid blocking the main event loop.
| 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 |
There was a problem hiding this comment.
The upload_tiktok function checks for the existence of a 'cookies.txt' file but does not handle the case where the file is not found, other than logging an error. It would be more robust to raise an exception or return an error value to allow the calling code to handle this situation.
| 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 |
There was a problem hiding this comment.
The setup_logger function sets up logging correctly. However, it changes the current working directory, which could have unintended side effects on other parts of the application that rely on the current directory. Consider using absolute paths for file handling instead of changing the working directory.
| def prepare_background(background_mp4: str, filename_mp3: str, filename_srt: str, verbose: bool = False) -> str: | ||
| video_duration = int(round(get_info(background_mp4).get('duration'), 0)) | ||
| audio_info = get_info(filename_mp3) | ||
| audio_duration = int(round(audio_info.get('duration'), 0)) | ||
| ss = random.randint(0, (video_duration-audio_duration)) | ||
| audio_duration = convert_time(audio_info.get('duration')) | ||
| if ss < 0: | ||
| ss = 0 | ||
|
|
||
| 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", | ||
| "-filter:v", | ||
| f"crop=ih/16*9:ih, scale=w=1080:h=1920:flags=bicubic, gblur=sigma=2, subtitles={srt_filename}:force_style=',Alignment=8,BorderStyle=7,Outline=3,Shadow=5,Blur=15,Fontsize=15,MarginL=45,MarginR=55,FontName=Lexend Bold'", | ||
| "-c:v", "libx264", "-preset", "5", | ||
| "-b:v", "5M", | ||
| "-c:a", "aac", "-ac", "1", | ||
| "-b:a", "96K", | ||
| f"{outfile}", "-y", | ||
| "-threads", f"{multiprocessing.cpu_count()//2}"] | ||
|
|
||
| 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 |
There was a problem hiding this comment.
The prepare_background function uses ffmpeg to process videos. It uses a blocking subprocess.run call within an asynchronous codebase. Consider using an asynchronous approach to running ffmpeg to avoid blocking the event loop.
| 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("--font_color", help="Subtitle font color in hex format: #FFF000", | ||
| default="#FFF000", type=str) | ||
| 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 | ||
|
|
||
| return args |
There was a problem hiding this comment.
The parse_args function is marked as asynchronous but does not perform any I/O operations that would benefit from being asynchronous. Consider removing the async keyword to simplify the code unless there is a specific reason for this function to be asynchronous.
| class VideoCreator: | ||
| def __init__(self, args): | ||
| self.args = args | ||
|
|
||
| self.series = jsonData.get('series', '') | ||
| self.part = jsonData.get('part', '') | ||
| self.text = jsonData.get('text', '') | ||
| self.tags = jsonData.get('tags', list()) | ||
| self.outro = jsonData.get('outro', '') | ||
| self.path = Path(media_folder).absolute() | ||
|
|
||
| def download_video(self, folder: str = '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]+bestaudio[ext=m4a]', | ||
| '--restrict-filenames', self.args.url], check=True) | ||
| 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} - {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): | ||
| srt_filename = srt_create( | ||
| self.model, self.path, self.series, self.part, self.text, self.mp3_file) | ||
| srt_filename = Path(srt_filename).absolute() | ||
|
|
||
| self.srt_file = srt_filename | ||
|
|
||
| highlight_words(self.srt_file, subtitle_format=self.args.sub_format, | ||
| font_color=self.args.font_color) | ||
| return srt_filename | ||
|
|
||
| def select_background(self): | ||
| background_mp4 = random_background() | ||
|
|
||
| self.mp4_backgroung = background_mp4 | ||
| return background_mp4 | ||
|
|
||
| def integrate_subtitles(self): | ||
| final_video = prepare_background( | ||
| self.mp4_backgroung, filename_mp3=self.mp3_file, filename_srt=self.srt_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 |
There was a problem hiding this comment.
The VideoCreator class methods are well-structured and modular. However, the class relies on global variables and reads from a JSON file upon instantiation. Consider passing the necessary data as parameters to the class constructor to improve testability and maintainability.
| """ | ||
| Returns the filename of a random file in the specified folder. | ||
|
|
||
| Args: | ||
| folder(str): The folder containing the files. | ||
|
|
||
| Returns: | ||
| str: The filename of a randomly selected file in the folder. | ||
| """ | ||
| directory = Path(folder).absolute() | ||
| if not directory.exists(): | ||
| directory.mkdir() | ||
|
|
||
| with KeepDir() as keep_dir: | ||
| keep_dir.chdir(folder) | ||
| files = os.listdir(".") | ||
| random_file = random.choice(files) | ||
| return Path(random_file).absolute() | ||
|
|
There was a problem hiding this comment.
The random_background function selects a random file from a specified directory. It assumes that all files in the directory are valid backgrounds. Consider adding a check to ensure that only appropriate file types (e.g., video files) are selected.
| """ | ||
| Get information about a video file. | ||
|
|
||
| Args: | ||
| filename (str): The path to the video file. | ||
| verbose (bool, optional): Whether to print verbose output. Defaults to False. | ||
|
|
||
| Returns: | ||
| dict: A dictionary containing information about the video file, including width, height, bit rate, and duration. | ||
| """ | ||
| try: | ||
| probe = ffmpeg.probe(filename) | ||
| video_stream = next( | ||
| (stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None) | ||
| audio_stream = next( | ||
| (stream for stream in probe['streams'] if stream['codec_type'] == 'audio'), None) | ||
| try: | ||
| duration = float(audio_stream['duration']) | ||
| except Exception: | ||
| if verbose: | ||
| console.log( | ||
| f"{msg.WARNING}MP4 default metadata not found") | ||
| logger.warning('MP4 default metadata not found') | ||
| duration = (datetime.datetime.strptime( | ||
| audio_stream['DURATION'], '%H:%M:%S.%f') - datetime.datetime.min).total_seconds() | ||
| if video_stream is None: | ||
| if verbose: | ||
| console.log( | ||
| f"{msg.WARNING}No video stream found") | ||
| logger.warning('No video stream found') | ||
| bit_rate = int(audio_stream['bit_rate']) | ||
| return {'bit_rate': bit_rate, 'duration': duration} | ||
|
|
||
| width = int(video_stream['width']) | ||
| height = int(video_stream['height']) | ||
| return {'width': width, 'height': height, 'duration': duration} | ||
|
|
||
| except ffmpeg.Error as e: | ||
| console.log(f"{msg.ERROR}{e.stderr}") | ||
| logger.exception(e.stderr) | ||
| sys.exit(1) |
There was a problem hiding this comment.
The get_info function retrieves video file information using ffmpeg. It exits the program if an error occurs, which is not ideal for a utility function. Instead, consider raising an exception to allow the caller to handle the error.
Summary by CodeRabbit
New Features
Enhancements
arg_parserandvideo_creatormodules.Dependencies
stable-tsandtiktok-uploaderpackages for advanced functionality.Refactor