lib2flac.py - Paranoid-convert your sound library to FLAC

lib2flac.py

#!/usr/local/bin/python3
"""lib2flac

Usage:
    lib2flac.py [DEST] [SOURCE]...

Options:
    -h --help   Show this message.

"""

from subprocess import run
from os import (walk, makedirs)
from os.path import (join, basename, dirname, relpath, isdir, exists, splitext)
import sys
import json
import datetime

from docopt import docopt

from tqdm import tqdm

FFPROBE="/usr/local/bin/ffprobe"
FFMPEG="/usr/local/bin/ffmpeg"
FLAC="/usr/local/bin/flac"

def files_to_process(search_dirs):
    """
    A generator for all files to process in the search_dirs
    yields (relative_path, full_path, is_audio)
    """
    for arg in search_dirs:
        for root, _, filenames in tqdm(walk(arg), desc="Folder Scan..."):
            for filename in filenames:
                fullpath = join(root, filename)
                relative_path = relpath(fullpath, dirname(arg))
                yield relative_path, fullpath


class Reporter:
    def __init__(self, stream):
        self.stream = stream

    def write_event(self, event_type='generic', **kwargs):
        rec = kwargs
        rec.update( {'event': event_type,
                     'time': datetime.datetime.now().isoformat() } )
        

        json.dump(rec, self.stream)
        self.stream.write("\n")

    def write_completed_process(self, status):
        self.write_event('process_completed',   
                        returncode=status.returncode, 
                        stderr=status.stderr.decode('utf-8', errors='replace'),
                        stdout=status.stdout.decode('utf-8', errors='replace'), 
                        args=status.args)
 


def validate_flac_file(path, report):
    status = run([FLAC, '-t', path], capture_output=True)
    report.write_completed_process(status)


def validate_copy(file1, file2, report):
    status = run(['/usr/bin/diff', file1, file2], capture_output=True)
    report.write_completed_process(status)


def convert_to_flac(source_path, dest_path, report):
    status = run([FFMPEG, '-hide_banner', '-i', source_path, '-acodec', 'flac', dest_path], capture_output=True)
    report.write_completed_process(status)
    validate_flac_file(dest_path, report)


def copy(source_path, dest_path, report):
    status = run(['/bin/cp', '-p', source_path, dest_path], capture_output=True)
    report.write_completed_process(status)
    validate_copy(source_path, dest_path, report)
 

def process_files(from_paths, to_path, report):
    report.write_event('start_run')
    
    files_list = list(files_to_process(from_paths))
    
    def is_audio(apath):
        status = run([FFPROBE, '-show_streams', '-of', 'json', apath], capture_output=True)
        if status.returncode == 0:
            streams = json.loads(status.stdout)["streams"]
            audio_streams = [s for s in streams if s["codec_type"] == "audio"]
            return len(audio_streams) > 0

        else:
            return False

    for relative_source, full_source in tqdm(files_list, desc="Converting", unit='Files'):
        target_path = join(to_path, relative_source)
        target_dir = dirname(target_path)
        
        if not isdir(target_dir):
            report.write_event('create_directory', path=target_dir)
            makedirs(target_dir)

        
        if is_audio(full_source):
            s = splitext(target_path)
            new_path = s[0] + '.flac'
            if exists(new_path):
                report.write_event('convert_target_exists', path=target_path)
                continue 

            convert_to_flac(full_source, new_path, report)
        else:
            if exists(target_path):
                report.write_event('copy_target_exists', path=target_path)
                continue

            copy(full_source, target_path, report)



if __name__ == "__main__":
    arguments = docopt(__doc__, version=1.0)
    report_name = 'lib2flac_' + datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + '.json'
    report_path = join(arguments['DEST'], report_name)
    with open(report_path, 'w') as r:
        process_files(from_paths=arguments['SOURCE'], 
                      to_path=arguments['DEST'],
                      report=Reporter(r))