Parallel flac to opus conversion script

transcode-opus.py

#!/usr/bin/env pipenv run python

import subprocess
from pathlib import Path
import sys
from tempfile import NamedTemporaryFile
import multiprocessing as mp


def usage():
    print("Usage: transcode-opus flac_path opus_path")
    sys.exit(1)


def transcode_worker(flac: Path, opus: Path):
    print(str(opus))
    opus.unlink(missing_ok=True)
    tmp = NamedTemporaryFile(suffix=".opus", delete=False)
    tmp_name = tmp.name
    subprocess.run(
        [
            "opusenc",
            "--quiet",
            "--bitrate",
            "128",
            "--discard-pictures",
            str(flac),
            str(tmp_name),
        ]
    )
    Path(tmp_name).rename(opus)


def prune_opus(flac_path: Path, opus_path: Path):
    opuses = opus_path.glob("**/*.opus")
    for opus in opuses:
        flac = flac_path.joinpath(*opus.parts[-3:]).with_suffix(".flac")
        if not flac.exists():
            print("clean", str(opus))
            opus.unlink()


def transcode(flac_path: Path, opus_path: Path):
    pool = mp.Pool(8)

    flacs = sorted(flac_path.glob("**/*.flac"))
    for flac in flacs:
        flac_time = flac.stat().st_mtime
        cover = opus_path.joinpath(*flac.parts[-3:-1], "cover.jpg")
        if not cover.exists():
            cover.parent.mkdir(parents=True, exist_ok=True)
            subprocess.run(["metaflac", "--export-picture-to", str(cover), str(flac)])
            print(cover)
        opus = opus_path.joinpath(*flac.parts[-3:]).with_suffix(".opus")
        if not opus.exists() or opus.stat().st_mtime < flac_time:
            opus.parent.mkdir(parents=True, exist_ok=True)
            pool.apply_async(transcode_worker, [flac, opus])

    pool.close()
    pool.join()


def main():
    if len(sys.argv) < 3:
        usage()
    [_, flac_root, opus_root] = sys.argv
    print(flac_root, opus_root)
    flac_path = Path(flac_root)
    opus_path = Path(opus_root)
    prune_opus(flac_path, opus_path)
    transcode(flac_path, opus_path)


if __name__ == "__main__":
    main()