From 641abcdd90cd25fe366c55065202693f54191f86 Mon Sep 17 00:00:00 2001 From: grglcy Date: Sat, 12 Jul 2025 12:45:50 +0100 Subject: [PATCH] remove transcode package --- src/transcode/__init__.py | 3 -- src/transcode/file.py | 59 ------------------------------ src/transcode/transcoder.py | 64 --------------------------------- src/transcode/worker.py | 71 ------------------------------------- 4 files changed, 197 deletions(-) delete mode 100644 src/transcode/__init__.py delete mode 100644 src/transcode/file.py delete mode 100644 src/transcode/transcoder.py delete mode 100644 src/transcode/worker.py diff --git a/src/transcode/__init__.py b/src/transcode/__init__.py deleted file mode 100644 index 50a1667..0000000 --- a/src/transcode/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .file import File -from .worker import Worker -from .transcoder import Transcoder diff --git a/src/transcode/file.py b/src/transcode/file.py deleted file mode 100644 index 11618ad..0000000 --- a/src/transcode/file.py +++ /dev/null @@ -1,59 +0,0 @@ -from pathlib import Path - - -audio_extensions = ['.flac'] -art_extensions = ['.jpg', '.jpeg', '.png'] - -class File: - def __init__(self, location: Path): - self.path = location - - def __str__(self): - return str(self.path) - - @property - def is_audio(self): - return self.extension in audio_extensions - - @property - def is_art(self): - return self.extension in art_extensions - - @property - def art_named_correctly(self): - return self.filename == "cover.jpg" - - @property - def filename(self) -> str: - return self.path.name - - @property - def stem(self) -> str: - return self.path.stem - - @property - def extension(self) -> str: - return self.path.suffix - - @property - def artist(self) -> Path: - return self.path.parent.parent - - def artist_out(self, root: Path) -> Path: - return root / self.artist.parts[-1] - - @property - def album(self) -> Path: - return self.path.parent - - def album_out(self, root: Path) -> Path: - return root /self.artist_out(root) / self.album.parts[-1] - - def artist_output(self, root: Path) -> Path: - return root / self.artist.parts[-1] - - def album_output(self, root: Path) -> Path: - return root / self.artist.parts[-1] / self.album.parts[-1] - - def output_file(self, root: Path, extension: str) -> Path: - return root / self.artist.parts[-1] / self.album.parts[-1] / f"{self.stem}.{extension}" diff --git a/src/transcode/transcoder.py b/src/transcode/transcoder.py deleted file mode 100644 index f511366..0000000 --- a/src/transcode/transcoder.py +++ /dev/null @@ -1,64 +0,0 @@ -from pathlib import Path -from multiprocessing import Pool, Manager, set_start_method -from log import Log, LogCat -from . import Worker - - -class Transcoder: - def __init__(self, encoder: Path, extension: str, input_root: Path, output_root: Path, log_path: Path): - self.encoder = encoder - self.extension = extension - self.input_root = input_root - self.output_root = output_root - - self.log_path = log_path - self.__log = Log(log_path) - self.log = LogCat(self.__log.queue, "TCD") - - def transcode(self): - transcode_list = [] - try: - for artist in self.input_root.iterdir(): - if artist.is_dir(): - for album in artist.iterdir(): - if album.is_dir(): - for file in album.iterdir(): - if file.is_file(): - if file.name == "DONE": - break - else: - transcode_list.append(file) - else: - self.log.warning("TRK", f"Warning, skipping non-dir '{album}' found in artist '{artist.parts[-1]}'") - continue - else: - self.log.warning("TRK", f"Warning, skipping non-dir '{artist}' found in root") - continue - self._transcode(transcode_list, self.encoder) - finally: - self.__log.stop() - - def _transcode(self, transcode_list: list, encoder: Path, workers=16): - manager = Manager() - queue = manager.Queue() - log = Log(self.log_path, queue) - logcat = LogCat(log.queue, "TCD") - args = [(str(self.output_root), self.extension, track, encoder, logcat) for track in transcode_list] - with Pool(workers) as pool: - pool.starmap(self.worker, args) - pool.close() - pool.join() - log.stop() - - def _transcode_single_thread(self, transcode_list: list, encoder: Path): - log = Log(self.log_path) - logcat = LogCat(log.queue, "TCD") - worker_args = [(track, encoder) for track in transcode_list] - for track, encoder in worker_args: - self.worker(str(self.output_root), self.extension, track, encoder, logcat) - log.stop() - - @staticmethod - def worker(output_root, extension, track, encoder, log): - w = Worker(output_root, extension) - w.transcode_worker(track, encoder, log) diff --git a/src/transcode/worker.py b/src/transcode/worker.py deleted file mode 100644 index 6e7b0a9..0000000 --- a/src/transcode/worker.py +++ /dev/null @@ -1,71 +0,0 @@ -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 \ No newline at end of file