89 lines
3.7 KiB
Python
89 lines
3.7 KiB
Python
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
from multiprocessing import Pool
|
|
from . import Track
|
|
|
|
|
|
class Transcoder:
|
|
def __init__(self, encoder: Path, extension: str, input_root: Path, output_root: Path):
|
|
self.encoder = encoder
|
|
self.extension = extension
|
|
self.input_root = input_root
|
|
self.output_root = output_root
|
|
|
|
def transcode(self):
|
|
transcode_list = []
|
|
for artist in self.input_root.iterdir():
|
|
if artist.is_dir():
|
|
artist_out = Path(self.output_root) / artist.name
|
|
artist_out.mkdir()
|
|
for album in artist.iterdir():
|
|
if album.is_dir():
|
|
album_out = artist_out / album.parts[-1]
|
|
if album_out.exists():
|
|
print(f"Skipping '{artist.parts[-1]} / {album.parts[-1]}'")
|
|
else:
|
|
album_out.mkdir()
|
|
done_file = album / 'DONE'
|
|
open(done_file, 'a').close()
|
|
self.copy_album_art(album, album_out)
|
|
for file in album.iterdir():
|
|
if file.is_file() and file.suffix.lower() == '.flac':
|
|
transcode_list.append(Track(file))
|
|
else:
|
|
print(f"Warning, skipping non-dir '{album}' found in artist '{artist.parts[-1]}'")
|
|
continue
|
|
else:
|
|
print(f"Warning, skipping non-dir '{artist}' found in root")
|
|
continue
|
|
self._transcode_single_thread(transcode_list, self.encoder)
|
|
#self._transcode(transcode_list, self.encoder)
|
|
|
|
def _transcode(self, transcode_list: list, encoder: Path, workers=16):
|
|
worker_args = [(track, encoder) for track in transcode_list]
|
|
with Pool(workers) as p:
|
|
results = p.starmap_async(self.transcode_worker, worker_args)
|
|
p.close()
|
|
p.join()
|
|
for result in results.get():
|
|
print(result)
|
|
|
|
def _transcode_single_thread(self, transcode_list: list, encoder: Path):
|
|
worker_args = [(track, encoder) for track in transcode_list]
|
|
for args in worker_args:
|
|
print(f"{args[0]}")
|
|
self.transcode_worker(*args)
|
|
|
|
|
|
def transcode_worker(self, track, encoder):
|
|
enc_filename = encoder.parts[-1]
|
|
track_output = track.output_file(self.output_root, self.extension)
|
|
if enc_filename == "opusenc.exe":
|
|
additional_args = ('--bitrate', '128', '--music')
|
|
subprocess_args = (str(encoder),) + additional_args + (str(track.path), str(track_output))
|
|
elif enc_filename == "qaac64.exe":
|
|
additional_args = (str(track), '-o', str(track_output), '--threading')
|
|
subprocess_args = (str(encoder),) + additional_args
|
|
try:
|
|
subprocess.run(subprocess_args, capture_output=True, text=True, check=True)
|
|
return f"Transcoded '{track}' successfully."
|
|
except Exception:
|
|
return f"ERROR: Transcoding of '{track}' failed."
|
|
|
|
@staticmethod
|
|
def copy_album_art(in_dir: Path, out_dir: Path, file_whitelist: list = None):
|
|
if file_whitelist is None:
|
|
file_whitelist = ['.jpg', '.jpeg', '.png']
|
|
art_file_name = "cover.jpg"
|
|
|
|
art_found = False
|
|
for file in in_dir.iterdir():
|
|
if file.is_file() and file.suffix.lower() in file_whitelist:
|
|
if file.name == art_file_name:
|
|
art_found = True
|
|
shutil.copy(file, out_dir / file.name)
|
|
|
|
if not art_found:
|
|
print(f"Warning: {art_file_name} not found in {in_dir}")
|