f2aif (8068B)
1 #!/usr/bin/env python3 2 3 import argparse 4 import subprocess 5 import sys 6 from concurrent.futures import ThreadPoolExecutor, as_completed 7 from dataclasses import dataclass 8 from pathlib import Path 9 10 11 @dataclass 12 class ConversionStats: 13 converted: int = 0 14 artwork_preserved: int = 0 15 no_artwork: int = 0 16 artwork_lost: int = 0 17 skipped: int = 0 18 errors: int = 0 19 deleted: int = 0 20 21 22 def check_artwork(file_path): 23 """Return True if the file contains an embedded artwork (video) stream.""" 24 result = subprocess.run( 25 [ 26 "ffprobe", 27 "-v", 28 "error", 29 "-select_streams", 30 "v", 31 "-show_entries", 32 "stream=codec_type", 33 "-of", 34 "default=noprint_wrappers=1:nokey=1", 35 str(file_path), 36 ], 37 capture_output=True, 38 text=True, 39 ) 40 return "video" in result.stdout 41 42 43 def convert_flac_to_aiff(input_path, delete_original=True): 44 """Convert a single FLAC file to AIFF. 45 46 Returns (success, status) where status is one of: 47 "artwork_preserved", "no_artwork", "artwork_lost", "skipped". 48 On failure returns (False, "error"). 49 """ 50 output_path = input_path.with_suffix(".aiff") 51 52 if output_path.exists(): 53 print(f"Skipping {input_path} - {output_path} already exists") 54 return True, "skipped" 55 56 print(f"Converting {input_path} to {output_path}") 57 58 try: 59 result = subprocess.run( 60 [ 61 "ffmpeg", 62 "-i", 63 str(input_path), 64 "-c:a", 65 "pcm_s16be", # 16-bit PCM (AIFF default audio codec) 66 "-map", 67 "0:a", # Map audio stream 68 "-map", 69 "0:v?", # Map video (cover art) stream if it exists 70 "-map_metadata", 71 "0", # Copy all metadata 72 "-write_id3v2", 73 "1", # Write ID3v2 chunk (required: AIFF muxer defaults to off) 74 "-id3v2_version", 75 "3", # ID3v2.3 for broad compatibility 76 "-f", 77 "aiff", 78 str(output_path), 79 ], 80 check=True, 81 capture_output=True, 82 ) 83 84 # Surface any non-fatal ffmpeg warnings (e.g. the optional 0:v? map 85 # producing a "no video stream" notice). 86 if result.stderr: 87 for line in result.stderr.decode(errors="replace").splitlines(): 88 if line.strip().lower().startswith("warning"): 89 print(f" ffmpeg warning: {line.strip()}") 90 91 had_artwork = check_artwork(input_path) 92 if had_artwork: 93 if check_artwork(output_path): 94 print(f"Successfully converted {input_path} (artwork preserved)") 95 status = "artwork_preserved" 96 else: 97 # Artwork existed in source but did not make it to the output. 98 # Keep the original so the user can investigate / re-encode. 99 print( 100 f"Warning: Artwork did not transfer for {input_path}; " 101 "keeping original" 102 ) 103 return True, "artwork_lost" 104 else: 105 print(f"Successfully converted {input_path}") 106 status = "no_artwork" 107 108 if delete_original: 109 try: 110 input_path.unlink() 111 print(f"Deleted original file: {input_path}") 112 except OSError as e: 113 print(f"Warning: Could not delete original file {input_path}: {str(e)}") 114 115 return True, status 116 117 except subprocess.CalledProcessError as e: 118 print(f"Error converting {input_path}:") 119 print(f"ffmpeg error: {e.stderr.decode(errors='replace')}") 120 return False, "error" 121 except Exception as e: 122 print(f"Unexpected error converting {input_path}: {str(e)}") 123 return False, "error" 124 125 126 def iter_flac_files(start_dir): 127 """Yield FLAC files under start_dir, case-insensitively. 128 129 pathlib's rglob is case-sensitive on Linux, so *.FLAC would be missed. 130 """ 131 lower = list(start_dir.rglob("*.flac")) 132 upper = list(start_dir.rglob("*.FLAC")) 133 # Deduplicate while preserving a stable order. 134 seen = set() 135 for p in lower + upper: 136 rp = p.resolve() 137 if rp not in seen: 138 seen.add(rp) 139 yield p 140 141 142 def main(): 143 parser = argparse.ArgumentParser( 144 description="Convert FLAC files to AIFF format recursively." 145 ) 146 parser.add_argument("folder_path", help="Path to the folder containing FLAC files") 147 parser.add_argument( 148 "--keep-original", 149 action="store_true", 150 help="Keep original FLAC files after conversion (default: delete)", 151 ) 152 parser.add_argument( 153 "--dry-run", 154 action="store_true", 155 help="Show what would be converted without writing or deleting anything", 156 ) 157 parser.add_argument( 158 "--jobs", 159 "-j", 160 type=int, 161 default=1, 162 help="Number of files to convert in parallel (default: 1)", 163 ) 164 args = parser.parse_args() 165 166 start_dir = Path(args.folder_path).resolve() 167 if not start_dir.exists(): 168 print(f"Error: The directory '{start_dir}' does not exist.") 169 return 1 170 if not start_dir.is_dir(): 171 print(f"Error: '{start_dir}' is not a directory.") 172 return 1 173 174 files = list(iter_flac_files(start_dir)) 175 if not files: 176 print(f"No FLAC files found under {start_dir}") 177 return 0 178 179 stats = ConversionStats() 180 delete_original = not args.keep_original 181 182 def handle(file_path): 183 return convert_flac_to_aiff(file_path, delete_original) 184 185 if args.dry_run: 186 for f in files: 187 print(f"[dry-run] Would convert {f} -> {f.with_suffix('.aiff')}") 188 print(f"\n{len(files)} file(s) would be converted.") 189 return 0 190 191 if args.jobs > 1: 192 with ThreadPoolExecutor(max_workers=args.jobs) as pool: 193 futures = {pool.submit(handle, f): f for f in files} 194 for fut in as_completed(futures): 195 f = futures[fut] 196 try: 197 success, status = fut.result() 198 except Exception as e: 199 print(f"Failed to process {f}: {str(e)}") 200 stats.errors += 1 201 continue 202 _tally(stats, success, status, f, delete_original) 203 else: 204 for f in files: 205 try: 206 success, status = handle(f) 207 except Exception as e: 208 print(f"Failed to process {f}: {str(e)}") 209 stats.errors += 1 210 continue 211 _tally(stats, success, status, f, delete_original) 212 213 print("\nConversion Complete!") 214 print(f"Successfully converted: {stats.converted} files") 215 if stats.artwork_preserved: 216 print(f" with artwork preserved: {stats.artwork_preserved}") 217 if stats.no_artwork: 218 print(f" without artwork: {stats.no_artwork}") 219 if stats.skipped: 220 print(f"Skipped (already exist): {stats.skipped} files") 221 if stats.artwork_lost: 222 print(f"Artwork transfer failed (originals kept): {stats.artwork_lost} files") 223 if not args.keep_original: 224 print(f"Original files deleted: {stats.deleted} files") 225 if stats.errors: 226 print(f"Errors encountered: {stats.errors} files") 227 228 return 1 if stats.errors else 0 229 230 231 def _tally(stats, success, status, file_path, delete_original): 232 if not success: 233 stats.errors += 1 234 return 235 if status == "skipped": 236 stats.skipped += 1 237 return 238 stats.converted += 1 239 if status == "artwork_preserved": 240 stats.artwork_preserved += 1 241 elif status == "no_artwork": 242 stats.no_artwork += 1 243 elif status == "artwork_lost": 244 stats.artwork_lost += 1 245 if delete_original and not file_path.exists() and status != "artwork_lost": 246 stats.deleted += 1 247 248 249 if __name__ == "__main__": 250 sys.exit(main())