ffmpeg

Module for converting audio files to different formats and checking for ffmpeg binary, and downloading it if not found.

FFmpegError

Bases: Exception

Base class for all exceptions related to FFmpeg.

async_convert(input_file, output_file, ffmpeg='ffmpeg', output_format='mp3', bitrate=None, ffmpeg_args=None, progress_handler=None) async

Convert the input file to the output file asynchronously with progress handler.

Arguments

• input_file: Path to input file or tuple of (url, file_format).
• output_file: Path to output file.
• ffmpeg: ffmpeg executable to use.
• output_format: output format.
• bitrate: constant/variable bitrate.
• ffmpeg_args: ffmpeg arguments.
• progress_handler: progress handler, has to accept an integer as argument.

Returns

• Tuple of conversion status and error dictionary.

Notes

• Make sure to check if ffmpeg is installed before calling this function.

Source code in spotdl/utils/ffmpeg.py

async def async_convert(
    input_file: Union[Path, Tuple[str, str]],
    output_file: Path,
    ffmpeg: str = "ffmpeg",
    output_format: str = "mp3",
    bitrate: Optional[str] = None,
    ffmpeg_args: Optional[str] = None,
    progress_handler: Optional[Callable[[int], None]] = None,
) -> Tuple[bool, Optional[Dict[str, Any]]]:
    """
    Convert the input file to the output file asynchronously with progress handler.

    ### Arguments
    - input_file: Path to input file or tuple of (url, file_format).
    - output_file: Path to output file.
    - ffmpeg: ffmpeg executable to use.
    - output_format: output format.
    - bitrate: constant/variable bitrate.
    - ffmpeg_args: ffmpeg arguments.
    - progress_handler: progress handler, has to accept an integer as argument.

    ### Returns
    - Tuple of conversion status and error dictionary.

    ### Notes
    - Make sure to check if ffmpeg is installed before calling this function.
    """

    loop = asyncio.get_running_loop()
    if sys.platform == "win32" and not isinstance(loop, asyncio.ProactorEventLoop):
        return await loop.run_in_executor(
            None,
            convert,
            input_file,
            output_file,
            ffmpeg,
            output_format,
            bitrate,
            ffmpeg_args,
            progress_handler,
        )

    arguments = _build_ffmpeg_arguments(
        input_file, output_file, output_format, bitrate, ffmpeg_args
    )

    process = await asyncio.create_subprocess_exec(
        ffmpeg,
        *arguments,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
        limit=2**20,
    )

    if not progress_handler:
        proc_out, _ = await process.communicate()

        if process.returncode != 0:
            version = get_ffmpeg_version(ffmpeg)
            message = proc_out.decode("utf-8", errors="replace") if proc_out else ""

            return False, {
                "return_code": process.returncode,
                "arguments": arguments,
                "ffmpeg": ffmpeg,
                "version": version[0],
                "build_year": version[1],
                "error": message,
            }

        return True, None

    progress_handler(0)

    out_buffer = []
    total_dur = None

    if process.stdout is not None:
        while True:
            try:
                out_line_bytes = await process.stdout.readline()
            except ValueError:
                # A single line exceeded the stream limit; readline drops the
                # buffered data, so skip the line and keep reading
                continue
            if not out_line_bytes:
                break

            out_line = out_line_bytes.decode("utf-8", errors="replace").strip()
            if out_line == "":
                continue

            out_buffer.append(out_line)

            total_dur_match = DUR_REGEX.search(out_line)
            if total_dur is None and total_dur_match:
                total_dur = to_ms(**total_dur_match.groupdict())  # type: ignore
                continue
            if total_dur:
                progress_time = TIME_REGEX.search(out_line)
                if progress_time:
                    elapsed_time = to_ms(**progress_time.groupdict())  # type: ignore
                    progress_handler(int(elapsed_time / total_dur * 100))  # type: ignore

    await process.wait()

    if process.returncode != 0:
        version = get_ffmpeg_version(ffmpeg)

        return False, {
            "return_code": process.returncode,
            "arguments": arguments,
            "ffmpeg": ffmpeg,
            "version": version[0],
            "build_year": version[1],
            "error": "\n".join(out_buffer),
        }

    progress_handler(100)

    return True, None

convert(input_file, output_file, ffmpeg='ffmpeg', output_format='mp3', bitrate=None, ffmpeg_args=None, progress_handler=None)

Convert the input file to the output file synchronously with progress handler.

Arguments

• input_file: Path to input file or tuple of (url, file_format).
• output_file: Path to output file.
• ffmpeg: ffmpeg executable to use.
• output_format: output format.
• bitrate: constant/variable bitrate.
• ffmpeg_args: ffmpeg arguments.
• progress_handler: progress handler, has to accept an integer as argument.

