Convert FLAC and other files to 192k MP3s in current dir using Python and FFmpeg.

convert_audio.py

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import json
import logging
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Optional


TARGET_BITRATE_BPS = 192_000
TARGET_BITRATE_LABEL = "192k"

# Broad extension filter to avoid probing every random file.
# ffprobe still decides whether the file is actually valid audio.
AUDIO_EXTENSIONS = {
    ".aac",
    ".aiff",
    ".alac",
    ".ape",
    ".flac",
    ".m4a",
    ".mka",
    ".mp2",
    ".mp3",
    ".ogg",
    ".opus",
    ".wav",
    ".webm",
    ".wma",
}


@dataclass(frozen=True)
class AudioInfo:
    path: Path
    codec_name: str
    bit_rate_bps: Optional[int]
    duration_seconds: Optional[float]


@dataclass(frozen=True)
class ConversionJob:
    source: Path
    output: Path
    reason: str


def setup_logging(verbose: bool) -> None:
    log_level = logging.DEBUG if verbose else logging.INFO

    logging.basicConfig(
        level=log_level,
        format="%(asctime)s | %(levelname)-8s | %(message)s",
        datefmt="%H:%M:%S",
    )


def require_executable(name: str) -> None:
    if shutil.which(name) is None:
        raise RuntimeError(f"Required executable not found on PATH: {name}")


def run_child_process(command: list[str]) -> subprocess.CompletedProcess[str]:
    logging.debug("Running command: %s", " ".join(command))

    return subprocess.run(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )


def ffprobe_audio(path: Path) -> Optional[AudioInfo]:
    command = [
        "ffprobe",
        "-v",
        "error",
        "-hide_banner",
        "-print_format",
        "json",
        "-show_entries",
        "format=duration,bit_rate:stream=index,codec_type,codec_name,bit_rate",
        str(path),
    ]

    result = run_child_process(command)

    if result.returncode != 0:
        logging.debug("ffprobe rejected file: %s | %s", path, result.stderr.strip())
        return None

    try:
        payload = json.loads(result.stdout)
    except json.JSONDecodeError:
        logging.warning("Could not parse ffprobe JSON for: %s", path)
        return None

    streams = payload.get("streams", [])
    audio_streams = [
        stream for stream in streams if stream.get("codec_type") == "audio"
    ]

    if not audio_streams:
        return None

    audio_stream = audio_streams[0]
    codec_name = str(audio_stream.get("codec_name", "")).lower().strip()

    if not codec_name:
        logging.warning("Audio stream has no codec name: %s", path)
        return None

    stream_bitrate = parse_int(audio_stream.get("bit_rate"))
    format_bitrate = parse_int(payload.get("format", {}).get("bit_rate"))
    duration = parse_float(payload.get("format", {}).get("duration"))

    return AudioInfo(
        path=path,
        codec_name=codec_name,
        bit_rate_bps=stream_bitrate or format_bitrate,
        duration_seconds=duration,
    )


def parse_int(value: object) -> Optional[int]:
    if value is None:
        return None

    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return None

    return parsed if parsed > 0 else None


def parse_float(value: object) -> Optional[float]:
    if value is None:
        return None

    try:
        parsed = float(value)
    except (TypeError, ValueError):
        return None

    return parsed if parsed > 0 else None


def find_candidate_files(root: Path) -> list[Path]:
    return sorted(
        path
        for path in root.iterdir()
        if path.is_file()
        and not path.name.startswith(".")
        and path.suffix.lower() in AUDIO_EXTENSIONS
    )


def build_output_path(
    source: Path,
    output_dir: Optional[Path],
    suffix: str,
) -> Path:
    target_directory = output_dir if output_dir is not None else source.parent
    output_name = f"{source.stem}{suffix}.mp3"
    return target_directory / output_name


