VienDesuPorting/VNTools
4
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

vnrecode: make it private!

Browse source
This commit is contained in:
OleSTEEP 2024-10-18 20:50:40 +03:00
parent d8e55bac9a
commit 9bb3cdcccb
4 changed files with 107 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

40
vnrecode/application.py
View file

@ -13,41 +13,37 @@ from .utils import Utils
class Application: class Application:
def __init__(self, params: Params, compress: Compress, printer: Printer, utils: Utils): def __init__(self, params: Params, compress: Compress, printer: Printer, utils: Utils):
self.params = params self.__params = params
self.compress = compress.compress self.__compress = compress.compress
self.printer = printer self.__printer = printer
self.utils = utils 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)
def run(self): def run(self):
start_time = datetime.now() 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"): if os.path.exists(self.__params.dest):
shutil.rmtree(f"{source}_compressed") shutil.rmtree(self.__params.dest)
self.printer.info("Creating folders...") self.__printer.info("Creating folders...")
for folder, folders, files in os.walk(source): for folder, folders, files in os.walk(source):
if not os.path.exists(folder.replace(source, f"{source}_compressed")): if not os.path.exists(folder.replace(source, self.__params.dest)):
os.mkdir(folder.replace(source, f"{source}_compressed")) os.mkdir(folder.replace(source, self.__params.dest))
self.printer.info(f'Compressing "{folder.replace(source, os.path.split(source)[-1])}" folder...') self.__printer.info(f'Compressing "{folder.replace(source, os.path.split(source)[-1])}" folder...')
output = folder.replace(source, f"{source}_compressed") 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 = [ 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 file in files if os.path.isfile(os.path.join(folder, file))
] ]
for future in as_completed(futures): for future in as_completed(futures):
future.result() future.result()
self.utils.print_duplicates() self.__utils.print_duplicates()
self.utils.get_compression_status(source) self.__utils.get_compression_status()
self.utils.sys_pause() self.__utils.sys_pause()
print(f"Time taken: {datetime.now() - start_time}") print(f"Time taken: {datetime.now() - start_time}")

98
vnrecode/compress.py
View file