Returns

• Tuple of conversion status and error dictionary.

Notes

• Make sure to check if ffmpeg is installed before calling this function.

Source code in spotdl/utils/ffmpeg.py

def convert(
    input_file: Union[Path, Tuple[str, str]],
    output_file: Path,
    ffmpeg: str = "ffmpeg",
    output_format: str = "mp3",
    bitrate: Optional[str] = None,
    ffmpeg_args: Optional[str] = None,
    progress_handler: Optional[Callable[[int], None]] = None,
) -> Tuple[bool, Optional[Dict[str, Any]]]:
    """
    Convert the input file to the output file synchronously with progress handler.

    ### Arguments
    - input_file: Path to input file or tuple of (url, file_format).
    - output_file: Path to output file.
    - ffmpeg: ffmpeg executable to use.
    - output_format: output format.
    - bitrate: constant/variable bitrate.
    - ffmpeg_args: ffmpeg arguments.
    - progress_handler: progress handler, has to accept an integer as argument.

    ### Returns
    - Tuple of conversion status and error dictionary.

    ### Notes
    - Make sure to check if ffmpeg is installed before calling this function.
    """

    arguments = _build_ffmpeg_arguments(
        input_file, output_file, output_format, bitrate, ffmpeg_args
    )

    # Run ffmpeg
    with subprocess.Popen(
        [ffmpeg, *arguments],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        universal_newlines=False,
    ) as process:
        if not progress_handler:
            # Wait for process to finish
            proc_out = process.communicate()

            if process.returncode != 0:
                # get version and build year
                version = get_ffmpeg_version(ffmpeg)

                # join stdout and stderr and decode to utf-8
                message = b"".join([out for out in proc_out if out]).decode("utf-8")

                # return error dictionary
                return False, {
                    "return_code": process.returncode,
                    "arguments": arguments,
                    "ffmpeg": ffmpeg,
                    "version": version[0],
                    "build_year": version[1],
                    "error": message,
                }

            return True, None

        progress_handler(0)

        out_buffer = []
        total_dur = None
        while True:
            if process.stdout is None:
                continue

            out_line = (
                process.stdout.readline().decode("utf-8", errors="replace").strip()
            )

            if out_line == "" and process.poll() is not None:
                break

            out_buffer.append(out_line.strip())

            total_dur_match = DUR_REGEX.search(out_line)
            if total_dur is None and total_dur_match:
                total_dur = to_ms(**total_dur_match.groupdict())  # type: ignore
                continue
            if total_dur:
                progress_time = TIME_REGEX.search(out_line)
                if progress_time:
                    elapsed_time = to_ms(**progress_time.groupdict())  # type: ignore
                    progress_handler(int(elapsed_time / total_dur * 100))  # type: ignore

        if process.returncode != 0:
            # get version and build year
            version = get_ffmpeg_version(ffmpeg)

            return False, {
                "return_code": process.returncode,
                "arguments": arguments,
                "ffmpeg": ffmpeg,
                "version": version[0],
                "build_year": version[1],
                "error": "\n".join(out_buffer),
            }

        progress_handler(100)

        return True, None

download_ffmpeg()

Download ffmpeg binary to spotdl directory.

Returns

• Path to ffmpeg binary.

Notes

• ffmpeg is downloaded from github releases for current platform and architecture.
• executable permission is set for ffmpeg binary.

Source code in spotdl/utils/ffmpeg.py

def download_ffmpeg() -> Path:
    """
    Download ffmpeg binary to spotdl directory.

    ### Returns
    - Path to ffmpeg binary.

    ### Notes
    - ffmpeg is downloaded from github releases
        for current platform and architecture.
    - executable permission is set for ffmpeg binary.
    """

    os_name = platform.system().lower()
    os_arch = platform.machine().lower()
    ffmpeg_url: Optional[str] = None

    # if platform.system() == "Darwin" and (
    #     platform.processor() == "arm"
    #     or subprocess.run(["sysctl", "-n", "sysctl.proc_translated"], check=False)
    # ):
    #     ffmpeg_url = FFMPEG_URLS["darwin"]["arm"]
    # else:
    #     ffmpeg_url = FFMPEG_URLS.get(os_name, {}).get(os_arch)

    ffmpeg_url = FFMPEG_URLS.get(os_name, {}).get(os_arch)

    if ffmpeg_url is None:
        raise FFmpegError("FFmpeg binary is not available for your system.")

    ffmpeg_path = Path(
        os.path.join(
            get_spotdl_path(), "ffmpeg" + (".exe" if os_name == "windows" else "")
        )
    )

    # Download binary and save it to a file in spotdl directory
    ffmpeg_binary = requests.get(ffmpeg_url, allow_redirects=True, timeout=10).content
    with open(ffmpeg_path, "wb") as ffmpeg_file:
        ffmpeg_file.write(ffmpeg_binary)

    # Set executable permission on linux and mac
    if os_name in ["linux", "darwin"]:
        ffmpeg_path.chmod(ffmpeg_path.stat().st_mode | stat.S_IEXEC)

    return ffmpeg_path