def classify_audio(
    audio_info: AudioInfo,
    output_dir: Optional[Path],
    replace_originals: bool,
) -> Optional[ConversionJob]:
    path = audio_info.path
    codec = audio_info.codec_name
    bitrate = audio_info.bit_rate_bps

    suffix = path.suffix.lower()

    is_mp3 = codec == "mp3" or suffix == ".mp3"
    is_flac = codec == "flac" or suffix == ".flac"
    is_webm_audio = suffix == ".webm" and codec in {"opus", "vorbis"}
    is_opus_file = suffix == ".opus" or codec == "opus"

    if is_mp3:
        if bitrate is not None and bitrate <= TARGET_BITRATE_BPS:
            logging.info(
                "Skipping MP3 already at/below 192 kbps: %s%s",
                path,
                f" ({bitrate // 1000} kbps)" if bitrate else "",
            )
            return None

        if replace_originals:
            output = path.with_suffix(".tmp-192k.mp3")
        else:
            output = build_output_path(path, output_dir, "-192k")

        if output.exists():
            logging.info("Skipping because output already exists: %s", output)
            return None

        return ConversionJob(
            source=path,
            output=output,
            reason=f"MP3 above 192 kbps ({bitrate // 1000} kbps)"
            if bitrate
            else "MP3 with unknown bitrate; converting conservatively",
        )

    if is_flac:
        if replace_originals:
            output = path.with_suffix(".mp3")
        else:
            output = build_output_path(path, output_dir, "")

        if output.exists():
            logging.info("Skipping because output already exists: %s", output)
            return None

        return ConversionJob(
            source=path,
            output=output,
            reason="FLAC converted to MP3",
        )

    if is_webm_audio or is_opus_file:
        output = build_output_path(path, output_dir, "")

        if output.exists():
            logging.info("Skipping because output already exists: %s", output)
            return None

        return ConversionJob(
            source=path,
            output=output,
            reason=f"{codec.upper()} audio converted to MP3"
            + (f" ({bitrate // 1000} kbps source)" if bitrate else ""),
        )

    logging.info("Skipping non-MP3/non-FLAC/non-WebM audio: %s (%s)", path, codec)
    return None


def convert_to_mp3(job: ConversionJob, replace_originals: bool) -> bool:
    logging.info("Processing: %s", job.source)
    logging.info("Reason: %s", job.reason)
    logging.info("Output: %s", job.output)

    job.output.parent.mkdir(parents=True, exist_ok=True)

    command = [
        "ffmpeg",
        "-hide_banner",
        "-loglevel",
        "error",
        "-nostats",
        "-y",
        "-i",
        str(job.source),
        "-map",
        "0:a:0",
        "-vn",
        "-codec:a",
        "libmp3lame",
        "-b:a",
        TARGET_BITRATE_LABEL,
        str(job.output),
    ]

    result = run_child_process(command)

    if result.returncode != 0:
        logging.error("Failed converting: %s", job.source)
        logging.error(result.stderr.strip() or "ffmpeg returned a non-zero exit code.")
        cleanup_partial_file(job.output)
        return False

    if replace_originals and job.source.suffix.lower() == ".mp3":
        try:
            job.output.replace(job.source)
            logging.info("Replaced original MP3: %s", job.source)
        except OSError as error:
            logging.error("Converted but failed replacing original: %s", error)
            return False

    logging.info("Converted successfully: %s", job.source)
    return True


def cleanup_partial_file(path: Path) -> None:
    try:
        if path.exists():
            path.unlink()
    except OSError as error:
        logging.warning("Could not delete partial output %s: %s", path, error)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Find valid audio files in the script's current directory only, "
            "then convert FLAC/high-bitrate MP3 files to 192 kbps MP3."
        )
    )

    parser.add_argument(
        "-o",
        "--output-dir",
        type=Path,
        default=None,
        help=(
            "Optional directory for converted files. "
            "If omitted, outputs are written next to source files."
        ),
    )

    parser.add_argument(
        "--replace-originals",
        action="store_true",
        help=(
            "Replace high-bitrate MP3 originals after successful conversion. "
            "FLAC files are not deleted; they produce .mp3 files."
        ),
    )

    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Show debug-level logging.",
    )

    return parser.parse_args()


def main() -> int:
    args = parse_args()
    setup_logging(args.verbose)

    try:
        require_executable("ffmpeg")
        require_executable("ffprobe")
    except RuntimeError as error:
        logging.error(str(error))
        return 1

    root = Path(__file__).resolve().parent

    output_dir = args.output_dir.expanduser().resolve() if args.output_dir else None

    logging.info("Scanning: %s", root)

    candidate_files = find_candidate_files(root)
    logging.info("Candidate files by extension: %d", len(candidate_files))

    valid_audio_files: list[AudioInfo] = []
    jobs: list[ConversionJob] = []

    for file_path in candidate_files:
        audio_info = ffprobe_audio(file_path)

        if audio_info is None:
            logging.info("Skipping invalid/non-audio file: %s", file_path)
            continue

        valid_audio_files.append(audio_info)

        job = classify_audio(
            audio_info=audio_info,
            output_dir=output_dir,
            replace_originals=args.replace_originals,
        )

        if job is not None:
            jobs.append(job)

    logging.info("Valid audio files found: %d", len(valid_audio_files))
    logging.info("\nTotal files to convert: %d", len(jobs))

    converted_count = 0
    failed_count = 0

    for job in jobs:
        if convert_to_mp3(job, replace_originals=args.replace_originals):
            converted_count += 1
        else:
            failed_count += 1

    logging.info("\nFinished.")
    logging.info("Total converted: %d", converted_count)
    logging.info("Total failed: %d", failed_count)
    logging.info("Total skipped: %d", len(valid_audio_files) - len(jobs))

    return 0 if failed_count == 0 else 2


if __name__ == "__main__":
    sys.exit(main())