vnrecode: rewrite duplicates processing
This commit is contained in:
parent
9bb3cdcccb
commit
4e6fd332c5
3 changed files with 46 additions and 54 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -1,5 +1,6 @@
|
|||
/output/
|
||||
/tests/
|
||||
/tests_compressed/
|
||||
/build/
|
||||
/dist/
|
||||
/vntools.egg-info/
|
||||
|
|
|
|||
|
|
@ -50,7 +50,8 @@ class Compress:
|
|||
|
||||
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}')
|
||||
prefix = self.__utils.get_hash(file)
|
||||
out_file = path.join(out_dir, f'{prefix}_{path.splitext(file)[0]}.{extension}')
|
||||
try:
|
||||
(FFmpeg()
|
||||
.input(path.join(in_dir, file))
|
||||
|
|
@ -59,16 +60,14 @@ 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.__utils.catch_unprocessed(path.join(in_dir, file), out_file, 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:
|
||||
prefix = self.__utils.get_hash(file)
|
||||
out_file = path.join(out_dir, f'{prefix}_{path.splitext(file)[0]}.{extension}')
|
||||
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
|
||||
|
||||
|
|
@ -82,18 +81,15 @@ class Compress:
|
|||
)
|
||||
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}")
|
||||
return out_file
|
||||
self.__utils.catch_unprocessed(path.join(in_dir, file), out_file, e)
|
||||
else:
|
||||
self.__utils.add_unprocessed_file(f'{in_dir}/{file}', f'{out_dir}/{file}')
|
||||
return f'{out_dir}/{path.splitext(file)[0]}.{extension}'
|
||||
self.__utils.copy_unprocessed(path.join(in_dir, file), out_file)
|
||||
return out_file
|
||||
|
||||
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}")
|
||||
prefix = self.__utils.get_hash(file)
|
||||
out_file = path.join(out_dir, f"{prefix}_{path.splitext(file)[0]}.{extension}")
|
||||
try:
|
||||
image = Image.open(path.join(in_dir, file))
|
||||
|
||||
|
|
@ -119,31 +115,25 @@ class Compress:
|
|||
minimize_size=True)
|
||||
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.catch_unprocessed(path.join(in_dir, file), out_file, e)
|
||||
return out_file
|
||||
|
||||
def unknown(self, in_dir: str, filename: str, out_dir: str) -> str:
|
||||
def unknown(self, in_dir: str, file: str, out_dir: str) -> str:
|
||||
prefix = self.__utils.get_hash(file)
|
||||
out_file = path.join(out_dir, f"{prefix}_{file}")
|
||||
if self.__params.force_compress:
|
||||
self.__printer.unknown_file(filename)
|
||||
out_file = self.__utils.check_duplicates(in_dir, out_dir, filename)
|
||||
self.__printer.unknown_file(file)
|
||||
try:
|
||||
(FFmpeg()
|
||||
.input(path.join(in_dir, filename))
|
||||
.input(path.join(in_dir, file))
|
||||
.output(out_file)
|
||||
.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}")
|
||||
return out_file
|
||||
self.__utils.catch_unprocessed(path.join(in_dir, file), out_file, e)
|
||||
else:
|
||||
self.__utils.add_unprocessed_file(path.join(in_dir, filename), path.join(out_dir, filename))
|
||||
return path.join(out_dir, filename)
|
||||
self.__utils.copy_unprocessed(path.join(in_dir, file), out_file)
|
||||
return out_file
|
||||
|
||||
def compress(self, dir_: str, filename: str, output: str):
|
||||
match File.get_type(filename):
|
||||
|
|
@ -156,8 +146,6 @@ class Compress:
|
|||
case "unknown":
|
||||
out_file = self.unknown(dir_, filename, output)
|
||||
|
||||
if self.__params.mimic_mode:
|
||||
self.__utils.mimic_rename(out_file, path.join(dir_, filename))
|
||||
|
||||
self.__utils.out_rename(out_file, filename)
|
||||
self.__printer.bar.update()
|
||||
self.__printer.bar.next()
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from shutil import copyfile
|
||||
import hashlib
|
||||
import sys
|
||||
import os
|
||||
import re
|
||||
|
||||
|
||||
class Utils:
|
||||
|
|
@ -26,6 +26,10 @@ class Utils:
|
|||
total_size += os.path.getsize(os.path.join(folder, file))
|
||||
return total_size
|
||||
|
||||
@staticmethod
|
||||
def get_hash(filename: str) -> str:
|
||||
return hashlib.md5(filename.encode()).hexdigest()[:8]
|
||||
|
||||
def get_compression_status(self):
|
||||
source_len = 0
|
||||
output_len = 0
|
||||
|
|
@ -54,22 +58,23 @@ class Utils:
|
|||
except ZeroDivisionError:
|
||||
self.__printer.warning("Nothing compressed!")
|
||||
|
||||
def add_unprocessed_file(self, source: str, output: str):
|
||||
def catch_unprocessed(self, source, output, error):
|
||||
self.copy_unprocessed(source, error)
|
||||
self.__errors += 1
|
||||
if not self.__params.hide_errors:
|
||||
self.__printer.error(f"File {os.path.split(source)[-1]} can't be processed! Error: {error}")
|
||||
|
||||
def copy_unprocessed(self, source, output):
|
||||
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 {os.path.split(source)[-1]} 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)
|
||||
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 catch_duplicates(self, path: str) -> str:
|
||||
if os.path.exists(path):
|
||||
new_path = os.path.splitext(path)[0] + "(vncopy)" + os.path.splitext(path)[1]
|
||||
self.__duplicates.append(new_path)
|
||||
return new_path
|
||||
return path
|
||||
|
||||
def print_duplicates(self):
|
||||
for filename in self.__duplicates:
|
||||
|
|
@ -78,11 +83,9 @@ class Utils:
|
|||
f'"{os.path.splitext(filename)[0] + "(vncopy)" + os.path.splitext(filename)[1]}"'
|
||||
)
|
||||
|
||||
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]
|
||||
target = os.path.splitext(target)[0] + "(vncopy)" + os.path.splitext(target)[1]
|
||||
|
||||
os.rename(filename, target.replace(self.__params.source, self.__params.dest))
|
||||
def out_rename(self, filename: str, target: str):
|
||||
if not self.__params.mimic_mode:
|
||||
dest_name = self.catch_duplicates(os.path.join(os.path.dirname(filename), target))
|
||||
os.rename(filename, dest_name)
|
||||
else:
|
||||
os.rename(filename, os.path.join(os.path.dirname(filename), target))
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue