utils

def calculate_file_checksum(
    filepath: str,
    checksum_algorithm: str,
    chunk_size: int = 10000,
) -> str:
    """Calculate file checksum for given algorithm.

    Args:
        filepath (str): Path to a file.
        checksum_algorithm (str): Algorithm to use. ('md5', 'sha1', 'sha256')
        chunk_size (int): Chunk size to read file.
            Defaults to 10000.

    Returns:
        str: Calculated checksum.

    Raises:
        ValueError: File not found or unknown checksum algorithm.

    """
    if not filepath:
        raise ValueError("Filepath is empty.")

    if not os.path.exists(filepath):
        raise ValueError(f"{filepath} doesn't exist.")

    if not os.path.isfile(filepath):
        raise ValueError(f"{filepath} is not a file.")

    func = getattr(hashlib, checksum_algorithm, None)
    if func is None:
        raise ValueError(
            f"Unknown checksum algorithm '{checksum_algorithm}'"
        )

    hash_obj = func()
    with open(filepath, "rb") as f:
        for chunk in iter(lambda: f.read(chunk_size), b""):
            hash_obj.update(chunk)
    return hash_obj.hexdigest()

def download_ffmpeg(
    progress: Optional[TransferProgress] = None,
):
    """Download ffmpeg from server.

    Todos:
        Add safeguard to avoid downloading of the file from multiple
            processes at once.

    Args:
        progress (ayon_api.TransferProgress): Keep track about download.

    """

    files_info = get_server_files_info()
    file_info = _find_file_info("ffmpeg", files_info)
    if file_info is None:
        raise ValueError((
            "Couldn't find ffmpeg source file for platform '{}'"
        ).format(platform.system()))

    dirpath = get_downloaded_ffmpeg_root()
    log.debug(f"Downloading ffmpeg into: '{dirpath}'")
    if not _download_file(file_info, dirpath, progress=progress):
        log.debug("Other processed already downloaded and extracted ffmpeg.")

    _FFmpegArgs.download_needed = False
    _FFmpegArgs.downloaded_root = NOT_SET

def extract_archive_file(
    archive_file: str,
    dst_folder: Optional[str] = None,
):
    """Extract archived file to a directory.

    Args:
        archive_file (str): Path to a archive file.
        dst_folder (Optional[str]): Directory where content will be extracted.
            By default, same folder where archive file is.

    """
    if not dst_folder:
        dst_folder = os.path.dirname(archive_file)

    archive_ext, archive_type = get_archive_ext_and_type(archive_file)

    print("Extracting {} -> {}".format(archive_file, dst_folder))
    if archive_type is None:
        _, ext = os.path.splitext(archive_file)
        raise ValueError((
            f"Invalid file extension \"{ext}\"."
            f" Expected {', '.join(IMPLEMENTED_ARCHIVE_FORMATS)}"
        ))

    if archive_type == "zip":
        zip_file = ZipFileLongPaths(archive_file)
        zip_file.extractall(dst_folder)
        zip_file.close()

    elif archive_type == "tar":
        if archive_ext == ".tar":
            tar_type = "r:"
        elif archive_ext.endswith(".xz"):
            tar_type = "r:xz"
        elif archive_ext.endswith(".gz"):
            tar_type = "r:gz"
        elif archive_ext.endswith(".bz2"):
            tar_type = "r:bz2"
        else:
            tar_type = "r:*"

        try:
            tar_file = tarfile.open(archive_file, tar_type)
        except tarfile.ReadError:
            raise ValueError("corrupted archive")

        tar_file.extractall(dst_folder)
        tar_file.close()

def get_archive_ext_and_type(
    archive_file: str
) -> Tuple[Optional[str], Optional[str]]:
    """Get archive extension and type.

    Args:
        archive_file (str): Path to archive file.

    Returns:
        Tuple[str, str]: Archive extension and type.

    """
    tmp_name = archive_file.lower()
    if tmp_name.endswith(".zip"):
        return ".zip", "zip"

    for ext in (
        ".tar",
        ".tgz",
        ".tar.gz",
        ".tar.xz",
        ".tar.bz2",
    ):
        if tmp_name.endswith(ext):
            return ext, "tar"

    return None, None

def get_ffmpeg_arguments(
    tool_name: "FFmpegToolname" = "ffmpeg"
) -> Optional[List[str]]:
    """Get arguments to run one of ffmpeg tools.

    Args:
        tool_name (FFmpegToolname): Name of
            tool for which arguments should be returned.

    Returns:
        list[str]: Path to OpenImageIO directory.

    """
    args = _FFmpegArgs.tools.get(tool_name, NOT_SET)
    if args is NOT_SET:
        args = _fill_ffmpeg_tool_args(tool_name)
    return copy.deepcopy(args)

def get_oiio_arguments(
    tool_name: "OIIOToolName" = "oiiotool"
) -> Optional[List[str]]:
    """Get arguments to run one of OpenImageIO tools.

    Possible OIIO tools:
        oiiotool, maketx, iv, iinfo, igrep, idiff, iconvert

    Args:
        tool_name (OIIOToolName): Name of OIIO tool.

    Returns:
        str: Path to zip info file.

    """
    args = _OIIOArgs.tools.get(tool_name, NOT_SET)
    if args is NOT_SET:
        args = _fill_oiio_tool_args(tool_name)
    return copy.deepcopy(args)

def get_server_files_info() -> List["ToolDownloadInfo"]:
    """Receive zip file info from server.

    Information must contain at least 'filename' and 'hash' with md5 zip
    file hash.

    Returns:
        list[dict[str, str]]: Information about files on server.

    """
    # Cache server files info, they won't change
    if _ThirdPartyCache.server_files_info is None:
        endpoint = _get_addon_endpoint()
        response = ayon_api.get(f"{endpoint}/files_info")
        response.raise_for_status()
        _ThirdPartyCache.server_files_info = response.data
    return copy.deepcopy(_ThirdPartyCache.server_files_info)

def is_ffmpeg_download_needed(
    addon_settings: Optional[Dict[str, Any]] = None
) -> bool:
    """Check if is download needed.

    Returns:
        bool: Should be config downloaded.

    """
    if _FFmpegArgs.download_needed is not None:
        return _FFmpegArgs.download_needed

    if addon_settings is None:
        addon_settings = get_addon_settings()
    ffmpeg_settings = addon_settings["ffmpeg"]
    download_needed = False
    if ffmpeg_settings["use_downloaded"]:
        # Check what is required by server
        ffmpeg_root = get_downloaded_ffmpeg_root()
        progress_info = {}
        if ffmpeg_root:
            progress_path = os.path.join(
                ffmpeg_root, DIST_PROGRESS_FILENAME
            )
            progress_info = _read_progress_file(progress_path)
        download_needed = progress_info.get("state") != "done"

    _FFmpegArgs.download_needed = download_needed
    return _FFmpegArgs.download_needed

def is_oiio_download_needed(
    addon_settings: Optional[Dict[str, Any]] = None
) -> bool:
    """Check if is download needed.

    Returns:
        bool: Should be config downloaded.

    """
    if _OIIOArgs.download_needed is not None:
        return _OIIOArgs.download_needed

    if addon_settings is None:
        addon_settings = get_addon_settings()
    oiio_settings = addon_settings["oiio"]

    download_needed = False
    if oiio_settings["use_downloaded"]:
        oiio_root = get_downloaded_oiio_root()
        progress_info = {}
        if oiio_root:
            progress_path = os.path.join(
                oiio_root, DIST_PROGRESS_FILENAME
            )
            progress_info = _read_progress_file(progress_path)
        download_needed = progress_info.get("state") != "done"
    _OIIOArgs.download_needed = download_needed
    return _OIIOArgs.download_needed

def validate_ffmpeg_args(args: List[str]) -> bool:
    """Validate ffmpeg arguments.

    Args:
        args (list[str]): ffmpeg arguments.

    Returns:
        bool: True if arguments are valid.

    """
    if not args:
        return False
    return _check_args_returncode(args + ["-version"])

def validate_file_checksum(
    filepath: str,
    checksum: str,
    checksum_algorithm: str,
) -> bool:
    """Validate file checksum.

    Args:
        filepath (str): Path to file.
        checksum (str): Hash of file.
        checksum_algorithm (str): Type of checksum.

    Returns:
        bool: Hash is valid/invalid.

    Raises:
        ValueError: File not found or unknown checksum algorithm.

    """
    return checksum == calculate_file_checksum(filepath, checksum_algorithm)

def validate_oiio_args(args: List[str]) -> bool:
    """Validate oiio arguments.

    Args:
        args (list[str]): oiio arguments.

    Returns:
        bool: True if arguments are valid.

    """
    if not args:
        return False
    return _check_args_returncode(args + ["--help"])