nix

o2aif (9519B)

---

 #!/usr/bin/env python3
 
 import argparse
 import subprocess
 import sys
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from pathlib import Path
 
 
 @dataclass
 class ConversionStats:
     converted: int = 0
     artwork_preserved: int = 0
     no_artwork: int = 0
     artwork_lost: int = 0
     skipped: int = 0
     errors: int = 0
     deleted: int = 0
 
 
 def check_artwork(file_path):
     """Return True if the file contains a video/artwork stream."""
     result = subprocess.run(
         [
             "ffprobe",
             "-v",
             "error",
             "-select_streams",
             "v",
             "-show_entries",
             "stream=codec_type,stream_disposition=attached_pic",
             "-of",
             "default=noprint_wrappers=1:nokey=1",
             str(file_path),
         ],
         capture_output=True,
         text=True,
     )
     output = result.stdout
     return "video" in output or "attached_pic=1" in output
 
 
 COVER_NAMES = ("cover", "folder", "front", "albumart", "album")
 COVER_EXTS = (".jpg", ".jpeg", ".png", ".bmp", ".gif", ".webp")
 
 
 def find_external_cover(input_path):
     """Return the path to a cover image next to the input, or None."""
     parent = input_path.parent
     for name in COVER_NAMES:
         for ext in COVER_EXTS:
             for variant in (name + ext, name.title() + ext, name.upper() + ext):
                 candidate = parent / variant
                 if candidate.is_file():
                     return candidate
     return None
 
 
 def convert_opus_to_aiff(input_path, delete_original=True):
     """Convert a single Opus file to AIFF.
 
     Returns (success, status) where status is one of:
     "artwork_preserved", "no_artwork", "artwork_lost", "skipped".
     On failure returns (False, "error").
     """
     output_path = input_path.with_suffix(".aiff")
 
     if output_path.exists():
         print(f"Skipping {input_path} - {output_path} already exists")
         return True, "skipped"
 
     print(f"Converting {input_path} to {output_path}")
 
     had_embedded_artwork = check_artwork(input_path)
     external_cover = None if had_embedded_artwork else find_external_cover(input_path)
 
     try:
         ffmpeg_cmd = [
             "ffmpeg",
             "-i",
             str(input_path),
         ]
 
         if external_cover:
             ffmpeg_cmd.extend(["-i", str(external_cover)])
 
         ffmpeg_cmd.extend(
             [
                 "-c:a",
                 "pcm_s16be",  # 16-bit PCM (AIFF default audio codec)
                 "-map",
                 "0:a",  # Map audio stream
                 "-map_metadata",
                 "0:s:0",  # Copy Vorbis-comment tags to global AIFF ID3 metadata
                 "-write_id3v2",
                 "1",  # Write ID3v2 chunk (required: AIFF muxer defaults to off)
                 "-id3v2_version",
                 "3",  # ID3v2.3 for broad compatibility
             ]
         )
 
         if had_embedded_artwork:
             ffmpeg_cmd.extend(["-map", "0:v?", "-c:v", "copy"])
         elif external_cover:
             ffmpeg_cmd.extend(["-map", "1:v", "-c:v", "copy"])
 
         ffmpeg_cmd.extend(["-f", "aiff", str(output_path)])
 
         result = subprocess.run(ffmpeg_cmd, check=True, capture_output=True)
 
         # Surface any non-fatal ffmpeg warnings.
         if result.stderr:
             for line in result.stderr.decode(errors="replace").splitlines():
                 if line.strip().lower().startswith("warning"):
                     print(f"  ffmpeg warning: {line.strip()}")
 
         has_output_artwork = check_artwork(output_path)
         if has_output_artwork:
             if had_embedded_artwork:
                 print(
                     f"Successfully converted {input_path} (embedded artwork preserved)"
                 )
                 status = "artwork_preserved"
             elif external_cover:
                 print(
                     f"Successfully converted {input_path} "
                     f"(external cover {external_cover.name} embedded)"
                 )
                 status = "artwork_preserved"
             else:
                 print(f"Successfully converted {input_path} (artwork preserved)")
                 status = "artwork_preserved"
         else:
             if had_embedded_artwork or external_cover:
                 # Artwork existed in source or as external file but did not
                 # make it to the output. Keep the original so the user can
                 # investigate / re-encode.
                 print(
                     f"Warning: Artwork did not transfer for {input_path}; "
                     "keeping original"
                 )
                 return True, "artwork_lost"
             print(f"Successfully converted {input_path}")
             status = "no_artwork"
 
         if delete_original:
             try:
                 input_path.unlink()
                 print(f"Deleted original file: {input_path}")
             except OSError as e:
                 print(f"Warning: Could not delete original file {input_path}: {str(e)}")
 
         return True, status
 
     except subprocess.CalledProcessError as e:
         print(f"Error converting {input_path}:")
         print(f"ffmpeg error: {e.stderr.decode(errors='replace')}")
         return False, "error"
     except Exception as e:
         print(f"Unexpected error converting {input_path}: {str(e)}")
         return False, "error"
 
 
 def iter_opus_files(start_dir):
     """Yield Opus files under start_dir, case-insensitively.
 
     pathlib's rglob is case-sensitive on Linux, so *.OPUS would be missed.
     """
     lower = list(start_dir.rglob("*.opus"))
     upper = list(start_dir.rglob("*.OPUS"))
     # Deduplicate while preserving a stable order.
     seen = set()
     for p in lower + upper:
         rp = p.resolve()
         if rp not in seen:
             seen.add(rp)
             yield p
 
 
 def main():
     parser = argparse.ArgumentParser(
         description="Convert Opus files to AIFF format recursively."
     )
     parser.add_argument("folder_path", help="Path to the folder containing Opus files")
     parser.add_argument(
         "--keep-original",
         action="store_true",
         help="Keep original Opus files after conversion (default: delete)",
     )
     parser.add_argument(
         "--dry-run",
         action="store_true",
         help="Show what would be converted without writing or deleting anything",
     )
     parser.add_argument(
         "--jobs",
         "-j",
         type=int,
         default=1,
         help="Number of files to convert in parallel (default: 1)",
     )
     args = parser.parse_args()
 
     start_dir = Path(args.folder_path).resolve()
     if not start_dir.exists():
         print(f"Error: The directory '{start_dir}' does not exist.")
         return 1
     if not start_dir.is_dir():
         print(f"Error: '{start_dir}' is not a directory.")
         return 1
 
     files = list(iter_opus_files(start_dir))
     if not files:
         print(f"No Opus files found under {start_dir}")
         return 0
 
     stats = ConversionStats()
     delete_original = not args.keep_original
 
     def handle(file_path):
         return convert_opus_to_aiff(file_path, delete_original)
 
     if args.dry_run:
         for f in files:
             print(f"[dry-run] Would convert {f} -> {f.with_suffix('.aiff')}")
         print(f"\n{len(files)} file(s) would be converted.")
         return 0
 
     if args.jobs > 1:
         with ThreadPoolExecutor(max_workers=args.jobs) as pool:
             futures = {pool.submit(handle, f): f for f in files}
             for fut in as_completed(futures):
                 f = futures[fut]
                 try:
                     success, status = fut.result()
                 except Exception as e:
                     print(f"Failed to process {f}: {str(e)}")
                     stats.errors += 1
                     continue
                 _tally(stats, success, status, f, delete_original)
     else:
         for f in files:
             try:
                 success, status = handle(f)
             except Exception as e:
                 print(f"Failed to process {f}: {str(e)}")
                 stats.errors += 1
                 continue
             _tally(stats, success, status, f, delete_original)
 
     print("\nConversion Complete!")
     print(f"Successfully converted: {stats.converted} files")
     if stats.artwork_preserved:
         print(f"  with artwork preserved: {stats.artwork_preserved}")
     if stats.no_artwork:
         print(f"  without artwork:        {stats.no_artwork}")
     if stats.skipped:
         print(f"Skipped (already exist): {stats.skipped} files")
     if stats.artwork_lost:
         print(f"Artwork transfer failed (originals kept): {stats.artwork_lost} files")
     if not args.keep_original:
         print(f"Original files deleted: {stats.deleted} files")
     if stats.errors:
         print(f"Errors encountered: {stats.errors} files")
 
     return 1 if stats.errors else 0
 
 
 def _tally(stats, success, status, file_path, delete_original):
     if not success:
         stats.errors += 1
         return
     if status == "skipped":
         stats.skipped += 1
         return
     stats.converted += 1
     if status == "artwork_preserved":
         stats.artwork_preserved += 1
     elif status == "no_artwork":
         stats.no_artwork += 1
     elif status == "artwork_lost":
         stats.artwork_lost += 1
     if delete_original and not file_path.exists() and status != "artwork_lost":
         stats.deleted += 1
 
 
 if __name__ == "__main__":
     sys.exit(main())