libconv/libconv.py

CUE_SHEET_EXTENSION: str = ".cue"

import os
import shutil
import logging
import subprocess
import sys
import re

from configparser import ConfigParser
from copy import deepcopy
from pathlib import Path
from typing import List, Optional, Tuple
from ffcuesplitter.cuesplitter import FFCueSplitter

CONFIG_FILE: str = sys.argv[1]

config = ConfigParser()
config.read(CONFIG_FILE)
SRC_FOLDER: str = config.get("source", "folder")
SRC_FILE_EXTS: List[str] = config.get("source", "file_exts").split(",")
SRC_ICON_EXTS: List[str] = config.get("source", "icon_exts").split(",")
DST_FOLDER: str = config.get("destination", "folder")
DST_FILE_EXT: str = config.get("destination", "file_ext")
DST_OVERWRITE: bool = config.get("libconv", "overwrite") == "true"
DRY_RUN: bool = config.get("libconv", "dry_run") != "false"
DEDUCE_METADATA: bool = config.get("libconv", "deduce_metadata") == "true"
FFMPEG_LOCATION: str = config.get("ffmpeg", "location")
FFMPEG_OPTS: str = config.get("ffmpeg", "options")

logging.basicConfig(level=logging.INFO)

# ----------------------------------------------------------------- #
# file and folder operations

def copy(src: str, dst: str):
    if (DRY_RUN):
        logging.info(f"copy {src} to {dst}")
    else:
        shutil.copyfile(src, dst, follow_symlinks=True)

def mkdir(path: str):
    if (DRY_RUN):
        logging.info(f"mkdir -r {path}")
    else:
        os.makedirs(path, exist_ok=True)

def execute(command: str):
    if (DRY_RUN):
        logging.info(f"execute: {command}")
    else:
        subprocess.run(command)

# ----------------------------------------------------------------- #
# cue sheet processing

def cue_sheet_processor(path: str):
    dst_folder = prepare_destination(path)
    sheet = get_files_with_ext(path, [CUE_SHEET_EXTENSION])

    if (len(sheet) != 1):
        logging.error(f"Expected exactly one but {path} contains {len(sheet)} cue sheets")
        return

    data = FFCueSplitter(sheet[0], dry=DRY_RUN, outputdir=dst_folder, suffix=DST_FILE_EXT, overwrite=DST_OVERWRITE, ffmpeg_cmd=FFMPEG_LOCATION, ffmpeg_add_params=FFMPEG_OPTS)
    data.open_cuefile()
    args = data.ffmpeg_arguments()["arguments"]
    dst_folder = prepare_destination(path)

    for arg in args:
        arg_end = arg.find("-y")
        filename = arg[arg_end + 5:-1]
        arg = arg[:arg_end]
        execute(f"{arg} \"{dst_folder}{filename}\"")

# ----------------------------------------------------------------- #
# basic "folder contains audio files" processing

def metadata_from_folder(path: str) -> str:
    # this method has to be adapted to your individual folder structure
    # if there is none then do nothing as by default this is not used
    if not DEDUCE_METADATA:
        return ""

    path = os.path.normpath(path)
    folders = path.split(os.sep)

    logging.debug(f"deducing metadata from folders: {folders}")

    album_name = re.sub(r'[ ]*\(.*?\)', '', folders[-1])
    comment = folders[-1].replace(album_name, "").replace("(", "").replace(")", "").replace(" ","")

    logging.debug(f"got metadata: album={album_name} comment={comment} artist={folders[-2]} genre={folders[-3]}")
    
    return f"-metadata ALBUM=\"{album_name}\" -metadata COMMENT=\"{comment}\" -metadata ARTIST=\"{folders[-2]}\" -metadata GENRE=\"{folders[-3]}\""

def metadata_from_file(filename: str) -> str:
    title=""
    number=""

    if re.match(r'[0-9]+[ ]*-[ ]*[.]*', filename):
        logging.debug(f"file {filename} matched metadata regex #1")
        number = re.search(r'[0-9]+[ ]*-[ ]*', filename).group()
        title = filename.replace(number, "")
        number = re.sub(r'[ ]*-[ ]*', '', number)
    elif re.match(r'[0-9]+\.[ ]*[.]*', filename):
        logging.info(f"file {filename} matched metadata regex #2")
        number = re.search(r'[0-9]+\.[ ]*', filename).group()
        title = filename.replace(number, "")
        number = re.sub(r'\.[ ]*', "", number)
    else:
        logging.debug(f"file {filename} matched no metadata regex")

    return f"-metadata TITLE=\"{title}\" -metadata TRACK=\"{number}\""

def file_processor(path: str):
    dst_folder = prepare_destination(path)
    files = get_files_with_ext(path, SRC_FILE_EXTS)
    album_metadata = metadata_from_folder(path)

    for file in files:
        filename = get_filename(file, path)
        file_metadata = metadata_from_file(filename[1:])
        execute(f"\"{FFMPEG_LOCATION}\" -i \"{file}\" {FFMPEG_OPTS} {album_metadata} {file_metadata} \"{dst_folder + filename}.{DST_FILE_EXT}\"")

# ----------------------------------------------------------------- #
# Iteration over library folders and preparation

def contains_extension(path: str, extensions: List[str]) -> bool:
    logging.debug(f"searching for extensions {extensions} in {path}")
    for root, dirs, files in os.walk(path):
        for file in files:
            for ext in extensions:
                if file.endswith(ext):
                    logging.debug(f"file {file} matched extension '{ext}'")
                    return True
        # needed to prevent recursive search
        break
    return False

def get_files_with_ext(path: str, extensions: List[str]) -> List[str]:
    logging.debug(f"obtainint all files with extensions {extensions} from {path}")
    ret_files = []
    for root, dirs, files in os.walk(path):
        for file in files:
            for ext in extensions:
                if file.endswith(ext):
                    logging.debug(f"file {file} matched extension '{ext}'")
                    ret_files.append(os.path.join(root, file))
                    # file is added now - no need to check the other extensions
                    break
        # needed to prevent recursive search
        break
    return ret_files

def prepare_destination(path: str) -> str:
    images = get_files_with_ext(path, SRC_ICON_EXTS)

    sub_path = path.replace(SRC_FOLDER, "")
    new_path = DST_FOLDER + sub_path
    mkdir(new_path)

    for img in images:
        img_name = img.replace(path, "")
        new_img_path = new_path + img_name
        copy(img, new_img_path)

    return new_path
        
def get_filename(file: str, path: str) -> str:
    fname = file.replace(path, "")
    for ext in SRC_FILE_EXTS:
        fname = fname.replace("." + ext, "")
    return fname

def scan_folder(path: str):
    for root, dirs, files in os.walk(path):
        for directory in dirs:
            process_folder(os.path.join(root, directory))
        # needed to prevent recursive search
        break

def process_folder(path: str):
    logging.info(f"Visiting folder '{path}'")
    if contains_extension(path, [CUE_SHEET_EXTENSION]):
        logging.info("found cuesheet")
        cue_sheet_processor(path)
    elif contains_extension(path, SRC_FILE_EXTS):
        logging.info("found audio files")
        file_processor(path)
    else:
        logging.info("scanning subfolders")
        scan_folder(path)

if __name__ == "__main__":
    logging.info(f"Using settings from {CONFIG_FILE}")
    if DRY_RUN:
        logging.info("Executing as DRY RUN")
    process_folder(SRC_FOLDER)