diff --git a/vnrecode/application.py b/vnrecode/application.py index 912c526..239bb05 100755 --- a/vnrecode/application.py +++ b/vnrecode/application.py @@ -13,41 +13,37 @@ from .utils import Utils class Application: def __init__(self, params: Params, compress: Compress, printer: Printer, utils: Utils): - self.params = params - self.compress = compress.compress - self.printer = printer - self.utils = utils - - def compress_worker(self, folder: str, file: str, source: str, output: str): - if os.path.isfile(os.path.join(folder, file)): - self.compress(folder, file, source, output) + self.__params = params + self.__compress = compress.compress + self.__printer = printer + self.__utils = utils def run(self): start_time = datetime.now() - self.printer.win_ascii_esc() + self.__printer.win_ascii_esc() - source = os.path.abspath(self.params.source) + source = self.__params.source - if os.path.exists(f"{source}_compressed"): - shutil.rmtree(f"{source}_compressed") + if os.path.exists(self.__params.dest): + shutil.rmtree(self.__params.dest) - self.printer.info("Creating folders...") + self.__printer.info("Creating folders...") for folder, folders, files in os.walk(source): - if not os.path.exists(folder.replace(source, f"{source}_compressed")): - os.mkdir(folder.replace(source, f"{source}_compressed")) + if not os.path.exists(folder.replace(source, self.__params.dest)): + os.mkdir(folder.replace(source, self.__params.dest)) - self.printer.info(f'Compressing "{folder.replace(source, os.path.split(source)[-1])}" folder...') - output = folder.replace(source, f"{source}_compressed") + self.__printer.info(f'Compressing "{folder.replace(source, os.path.split(source)[-1])}" folder...') + output = folder.replace(source, self.__params.dest) - with ThreadPoolExecutor(max_workers=self.params.workers) as executor: + with ThreadPoolExecutor(max_workers=self.__params.workers) as executor: futures = [ - executor.submit(self.compress, folder, file, source, output) + executor.submit(self.__compress, folder, file, output) for file in files if os.path.isfile(os.path.join(folder, file)) ] for future in as_completed(futures): future.result() - self.utils.print_duplicates() - self.utils.get_compression_status(source) - self.utils.sys_pause() + self.__utils.print_duplicates() + self.__utils.get_compression_status() + self.__utils.sys_pause() print(f"Time taken: {datetime.now() - start_time}") \ No newline at end of file diff --git a/vnrecode/compress.py b/vnrecode/compress.py index 18fb20e..c484243 100644 --- a/vnrecode/compress.py +++ b/vnrecode/compress.py @@ -44,13 +44,13 @@ class File: class Compress: def __init__(self, params: Params, printer: Printer, utils: Utils): - self.params = params - self.printer = printer - self.utils = utils + self.__params = params + self.__printer = printer + self.__utils = utils def audio(self, in_dir: str, file: str, out_dir: str, extension: str) -> str: - bit_rate = self.params.audio_bitrate - out_file = self.utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}') + bit_rate = self.__params.audio_bitrate + out_file = self.__utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}') try: (FFmpeg() .input(path.join(in_dir, file)) @@ -59,18 +59,18 @@ class Compress: .execute() ) except FFmpegError as e: - self.utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file)) - self.utils.errors += 1 - if not self.params.hide_errors: - self.printer.error(f"File {file} can't be processed! Error: {e}") - self.printer.files(file, path.splitext(file)[0], extension, f"{bit_rate}") + self.__utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file)) + self.__utils.errors += 1 + if not self.__params.hide_errors: + self.__printer.error(f"File {file} can't be processed! Error: {e}") + self.__printer.files(file, path.splitext(file)[0], extension, f"{bit_rate}") return out_file def video(self, in_dir: str, file: str, out_dir: str, extension: str) -> str: - if not self.params.video_skip: - out_file = self.utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}') - codec = self.params.video_codec - crf = self.params.video_crf + if not self.__params.video_skip: + out_file = self.__utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}') + codec = self.__params.video_codec + crf = self.__params.video_crf try: (FFmpeg() @@ -80,33 +80,33 @@ class Compress: .output(out_file,{"codec:v": codec, "v:b": 0, "loglevel": "error"}, crf=crf) .execute() ) - self.printer.files(file, path.splitext(file)[0], extension, codec) + self.__printer.files(file, path.splitext(file)[0], extension, codec) except FFmpegError as e: - self.utils.add_unprocessed_file(f'{in_dir}/{file}', f'{out_dir}/{file}') - self.utils.errors += 1 - if not self.params.hide_errors: - self.printer.error(f"File {file} can't be processed! Error: {e}") + self.__utils.add_unprocessed_file(f'{in_dir}/{file}', f'{out_dir}/{file}') + self.__utils.errors += 1 + if not self.__params.hide_errors: + self.__printer.error(f"File {file} can't be processed! Error: {e}") return out_file else: - self.utils.add_unprocessed_file(f'{in_dir}/{file}', f'{out_dir}/{file}') + self.__utils.add_unprocessed_file(f'{in_dir}/{file}', f'{out_dir}/{file}') return f'{out_dir}/{path.splitext(file)[0]}.{extension}' def image(self, in_dir: str, file: str, out_dir: str, extension: str) -> str: - quality = self.params.image_quality - out_file = self.utils.check_duplicates(in_dir, out_dir, f"{path.splitext(file)[0]}.{extension}") + quality = self.__params.image_quality + out_file = self.__utils.check_duplicates(in_dir, out_dir, f"{path.splitext(file)[0]}.{extension}") try: image = Image.open(path.join(in_dir, file)) if (extension == "jpg" or extension == "jpeg" or - (extension == "webp" and not self.params.webp_rgba)): + (extension == "webp" and not self.__params.webp_rgba)): if File.has_transparency(image): - self.printer.warning(f"{file} has transparency. Changing to fallback...") - extension = self.params.image_fall_ext + self.__printer.warning(f"{file} has transparency. Changing to fallback...") + extension = self.__params.image_fall_ext if File.has_transparency(image): image.convert('RGBA') - res_downscale = self.params.image_downscale + res_downscale = self.__params.image_downscale if res_downscale != 1: width, height = image.size new_size = (int(width / res_downscale), int(height / res_downscale)) @@ -114,21 +114,21 @@ class Compress: image.save(out_file, optimize=True, - lossless=self.params.image_lossless, + lossless=self.__params.image_lossless, quality=quality, minimize_size=True) - self.printer.files(file, path.splitext(file)[0], extension, f"{quality}%") + self.__printer.files(file, path.splitext(file)[0], extension, f"{quality}%") except Exception as e: - self.utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file)) - self.utils.errors += 1 - if not self.params.hide_errors: - self.printer.error(f"File {file} can't be processed! Error: {e}") + self.__utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file)) + self.__utils.errors += 1 + if not self.__params.hide_errors: + self.__printer.error(f"File {file} can't be processed! Error: {e}") return out_file def unknown(self, in_dir: str, filename: str, out_dir: str) -> str: - if self.params.force_compress: - self.printer.unknown_file(filename) - out_file = self.utils.check_duplicates(in_dir, out_dir, filename) + if self.__params.force_compress: + self.__printer.unknown_file(filename) + out_file = self.__utils.check_duplicates(in_dir, out_dir, filename) try: (FFmpeg() .input(path.join(in_dir, filename)) @@ -136,28 +136,28 @@ class Compress: .execute() ) except FFmpegError as e: - self.utils.add_unprocessed_file(path.join(in_dir, filename), path.join(out_dir, filename)) - self.utils.errors += 1 - if not self.params.hide_errors: - self.printer.error(f"File {filename} can't be processed! Error: {e}") + self.__utils.add_unprocessed_file(path.join(in_dir, filename), path.join(out_dir, filename)) + self.__utils.errors += 1 + if not self.__params.hide_errors: + self.__printer.error(f"File {filename} can't be processed! Error: {e}") return out_file else: - self.utils.add_unprocessed_file(path.join(in_dir, filename), path.join(out_dir, filename)) + self.__utils.add_unprocessed_file(path.join(in_dir, filename), path.join(out_dir, filename)) return path.join(out_dir, filename) - def compress(self, _dir: str, filename: str, source: str, output: str): + def compress(self, dir_: str, filename: str, output: str): match File.get_type(filename): case "audio": - out_file = self.audio(_dir, filename, output, self.params.audio_ext) + out_file = self.audio(dir_, filename, output, self.__params.audio_ext) case "image": - out_file = self.image(_dir, filename, output, self.params.image_ext) + out_file = self.image(dir_, filename, output, self.__params.image_ext) case "video": - out_file = self.video(_dir, filename, output, self.params.video_ext) + out_file = self.video(dir_, filename, output, self.__params.video_ext) case "unknown": - out_file = self.unknown(_dir, filename, output) + out_file = self.unknown(dir_, filename, output) - if self.params.mimic_mode: - self.utils.mimic_rename(out_file, path.join(_dir, filename), source) + if self.__params.mimic_mode: + self.__utils.mimic_rename(out_file, path.join(dir_, filename)) - self.printer.bar.update() - self.printer.bar.next() + self.__printer.bar.update() + self.__printer.bar.next() diff --git a/vnrecode/params.py b/vnrecode/params.py index ea15691..ad4ec59 100644 --- a/vnrecode/params.py +++ b/vnrecode/params.py @@ -29,6 +29,7 @@ class Params: video_codec: str source: str + dest: str @classmethod def setup(cls) -> Self: @@ -59,12 +60,13 @@ class Params: video_ext = config["VIDEO"]["Extension"] if args.config else args.v_ext video_codec = config["VIDEO"]["Codec"] if args.config else args.v_codec source = args.source + dest = f"{source}_compressed" return cls( copy_unprocessed, force_compress, mimic_mode, hide_errors, webp_rgba, workers, audio_ext, audio_bitrate, image_downscale, image_ext, image_fall_ext, image_lossless, image_quality, - video_crf, video_skip, video_ext, video_codec, source + video_crf, video_skip, video_ext, video_codec, source, dest ) @staticmethod diff --git a/vnrecode/utils.py b/vnrecode/utils.py index 3410eae..a75e4c3 100644 --- a/vnrecode/utils.py +++ b/vnrecode/utils.py @@ -1,19 +1,16 @@ from shutil import copyfile -from glob import glob import sys import os import re -import fnmatch - class Utils: - def __init__(self, params, printer): - self.errors = 0 - self.params = params - self.printer = printer - self.duplicates = [] + def __init__(self, params_inst, printer_inst): + self.__errors = 0 + self.__params = params_inst + self.__printer = printer_inst + self.__duplicates = [] @staticmethod def sys_pause(): @@ -29,66 +26,63 @@ class Utils: total_size += os.path.getsize(os.path.join(folder, file)) return total_size - def get_compression(self, source: str, output: str): + def get_compression_status(self): + source_len = 0 + output_len = 0 + + for folder, folders, files in os.walk(self.__params.source): + source_len += len(files) + + for folder, folders, files in os.walk(self.__params.dest): + for file in files: + if not os.path.splitext(file)[1].count("(vncopy)"): + output_len += 1 + + if self.__errors != 0: + self.__printer.warning("Some files failed to compress!") + + if source_len == output_len: + self.__printer.info("Success!") + else: + self.__printer.warning("Original and compressed folders are not identical!") try: - source = self.get_size(source) - output = self.get_size(output) + source = self.get_size(self.__params.source) + output = self.get_size(self.__params.dest) print(f"\nResult: {source/1024/1024:.2f}MB -> " f"{output/1024/1024:.2f}MB ({(output - source)/1024/1024:.2f}MB)") except ZeroDivisionError: - self.printer.warning("Nothing compressed!") - - def get_compression_status(self, source: str): - source_len = 0 - output_len = 0 - - for folder, folders, files in os.walk(source): - source_len += len(files) - - for folder, folders, files in os.walk(f'{source}_compressed'): - for file in files: - if not os.path.splitext(file)[1].count("(copy)"): - output_len += 1 - - if self.errors != 0: - self.printer.warning("Some files failed to compress!") - - if source_len == output_len: - self.printer.info("Success!") - else: - self.printer.warning("Original and compressed folders are not identical!") - self.get_compression(source, f"{source}_compressed") + self.__printer.warning("Nothing compressed!") def add_unprocessed_file(self, source: str, output: str): - if self.params.copy_unprocessed: + if self.__params.copy_unprocessed: filename = os.path.split(source)[-1] copyfile(source, output) - self.printer.info(f"File {filename} copied to compressed folder.") + self.__printer.info(f"File {filename} copied to compressed folder.") def check_duplicates(self, source: str, output: str, filename: str) -> str: re_pattern = re.compile(os.path.splitext(filename)[0]+r".[a-zA-Z0-9]+$", re.IGNORECASE) duplicates = [name for name in os.listdir(source) if re_pattern.match(name)] if len(duplicates) > 1: - if filename.lower() not in (duplicate.lower() for duplicate in self.duplicates): - self.duplicates.append(filename) + if filename.lower() not in (duplicate.lower() for duplicate in self.__duplicates): + self.__duplicates.append(filename) new_name = os.path.splitext(filename)[0] + "(vncopy)" + os.path.splitext(filename)[1] return os.path.join(output, new_name) return os.path.join(output, filename) def print_duplicates(self): - for filename in self.duplicates: - self.printer.warning( + for filename in self.__duplicates: + self.__printer.warning( f'Duplicate file has been found! Check manually this files - "{filename}", ' f'"{os.path.splitext(filename)[0] + "(vncopy)" + os.path.splitext(filename)[1]}"' ) - def mimic_rename(self, filename: str, target: str, source: str): + def mimic_rename(self, filename: str, target: str): if filename.count("(vncopy)"): orig_name = filename.replace("(vncopy)", "") - index = self.duplicates.index(os.path.split(orig_name)[-1]) - self.duplicates[index] = os.path.split(target)[-1] + index = self.__duplicates.index(os.path.split(orig_name)[-1]) + self.__duplicates[index] = os.path.split(target)[-1] target = os.path.splitext(target)[0] + "(vncopy)" + os.path.splitext(target)[1] - os.rename(filename, target.replace(source, f"{source}_compressed")) \ No newline at end of file + os.rename(filename, target.replace(self.__params.source, self.__params.dest)) \ No newline at end of file