@ -44,13 +44,13 @@ class File:
class Compress: class Compress:
def __init__(self, params: Params, printer: Printer, utils: Utils): def __init__(self, params: Params, printer: Printer, utils: Utils):
self.params = params self.__params = params
self.printer = printer self.__printer = printer
self.utils = utils self.__utils = utils
def audio(self, in_dir: str, file: str, out_dir: str, extension: str) -> str: def audio(self, in_dir: str, file: str, out_dir: str, extension: str) -> str:
bit_rate = self.params.audio_bitrate bit_rate = self.__params.audio_bitrate
out_file = self.utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}') out_file = self.__utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}')
try: try:
(FFmpeg() (FFmpeg()
.input(path.join(in_dir, file)) .input(path.join(in_dir, file))
@ -59,18 +59,18 @@ class Compress:
.execute() .execute()
) )
except FFmpegError as e: except FFmpegError as e:
self.utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file)) self.__utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file))
self.utils.errors += 1 self.__utils.errors += 1
if not self.params.hide_errors: if not self.__params.hide_errors:
self.printer.error(f"File {file} can't be processed! Error: {e}") 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.__printer.files(file, path.splitext(file)[0], extension, f"{bit_rate}")
return out_file return out_file
def video(self, in_dir: str, file: str, out_dir: str, extension: str) -> str: def video(self, in_dir: str, file: str, out_dir: str, extension: str) -> str:
if not self.params.video_skip: if not self.__params.video_skip:
out_file = self.utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}') out_file = self.__utils.check_duplicates(in_dir, out_dir, f'{path.splitext(file)[0]}.{extension}')
codec = self.params.video_codec codec = self.__params.video_codec
crf = self.params.video_crf crf = self.__params.video_crf
try: try:
(FFmpeg() (FFmpeg()
@ -80,33 +80,33 @@ class Compress:
.output(out_file,{"codec:v": codec, "v:b": 0, "loglevel": "error"}, crf=crf) .output(out_file,{"codec:v": codec, "v:b": 0, "loglevel": "error"}, crf=crf)
.execute() .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: except FFmpegError as e:
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}')
self.utils.errors += 1 self.__utils.errors += 1
if not self.params.hide_errors: if not self.__params.hide_errors:
self.printer.error(f"File {file} can't be processed! Error: {e}") self.__printer.error(f"File {file} can't be processed! Error: {e}")
return out_file return out_file
else: 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}' return f'{out_dir}/{path.splitext(file)[0]}.{extension}'
def image(self, in_dir: str, file: str, out_dir: str, extension: str) -> str: def image(self, in_dir: str, file: str, out_dir: str, extension: str) -> str:
quality = self.params.image_quality quality = self.__params.image_quality
out_file = self.utils.check_duplicates(in_dir, out_dir, f"{path.splitext(file)[0]}.{extension}") out_file = self.__utils.check_duplicates(in_dir, out_dir, f"{path.splitext(file)[0]}.{extension}")
try: try:
image = Image.open(path.join(in_dir, file)) image = Image.open(path.join(in_dir, file))
if (extension == "jpg" or extension == "jpeg" or 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): if File.has_transparency(image):
self.printer.warning(f"{file} has transparency. Changing to fallback...") self.__printer.warning(f"{file} has transparency. Changing to fallback...")
extension = self.params.image_fall_ext extension = self.__params.image_fall_ext
if File.has_transparency(image): if File.has_transparency(image):
image.convert('RGBA') image.convert('RGBA')
res_downscale = self.params.image_downscale res_downscale = self.__params.image_downscale
if res_downscale != 1: if res_downscale != 1:
width, height = image.size width, height = image.size
new_size = (int(width / res_downscale), int(height / res_downscale)) new_size = (int(width / res_downscale), int(height / res_downscale))
@ -114,21 +114,21 @@ class Compress:
image.save(out_file, image.save(out_file,
optimize=True, optimize=True,
lossless=self.params.image_lossless, lossless=self.__params.image_lossless,
quality=quality, quality=quality,
minimize_size=True) 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: except Exception as e:
self.utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file)) self.__utils.add_unprocessed_file(path.join(in_dir, file), path.join(out_dir, file))
self.utils.errors += 1 self.__utils.errors += 1
if not self.params.hide_errors: if not self.__params.hide_errors:
self.printer.error(f"File {file} can't be processed! Error: {e}") self.__printer.error(f"File {file} can't be processed! Error: {e}")
return out_file return out_file
def unknown(self, in_dir: str, filename: str, out_dir: str) -> str: def unknown(self, in_dir: str, filename: str, out_dir: str) -> str:
if self.params.force_compress: if self.__params.force_compress:
self.printer.unknown_file(filename) self.__printer.unknown_file(filename)
out_file = self.utils.check_duplicates(in_dir, out_dir, filename) out_file = self.__utils.check_duplicates(in_dir, out_dir, filename)
try: try:
(FFmpeg() (FFmpeg()
.input(path.join(in_dir, filename)) .input(path.join(in_dir, filename))
@ -136,28 +136,28 @@ class Compress:
.execute() .execute()
) )
except FFmpegError as e: except FFmpegError as e:
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))
self.utils.errors += 1 self.__utils.errors += 1
if not self.params.hide_errors: if not self.__params.hide_errors:
self.printer.error(f"File {filename} can't be processed! Error: {e}") self.__printer.error(f"File {filename} can't be processed! Error: {e}")
return out_file return out_file
else: 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) 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): match File.get_type(filename):
case "audio": 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": 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": 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": case "unknown":
out_file = self.unknown(_dir, filename, output) out_file = self.unknown(dir_, filename, output)
if self.params.mimic_mode: if self.__params.mimic_mode:
self.utils.mimic_rename(out_file, path.join(_dir, filename), source) self.__utils.mimic_rename(out_file, path.join(dir_, filename))
self.printer.bar.update() self.__printer.bar.update()
self.printer.bar.next() self.__printer.bar.next()

4
vnrecode/params.py
View file