get_ffmpeg_path()

Get path to global ffmpeg binary or a local ffmpeg binary.

Returns

• Path to ffmpeg binary or None if not found.

Source code in spotdl/utils/ffmpeg.py

def get_ffmpeg_path() -> Optional[Path]:
    """
    Get path to global ffmpeg binary or a local ffmpeg binary.

    ### Returns
    - Path to ffmpeg binary or None if not found.
    """

    # Check if ffmpeg is installed
    global_ffmpeg = shutil.which("ffmpeg")
    if global_ffmpeg:
        return Path(global_ffmpeg)

    # Get local ffmpeg path
    return get_local_ffmpeg()

get_ffmpeg_version(ffmpeg='ffmpeg') cached

Get ffmpeg version.

Arguments

• ffmpeg: ffmpeg executable to check

Returns

• Tuple of optional version and optional year.

Errors

• FFmpegError if ffmpeg is not installed.
• FFmpegError if ffmpeg version is not found.

Source code in spotdl/utils/ffmpeg.py

@functools.lru_cache(maxsize=None)
def get_ffmpeg_version(ffmpeg: str = "ffmpeg") -> Tuple[Optional[float], Optional[int]]:
    """
    Get ffmpeg version.

    ### Arguments
    - ffmpeg: ffmpeg executable to check

    ### Returns
    - Tuple of optional version and optional year.

    ### Errors
    - FFmpegError if ffmpeg is not installed.
    - FFmpegError if ffmpeg version is not found.
    """

    # Check if ffmpeg is installed
    if not is_ffmpeg_installed(ffmpeg):
        if ffmpeg == "ffmpeg":
            raise FFmpegError("ffmpeg is not installed.")

        raise FFmpegError(f"{ffmpeg} is not a valid ffmpeg executable.")

    with subprocess.Popen(
        [ffmpeg, "-version"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding="utf-8",
    ) as process:
        output = "".join(process.communicate())

    # Search for version and build year in output
    version_result = VERSION_REGEX.search(output)
    year_result = YEAR_REGEX.search(output)

    build_year = None
    version = None

    if version_result is not None:
        # remove all non numeric characters from string example: n4.3
        version_str = re.sub(r"[a-zA-Z]", "", version_result.group(0))

        # parse version string to float
        version = float(version_str) if version_str else None

    if year_result is not None:
        # get build years from string example: Copyright (c) 2019-2020
        build_years = [
            int(
                re.sub(r"[^0-9]", "", year)
            )  # remove all non numeric characters from string
            for year in year_result.group(0).split(
                "-"
            )  # split string into list of years
        ]

        # get the highest build year
        build_year = max(build_years)

    return (version, build_year)

get_local_ffmpeg()

Get local ffmpeg binary path.

Returns

• Path to ffmpeg binary or None if not found.

Source code in spotdl/utils/ffmpeg.py

def get_local_ffmpeg() -> Optional[Path]:
    """
    Get local ffmpeg binary path.

    ### Returns
    - Path to ffmpeg binary or None if not found.
    """

    ffmpeg_path = Path(get_spotdl_path()) / (
        "ffmpeg" + (".exe" if platform.system() == "Windows" else "")
    )

    if ffmpeg_path.is_file():
        return ffmpeg_path

    return None

is_ffmpeg_installed(ffmpeg='ffmpeg')

Check if ffmpeg is installed.

Arguments

• ffmpeg: ffmpeg executable to check

Returns

• True if ffmpeg is installed, False otherwise.

Source code in spotdl/utils/ffmpeg.py

def is_ffmpeg_installed(ffmpeg: str = "ffmpeg") -> bool:
    """
    Check if ffmpeg is installed.

    ### Arguments
    - ffmpeg: ffmpeg executable to check

    ### Returns
    - True if ffmpeg is installed, False otherwise.
    """

    if ffmpeg == "ffmpeg":
        global_ffmpeg = shutil.which("ffmpeg")
        if global_ffmpeg is None:
            ffmpeg_path = get_ffmpeg_path()
        else:
            ffmpeg_path = Path(global_ffmpeg)
    else:
        ffmpeg_path = Path(ffmpeg)

    if ffmpeg_path is None:
        return False

    # else check if path to ffmpeg is valid
    # and if ffmpeg has the correct access rights
    return ffmpeg_path.exists() and os.access(ffmpeg_path, os.X_OK)