nix

configuration files that power my machines
Log | Files | Refs | README | LICENSE

f2opus (2790B)

      1 #!/usr/bin/env python3
      2 import argparse
      3 import concurrent.futures
      4 import os
      5 import subprocess
      6 from itertools import count
      7 from pathlib import Path
      8 
      9 FFMPEG_IN = (
     10     "ffmpeg",
     11     "-hide_banner",
     12     "-loglevel",
     13     "error",
     14     "-nostdin",
     15     "-threads",
     16     "1",
     17 )
     18 FFMPEG_OUT = (
     19     "-c:a",
     20     "libopus",
     21     "-b:a",
     22     "160k",
     23     "-vbr",
     24     "on",
     25     "-map_metadata",
     26     "0",
     27     "-f",
     28     "opus",
     29     "-y",
     30 )
     31 
     32 
     33 def convert(in_path, out_path, total, counter):
     34     tmp = out_path.with_name(out_path.name + ".part")
     35     if out_path.exists() and out_path.stat().st_size > 0:
     36         tmp.unlink(missing_ok=True)
     37         in_path.unlink()
     38         i = next(counter)
     39         print(f"[{i}/{total}] SKIP {in_path}")
     40         return True
     41 
     42     tmp.unlink(missing_ok=True)
     43     result = subprocess.run(
     44         [*FFMPEG_IN, "-i", str(in_path), *FFMPEG_OUT, str(tmp)],
     45         stdin=subprocess.DEVNULL,
     46         stdout=subprocess.DEVNULL,
     47         stderr=subprocess.PIPE,
     48     )
     49     i = next(counter)
     50     if result.returncode == 0 and tmp.exists() and tmp.stat().st_size > 0:
     51         os.replace(tmp, out_path)
     52         in_path.unlink()
     53         print(f"[{i}/{total}] OK   {in_path}")
     54         return True
     55 
     56     tmp.unlink(missing_ok=True)
     57     print(f"[{i}/{total}] FAIL {in_path}")
     58     if result.stderr:
     59         print(f"       {result.stderr.decode().strip().splitlines()[-1]}")
     60     return False
     61 
     62 
     63 def main():
     64     parser = argparse.ArgumentParser(
     65         description="Convert FLAC files to Opus in using ffmpeg"
     66     )
     67     parser.add_argument(
     68         "root", nargs="?", default=".", help="directory to search for FLAC files"
     69     )
     70     parser.add_argument(
     71         "workers",
     72         nargs="?",
     73         type=int,
     74         default=2 * (os.cpu_count() or 1),
     75         help="ffmpeg workers count",
     76     )
     77     args = parser.parse_args()
     78 
     79     root = Path(args.root).resolve()
     80     jobs = sorted(
     81         (p.resolve(), p.with_suffix(".opus").resolve()) for p in root.rglob("*.flac")
     82     )
     83     total = len(jobs)
     84     if total == 0:
     85         print("No FLAC files found")
     86         return
     87 
     88     print(f"Found {total} FLAC files in {root}")
     89     print(f"Workers: {args.workers}\n")
     90 
     91     counter = count(1)
     92     failed = []
     93 
     94     with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as ex:
     95         future_to_in = {
     96             ex.submit(convert, in_path, out_path, total, counter): in_path
     97             for in_path, out_path in jobs
     98         }
     99         for future in concurrent.futures.as_completed(future_to_in):
    100             if not future.result():
    101                 failed.append(future_to_in[future])
    102 
    103     print(f"\nDone: {total - len(failed)}/{total} converted")
    104     if failed:
    105         print(f"{len(failed)} failed:")
    106         for f in failed:
    107             print(f"  {f}")
    108 
    109 
    110 if __name__ == "__main__":
    111     main()