@ -29,6 +29,7 @@ class Params:
video_codec: str video_codec: str
source: str source: str
dest: str
@classmethod @classmethod
def setup(cls) -> Self: def setup(cls) -> Self:
@ -59,12 +60,13 @@ class Params:
video_ext = config["VIDEO"]["Extension"] if args.config else args.v_ext video_ext = config["VIDEO"]["Extension"] if args.config else args.v_ext
video_codec = config["VIDEO"]["Codec"] if args.config else args.v_codec video_codec = config["VIDEO"]["Codec"] if args.config else args.v_codec
source = args.source source = args.source
dest = f"{source}_compressed"
return cls( return cls(
copy_unprocessed, force_compress, mimic_mode, hide_errors, webp_rgba, workers, copy_unprocessed, force_compress, mimic_mode, hide_errors, webp_rgba, workers,
audio_ext, audio_bitrate, audio_ext, audio_bitrate,
image_downscale, image_ext, image_fall_ext, image_lossless, image_quality, 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 @staticmethod

80
vnrecode/utils.py
View file

@ -1,19 +1,16 @@
from shutil import copyfile from shutil import copyfile
from glob import glob
import sys import sys
import os import os
import re import re
import fnmatch
class Utils: class Utils:
def __init__(self, params, printer): def __init__(self, params_inst, printer_inst):
self.errors = 0 self.__errors = 0
self.params = params self.__params = params_inst
self.printer = printer self.__printer = printer_inst
self.duplicates = [] self.__duplicates = []
@staticmethod @staticmethod
def sys_pause(): def sys_pause():
@ -29,66 +26,63 @@ class Utils:
total_size += os.path.getsize(os.path.join(folder, file)) total_size += os.path.getsize(os.path.join(folder, file))
return total_size 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: try:
source = self.get_size(source) source = self.get_size(self.__params.source)
output = self.get_size(output) output = self.get_size(self.__params.dest)
print(f"\nResult: {source/1024/1024:.2f}MB -> " print(f"\nResult: {source/1024/1024:.2f}MB -> "
f"{output/1024/1024:.2f}MB ({(output - source)/1024/1024:.2f}MB)") f"{output/1024/1024:.2f}MB ({(output - source)/1024/1024:.2f}MB)")
except ZeroDivisionError: except ZeroDivisionError:
self.printer.warning("Nothing compressed!") 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")
def add_unprocessed_file(self, source: str, output: str): 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] filename = os.path.split(source)[-1]
copyfile(source, output) 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: 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) 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)] duplicates = [name for name in os.listdir(source) if re_pattern.match(name)]
if len(duplicates) > 1: if len(duplicates) > 1:
if filename.lower() not in (duplicate.lower() for duplicate in self.duplicates): if filename.lower() not in (duplicate.lower() for duplicate in self.__duplicates):
self.duplicates.append(filename) self.__duplicates.append(filename)
new_name = os.path.splitext(filename)[0] + "(vncopy)" + os.path.splitext(filename)[1] 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, new_name)
return os.path.join(output, filename) return os.path.join(output, filename)
def print_duplicates(self): def print_duplicates(self):
for filename in self.duplicates: for filename in self.__duplicates:
self.printer.warning( self.__printer.warning(
f'Duplicate file has been found! Check manually this files - "{filename}", ' f'Duplicate file has been found! Check manually this files - "{filename}", '
f'"{os.path.splitext(filename)[0] + "(vncopy)" + os.path.splitext(filename)[1]}"' 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)"): if filename.count("(vncopy)"):
orig_name = filename.replace("(vncopy)", "") orig_name = filename.replace("(vncopy)", "")
index = self.duplicates.index(os.path.split(orig_name)[-1]) index = self.__duplicates.index(os.path.split(orig_name)[-1])
self.duplicates[index] = os.path.split(target)[-1] self.__duplicates[index] = os.path.split(target)[-1]
target = os.path.splitext(target)[0] + "(vncopy)" + os.path.splitext(target)[1] target = os.path.splitext(target)[0] + "(vncopy)" + os.path.splitext(target)[1]
os.rename(filename, target.replace(source, f"{source}_compressed")) os.rename(filename, target.replace(self.__params.source, self.__params.dest))