206 lines
6.5 KiB
Python
206 lines
6.5 KiB
Python
from musicutil.utils.do_replaygain import do_replaygain
|
|
from ..types import File
|
|
from ..utils.scan_for_music import scan_for_music
|
|
from ..utils.load_tag_information import load_tag_information
|
|
from ..transcode_presets import print_transcode_presets, transcode_presets_list
|
|
from ..utils.transcoder import TranscodeConfig, transcode, get_transcode_config
|
|
|
|
from os import makedirs as make_directories
|
|
from os.path import exists as path_exists
|
|
from shutil import copy as copy_file
|
|
from copy import deepcopy as deep_copy
|
|
|
|
|
|
class CopyCommandState:
|
|
files: list[File] = []
|
|
transcoded_files: list[File] = []
|
|
|
|
class CopyCommandArgs():
|
|
src: str
|
|
dest: str
|
|
transcode_preset: str
|
|
custom_transcoder_config: str
|
|
single_directory: bool
|
|
skip_existing: bool
|
|
skip_replaygain: bool
|
|
|
|
def add_copy_command(subparsers):
|
|
copy_parser = subparsers.add_parser('copy')
|
|
copy_parser.add_argument(
|
|
'src',
|
|
type=str,
|
|
help='src base music directory')
|
|
copy_parser.add_argument(
|
|
'dest',
|
|
type=str,
|
|
help='dest music directory')
|
|
copy_parser.add_argument(
|
|
'--transcode-preset',
|
|
type=str,
|
|
help='transcode preset (copy = no transcode)',
|
|
default="copy")
|
|
copy_parser.add_argument(
|
|
'--custom-transcoder-config',
|
|
type=str,
|
|
help='custom transcoder config')
|
|
copy_parser.add_argument(
|
|
'--skip-existing',
|
|
action='store_true')
|
|
copy_parser.add_argument(
|
|
'--skip-replaygain',
|
|
action='store_true')
|
|
copy_parser.add_argument(
|
|
'--single-directory',
|
|
action='store_true')
|
|
|
|
def get_copy_args(args) -> CopyCommandArgs:
|
|
print(args)
|
|
command_args = CopyCommandArgs()
|
|
command_args.src = args.src
|
|
command_args.dest = args.dest
|
|
command_args.transcode_preset = args.transcode_preset
|
|
command_args.custom_transcoder_config = args.custom_transcoder_config
|
|
command_args.single_directory = args.single_directory
|
|
command_args.skip_existing = args.skip_existing
|
|
command_args.skip_replaygain = args.skip_replaygain
|
|
return command_args
|
|
|
|
class CopyCommand():
|
|
def __init__(self, args: CopyCommandArgs):
|
|
self.args = args
|
|
self.state = CopyCommandState()
|
|
|
|
def run(self):
|
|
if self.args.transcode_preset == "list":
|
|
print_transcode_presets()
|
|
exit()
|
|
|
|
print("Copying")
|
|
self.scan_for_music()
|
|
self.load_tag_information()
|
|
if self.args.single_directory:
|
|
self.check_for_collisions()
|
|
self.transcode_files()
|
|
if self.args.single_directory:
|
|
self.create_mappings()
|
|
if not self.args.skip_replaygain:
|
|
self.add_replaygain_tags()
|
|
|
|
def scan_for_music(self):
|
|
print("Scanning For Music")
|
|
self.state.files = scan_for_music(self.args.src)
|
|
|
|
def load_tag_information(self):
|
|
print("Loading Tag Information")
|
|
|
|
for file in self.state.files:
|
|
file.tags = load_tag_information(file)
|
|
|
|
def check_for_collisions(self):
|
|
print("Checking For Colisions")
|
|
seen = set()
|
|
dupes = []
|
|
|
|
for file in self.state.files:
|
|
filename = file.filename
|
|
if filename in seen:
|
|
dupes.append(filename)
|
|
else:
|
|
seen.add(filename)
|
|
|
|
if len(dupes) > 0:
|
|
print("Dupes Found:", dupes)
|
|
print("Cannot continue using --single-directory")
|
|
print("Please rename or remove duplicates")
|
|
exit()
|
|
|
|
def _transcode_copy(self, file: File):
|
|
src = file.join_path_to()
|
|
dest = file.join_filename(
|
|
) if self.args.single_directory else file.join_path_from_src()
|
|
dest = self.args.dest + "/" + dest
|
|
|
|
exists = path_exists(dest)
|
|
|
|
if (self.args.skip_existing and not exists) or not self.args.skip_existing:
|
|
print("Copying", src, "to", dest)
|
|
copy_file(
|
|
src,
|
|
dest,
|
|
)
|
|
else:
|
|
print(
|
|
"Skipping",
|
|
src,
|
|
"as already is copied at",
|
|
dest)
|
|
|
|
self.state.transcoded_files.append(file)
|
|
|
|
def _transcode_with_config(self, file: File, trans_config: TranscodeConfig):
|
|
src = file.join_path_to()
|
|
|
|
new_file = deep_copy(file)
|
|
new_file.extension = trans_config.file_extension
|
|
|
|
dest_filepath = new_file.join_filename(
|
|
) if self.args.single_directory else new_file.join_path_from_src()
|
|
dest_filepath = self.args.dest + "/" + dest_filepath
|
|
|
|
if (self.args.skip_existing and path_exists(dest_filepath)):
|
|
print("Skipping transcoding", dest_filepath)
|
|
self.state.transcoded_files.append(new_file)
|
|
return
|
|
|
|
print("Transcoding", src, "to", dest_filepath)
|
|
|
|
transcode(file, trans_config, dest_filepath)
|
|
|
|
self.state.transcoded_files.append(new_file)
|
|
|
|
def transcode_files(self):
|
|
print("Transcoding Files")
|
|
|
|
if not self.args.single_directory:
|
|
directories = set()
|
|
for file in self.state.files:
|
|
directories.add(file.path_from_src)
|
|
for dir in directories:
|
|
make_directories(
|
|
self.args.dest + "/" + dir, exist_ok=True)
|
|
|
|
is_transcode_config_set = self.args.custom_transcoder_config is not None
|
|
|
|
if self.args.transcode_preset == "copy" and not is_transcode_config_set:
|
|
for file in self.state.files:
|
|
self._transcode_copy(file)
|
|
return
|
|
else:
|
|
global trans_config
|
|
trans_config = None
|
|
if is_transcode_config_set:
|
|
with open(self.args.custom_transcoder_config, "r+") as f:
|
|
trans_config = TranscodeConfig()
|
|
trans_config.load_from_file(f)
|
|
else:
|
|
trans_config = get_transcode_config(self.args.transcode_preset)
|
|
|
|
print(trans_config)
|
|
|
|
for file in self.state.files:
|
|
self._transcode_with_config(
|
|
file, trans_config)
|
|
|
|
def create_mappings(self):
|
|
with open(self.dest + "/" + "mappings.txt", "w") as f:
|
|
f.write("\n".join([
|
|
f"{file.path_from_src} <- {file.filename}" for file in self.state.files
|
|
]))
|
|
|
|
def add_replaygain_tags(self):
|
|
print("Adding ReplayGain Tags")
|
|
|
|
for file in self.state.files:
|
|
print(f"Adding ReplayGain Tags to \"{file.filename}.{file.extension}\"")
|
|
|
|
do_replaygain(file) |