Compare commits
24 commits
0a9114ff64
...
626eaae5e2
Author | SHA1 | Date | |
---|---|---|---|
626eaae5e2 | |||
bc84703b73 | |||
a75314d2ad | |||
407ab98000 | |||
df20bd3636 | |||
1c1e8a9292 | |||
4e6fd332c5 | |||
9bb3cdcccb | |||
d8e55bac9a | |||
0b43756ef5 | |||
90a6b4e0c1 | |||
7433027cf3 | |||
8f9db132e6 | |||
b534214be9 | |||
a9aeb52506 | |||
44c12a5688 | |||
92474b4aa4 | |||
03647d4b84 | |||
a69b17c624 | |||
71c5764f26 | |||
f240fdca5f | |||
e5fa49ad53 | |||
85df574d3c | |||
7487eb94bd |
33 changed files with 905 additions and 507 deletions
5
.gitignore
vendored
5
.gitignore
vendored
|
@ -1 +1,6 @@
|
|||
/output/
|
||||
/tests/
|
||||
/tests_compressed/
|
||||
/build/
|
||||
/dist/
|
||||
/vntools.egg-info/
|
||||
|
|
|
@ -1,59 +0,0 @@
|
|||
#!/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}")
|
|
@ -1,160 +0,0 @@
|
|||
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()
|
|
@ -1,11 +0,0 @@
|
|||
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()
|
|
@ -1,51 +0,0 @@
|
|||
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()
|
|
@ -1,79 +0,0 @@
|
|||
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}"
|
27
README.md
27
README.md
|
@ -1,8 +1,25 @@
|
|||
## VNTools
|
||||
Collection of tools used by administrators from VN Telegram Channel
|
||||
Collection of tools used by VienDesu! Porting Team
|
||||
|
||||
### Tools
|
||||
* `FFMpeg-Compressor` - Python utility uses ffmpeg to compress Visual Novel Resources
|
||||
* `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-Unpacker` - Simple .rpy script that will make any RenPy game unpack itself
|
||||
* `VNDS-to-RenPy` - Simple script for converting vnds scripts to rpy
|
||||
* `vnrecode` - Python utility to compress Visual Novel Resources
|
||||
* `unrenapk` - A Python script for extracting game project from Ren'Py based .apk and .obb files
|
||||
* `renpy-ripper` - Simple .rpy script that will make any RenPy game unpack itself
|
||||
* `vnds2renpy` - Simple script for converting VNDS engine scripts to .rpy ones
|
||||
|
||||
### 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,2 +0,0 @@
|
|||
Pillow==9.5.0
|
||||
colorama==0.4.6
|
|
@ -1,108 +0,0 @@
|
|||
#!/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])
|
16
build.bat
Normal file → Executable file
16
build.bat
Normal file → Executable file
|
@ -3,16 +3,16 @@ if not defined VIRTUAL_ENV goto :venv_error
|
|||
|
||||
mkdir output
|
||||
mkdir output\bin
|
||||
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 -r requirements.txt || goto :exit
|
||||
python -m pip install Nuitka || 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 FFMpeg-Compressor\ffmpeg-comp.toml output\bin /Y
|
||||
move /Y output\ffmpeg-comp.exe output\bin
|
||||
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\rendroid-unpack.exe output\bin
|
||||
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=vnds2renpy VNDS-to-RenPy/convert.py || goto :exit
|
||||
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=vnrecode vnrecode\__main__.py || goto :exit
|
||||
xcopy vnrecode\vnrecode.toml output\bin /Y
|
||||
move /Y output\vnrecode.exe output\bin
|
||||
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=unrenapk unrenapk\__main__.py || goto :exit
|
||||
move /Y output\unrenapk.exe output\bin
|
||||
python -m nuitka --jobs=%NUMBER_OF_PROCESSORS% --output-dir=output --follow-imports --onefile --output-filename=vnds2renpy vnds2renpy/__main__.py || goto :exit
|
||||
move /Y output\vnds2renpy.exe output\bin
|
||||
echo "Done! You can get binaries into output\bin directory"
|
||||
|
||||
:venv_error
|
||||
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/bin
|
||||
python3 -m pip install -r FFMpeg-Compressor/requirements.txt
|
||||
python3 -m pip install -r RenPy-Android-Unpack/requirements.txt
|
||||
python3 -m pip install -r requirements.txt
|
||||
python3 -m pip install Nuitka
|
||||
case "$(uname -s)" in
|
||||
Linux*) jobs="--jobs=$(nproc)";;
|
||||
Darwin*) jobs="--jobs=$(sysctl -n hw.ncpu)";;
|
||||
esac
|
||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=ffmpeg-comp FFMpeg-Compressor/main.py
|
||||
cp FFMpeg-Compressor/ffmpeg-comp.toml output/bin
|
||||
mv output/ffmpeg-comp output/bin
|
||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=rendroid-unpack RenPy-Android-Unpack/unpack.py
|
||||
mv output/rendroid-unpack output/bin
|
||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=vnds2renpy VNDS-to-RenPy/convert.py
|
||||
mv output/vnds2renpy output/bin
|
||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=vnrecode vnrecode/__main__.py
|
||||
cp vnrecode/vnrecode.toml output/bin
|
||||
mv output/vnrecode output/bin
|
||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=unrenapk unrenapk/__main__.py
|
||||
mv output/unrenapk output/bin
|
||||
python3 -m nuitka "${jobs}" --output-dir=output --onefile --follow-imports --output-filename=vnds2renpy vnds2renpy/__main__.py
|
||||
mv output/vnds2renpy output/bin
|
||||
echo "Done! You can get binaries into output/bin directory"
|
32
pyproject.toml
Normal file
32
pyproject.toml
Normal file
|
@ -0,0 +1,32 @@
|
|||
[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"
|
||||
]
|
|
@ -2,4 +2,5 @@ Pillow==10.3.0
|
|||
pillow-avif-plugin==1.4.3
|
||||
python-ffmpeg==2.0.12
|
||||
progress==1.6
|
||||
colorama==0.4.6
|
||||
colorama==0.4.6
|
||||
argparse~=1.4.0
|
|
@ -1,6 +1,6 @@
|
|||
## RenPy-Android-Unpack
|
||||
## unrenapk
|
||||
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
|
||||
* Put some .apk & .obb files in folder
|
||||
* `rendroid-unpack` (It unpacks all .apk and .obb files in the directory where it is located)
|
||||
* `unrenapk` (It unpacks all .apk and .obb files in the directory where it is located)
|
0
unrenapk/__init__.py
Normal file
0
unrenapk/__init__.py
Normal file
6
unrenapk/__main__.py
Executable file
6
unrenapk/__main__.py
Executable file
|
@ -0,0 +1,6 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
from . import application
|
||||
|
||||
if __name__ == '__main__':
|
||||
application.launch()
|
98
unrenapk/actions.py
Executable file
98
unrenapk/actions.py
Executable file
|
@ -0,0 +1,98 @@
|
|||
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!")
|
51
unrenapk/application.py
Normal file
51
unrenapk/application.py
Normal file
|
@ -0,0 +1,51 @@
|
|||
#!/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!")
|
14
unrenapk/printer.py
Normal file
14
unrenapk/printer.py
Normal file
|
@ -0,0 +1,14 @@
|
|||
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,7 +1,7 @@
|
|||
## VNDS-to-RenPy
|
||||
## vnds2renpy
|
||||
Simple script for converting vnds scripts to rpy
|
||||
|
||||
### How to use
|
||||
|
||||
* Extract VNDS visual novel's archive and get `scripts` folder from these (or `script.zip` if scripts folder is empty)
|
||||
* Launch `convert.py` (It will automatically extract `scripts.zip` archive (if it needed) and converts .scr scripts to .rpy)
|
||||
* Launch `vnds2renpy` (It will automatically extract `scripts.zip` archive (if it needed) and converts .scr scripts to .rpy)
|
0
vnds2renpy/__init__.py
Normal file
0
vnds2renpy/__init__.py
Normal file
|
@ -1,13 +1,7 @@
|
|||
## FFMpeg-Compressor
|
||||
Python utility uses ffmpeg to compress Visual Novel Resources
|
||||
## vnrecode
|
||||
Python utility uses Pillow and ffmpeg to compress Visual Novel Resources
|
||||
|
||||
### 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
|
||||
### Configuration file
|
||||
#### 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`)
|
||||
* ForceCompress - Force try to compress all files in directory via ffmpeg. (default: `false`)
|
||||
|
@ -32,9 +26,34 @@ Python utility uses 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`
|
||||
* 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)
|
||||
* [x] Recreate whole game directory with compressed files
|
||||
* [x] Cross-platform (Easy Windows usage and binaries, macOS binaries)
|
||||
* [x] Use ffmpeg python bindings instead of cli commands
|
||||
* [x] Use ffmpeg python bindings instead of os.system
|
||||
* [x] Multithread
|
||||
* [ ] Reorganize code
|
0
vnrecode/__init__.py
Normal file
0
vnrecode/__init__.py
Normal file
23
vnrecode/__main__.py
Normal file
23
vnrecode/__main__.py
Normal file
|
@ -0,0 +1,23 @@
|
|||
#!/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()
|
59
vnrecode/application.py
Executable file
59
vnrecode/application.py
Executable file
|
@ -0,0 +1,59 @@
|
|||
#!/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}")
|
197
vnrecode/compress.py
Normal file
197
vnrecode/compress.py
Normal file
|
@ -0,0 +1,197 @@
|
|||
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()
|
109
vnrecode/params.py
Normal file
109
vnrecode/params.py
Normal file
|
@ -0,0 +1,109 @@
|
|||
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
|
93
vnrecode/printer.py
Normal file
93
vnrecode/printer.py
Normal file
|
@ -0,0 +1,93 @@
|
|||
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)"))
|
144
vnrecode/utils.py
Normal file
144
vnrecode/utils.py
Normal file
|
@ -0,0 +1,144 @@
|
|||
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