71 lines
2.8 KiB
Python
71 lines
2.8 KiB
Python
import shutil
|
|
import subprocess
|
|
from . import File
|
|
from pathlib import Path
|
|
|
|
class Worker:
|
|
def __init__(self, output_root, extension):
|
|
self.output_root = Path(output_root)
|
|
self.extension = extension
|
|
|
|
def transcode_worker(self, track, encoder, log):
|
|
track = File(track)
|
|
if track.is_art:
|
|
return self.copy_album_art(track, log)
|
|
elif track.is_audio:
|
|
return self.transcode_audio(track, encoder, log)
|
|
else:
|
|
log.info("WRK", f"File {track.path} ignored")
|
|
|
|
def copy_album_art(self, file: File, log):
|
|
self.create_directories(file)
|
|
output_path = file.output_file(self.output_root, file.extension[1:])
|
|
if output_path.exists():
|
|
log.info("ART", f"Skipped {output_path}")
|
|
return
|
|
try:
|
|
shutil.copy(file.path, output_path)
|
|
log.info("ART", f"Successfully copied {output_path}")
|
|
except FileExistsError:
|
|
log.warning("ART", f"File exists: {output_path}")
|
|
except Exception:
|
|
log.error("ART", f"{file.path} failed to copy")
|
|
|
|
def transcode_audio(self, track, encoder, log):
|
|
track_output = track.output_file(self.output_root, self.extension)
|
|
if track_output.exists():
|
|
log.info("AUD", f"Skipped {track_output}")
|
|
return
|
|
self.create_directories(track)
|
|
encoder_args = self.encoder_args(encoder, track, log)
|
|
try:
|
|
subprocess.run(encoder_args, capture_output=True, text=True, check=True)
|
|
log.info("AUD", f"Transcoded '{track}' successfully.")
|
|
except Exception:
|
|
log.error("AUD", f"ERROR: Transcoding of '{track}' failed.")
|
|
|
|
def encoder_args(self, encoder, track, log):
|
|
track_output = track.output_file(self.output_root, self.extension)
|
|
enc_filename = encoder.parts[-1]
|
|
encoder_args = ""
|
|
if enc_filename == "opusenc.exe":
|
|
additional_args = ('--bitrate', '128', '--music')
|
|
encoder_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')
|
|
encoder_args = (str(encoder),) + additional_args
|
|
else:
|
|
log.error("AUD", f"Encoder {encoder} not recognised.")
|
|
return encoder_args
|
|
|
|
def create_directories(self, track: File):
|
|
if not track.artist_out(self.output_root).exists():
|
|
try:
|
|
track.artist_out(self.output_root).mkdir()
|
|
except FileExistsError:
|
|
pass
|
|
if not track.album_out(self.output_root).exists():
|
|
try:
|
|
track.album_out(self.output_root).mkdir()
|
|
except FileExistsError:
|
|
pass |