Compare commits
No commits in common. "626eaae5e2ee78504785c5f399bb6440602fb7bb" and "0a9114ff64dbf9c88dbe4661d422985918027b0c" have entirely different histories.
626eaae5e2
...
0a9114ff64
33 changed files with 507 additions and 905 deletions
5
.gitignore
vendored
5
.gitignore
vendored
|
@ -1,6 +1 @@
|
||||||
/output/
|
/output/
|
||||||
/tests/
|
|
||||||
/tests_compressed/
|
|
||||||
/build/
|
|
||||||
/dist/
|
|
||||||
/vntools.egg-info/
|
|
||||||
|
|
|
@ -1,7 +1,13 @@
|
||||||
## vnrecode
|
## FFMpeg-Compressor
|
||||||
Python utility uses Pillow and ffmpeg to compress Visual Novel Resources
|
Python utility uses ffmpeg to compress Visual Novel Resources
|
||||||
|
|
||||||
### Configuration file
|
### How to use
|
||||||
|
* Download `ffmpeg-comp.toml` and put in next to binary or in to `/etc` folder
|
||||||
|
* Change the configuration of the utility in `ffmpeg-comp.toml` for yourself
|
||||||
|
* `ffmpeg-comp {folder}`
|
||||||
|
* In result, you get `{folder-compressed}` near with original `{folder}`
|
||||||
|
|
||||||
|
### Configuration
|
||||||
#### FFMPEG section
|
#### FFMPEG section
|
||||||
* CopyUnprocessed - Copy all files that failed to compress by ffmpeg to destination folder. In can help to recreate original folder, but with compressed files. (default: `true`)
|
* CopyUnprocessed - Copy all files that failed to compress by ffmpeg to destination folder. In can help to recreate original folder, but with compressed files. (default: `true`)
|
||||||
* ForceCompress - Force try to compress all files in directory via ffmpeg. (default: `false`)
|
* ForceCompress - Force try to compress all files in directory via ffmpeg. (default: `false`)
|
||||||
|
@ -26,34 +32,9 @@ Python utility uses Pillow and ffmpeg to compress Visual Novel Resources
|
||||||
* Extension - Required image file extension. It supports: `.3gp`, `.amv`, `.avi`, `.gif`, `.m2l`, `.m4v`, `.mkv`, `.mov`, `.mp4`, `.m4v`, `.mpeg`, `.mpv`, `.webm`, `.ogv`
|
* Extension - Required image file extension. It supports: `.3gp`, `.amv`, `.avi`, `.gif`, `.m2l`, `.m4v`, `.mkv`, `.mov`, `.mp4`, `.m4v`, `.mpeg`, `.mpv`, `.webm`, `.ogv`
|
||||||
* Codec - (Maybe optional in future) Required video codec. (See official ffmpeg documentation for supported codecs)
|
* Codec - (Maybe optional in future) Required video codec. (See official ffmpeg documentation for supported codecs)
|
||||||
|
|
||||||
### CLI Parameters
|
|
||||||
##### positional arguments:
|
|
||||||
* source - Directory with game files to recode
|
|
||||||
|
|
||||||
##### options:
|
|
||||||
* ` -h, --help ` - show this help message and exit
|
|
||||||
* ` -c CONFIG ` - Utility config file
|
|
||||||
* ` -nu ` - Don't copy unprocessed
|
|
||||||
* ` -f, --force ` - Try to recode unknown files
|
|
||||||
* ` -nm, --no-mimic ` - Disable mimic mode
|
|
||||||
* ` -v, --show_errors ` - Show recode errors
|
|
||||||
* ` --webp-rgb ` - Recode .webp without alpha channel
|
|
||||||
* ` -j JOBS, --jobs JOBS ` - Number of threads
|
|
||||||
* ` -ae A_EXT ` - Audio extension
|
|
||||||
* ` -ab A_BIT ` - Audio bit rate
|
|
||||||
* ` -id I_DOWN ` - Image resolution downscale multiplier
|
|
||||||
* ` -ie I_EXT ` - Image extension
|
|
||||||
* ` -ife I_FALLEXT ` - Image fallback extension
|
|
||||||
* ` -il ` - Image losing compression mode
|
|
||||||
* ` -iq I_QUALITY ` - Image quality
|
|
||||||
* ` --v_crf V_CRF ` - Video CRF number
|
|
||||||
* ` -vs ` - Skip video recoding
|
|
||||||
* ` -ve V_EXT ` - Video extension
|
|
||||||
* ` -vc V_CODEC ` - Video codec name
|
|
||||||
|
|
||||||
### TODO (for testing branch)
|
### TODO (for testing branch)
|
||||||
* [x] Recreate whole game directory with compressed files
|
* [x] Recreate whole game directory with compressed files
|
||||||
* [x] Cross-platform (Easy Windows usage and binaries, macOS binaries)
|
* [x] Cross-platform (Easy Windows usage and binaries, macOS binaries)
|
||||||
* [x] Use ffmpeg python bindings instead of os.system
|
* [x] Use ffmpeg python bindings instead of cli commands
|
||||||
* [x] Multithread
|
* [x] Multithread
|
||||||
* [ ] Reorganize code
|
* [ ] Reorganize code
|
59
FFMpeg-Compressor/main.py
Executable file
59
FFMpeg-Compressor/main.py
Executable file
|
@ -0,0 +1,59 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||||
|
from modules.configloader import config
|
||||||
|
from modules import compressor
|
||||||
|
from modules import printer
|
||||||
|
from modules import utils
|
||||||
|
from datetime import datetime
|
||||||
|
import shutil
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
|
||||||
|
|
||||||
|
def get_args():
|
||||||
|
try:
|
||||||
|
if sys.argv[1][len(sys.argv[1])-1] == "/":
|
||||||
|
path = sys.argv[1][:len(sys.argv[1])-1]
|
||||||
|
else:
|
||||||
|
path = sys.argv[1]
|
||||||
|
return path
|
||||||
|
except IndexError:
|
||||||
|
print(utils.help_message())
|
||||||
|
exit()
|
||||||
|
|
||||||
|
|
||||||
|
def compress_worker(folder, file, target_folder, req_folder):
|
||||||
|
if os.path.isfile(f'{folder}/{file}'):
|
||||||
|
compressor.compress_file(folder, file, target_folder, req_folder)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
start_time = datetime.now()
|
||||||
|
printer.win_ascii_esc()
|
||||||
|
req_folder = os.path.abspath(get_args())
|
||||||
|
|
||||||
|
printer.bar_init(req_folder)
|
||||||
|
|
||||||
|
if os.path.exists(f"{req_folder}_compressed"):
|
||||||
|
shutil.rmtree(f"{req_folder}_compressed")
|
||||||
|
|
||||||
|
printer.info("Creating folders...")
|
||||||
|
for folder, folders, files in os.walk(req_folder):
|
||||||
|
if not os.path.exists(folder.replace(req_folder, f"{req_folder}_compressed")):
|
||||||
|
os.mkdir(folder.replace(req_folder, f"{req_folder}_compressed"))
|
||||||
|
|
||||||
|
printer.info(f"Compressing \"{folder.replace(req_folder, req_folder.split('/').pop())}\" folder...")
|
||||||
|
target_folder = folder.replace(req_folder, f"{req_folder}_compressed")
|
||||||
|
|
||||||
|
with ThreadPoolExecutor(max_workers=config["FFMPEG"]["Workers"]) as executor:
|
||||||
|
futures = [
|
||||||
|
executor.submit(compress_worker, folder, file, target_folder, req_folder)
|
||||||
|
for file in files
|
||||||
|
]
|
||||||
|
for future in as_completed(futures):
|
||||||
|
future.result()
|
||||||
|
|
||||||
|
utils.get_compression_status(req_folder)
|
||||||
|
utils.sys_pause()
|
||||||
|
print(f"Time taken: {datetime.now() - start_time}")
|
160
FFMpeg-Compressor/modules/compressor.py
Normal file
160
FFMpeg-Compressor/modules/compressor.py
Normal file
|
@ -0,0 +1,160 @@
|
||||||
|
from modules import configloader
|
||||||
|
from modules import printer
|
||||||
|
from modules import utils
|
||||||
|
from PIL import Image
|
||||||
|
import pillow_avif
|
||||||
|
from ffmpeg import FFmpeg, FFmpegError
|
||||||
|
import os
|
||||||
|
|
||||||
|
|
||||||
|
def get_file_type(filename):
|
||||||
|
audio_ext = ['.aac', '.flac', '.m4a', '.mp3', '.ogg', '.opus', '.raw', '.wav', '.wma']
|
||||||
|
image_ext = ['.apng', '.avif', '.bmp', '.tga', '.tiff', '.dds', '.svg', '.webp', '.jpg', '.jpeg', '.png']
|
||||||
|
video_ext = ['.3gp' '.amv', '.avi', '.m2t', '.m4v', '.mkv', '.mov', '.mp4', '.m4v', '.mpeg', '.mpv',
|
||||||
|
'.webm', '.ogv']
|
||||||
|
|
||||||
|
if os.path.splitext(filename)[1] in audio_ext:
|
||||||
|
return "audio"
|
||||||
|
elif os.path.splitext(filename)[1] in image_ext:
|
||||||
|
return "image"
|
||||||
|
elif os.path.splitext(filename)[1] in video_ext:
|
||||||
|
return "video"
|
||||||
|
else:
|
||||||
|
return "unknown"
|
||||||
|
|
||||||
|
|
||||||
|
def has_transparency(img):
|
||||||
|
if img.info.get("transparency", None) is not None:
|
||||||
|
return True
|
||||||
|
if img.mode == "P":
|
||||||
|
transparent = img.info.get("transparency", -1)
|
||||||
|
for _, index in img.getcolors():
|
||||||
|
if index == transparent:
|
||||||
|
return True
|
||||||
|
elif img.mode == "RGBA":
|
||||||
|
extrema = img.getextrema()
|
||||||
|
if extrema[3][0] < 255:
|
||||||
|
return True
|
||||||
|
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def compress_audio(folder, file, target_folder, extension):
|
||||||
|
bitrate = configloader.config['AUDIO']['BitRate']
|
||||||
|
try:
|
||||||
|
(FFmpeg()
|
||||||
|
.input(f'{folder}/{file}')
|
||||||
|
.option("hide_banner")
|
||||||
|
.output(utils.check_duplicates(f'{target_folder}/{os.path.splitext(file)[0]}.{extension}'),
|
||||||
|
{"b:a": bitrate, "loglevel": "error"})
|
||||||
|
.execute()
|
||||||
|
)
|
||||||
|
except FFmpegError as e:
|
||||||
|
utils.add_unprocessed_file(f'{folder}/{file}', f'{target_folder}/{file}')
|
||||||
|
utils.errors_count += 1
|
||||||
|
if not configloader.config['FFMPEG']['HideErrors']:
|
||||||
|
printer.error(f"File {file} can't be processed! Error: {e}")
|
||||||
|
printer.files(file, os.path.splitext(file)[0], extension, f"{bitrate}")
|
||||||
|
return f'{target_folder}/{os.path.splitext(file)[0]}.{extension}'
|
||||||
|
|
||||||
|
|
||||||
|
def compress_video(folder, file, target_folder, extension):
|
||||||
|
if not configloader.config['VIDEO']['SkipVideo']:
|
||||||
|
codec = configloader.config['VIDEO']['Codec']
|
||||||
|
crf = configloader.config['VIDEO']['CRF']
|
||||||
|
|
||||||
|
try:
|
||||||
|
(FFmpeg()
|
||||||
|
.input(f'{folder}/{file}')
|
||||||
|
.option("hide_banner")
|
||||||
|
.option("hwaccel", "auto")
|
||||||
|
.output(utils.check_duplicates(f'{target_folder}/{os.path.splitext(file)[0]}.{extension}'),
|
||||||
|
{"codec:v": codec, "v:b": 0, "loglevel": "error"}, crf=crf)
|
||||||
|
.execute()
|
||||||
|
)
|
||||||
|
printer.files(file, os.path.splitext(file)[0], extension, codec)
|
||||||
|
except FFmpegError as e:
|
||||||
|
utils.add_unprocessed_file(f'{folder}/{file}', f'{target_folder}/{file}')
|
||||||
|
utils.errors_count += 1
|
||||||
|
if not configloader.config['FFMPEG']['HideErrors']:
|
||||||
|
printer.error(f"File {file} can't be processed! Error: {e}")
|
||||||
|
else:
|
||||||
|
utils.add_unprocessed_file(f'{folder}/{file}', f'{target_folder}/{file}')
|
||||||
|
return f'{target_folder}/{os.path.splitext(file)[0]}.{extension}'
|
||||||
|
|
||||||
|
|
||||||
|
def compress_image(folder, file, target_folder, extension):
|
||||||
|
quality = configloader.config['IMAGE']['Quality']
|
||||||
|
try:
|
||||||
|
image = Image.open(f'{folder}/{file}')
|
||||||
|
|
||||||
|
if (extension == "jpg" or extension == "jpeg" or
|
||||||
|
(extension == "webp" and not configloader.config['FFMPEG']['WebpRGBA'])):
|
||||||
|
if has_transparency(image):
|
||||||
|
printer.warning(f"{file} has transparency. Changing to fallback...")
|
||||||
|
extension = configloader.config['IMAGE']['FallBackExtension']
|
||||||
|
|
||||||
|
if has_transparency(image):
|
||||||
|
image.convert('RGBA')
|
||||||
|
|
||||||
|
res_downscale = configloader.config['IMAGE']['ResDownScale']
|
||||||
|
if res_downscale != 1:
|
||||||
|
width, height = image.size
|
||||||
|
new_size = (int(width / res_downscale), int(height / res_downscale))
|
||||||
|
image = image.resize(new_size)
|
||||||
|
|
||||||
|
image.save(utils.check_duplicates(f"{target_folder}/{os.path.splitext(file)[0]}.{extension}"),
|
||||||
|
optimize=True,
|
||||||
|
lossless=configloader.config['IMAGE']['Lossless'],
|
||||||
|
quality=quality,
|
||||||
|
minimize_size=True)
|
||||||
|
printer.files(file, os.path.splitext(file)[0], extension, f"{quality}%")
|
||||||
|
except Exception as e:
|
||||||
|
utils.add_unprocessed_file(f'{folder}/{file}', f'{target_folder}/{file}')
|
||||||
|
utils.errors_count += 1
|
||||||
|
if not configloader.config['FFMPEG']['HideErrors']:
|
||||||
|
printer.error(f"File {file} can't be processed! Error: {e}")
|
||||||
|
return f'{target_folder}/{os.path.splitext(file)[0]}.{extension}'
|
||||||
|
|
||||||
|
|
||||||
|
def compress(folder, file, target_folder):
|
||||||
|
if configloader.config["FFMPEG"]["ForceCompress"]:
|
||||||
|
printer.unknown_file(file)
|
||||||
|
try:
|
||||||
|
(FFmpeg()
|
||||||
|
.input(f'{folder}/{file}')
|
||||||
|
.output(f'{target_folder}/{file}')
|
||||||
|
.execute()
|
||||||
|
)
|
||||||
|
except FFmpegError as e:
|
||||||
|
utils.add_unprocessed_file(f'{folder}/{file}', f'{target_folder}/{file}')
|
||||||
|
utils.errors_count += 1
|
||||||
|
if not configloader.config['FFMPEG']['HideErrors']:
|
||||||
|
printer.error(f"File {file} can't be processed! Error: {e}")
|
||||||
|
else:
|
||||||
|
utils.add_unprocessed_file(f'{folder}/{file}', f'{target_folder}/{file}')
|
||||||
|
return f'{target_folder}/{file}'
|
||||||
|
|
||||||
|
|
||||||
|
def compress_file(_dir, filename, target_dir, source):
|
||||||
|
match get_file_type(filename):
|
||||||
|
case "audio":
|
||||||
|
comp_file = compress_audio(_dir, filename, target_dir,
|
||||||
|
configloader.config['AUDIO']['Extension'])
|
||||||
|
case "image":
|
||||||
|
comp_file = compress_image(_dir, filename, target_dir,
|
||||||
|
configloader.config['IMAGE']['Extension'])
|
||||||
|
case "video":
|
||||||
|
comp_file = compress_video(_dir, filename, target_dir,
|
||||||
|
configloader.config['VIDEO']['Extension'])
|
||||||
|
case "unknown":
|
||||||
|
comp_file = compress(_dir, filename, target_dir)
|
||||||
|
|
||||||
|
if configloader.config['FFMPEG']['MimicMode']:
|
||||||
|
try:
|
||||||
|
os.rename(comp_file, f'{_dir}/{filename}'.replace(source, f"{source}_compressed"))
|
||||||
|
except FileNotFoundError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
printer.bar.update()
|
||||||
|
printer.bar.next()
|
11
FFMpeg-Compressor/modules/configloader.py
Normal file
11
FFMpeg-Compressor/modules/configloader.py
Normal file
|
@ -0,0 +1,11 @@
|
||||||
|
import tomllib
|
||||||
|
from modules import printer
|
||||||
|
|
||||||
|
try:
|
||||||
|
config = tomllib.load(open("ffmpeg-comp.toml", "rb"))
|
||||||
|
except FileNotFoundError:
|
||||||
|
try:
|
||||||
|
config = tomllib.load(open("/etc/ffmpeg-comp.toml", "rb"))
|
||||||
|
except FileNotFoundError:
|
||||||
|
printer.error("Config file not found. Please put it next to binary or in to /etc folder.")
|
||||||
|
exit()
|
51
FFMpeg-Compressor/modules/printer.py
Normal file
51
FFMpeg-Compressor/modules/printer.py
Normal file
|
@ -0,0 +1,51 @@
|
||||||
|
from progress.bar import IncrementalBar
|
||||||
|
import colorama
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
|
||||||
|
|
||||||
|
def bar_init(folder):
|
||||||
|
file_count = 0
|
||||||
|
for folder, folders, file in os.walk(folder):
|
||||||
|
file_count += len(file)
|
||||||
|
global bar
|
||||||
|
bar = IncrementalBar('Compressing', max=file_count, suffix='[%(index)d/%(max)d] (%(percent).1f%%)')
|
||||||
|
bar.update()
|
||||||
|
|
||||||
|
|
||||||
|
def bar_print(string):
|
||||||
|
print(string)
|
||||||
|
bar.update()
|
||||||
|
|
||||||
|
|
||||||
|
# Fill whole string with spaces for cleaning progress bar
|
||||||
|
def clean_str(string):
|
||||||
|
return string + " " * (os.get_terminal_size().columns - len(string))
|
||||||
|
|
||||||
|
|
||||||
|
def info(string):
|
||||||
|
bar_print(clean_str(f"\r\033[100m- {string}\033[49m"))
|
||||||
|
|
||||||
|
|
||||||
|
def warning(string):
|
||||||
|
bar_print(clean_str(f"\r\033[93m!\033[0m {string}\033[49m"))
|
||||||
|
|
||||||
|
|
||||||
|
def error(string):
|
||||||
|
bar_print(clean_str(f"\r\033[31m\u2715\033[0m {string}\033[49m"))
|
||||||
|
|
||||||
|
|
||||||
|
def files(source, dest, dest_ext, comment):
|
||||||
|
source_ext = os.path.splitext(source)[1]
|
||||||
|
source_name = os.path.splitext(source)[0]
|
||||||
|
|
||||||
|
bar_print(clean_str(f"\r\033[0;32m\u2713\033[0m \033[0;37m{source_name}\033[0m{source_ext}\033[0;37m -> {dest}\033[0m.{dest_ext}\033[0;37m ({comment})\033[0m"))
|
||||||
|
|
||||||
|
|
||||||
|
def unknown_file(file):
|
||||||
|
bar_print(clean_str(f"\r* \033[0;33m{file}\033[0m (File will be force compressed via ffmpeg)"))
|
||||||
|
|
||||||
|
|
||||||
|
def win_ascii_esc():
|
||||||
|
if sys.platform == "win32":
|
||||||
|
colorama.init()
|
79
FFMpeg-Compressor/modules/utils.py
Normal file
79
FFMpeg-Compressor/modules/utils.py
Normal file
|
@ -0,0 +1,79 @@
|
||||||
|
from modules import configloader
|
||||||
|
from modules import printer
|
||||||
|
from shutil import copyfile
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
|
||||||
|
errors_count = 0
|
||||||
|
|
||||||
|
|
||||||
|
def get_dir_size(directory, files):
|
||||||
|
total_size = 0
|
||||||
|
for folder, folders, files in os.walk(directory):
|
||||||
|
for file in files:
|
||||||
|
if not os.path.islink(f"{folder}/{file}"):
|
||||||
|
total_size += os.path.getsize(f"{folder}/{file}")
|
||||||
|
return total_size
|
||||||
|
|
||||||
|
|
||||||
|
def get_compression(orig, comp):
|
||||||
|
processed_files = []
|
||||||
|
for folder, folders, files in os.walk(comp):
|
||||||
|
for file in files:
|
||||||
|
processed_files.append(file)
|
||||||
|
try:
|
||||||
|
orig = get_dir_size(orig, processed_files)
|
||||||
|
comp = get_dir_size(comp, processed_files)
|
||||||
|
|
||||||
|
print(f"\nResult: {orig/1024/1024:.2f}MB -> {comp/1024/1024:.2f}MB ({(comp - orig)/1024/1024:.2f}MB)")
|
||||||
|
except ZeroDivisionError:
|
||||||
|
printer.warning("Nothing compressed!")
|
||||||
|
|
||||||
|
|
||||||
|
def get_compression_status(orig_folder):
|
||||||
|
orig_folder_len = 0
|
||||||
|
comp_folder_len = 0
|
||||||
|
|
||||||
|
for folder, folders, files in os.walk(orig_folder):
|
||||||
|
orig_folder_len += len(files)
|
||||||
|
|
||||||
|
for folder, folders, files in os.walk(f'{orig_folder}_compressed'):
|
||||||
|
for file in files:
|
||||||
|
if not os.path.splitext(file)[1].count(" (copy)"):
|
||||||
|
comp_folder_len += 1
|
||||||
|
|
||||||
|
if errors_count != 0:
|
||||||
|
printer.warning("Some files failed to compress!")
|
||||||
|
|
||||||
|
if orig_folder_len == comp_folder_len:
|
||||||
|
printer.info("Success!")
|
||||||
|
get_compression(orig_folder, f"{orig_folder}_compressed")
|
||||||
|
else:
|
||||||
|
printer.warning("Original and compressed folders are not identical!")
|
||||||
|
get_compression(orig_folder, f"{orig_folder}_compressed")
|
||||||
|
|
||||||
|
|
||||||
|
def add_unprocessed_file(orig_folder, new_folder):
|
||||||
|
if configloader.config['FFMPEG']['CopyUnprocessed']:
|
||||||
|
filename = orig_folder.split("/").pop()
|
||||||
|
copyfile(orig_folder, new_folder)
|
||||||
|
printer.info(f"File {filename} copied to compressed folder.")
|
||||||
|
|
||||||
|
|
||||||
|
def check_duplicates(new_folder):
|
||||||
|
filename = new_folder.split().pop()
|
||||||
|
if os.path.exists(new_folder):
|
||||||
|
printer.warning(
|
||||||
|
f'Duplicate file has been found! Check manually this files - "{filename}", '
|
||||||
|
f'"{os.path.splitext(filename)[0] + "(copy)" + os.path.splitext(filename)[1]}"')
|
||||||
|
return os.path.splitext(new_folder)[0] + "(copy)" + os.path.splitext(new_folder)[1]
|
||||||
|
return new_folder
|
||||||
|
|
||||||
|
|
||||||
|
def sys_pause():
|
||||||
|
if sys.platform == "win32":
|
||||||
|
os.system("pause")
|
||||||
|
|
||||||
|
|
||||||
|
def help_message():
|
||||||
|
return "Usage: ffmpeg-comp {folder}"
|
|
@ -2,5 +2,4 @@ Pillow==10.3.0
|
||||||
pillow-avif-plugin==1.4.3
|
pillow-avif-plugin==1.4.3
|
||||||
python-ffmpeg==2.0.12
|
python-ffmpeg==2.0.12
|
||||||
progress==1.6
|
progress==1.6
|
||||||
colorama==0.4.6
|
colorama==0.4.6
|
||||||
argparse~=1.4.0
|
|
27
README.md
27
README.md
|
@ -1,25 +1,8 @@
|
||||||
## VNTools
|
## VNTools
|
||||||
Collection of tools used by VienDesu! Porting Team
|
Collection of tools used by administrators from VN Telegram Channel
|
||||||
|
|
||||||
### Tools
|
### Tools
|
||||||
* `vnrecode` - Python utility to compress Visual Novel Resources
|
* `FFMpeg-Compressor` - Python utility uses ffmpeg to compress Visual Novel Resources
|
||||||
* `unrenapk` - A Python script for extracting game project from Ren'Py based .apk and .obb files
|
* `RenPy-Android-Unpack` - A simple Python script for unpacking Ren'Py based .apk and .obb files to ready to use Ren'Py SDK's Project
|
||||||
* `renpy-ripper` - Simple .rpy script that will make any RenPy game unpack itself
|
* `RenPy-Unpacker` - Simple .rpy script that will make any RenPy game unpack itself
|
||||||
* `vnds2renpy` - Simple script for converting VNDS engine scripts to .rpy ones
|
* `VNDS-to-RenPy` - Simple script for converting vnds scripts to rpy
|
||||||
|
|
||||||
### Installation
|
|
||||||
#### Download from releases:
|
|
||||||
* Windows - `TODO`
|
|
||||||
* Linux - `TODO`
|
|
||||||
* MacOS - `TODO`
|
|
||||||
|
|
||||||
#### Build tools as binaries:
|
|
||||||
* Run `./build.sh` on UNIX
|
|
||||||
* Run `.\build.bat` for Windows
|
|
||||||
* Arch Linux - `TODO`
|
|
||||||
* NixOS - `TODO`
|
|
||||||
|
|
||||||
#### Install as python package:
|
|
||||||
* Run `pip install -U .` command in project folder
|
|
||||||
* Arch Linux - `TODO`
|
|
||||||
* NixOS - `TODO`
|
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
## unrenapk
|
## RenPy-Android-Unpack
|
||||||
A simple Python script for unpacking Ren'Py based .apk and .obb files to ready to use Ren'Py SDK's Project
|
A simple Python script for unpacking Ren'Py based .apk and .obb files to ready to use Ren'Py SDK's Project
|
||||||
|
|
||||||
### How to use
|
### How to use
|
||||||
* Put some .apk & .obb files in folder
|
* Put some .apk & .obb files in folder
|
||||||
* `unrenapk` (It unpacks all .apk and .obb files in the directory where it is located)
|
* `rendroid-unpack` (It unpacks all .apk and .obb files in the directory where it is located)
|
2
RenPy-Android-Unpack/requirements.txt
Normal file
2
RenPy-Android-Unpack/requirements.txt
Normal file
|
@ -0,0 +1,2 @@
|
||||||
|
Pillow==9.5.0
|
||||||
|
colorama==0.4.6
|
108
RenPy-Android-Unpack/unpack.py
Executable file
108
RenPy-Android-Unpack/unpack.py
Executable file
|
@ -0,0 +1,108 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
from PIL import Image
|
||||||
|
import colorama
|
||||||
|
import zipfile
|
||||||
|
import shutil
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
|
||||||
|
|
||||||
|
def printer(msg, level):
|
||||||
|
match level:
|
||||||
|
case "info":
|
||||||
|
print(f"\033[100m[INFO] {msg}\033[49m")
|
||||||
|
case "warn":
|
||||||
|
print(f"\033[93m[WARN]\033[0m {msg}\033[49m")
|
||||||
|
case "err":
|
||||||
|
print(f"\033[31m[ERROR]\033[0m {msg} Exiting...\033[49m")
|
||||||
|
exit()
|
||||||
|
|
||||||
|
|
||||||
|
def extract_folder(zip_ref, path, dest):
|
||||||
|
for content in zip_ref.namelist():
|
||||||
|
if content.split('/')[0] == path:
|
||||||
|
zip_ref.extract(content, dest)
|
||||||
|
|
||||||
|
|
||||||
|
def find_modern_icon(directory):
|
||||||
|
icons = []
|
||||||
|
for folder, folders, files in os.walk(directory):
|
||||||
|
for file in os.listdir(folder):
|
||||||
|
if os.path.splitext(file)[1] == ".png":
|
||||||
|
image = Image.open(f"{folder}/{file}")
|
||||||
|
if image.size[0] == 432 and image.size[1] == 432:
|
||||||
|
icons.append(f"{folder}/{file}")
|
||||||
|
if len(icons) == 0:
|
||||||
|
raise KeyError
|
||||||
|
return icons
|
||||||
|
|
||||||
|
|
||||||
|
def extract_assets(file):
|
||||||
|
try:
|
||||||
|
with zipfile.ZipFile(file, 'r') as zip_ref:
|
||||||
|
extract_folder(zip_ref, 'assets', '')
|
||||||
|
if os.path.splitext(file)[1] == '.apk':
|
||||||
|
try:
|
||||||
|
# ~Ren'Py 8, 7
|
||||||
|
extract_folder(zip_ref, 'res', 'assets')
|
||||||
|
for icon in find_modern_icon('assets/res'):
|
||||||
|
os.rename(icon, f"assets/{os.path.split(icon)[1]}")
|
||||||
|
except KeyError:
|
||||||
|
try:
|
||||||
|
# ~Ren'Py 6
|
||||||
|
zip_ref.extract('res/drawable/icon.png', 'assets')
|
||||||
|
os.rename('assets/res/drawable/icon.png', 'assets/icon.png')
|
||||||
|
except KeyError:
|
||||||
|
printer("Icon not found. Maybe it is not supported apk?", "warn")
|
||||||
|
except zipfile.BadZipFile:
|
||||||
|
return printer("Cant extract .apk file!", "err")
|
||||||
|
|
||||||
|
|
||||||
|
def rename_files(directory):
|
||||||
|
for dir_ in os.walk(directory):
|
||||||
|
for file in dir_[2]:
|
||||||
|
path = f'{dir_[0]}/{file}'
|
||||||
|
folder = '/'.join(path.split('/')[:len(path.split('/')) - 1])
|
||||||
|
newname = f'{path.split("/").pop().replace("x-", "")}'
|
||||||
|
os.rename(path, f'{folder}/{newname}')
|
||||||
|
|
||||||
|
|
||||||
|
def rename_dirs(directory):
|
||||||
|
dirs = []
|
||||||
|
for dir_ in os.walk(directory):
|
||||||
|
dirs.append(dir_[0])
|
||||||
|
dirs.reverse()
|
||||||
|
dirs.pop()
|
||||||
|
for dir__ in dirs:
|
||||||
|
folder = '/'.join(dir__.split('/')[:len(dir__.split('/')) - 1])
|
||||||
|
newname = f'{dir__.split("/").pop().replace("x-", "")}'
|
||||||
|
os.rename(dir__, f'{folder}/{newname}')
|
||||||
|
|
||||||
|
|
||||||
|
def remove_unneeded(names, ignore):
|
||||||
|
for name in names:
|
||||||
|
try:
|
||||||
|
shutil.rmtree(name)
|
||||||
|
except FileNotFoundError:
|
||||||
|
if not ignore:
|
||||||
|
printer(f"Path {name} not found!", "warn")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
if sys.platform == "win32":
|
||||||
|
colorama.init()
|
||||||
|
for filename in os.listdir(os.getcwd()):
|
||||||
|
if os.path.splitext(filename)[1] == '.apk' or os.path.splitext(filename)[1] == '.obb':
|
||||||
|
remove_unneeded(['assets'], True)
|
||||||
|
printer(f'Extracting assets from {filename}... ', "info")
|
||||||
|
extract_assets(filename)
|
||||||
|
printer('Renaming game assets... ', "info")
|
||||||
|
rename_files('assets')
|
||||||
|
rename_dirs('assets')
|
||||||
|
printer('Removing unneeded files... ', "info")
|
||||||
|
if os.path.splitext(filename)[1] == '.apk':
|
||||||
|
remove_unneeded(['assets/renpy', 'assets/res'], False)
|
||||||
|
remove_unneeded(['assets/dexopt'], True)
|
||||||
|
printer('Renaming directory... ', "info")
|
||||||
|
remove_unneeded([os.path.splitext(filename)[0]], True)
|
||||||
|
os.rename('assets', os.path.splitext(filename)[0])
|
|
@ -1,7 +1,7 @@
|
||||||
## vnds2renpy
|
## VNDS-to-RenPy
|
||||||
Simple script for converting vnds scripts to rpy
|
Simple script for converting vnds scripts to rpy
|
||||||
|
|
||||||
### How to use
|
### How to use
|
||||||
|
|
||||||
* Extract VNDS visual novel's archive and get `scripts` folder from these (or `script.zip` if scripts folder is empty)
|
* Extract VNDS visual novel's archive and get `scripts` folder from these (or `script.zip` if scripts folder is empty)
|
||||||
* Launch `vnds2renpy` (It will automatically extract `scripts.zip` archive (if it needed) and converts .scr scripts to .rpy)
|
* Launch `convert.py` (It will automatically extract `scripts.zip` archive (if it needed) and converts .scr scripts to .rpy)
|
16
build.bat
Executable file → Normal file
16
build.bat
Executable file → Normal file
|
@ -3,16 +3,16 @@ if not defined VIRTUAL_ENV goto :venv_error
|
||||||
|
|
||||||
mkdir output
|
mkdir output
|
||||||
mkdir output\bin
|
mkdir output\bin
|
||||||
python -m pip install -r requirements.txt || goto :exit
|
python -m pip install -r FFMpeg-Compressor\requirements.txt || goto :exit
|
||||||
|
python -m pip install -r RenPy-Android-Unpack\requirements.txt || goto :exit
|
||||||
python -m pip install Nuitka || goto :exit
|
python -m pip install Nuitka || goto :exit
|
||||||
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=vnrecode vnrecode\__main__.py || goto :exit
|
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=ffmpeg-comp FFMpeg-Compressor\main.py || goto :exit
|
||||||
xcopy vnrecode\vnrecode.toml output\bin /Y
|
xcopy FFMpeg-Compressor\ffmpeg-comp.toml output\bin /Y
|
||||||
move /Y output\vnrecode.exe output\bin
|
move /Y output\ffmpeg-comp.exe output\bin
|
||||||
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=unrenapk unrenapk\__main__.py || goto :exit
|
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=rendroid-unpack RenPy-Android-Unpack\unpack.py || goto :exit
|
||||||
move /Y output\unrenapk.exe output\bin
|
move /Y output\rendroid-unpack.exe output\bin
|
||||||
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=vnds2renpy vnds2renpy/__main__.py || goto :exit
|
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=vnds2renpy VNDS-to-RenPy/convert.py || goto :exit
|
||||||
move /Y output\vnds2renpy.exe output\bin
|
move /Y output\vnds2renpy.exe output\bin
|
||||||
echo "Done! You can get binaries into output\bin directory"
|
|
||||||
|
|
||||||
:venv_error
|
:venv_error
|
||||||
echo "Please create and activate venv before running this script: python -m venv .\venv && .\venv\Scripts\activate.bat"
|
echo "Please create and activate venv before running this script: python -m venv .\venv && .\venv\Scripts\activate.bat"
|
||||||
|
|
18
build.sh
18
build.sh
|
@ -8,17 +8,17 @@ fi
|
||||||
|
|
||||||
mkdir -p output
|
mkdir -p output
|
||||||
mkdir -p output/bin
|
mkdir -p output/bin
|
||||||
python3 -m pip install -r requirements.txt
|
python3 -m pip install -r FFMpeg-Compressor/requirements.txt
|
||||||
|
python3 -m pip install -r RenPy-Android-Unpack/requirements.txt
|
||||||
python3 -m pip install Nuitka
|
python3 -m pip install Nuitka
|
||||||
case "$(uname -s)" in
|
case "$(uname -s)" in
|
||||||
Linux*) jobs="--jobs=$(nproc)";;
|
Linux*) jobs="--jobs=$(nproc)";;
|
||||||
Darwin*) jobs="--jobs=$(sysctl -n hw.ncpu)";;
|
Darwin*) jobs="--jobs=$(sysctl -n hw.ncpu)";;
|
||||||
esac
|
esac
|
||||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=vnrecode vnrecode/__main__.py
|
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=ffmpeg-comp FFMpeg-Compressor/main.py
|
||||||
cp vnrecode/vnrecode.toml output/bin
|
cp FFMpeg-Compressor/ffmpeg-comp.toml output/bin
|
||||||
mv output/vnrecode output/bin
|
mv output/ffmpeg-comp output/bin
|
||||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=unrenapk unrenapk/__main__.py
|
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=rendroid-unpack RenPy-Android-Unpack/unpack.py
|
||||||
mv output/unrenapk output/bin
|
mv output/rendroid-unpack output/bin
|
||||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=vnds2renpy vnds2renpy/__main__.py
|
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=vnds2renpy VNDS-to-RenPy/convert.py
|
||||||
mv output/vnds2renpy output/bin
|
mv output/vnds2renpy output/bin
|
||||||
echo "Done! You can get binaries into output/bin directory"
|
|
|
@ -1,32 +0,0 @@
|
||||||
[build-system]
|
|
||||||
requires = [
|
|
||||||
"setuptools >= 61.0"
|
|
||||||
]
|
|
||||||
build-backend = "setuptools.build_meta"
|
|
||||||
|
|
||||||
[tool.setuptools]
|
|
||||||
packages = ["vnrecode", "unrenapk", "vnds2renpy"]
|
|
||||||
include-package-data = true
|
|
||||||
|
|
||||||
[tool.setuptools.package-data]
|
|
||||||
'vnrecode' = ['*.py']
|
|
||||||
'vnds2renpy' = ['*.py']
|
|
||||||
'unrenapk' = ['*.py']
|
|
||||||
|
|
||||||
[project.scripts]
|
|
||||||
vnrecode = "vnrecode.__main__:init"
|
|
||||||
vnds2renpy = "vnds2renpy.__main__:main"
|
|
||||||
unrenapk = "unrenapk.application:launch"
|
|
||||||
|
|
||||||
[project]
|
|
||||||
name = "vntools"
|
|
||||||
version = "2.0-dev"
|
|
||||||
requires-python = ">= 3.11"
|
|
||||||
dependencies = [
|
|
||||||
"Pillow==10.3.0",
|
|
||||||
"pillow-avif-plugin==1.4.3",
|
|
||||||
"python-ffmpeg==2.0.12",
|
|
||||||
"progress==1.6",
|
|
||||||
"colorama==0.4.6",
|
|
||||||
"argparse~=1.4.0"
|
|
||||||
]
|
|
|
@ -1,6 +0,0 @@
|
||||||
#!/usr/bin/env python3
|
|
||||||
|
|
||||||
from . import application
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
|
||||||
application.launch()
|
|
|
@ -1,98 +0,0 @@
|
||||||
from zipfile import ZipFile, BadZipFile
|
|
||||||
from PIL import Image
|
|
||||||
import shutil
|
|
||||||
import os
|
|
||||||
|
|
||||||
from .printer import Printer
|
|
||||||
|
|
||||||
|
|
||||||
class Extract:
|
|
||||||
|
|
||||||
def __init__(self, output: str):
|
|
||||||
self.output = output
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def folder(zip_ref: ZipFile, path: str, dest: str):
|
|
||||||
for content in zip_ref.namelist():
|
|
||||||
if content.split('/')[0] == path:
|
|
||||||
zip_ref.extract(content, dest)
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def icon(directory: str):
|
|
||||||
icons = []
|
|
||||||
for folder, folders, files in os.walk(directory):
|
|
||||||
for file in os.listdir(folder):
|
|
||||||
if os.path.splitext(file)[1] == ".png":
|
|
||||||
image = Image.open(f"{folder}/{file}")
|
|
||||||
if image.size[0] == 432 and image.size[1] == 432:
|
|
||||||
icons.append(f"{folder}/{file}")
|
|
||||||
if len(icons) == 0:
|
|
||||||
raise KeyError
|
|
||||||
return icons
|
|
||||||
|
|
||||||
def assets(self, file: str):
|
|
||||||
try:
|
|
||||||
with ZipFile(file, 'r') as zip_ref:
|
|
||||||
self.folder(zip_ref, 'assets', self.output)
|
|
||||||
if os.path.splitext(file)[1] == '.apk':
|
|
||||||
try:
|
|
||||||
# ~Ren'Py 8, 7
|
|
||||||
self.folder(zip_ref, 'res', os.path.join(self.output, 'assets'))
|
|
||||||
for icon in self.icon(os.path.join(self.output, 'assets/res')):
|
|
||||||
os.rename(icon, os.path.join(self.output, "assets", os.path.split(icon)[1]))
|
|
||||||
except KeyError:
|
|
||||||
try:
|
|
||||||
# ~Ren'Py 6
|
|
||||||
zip_ref.extract('res/drawable/icon.png', os.path.join(self.output, 'assets'))
|
|
||||||
os.rename(os.path.join(self.output, 'assets/res/drawable/icon.png'),
|
|
||||||
os.path.join(self.output, 'assets/icon.png'))
|
|
||||||
except KeyError:
|
|
||||||
Printer.warn("Icon not found. Maybe it is not supported apk?")
|
|
||||||
except BadZipFile:
|
|
||||||
Printer.err("Cant extract .apk file!")
|
|
||||||
|
|
||||||
|
|
||||||
class Rename:
|
|
||||||
|
|
||||||
def __init__(self, output):
|
|
||||||
self.output = output
|
|
||||||
|
|
||||||
def files(self, directory: str):
|
|
||||||
for dir_ in os.walk(os.path.join(self.output, directory)):
|
|
||||||
for file in dir_[2]:
|
|
||||||
path = f'{dir_[0]}/{file}'
|
|
||||||
folder = '/'.join(path.split('/')[:len(path.split('/')) - 1])
|
|
||||||
newname = f'{path.split("/").pop().replace("x-", "")}'
|
|
||||||
os.rename(path, f'{folder}/{newname}')
|
|
||||||
|
|
||||||
def dirs(self, directory: str):
|
|
||||||
dirs = []
|
|
||||||
for dir_ in os.walk(os.path.join(self.output, directory)):
|
|
||||||
dirs.append(dir_[0])
|
|
||||||
dirs.reverse()
|
|
||||||
dirs.pop()
|
|
||||||
for dir__ in dirs:
|
|
||||||
folder = '/'.join(dir__.split('/')[:len(dir__.split('/')) - 1])
|
|
||||||
newname = f'{dir__.split("/").pop().replace("x-", "")}'
|
|
||||||
os.rename(dir__, f'{folder}/{newname}')
|
|
||||||
|
|
||||||
|
|
||||||
class Actions:
|
|
||||||
|
|
||||||
def __init__(self, output: str):
|
|
||||||
self.output = output
|
|
||||||
|
|
||||||
def extract(self) -> Extract:
|
|
||||||
return Extract(self.output)
|
|
||||||
|
|
||||||
def rename(self) -> Rename:
|
|
||||||
return Rename(self.output)
|
|
||||||
|
|
||||||
def clean(self, names: list, ignore: bool):
|
|
||||||
for name in names:
|
|
||||||
name = os.path.join(self.output, name)
|
|
||||||
try:
|
|
||||||
shutil.rmtree(name)
|
|
||||||
except FileNotFoundError:
|
|
||||||
if not ignore:
|
|
||||||
Printer.warn(f"Path {name} not found!")
|
|
|
@ -1,51 +0,0 @@
|
||||||
#!/usr/bin/env python3
|
|
||||||
import colorama
|
|
||||||
import argparse
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
|
|
||||||
from .printer import Printer
|
|
||||||
from .actions import Actions
|
|
||||||
|
|
||||||
|
|
||||||
def args_init():
|
|
||||||
parser = argparse.ArgumentParser(
|
|
||||||
prog='unrenapk',
|
|
||||||
description='Extract Ren\'Py .apk and .obb files into Ren\'Py SDK\'s project'
|
|
||||||
)
|
|
||||||
parser.add_argument('path')
|
|
||||||
parser.add_argument('-o', '--output')
|
|
||||||
return parser.parse_args()
|
|
||||||
|
|
||||||
def launch():
|
|
||||||
if sys.platform == "win32":
|
|
||||||
colorama.init()
|
|
||||||
args = args_init()
|
|
||||||
if args.output:
|
|
||||||
output = args.output
|
|
||||||
else:
|
|
||||||
output = ''
|
|
||||||
actions = Actions(output)
|
|
||||||
printer = Printer()
|
|
||||||
|
|
||||||
filename = args.path
|
|
||||||
if os.path.splitext(filename)[1] == '.apk' or os.path.splitext(filename)[1] == '.obb':
|
|
||||||
actions.clean(['assets'], True)
|
|
||||||
|
|
||||||
printer.info(f'Extracting assets from {filename}... ')
|
|
||||||
actions.extract().assets(filename)
|
|
||||||
|
|
||||||
printer.info('Renaming game assets... ')
|
|
||||||
actions.rename().files('assets')
|
|
||||||
actions.rename().dirs('assets')
|
|
||||||
|
|
||||||
printer.info('Removing unneeded files... ')
|
|
||||||
if os.path.splitext(filename)[1] == '.apk':
|
|
||||||
actions.clean(['assets/renpy', 'assets/res'], False)
|
|
||||||
actions.clean(['assets/dexopt'], True)
|
|
||||||
|
|
||||||
printer.info('Renaming directory... ')
|
|
||||||
actions.clean([os.path.splitext(filename)[0]], True)
|
|
||||||
os.rename(os.path.join(output, 'assets'), os.path.splitext(filename)[0])
|
|
||||||
else:
|
|
||||||
Printer.err("It's not an .apk or .obb file!")
|
|
|
@ -1,14 +0,0 @@
|
||||||
class Printer:
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def info(msg: str):
|
|
||||||
print(f"\033[100m[INFO] {msg}\033[49m")
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def warn(msg: str):
|
|
||||||
print(f"\033[93m[WARN]\033[0m {msg}\033[49m")
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def err(msg: str):
|
|
||||||
print(f"\033[31m[ERROR]\033[0m {msg} Exiting...\033[49m")
|
|
||||||
exit()
|
|
|
@ -1,23 +0,0 @@
|
||||||
#!/usr/bin/env python3
|
|
||||||
from .application import Application
|
|
||||||
from .compress import Compress
|
|
||||||
from .printer import Printer
|
|
||||||
from .params import Params
|
|
||||||
from .utils import Utils
|
|
||||||
|
|
||||||
|
|
||||||
def init():
|
|
||||||
"""
|
|
||||||
This function creates all needed class instances and run utility
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
params = Params.setup()
|
|
||||||
printer = Printer(params.source)
|
|
||||||
utils = Utils(params, printer)
|
|
||||||
compress = Compress(params, printer, utils)
|
|
||||||
|
|
||||||
Application(params, compress, printer, utils).run()
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
init()
|
|
|
@ -1,59 +0,0 @@
|
||||||
#!/usr/bin/env python3
|
|
||||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
||||||
from datetime import datetime
|
|
||||||
from pathlib import Path
|
|
||||||
import shutil
|
|
||||||
import os
|
|
||||||
|
|
||||||
from .compress import Compress
|
|
||||||
from .printer import Printer
|
|
||||||
from .params import Params
|
|
||||||
from .utils import Utils
|
|
||||||
|
|
||||||
|
|
||||||
class Application:
|
|
||||||
"""
|
|
||||||
Main class for utility
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, params_inst: Params, compress_inst: Compress, printer_inst: Printer, utils_inst: Utils):
|
|
||||||
self.__params = params_inst
|
|
||||||
self.__compress = compress_inst.compress
|
|
||||||
self.__printer = printer_inst
|
|
||||||
self.__utils = utils_inst
|
|
||||||
|
|
||||||
def run(self):
|
|
||||||
"""
|
|
||||||
Method creates a folder in which all the recoded files will be placed,
|
|
||||||
creates a queue of recoding processes for each file and, when the files are run out in the original folder,
|
|
||||||
calls functions to display the result
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
start_time = datetime.now()
|
|
||||||
self.__printer.win_ascii_esc()
|
|
||||||
|
|
||||||
source = self.__params.source
|
|
||||||
|
|
||||||
if self.__params.dest.exists():
|
|
||||||
shutil.rmtree(self.__params.dest)
|
|
||||||
|
|
||||||
self.__printer.info("Creating folders...")
|
|
||||||
for folder, folders, files in os.walk(source):
|
|
||||||
output = self.__utils.get_comp_subdir(folder)
|
|
||||||
if not output.exists():
|
|
||||||
os.mkdir(output)
|
|
||||||
|
|
||||||
self.__printer.info(f'Compressing "{folder}" folder...')
|
|
||||||
|
|
||||||
with ThreadPoolExecutor(max_workers=self.__params.workers) as executor:
|
|
||||||
futures = [
|
|
||||||
executor.submit(self.__compress, Path(folder, file), Path(output))
|
|
||||||
for file in files if Path(folder, file).is_file()
|
|
||||||
]
|
|
||||||
for future in as_completed(futures):
|
|
||||||
future.result()
|
|
||||||
|
|
||||||
self.__utils.print_duplicates()
|
|
||||||
self.__utils.get_recode_status()
|
|
||||||
self.__utils.sys_pause()
|
|
||||||
print(f"Time taken: {datetime.now() - start_time}")
|
|
|
@ -1,197 +0,0 @@
|
||||||
from ffmpeg import FFmpeg, FFmpegError
|
|
||||||
from pathlib import Path
|
|
||||||
from PIL import Image
|
|
||||||
import pillow_avif
|
|
||||||
|
|
||||||
from .printer import Printer
|
|
||||||
from .params import Params
|
|
||||||
from .utils import Utils
|
|
||||||
|
|
||||||
|
|
||||||
class File:
|
|
||||||
"""
|
|
||||||
Class contains some methods to work with files
|
|
||||||
"""
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def get_type(path: Path) -> str:
|
|
||||||
"""
|
|
||||||
Method returns filetype string for file
|
|
||||||
:param path: Path of file to determine type
|
|
||||||
:return: filetype string: audio, image, video, unknown
|
|
||||||
"""
|
|
||||||
extensions = {
|
|
||||||
"audio": ['.aac', '.flac', '.m4a', '.mp3', '.ogg', '.opus', '.raw', '.wav', '.wma'],
|
|
||||||
"image": ['.apng', '.avif', '.bmp', '.tga', '.tiff', '.dds', '.svg', '.webp', '.jpg', '.jpeg', '.png'],
|
|
||||||
"video": ['.3gp' '.amv', '.avi', '.m2t', '.m4v', '.mkv', '.mov', '.mp4', '.m4v', '.mpeg', '.mpv',
|
|
||||||
'.webm', '.ogv']
|
|
||||||
}
|
|
||||||
|
|
||||||
for file_type in extensions:
|
|
||||||
if path.suffix in extensions[file_type]:
|
|
||||||
return file_type
|
|
||||||
return "unknown"
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def has_transparency(img: Image) -> bool:
|
|
||||||
"""
|
|
||||||
Method checks if image has transparency
|
|
||||||
:param img: Pillow Image
|
|
||||||
:return: bool
|
|
||||||
"""
|
|
||||||
if img.info.get("transparency", None) is not None:
|
|
||||||
return True
|
|
||||||
if img.mode == "P":
|
|
||||||
transparent = img.info.get("transparency", -1)
|
|
||||||
for _, index in img.getcolors():
|
|
||||||
if index == transparent:
|
|
||||||
return True
|
|
||||||
elif img.mode == "RGBA":
|
|
||||||
extrema = img.getextrema()
|
|
||||||
if extrema[3][0] < 255:
|
|
||||||
return True
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
class Compress:
|
|
||||||
|
|
||||||
def __init__(self, params_inst: Params, printer_inst: Printer, utils_inst: Utils):
|
|
||||||
self.__params = params_inst
|
|
||||||
self.__printer = printer_inst
|
|
||||||
self.__utils = utils_inst
|
|
||||||
|
|
||||||
def audio(self, input_path: Path, output_dir: Path, extension: str) -> Path:
|
|
||||||
"""
|
|
||||||
Method recodes audio files to another format using ffmpeg utility
|
|
||||||
:param input_path: Path of the original audio file
|
|
||||||
:param output_dir: Path of the output (compression) folder
|
|
||||||
:param extension: Extension of the new audio file
|
|
||||||
:return: Path of compressed audio file with md5 hash as prefix
|
|
||||||
"""
|
|
||||||
bit_rate = self.__params.audio_bitrate
|
|
||||||
prefix = self.__utils.get_hash(input_path.name)
|
|
||||||
out_file = Path(output_dir, f'{prefix}_{input_path.stem}.{extension}')
|
|
||||||
try:
|
|
||||||
(FFmpeg()
|
|
||||||
.input(input_path)
|
|
||||||
.option("hide_banner")
|
|
||||||
.output(out_file,{"b:a": bit_rate, "loglevel": "error"})
|
|
||||||
.execute()
|
|
||||||
)
|
|
||||||
except FFmpegError as e:
|
|
||||||
self.__utils.catch_unprocessed(input_path, out_file, e)
|
|
||||||
self.__printer.files(input_path, out_file, f"{bit_rate}")
|
|
||||||
return out_file
|
|
||||||
|
|
||||||
def image(self, input_path: Path, output_dir: Path, extension: str) -> Path:
|
|
||||||
"""
|
|
||||||
Method recodes image files to another format using Pillow
|
|
||||||
:param input_path: Path of the original image file
|
|
||||||
:param output_dir: Path of the output (compression) folder
|
|
||||||
:param extension: Extension of the new image file
|
|
||||||
:return: Path of compressed image file with md5 hash as prefix
|
|
||||||
"""
|
|
||||||
quality = self.__params.image_quality
|
|
||||||
prefix = self.__utils.get_hash(input_path.name)
|
|
||||||
out_file = Path(output_dir, f"{prefix}_{input_path.stem}.{extension}")
|
|
||||||
try:
|
|
||||||
image = Image.open(input_path)
|
|
||||||
|
|
||||||
if (extension == "jpg" or extension == "jpeg" or
|
|
||||||
(extension == "webp" and not self.__params.webp_rgba)):
|
|
||||||
if File.has_transparency(image):
|
|
||||||
self.__printer.warning(f"{input_path.name} has transparency. Changing to fallback...")
|
|
||||||
out_file = Path(output_dir, f"{prefix}_{input_path.stem}.{self.__params.image_fall_ext}")
|
|
||||||
|
|
||||||
if File.has_transparency(image):
|
|
||||||
image.convert('RGBA')
|
|
||||||
|
|
||||||
res_downscale = self.__params.image_downscale
|
|
||||||
if res_downscale != 1:
|
|
||||||
width, height = image.size
|
|
||||||
new_size = (int(width / res_downscale), int(height / res_downscale))
|
|
||||||
image = image.resize(new_size)
|
|
||||||
|
|
||||||
image.save(out_file,
|
|
||||||
optimize=True,
|
|
||||||
lossless=self.__params.image_lossless,
|
|
||||||
quality=quality,
|
|
||||||
minimize_size=True)
|
|
||||||
self.__printer.files(input_path, out_file, f"{quality}%")
|
|
||||||
except Exception as e:
|
|
||||||
self.__utils.catch_unprocessed(input_path, out_file, e)
|
|
||||||
return out_file
|
|
||||||
|
|
||||||
def video(self, input_path: Path, output_dir: Path, extension: str) -> Path:
|
|
||||||
"""
|
|
||||||
Method recodes video files to another format using ffmpeg utility
|
|
||||||
:param input_path: Path of the original video file
|
|
||||||
:param output_dir: Path of the output (compression) folder
|
|
||||||
:param extension: Extension of the new video file
|
|
||||||
:return: Path of compressed video file with md5 hash as prefix
|
|
||||||
"""
|
|
||||||
prefix = self.__utils.get_hash(input_path.name)
|
|
||||||
out_file = Path(output_dir, f'{prefix}_{input_path.stem}.{extension}')
|
|
||||||
if not self.__params.video_skip:
|
|
||||||
codec = self.__params.video_codec
|
|
||||||
crf = self.__params.video_crf
|
|
||||||
|
|
||||||
try:
|
|
||||||
(FFmpeg()
|
|
||||||
.input(input_path)
|
|
||||||
.option("hide_banner")
|
|
||||||
.option("hwaccel", "auto")
|
|
||||||
.output(out_file,{"codec:v": codec, "v:b": 0, "loglevel": "error"}, crf=crf)
|
|
||||||
.execute()
|
|
||||||
)
|
|
||||||
self.__printer.files(input_path, out_file, codec)
|
|
||||||
except FFmpegError as e:
|
|
||||||
self.__utils.catch_unprocessed(input_path, out_file, e)
|
|
||||||
else:
|
|
||||||
self.__utils.copy_unprocessed(input_path, out_file)
|
|
||||||
return out_file
|
|
||||||
|
|
||||||
def unknown(self, input_path: Path, output_dir: Path) -> Path:
|
|
||||||
"""
|
|
||||||
Method recodes files with "unknown" file format using ffmpeg,
|
|
||||||
in the hope that ffmpeg supports this file type and the default settings for it will reduce its size
|
|
||||||
:param input_path: Path of the original file
|
|
||||||
:param output_dir: Path of the output (compression) folder
|
|
||||||
:return: Path of compressed file with md5 hash as prefix
|
|
||||||
"""
|
|
||||||
prefix = self.__utils.get_hash(input_path.name)
|
|
||||||
out_file = Path(output_dir, f"{prefix}_{input_path.name}")
|
|
||||||
if self.__params.force_compress:
|
|
||||||
self.__printer.unknown_file(input_path.name)
|
|
||||||
try:
|
|
||||||
(FFmpeg()
|
|
||||||
.input(input_path)
|
|
||||||
.output(out_file)
|
|
||||||
.execute()
|
|
||||||
)
|
|
||||||
except FFmpegError as e:
|
|
||||||
self.__utils.catch_unprocessed(input_path, out_file, e)
|
|
||||||
else:
|
|
||||||
self.__utils.copy_unprocessed(input_path, out_file)
|
|
||||||
return out_file
|
|
||||||
|
|
||||||
def compress(self, source: Path, output: Path):
|
|
||||||
"""
|
|
||||||
It the core method for this program. Method determines file type and call compress function for it
|
|
||||||
:param source: Path of file to compress
|
|
||||||
:param output: Path of output file
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
match File.get_type(source):
|
|
||||||
case "audio":
|
|
||||||
out_file = self.audio(source, output, self.__params.audio_ext)
|
|
||||||
case "image":
|
|
||||||
out_file = self.image(source, output, self.__params.image_ext)
|
|
||||||
case "video":
|
|
||||||
out_file = self.video(source, output, self.__params.video_ext)
|
|
||||||
case "unknown":
|
|
||||||
out_file = self.unknown(source, output)
|
|
||||||
|
|
||||||
self.__utils.out_rename(out_file, source)
|
|
||||||
self.__printer.bar.update()
|
|
||||||
self.__printer.bar.next()
|
|
|
@ -1,109 +0,0 @@
|
||||||
from argparse import ArgumentParser, Namespace
|
|
||||||
from dataclasses import dataclass
|
|
||||||
from pathlib import Path
|
|
||||||
from typing import Self
|
|
||||||
import tomllib
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Params:
|
|
||||||
|
|
||||||
"""
|
|
||||||
This dataclass contains all parameters for utility
|
|
||||||
"""
|
|
||||||
|
|
||||||
copy_unprocessed: bool
|
|
||||||
force_compress: bool
|
|
||||||
mimic_mode: bool
|
|
||||||
hide_errors: bool
|
|
||||||
webp_rgba: bool
|
|
||||||
workers: int
|
|
||||||
|
|
||||||
audio_ext: str
|
|
||||||
audio_bitrate: str
|
|
||||||
|
|
||||||
image_downscale: int
|
|
||||||
image_ext: str
|
|
||||||
image_fall_ext: str
|
|
||||||
image_lossless: str
|
|
||||||
image_quality: int
|
|
||||||
|
|
||||||
video_crf: int
|
|
||||||
video_skip: bool
|
|
||||||
video_ext: str
|
|
||||||
video_codec: str
|
|
||||||
|
|
||||||
source: Path
|
|
||||||
dest: Path
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def setup(cls) -> Self:
|
|
||||||
"""
|
|
||||||
Method initialize all parameters and returns class instance
|
|
||||||
:return: Params instance
|
|
||||||
"""
|
|
||||||
args = cls.get_args()
|
|
||||||
if args.config is not None:
|
|
||||||
if Path(args.config).is_file():
|
|
||||||
with open(args.config, "rb") as cfile:
|
|
||||||
config = tomllib.load(cfile)
|
|
||||||
else:
|
|
||||||
print("Failed to find config. Check `vnrecode -h` to more info")
|
|
||||||
exit(255)
|
|
||||||
|
|
||||||
copy_unprocessed = config["FFMPEG"]["CopyUnprocessed"] if args.config else args.unproc
|
|
||||||
force_compress = config["FFMPEG"]["ForceCompress"] if args.config else args.force
|
|
||||||
mimic_mode = config["FFMPEG"]["MimicMode"] if args.config else args.mimic
|
|
||||||
hide_errors = config["FFMPEG"]["HideErrors"] if args.config else args.show_errors
|
|
||||||
workers = config["FFMPEG"]["Workers"] if args.config else args.jobs
|
|
||||||
webp_rgba = config["FFMPEG"]["WebpRGBA"] if args.config else args.webp_rgba
|
|
||||||
audio_ext = config["AUDIO"]["Extension"] if args.config else args.a_ext
|
|
||||||
audio_bitrate = config["AUDIO"]["BitRate"] if args.config else args.a_bit
|
|
||||||
image_downscale = config["IMAGE"]["ResDownScale"] if args.config else args.i_down
|
|
||||||
image_ext = config["IMAGE"]["Extension"] if args.config else args.i_ext
|
|
||||||
image_fall_ext = config["IMAGE"]["FallBackExtension"] if args.config else args.i_fallext
|
|
||||||
image_lossless = config["IMAGE"]["Lossless"] if args.config else args.i_lossless
|
|
||||||
image_quality = config["IMAGE"]["Quality"] if args.config else args.i_quality
|
|
||||||
video_crf = config["VIDEO"]["CRF"] if args.config else args.v_crf
|
|
||||||
video_skip = config["VIDEO"]["SkipVideo"] if args.config else args.v_skip
|
|
||||||
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 = Path(args.source)
|
|
||||||
dest = Path(f"{args.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, dest
|
|
||||||
)
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def get_args() -> Namespace:
|
|
||||||
"""
|
|
||||||
Method gets CLI arguments and returns argparse.Namespace instance
|
|
||||||
:return: argparse.Namespace of CLI args
|
|
||||||
"""
|
|
||||||
parser = ArgumentParser(prog="vnrecode",
|
|
||||||
description="Python utility to compress Visual Novel Resources"
|
|
||||||
)
|
|
||||||
parser.add_argument("source", help="Directory with game files to recode")
|
|
||||||
parser.add_argument("-c", dest='config', help="Utility config file")
|
|
||||||
parser.add_argument("-nu", dest='unproc', action='store_false', help="Don't copy unprocessed")
|
|
||||||
parser.add_argument("-f", "--force", action='store_true', help="Try to recode unknown files")
|
|
||||||
parser.add_argument("-nm", "--no-mimic", dest='mimic', action='store_false', help="Disable mimic mode")
|
|
||||||
parser.add_argument("-v", "--show_errors", action='store_false', help="Show recode errors")
|
|
||||||
parser.add_argument("--webp-rgb", dest='webp_rgba', action='store_false', help="Recode .webp without alpha channel")
|
|
||||||
parser.add_argument("-j", "--jobs", type=int, help="Number of threads", default=16)
|
|
||||||
parser.add_argument("-ae", dest="a_ext", help="Audio extension", default="opus")
|
|
||||||
parser.add_argument("-ab", dest="a_bit", help="Audio bit rate", default="128k")
|
|
||||||
parser.add_argument("-id", dest="i_down", type=int, help="Image resolution downscale multiplier", default=1)
|
|
||||||
parser.add_argument("-ie", dest="i_ext", help="Image extension", default="avif")
|
|
||||||
parser.add_argument("-ife", dest="i_fallext", help="Image fallback extension", default="webp")
|
|
||||||
parser.add_argument("-il", dest='i_lossless', action='store_false', help="Image losing compression mode")
|
|
||||||
parser.add_argument("-iq", dest="i_quality", type=int, help="Image quality", default=100)
|
|
||||||
parser.add_argument("--v_crf", help="Video CRF number", type=int, default=27)
|
|
||||||
parser.add_argument("-vs", dest="v_skip", action='store_true', help="Skip video recoding")
|
|
||||||
parser.add_argument("-ve", dest="v_ext", help="Video extension", default="webm")
|
|
||||||
parser.add_argument("-vc", dest="v_codec", help="Video codec name", default="libvpx-vp9")
|
|
||||||
args = parser.parse_args()
|
|
||||||
return args
|
|
|
@ -1,93 +0,0 @@
|
||||||
from progress.bar import IncrementalBar
|
|
||||||
from pathlib import Path
|
|
||||||
import colorama
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
import re
|
|
||||||
|
|
||||||
class Printer:
|
|
||||||
"""
|
|
||||||
Class implements CLI UI for this utility
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, source: Path):
|
|
||||||
"""
|
|
||||||
:param source: Path of original (compressing) folder to count its files for progress bar
|
|
||||||
"""
|
|
||||||
file_count = 0
|
|
||||||
for folder, folders, file in os.walk(source):
|
|
||||||
file_count += len(file)
|
|
||||||
self.bar = IncrementalBar('Compressing', max=file_count, suffix='[%(index)d/%(max)d] (%(percent).1f%%)')
|
|
||||||
self.bar.update()
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def clean_str(string: str) -> str:
|
|
||||||
"""
|
|
||||||
Method fills end of string with spaces to remove progress bar garbage from console
|
|
||||||
:param string: String to "clean"
|
|
||||||
:return: "Clean" string
|
|
||||||
"""
|
|
||||||
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
|
|
||||||
return string + " " * (os.get_terminal_size().columns - len(ansi_escape.sub('', string)))
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def win_ascii_esc():
|
|
||||||
"""
|
|
||||||
Method setups colorama for cmd
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if sys.platform == "win32":
|
|
||||||
colorama.init()
|
|
||||||
|
|
||||||
def bar_print(self, string: str):
|
|
||||||
"""
|
|
||||||
Method prints some string in console and updates progress bar
|
|
||||||
:param string: String to print
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
print(string)
|
|
||||||
self.bar.update()
|
|
||||||
|
|
||||||
def info(self, string: str):
|
|
||||||
"""
|
|
||||||
Method prints string with decor for info messages
|
|
||||||
:param string: String to print
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self.bar_print(self.clean_str(f"\r\033[100m- {string}\033[49m"))
|
|
||||||
|
|
||||||
def warning(self, string: str):
|
|
||||||
"""
|
|
||||||
Method prints string with decor for warning messages
|
|
||||||
:param string: String to print
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self.bar_print(self.clean_str(f"\r\033[93m!\033[0m {string}\033[49m"))
|
|
||||||
|
|
||||||
def error(self, string: str):
|
|
||||||
"""
|
|
||||||
Method prints string with decor for error messages
|
|
||||||
:param string: String to print
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self.bar_print(self.clean_str(f"\r\033[31m\u2715\033[0m {string}\033[49m"))
|
|
||||||
|
|
||||||
def files(self, source_path: Path, output_path: Path, comment: str):
|
|
||||||
"""
|
|
||||||
Method prints the result of recoding a file with some decorations in the form:
|
|
||||||
input file name -> output file name (quality setting)
|
|
||||||
:param source_path: Input file Path
|
|
||||||
:param output_path: Output file Path
|
|
||||||
:param comment: Comment about recode quality setting
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self.bar_print(self.clean_str(f"\r\033[0;32m\u2713\033[0m \033[0;37m{source_path.stem}\033[0m{source_path.suffix}\033[0;37m -> "
|
|
||||||
f"{source_path.stem}\033[0m{output_path.suffix}\033[0;37m ({comment})\033[0m"))
|
|
||||||
|
|
||||||
def unknown_file(self, filename: str):
|
|
||||||
"""
|
|
||||||
Method prints the result of recoding unknown file
|
|
||||||
:param filename: Name of unknown file
|
|
||||||
:return:
|
|
||||||
"""
|
|
||||||
self.bar_print(self.clean_str(f"\r\u2713 \033[0;33m{filename}\033[0m (File will be force compressed via ffmpeg)"))
|
|
|
@ -1,144 +0,0 @@
|
||||||
from shutil import copyfile
|
|
||||||
from pathlib import Path
|
|
||||||
import hashlib
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
|
|
||||||
from vnrecode.printer import Printer
|
|
||||||
from vnrecode.params import Params
|
|
||||||
|
|
||||||
|
|
||||||
class Utils:
|
|
||||||
"""
|
|
||||||
Class contains various methods for internal utility use
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, params_inst: Params, printer_inst: Printer):
|
|
||||||
self.__errors = 0
|
|
||||||
self.__params = params_inst
|
|
||||||
self.__printer = printer_inst
|
|
||||||
self.__duplicates = {}
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def sys_pause():
|
|
||||||
"""
|
|
||||||
Method calls pause for Windows cmd shell
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if sys.platform == "win32":
|
|
||||||
os.system("pause")
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
def get_hash(filename: str) -> str:
|
|
||||||
"""
|
|
||||||
Method returns 8 chars of md5 hash for filename
|
|
||||||
:param filename: File name to get md5
|
|
||||||
:return: 8 chars of md5 hash
|
|
||||||
"""
|
|
||||||
return hashlib.md5(filename.encode()).hexdigest()[:8]
|
|
||||||
|
|
||||||
def get_comp_subdir(self, folder: str) -> Path:
|
|
||||||
"""
|
|
||||||
Method returns the Path from str, changing the source folder in it to a compressed one
|
|
||||||
:param folder: source subfolder
|
|
||||||
:return: Path object with compressed subfolder
|
|
||||||
"""
|
|
||||||
return Path(folder.replace(str(self.__params.source), str(self.__params.dest), 1))
|
|
||||||
|
|
||||||
def get_recode_status(self):
|
|
||||||
"""
|
|
||||||
Method prints recoding results
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
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 file.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 = sum(file.stat().st_size for file in self.__params.source.glob('**/*') if file.is_file())
|
|
||||||
output = sum(file.stat().st_size for file in self.__params.dest.glob('**/*') if file.is_file())
|
|
||||||
|
|
||||||
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 catch_unprocessed(self, input_path: Path, output_path: Path, error):
|
|
||||||
"""
|
|
||||||
Method processes files that have not been recoded due to an error and prints error to console
|
|
||||||
if hide_errors parameter is False
|
|
||||||
:param input_path: Path of unprocessed file
|
|
||||||
:param output_path: Destination path of unprocessed file
|
|
||||||
:param error: Recoding exception
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
self.copy_unprocessed(input_path, output_path)
|
|
||||||
self.__errors += 1
|
|
||||||
if not self.__params.hide_errors:
|
|
||||||
self.__printer.error(f"File {input_path.name} can't be processed! Error: {error}")
|
|
||||||
|
|
||||||
def copy_unprocessed(self, input_path: Path, output_path: Path):
|
|
||||||
"""
|
|
||||||
Method copies an unprocessed file from the source folder to the destination folder
|
|
||||||
:param input_path: Path of unprocessed file
|
|
||||||
:param output_path: Destination path of unprocessed file
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if self.__params.copy_unprocessed:
|
|
||||||
copyfile(input_path, output_path)
|
|
||||||
self.__printer.info(f"File {input_path.name} copied to compressed folder.")
|
|
||||||
|
|
||||||
def catch_duplicates(self, path: Path) -> Path:
|
|
||||||
"""
|
|
||||||
Method checks if file path exists and returns folder/filename(vncopy).ext path
|
|
||||||
if duplicate founded
|
|
||||||
:param path: Some file Path
|
|
||||||
:return: Duplicate path name with (vncopy) on end
|
|
||||||
"""
|
|
||||||
if path.is_file() and path.exists():
|
|
||||||
orig_name = path.name.replace("(vncopy)", "")
|
|
||||||
new_path = Path(path.parent, path.stem + "(vncopy)" + path.suffix)
|
|
||||||
try: self.__duplicates[orig_name]
|
|
||||||
except KeyError: self.__duplicates[orig_name] = []
|
|
||||||
if not new_path.name in self.__duplicates[orig_name]:
|
|
||||||
self.__duplicates[orig_name].append(new_path.name)
|
|
||||||
return self.catch_duplicates(new_path)
|
|
||||||
return path
|
|
||||||
|
|
||||||
def print_duplicates(self):
|
|
||||||
"""
|
|
||||||
Method prints message about all duplicates generated during recode process
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
for filename in self.__duplicates.keys():
|
|
||||||
self.__printer.warning(
|
|
||||||
f'Duplicate file has been found! Check manually this files - "{filename}", ' +
|
|
||||||
', '.join(self.__duplicates[filename])
|
|
||||||
)
|
|
||||||
|
|
||||||
def out_rename(self, out_path: Path, target: Path):
|
|
||||||
"""
|
|
||||||
Method removes md5 hash from file name and changes file extension in dependence of mimic mode
|
|
||||||
:param out_path: Recoded file Path
|
|
||||||
:param target: Target filename
|
|
||||||
:return: None
|
|
||||||
"""
|
|
||||||
if not self.__params.mimic_mode:
|
|
||||||
dest_name = self.catch_duplicates(Path(out_path.parent, target.stem+out_path.suffix))
|
|
||||||
os.rename(out_path, dest_name)
|
|
||||||
else:
|
|
||||||
os.rename(out_path, Path(out_path.parent, target.name))
|
|
Loading…
Add table
Add a link
Reference in a new issue