nixfiles/musicutil/commands/copy_command.py

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:
    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)

        should_skip = False
        if self.args.skip_existing and exists:
            should_skip = True

        if should_skip:
            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 = ""

        if self.args.single_directory:
            dest_filepath = new_file.join_filename()
        else:
            dest_filepath = new_file.join_path_from_src()

        dest_filepath = self.args.dest + "/" + dest_filepath

        should_skip_transcode = self.args.skip_existing and path_exists(
            dest_filepath)

        if should_skip_transcode:
            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:
            # set can't contain duplicates
            # so it only creates a dir once
            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)

            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)