george/whatopus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Separate transcoding functionality into class

Browse Source
This commit is contained in:
George Lacey 2025-01-18 10:31:26 +00:00
parent 422e7eb82b
commit c73217843b
3 changed files with 77 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/main.py
View File

@ -1,10 +1,6 @@
import argparse
import subprocess
import shutil
from subprocess import CalledProcessError
from pathlib import Path
from multiprocessing import Pool
from transcode import Transcoder
def get_args():
parser = argparse.ArgumentParser()
@ -14,62 +10,11 @@ def get_args():
return parser.parse_args()
def transcode(transcode_list: list, encoder: Path, workers=16):
worker_args = [(row[0], row[1], encoder) for row in transcode_list]
with Pool(workers) as p:
results = p.starmap_async(transcode_worker, worker_args)
p.close()
p.join()
for result in results.get():
print(result)
def transcode_worker(in_track, out_track, encoder):
enc_filename = encoder.parts[-1]
if enc_filename == "opusenc.exe":
additional_args = ('--bitrate', '128', '--music')
subprocess_args = (str(encoder),) + additional_args + (in_track, out_track)
elif enc_filename == "qaac64.exe":
additional_args = (in_track, '-o', out_track, '--threading')
subprocess_args = (str(encoder),) + additional_args
try:
subprocess.run(subprocess_args, capture_output=True, text=True, check=True)
return f"Transcoded '{in_track}' successfully."
except Exception:
return f"ERROR: Transcoding of '{in_track}' failed."
def main(input_dir: Path, output_dir: Path, encoder: Path, out_extension: str = 'opus'):
enc_filename = encoder.parts[-1]
if enc_filename == "qaac64.exe":
if encoder.parts[-1] == "qaac64.exe":
out_extension = "m4a"
transcode_list = []
file_whitelist = ['.jpg', '.jpeg', '.png']
for artist in input_dir.iterdir():
if artist.is_dir():
artist_out = Path(output_dir) / 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()
for file in album.iterdir():
if file.is_file() and file.suffix.lower() == '.flac':
transcode_list.append((str(file), str(album_out / f"{file.stem}.{out_extension}")))
elif file.is_file() and file.suffix.lower() in file_whitelist:
shutil.copy(file, album_out / file.name)
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
transcode(transcode_list, encoder)
transcoder = Transcoder(encoder, out_extension, input_dir, output_dir)
transcoder.transcode()
if __name__ == '__main__':

1
src/transcode/__init__.py Normal file
View File

@ -0,0 +1 @@
from .transcoder import Transcoder

72
src/transcode/transcoder.py Normal file
View File

@ -0,0 +1,72 @@
from pathlib import Path
import shutil
import subprocess
from multiprocessing import Pool
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 = []
file_whitelist = ['.jpg', '.jpeg', '.png']
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()
for file in album.iterdir():
if file.is_file() and file.suffix.lower() == '.flac':
transcode_list.append((str(file), str(album_out / f"{file.stem}.{self.extension}")))
elif file.is_file() and file.suffix.lower() in file_whitelist:
shutil.copy(file, album_out / file.name)
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(transcode_list, self.encoder)
def _transcode(self, transcode_list: list, encoder: Path, workers=16):
worker_args = [(row[0], row[1], encoder) for row 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 = [(row[0], row[1], encoder) for row in transcode_list]
for args in worker_args:
print(f"{args[0]}")
self.transcode_worker(*args)
def transcode_worker(self, in_track, out_track, encoder):
enc_filename = encoder.parts[-1]
if enc_filename == "opusenc.exe":
additional_args = ('--bitrate', '128', '--music')
subprocess_args = (str(encoder),) + additional_args + (in_track, out_track)
elif enc_filename == "qaac64.exe":
additional_args = (in_track, '-o', out_track, '--threading')
subprocess_args = (str(encoder),) + additional_args
try:
subprocess.run(subprocess_args, capture_output=True, text=True, check=True)
return f"Transcoded '{in_track}' successfully."
except Exception:
return f"ERROR: Transcoding of '{in_track}' failed."