utils

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

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

    Returns:
        str: Calculated checksum.

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

    """
    import hashlib

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

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

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

    func = getattr(hashlib, checksum_algorithm, None)
    if func is None:
        raise ValueError(
            "Unknown checksum algorithm '{}'".format(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 cleanup_executables_info():
    """Remove executables that do not exist anymore."""

    info = get_executables_info(check_cleanup=False)
    available_versions = info.setdefault("available_versions", [])

    new_executables = []
    for item in available_versions:
        executable = item.get("executable")
        if not executable or not os.path.exists(executable):
            continue

        version = load_executable_version(executable)
        if version and item.get("version") != version:
            item["version"] = version
        new_executables.append(item)

    info["available_versions"] = new_executables
    info["last_cleanup"] = {
        "value": datetime.datetime.now().strftime(DATE_FMT),
        "fmt": DATE_FMT,
    }
    store_executables_info(info)

def deploy_ayon_launcher_shims(
    ensure_protocol_is_registered: bool = False,
    create_desktop_icons: bool = False,
):
    """Deploy shim executables for AYON launcher.

    Argument 'validate_registers' is to fix registered protocol on Windows,
        issue caused in v1.1.0, since 1.1.1 is used different registry path.

    Args:
        ensure_protocol_is_registered (bool): Validate if protocol is
            registered on windows.
        create_desktop_icons (bool): Create desktop shortcuts. Used
            only on windows.

    """
    if not IS_BUILT_APPLICATION:
        return

    # Validate platform name
    platform_name = platform.system().lower()
    if platform_name not in ("windows", "linux", "darwin"):
        raise ValueError("Unsupported platform {}".format(platform_name))

    _cleanup_shims()

    executable_root = os.path.dirname(sys.executable)
    installer_shim_root = os.path.join(executable_root, "shim")

    with open(os.path.join(installer_shim_root, "shim.json"), "r") as stream:
        shim_data = json.load(stream)

    src_shim_version = semver.VersionInfo.parse(shim_data["version"])

    # Read existing shim version (if there is any)
    dst_shim_version = _get_installed_shim_version()

    # Skip if shim is same or lower
    if src_shim_version <= semver.VersionInfo.parse(dst_shim_version):
        # Make sure windows registers are correctly set for each user
        if ensure_protocol_is_registered:
            register_ayon_launcher_protocol()
        if platform_name == "windows" and create_desktop_icons:
            _create_windows_shortcut()
        return

    platform_name = platform.system().lower()
    if platform_name == "windows":
        _deploy_shim_windows(
            installer_shim_root,
            create_desktop_icons,
        )

    elif platform_name == "linux":
        _deploy_shim_linux(installer_shim_root)

    elif platform_name == "darwin":
        _deploy_shim_macos(installer_shim_root)
    register_ayon_launcher_protocol()

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)
        try:
            zip_file.extractall(dst_folder)
        finally:
            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 SystemExit("corrupted archive")

        try:
            tar_file.extractall(dst_folder)
        finally:
            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_ayon_appdirs(*args):
    """Local app data directory of AYON launcher.

    Deprecated:
        The function was replaced with 'get_launcher_local_dir'
            or 'get_launcher_storage_dir' based on usage.
        Deprecated since 1.1.0 .

    Args:
        *args (Iterable[str]): Subdirectories/files in local app data dir.

    Returns:
        str: Path to directory/file in local app data dir.

    """
    warnings.warn(
        (
            "Function 'get_ayon_appdirs' is deprecated. Should be replaced"
            " with 'get_launcher_local_dir' or 'get_launcher_storage_dir'"
            " based on use-case."
        ),
        DeprecationWarning
    )
    return _get_ayon_appdirs(*args)

def get_ayon_launch_args(*args: str) -> List[str]:
    """Launch arguments that can be used to launch ayon process.

    Args:
        *args (str): Additional arguments.

    Returns:
        list[str]: Launch arguments.

    """
    output = [sys.executable]
    if not IS_BUILT_APPLICATION:
        output.append(os.path.join(os.environ["AYON_ROOT"], "start.py"))
    output.extend(args)
    return output

def get_downloads_dir() -> str:
    """Downloads directory path.

    Each platform may use different approach how the downloads directory is
    received. This function will try to find the directory and return it.

    Returns:
        Union[str, None]: Path to downloads directory or None if not found.

    """
    if _Cache.downloads_dir != 0:
        return _Cache.downloads_dir

    path = None
    try:
        platform_name = platform.system().lower()
        if platform_name == "linux":
            path = _get_linux_downloads_dir()
        elif platform_name == "windows":
            path = _get_windows_downloads_dir()
        elif platform_name == "darwin":
            path = _get_macos_downloads_dir()

    except Exception:
        pass

    if path is None:
        default = os.path.expanduser("~/Downloads")
        if os.path.exists(default):
            path = default

    _Cache.downloads_dir = path
    return path

def get_executable_paths_by_version(version: str) -> List[str]:
    """Get executable paths by version.

    Returns:
        list[str]: Paths to executables.

    """
    return [
        item["executable"]
        for item in get_executables_info_by_version(version, validate=True)
    ]

def get_executables_info_by_version(
    version: str, validate: Optional[bool] = True
) -> List[Dict[str, Any]]:
    """Get available executable info by version.

    Args:
        version (str): Version of executable.
        validate (bool): Validate if 'version.py' contains same version.

    Returns:
        list[dict[str, Any]]: Executable info matching version.

    """
    info = get_executables_info()
    available_versions = info.setdefault("available_versions", [])
    if validate:
        _available_versions = []
        for item in available_versions:
            executable = item.get("executable")
            if not executable or not os.path.exists(executable):
                continue

            executable_version = load_executable_version(executable)
            if executable_version == version:
                _available_versions.append(item)
        available_versions = _available_versions
    return [
        item
        for item in available_versions
        if item.get("version") == version
    ]

def get_executables_info_filepath() -> str:
    """Get path to file where information about executables is stored.

    Returns:
        str: Path to json file where executables info are stored.

    """
    return get_launcher_local_dir("executables.json")

def get_launcher_local_dir(
    *subdirs: str,
    create: Optional[bool] = False
) -> str:
    """Get local directory for launcher.

    Local directory is used for storing machine or user specific data.

    The location is user specific.

    Note:
        This function should be called at least once on bootstrap.

    Args:
        *subdirs (str): Subdirectories relative to local dir.
        create (Optional[bool]): Create the folder if it does not exist.

    Returns:
        str: Path to local directory.

    """
    storage_dir = os.getenv("AYON_LAUNCHER_LOCAL_DIR")
    if not storage_dir:
        storage_dir = _get_ayon_appdirs()
        os.environ["AYON_LAUNCHER_LOCAL_DIR"] = storage_dir

    path = os.path.join(storage_dir, *subdirs)
    if create:
        os.makedirs(path, exist_ok=True)
    return path

def get_launcher_storage_dir(
    *subdirs: str,
    create: Optional[bool] = False
) -> str:
    """Get storage directory for launcher.

    Storage directory is used for storing shims, addons, dependencies, etc.

    It is not recommended, but the location can be shared across
        multiple machines.

    Note:
        This function should be called at least once on bootstrap.

    Args:
        *subdirs (str): Subdirectories relative to storage dir.
        create (Optional[bool]): Create the folder if it does not exist.

    Returns:
        str: Path to storage directory.

    """
    storage_dir = os.getenv("AYON_LAUNCHER_STORAGE_DIR")
    if not storage_dir:
        storage_dir = _get_ayon_appdirs()
        os.environ["AYON_LAUNCHER_STORAGE_DIR"] = storage_dir

    path = os.path.join(storage_dir, *subdirs)
    if create and not os.path.exists(path):
        os.makedirs(path, exist_ok=True)
    return path

def get_local_site_id() -> str:
    """Get local site identifier.

    Site id is created if does not exist yet.

    Returns:
        str: Site id.

    """
    # used for background syncing
    site_id = os.environ.get(SITE_ID_ENV_KEY)
    if site_id:
        return site_id

    site_id_path = get_launcher_local_dir("site_id")
    if os.path.exists(site_id_path):
        with open(site_id_path, "r") as stream:
            site_id = stream.read()

    if not site_id:
        os.makedirs(os.path.dirname(site_id_path), exist_ok=True)
        site_id = _create_local_site_id()
        with open(site_id_path, "w") as stream:
            stream.write(site_id)
    return site_id

def get_shim_executable_path() -> str:
    """Path to shim executable.

    It is not validated if shim exists.

    Returns:
        str: Path where shim executable should be found.

    """
    filename = "ayon"
    if platform.system().lower() == "windows":
        filename += ".exe"
    return os.path.join(_get_shim_executable_root(), filename)

def is_dev_mode_enabled() -> bool:
    """Check if dev is enabled.

    A dev bundle is used when dev is enabled.

    Returns:
        bool: Dev is enabled.

    """
    return os.getenv("AYON_USE_DEV") == "1"

def is_staging_enabled() -> bool:
    """Check if staging is enabled.

    Returns:
        bool: True if staging is enabled.

    """
    return os.getenv("AYON_USE_STAGING") == "1"

def load_executable_version(executable: str) -> Optional[str]:
    """Get version of executable.

    Args:
        executable (str): Path to executable.

    Returns:
        Union[str, None]: Version of executable.

    """
    if not executable:
        return None
    return load_version_from_root(os.path.dirname(executable))

def load_version_from_file(filepath: Path) -> str:
    """Execute python file and return '__version__' variable."""
    version_content = filepath.read_text(encoding="utf-8")
    version_globals = {}
    exec(version_content, version_globals)
    return version_globals["__version__"]

def load_version_from_root(root: str) -> Optional[str]:
    """Get version of executable.

    Args:
        root (str): Path to executable.

    Returns:
        Union[str, None]: Version of executable.

    """
    if not root:
        return None

    p_root: Path = Path(root)
    if not p_root.exists():
        return None

    roots: list[Path] = [p_root]

    # On macOS, the version file is located in 'Resources' folder inside
    #   the 'MacOs' where executable is. Can be next to executable if is not
    #   yet bundled (dev of build).
    if platform.system().lower() == "darwin":
        roots.insert(0, p_root.parent / "Resources")

    for root_obj in roots:
        version_filepath = root_obj / "version"
        if version_filepath.exists():
            content = version_filepath.read_text(encoding="utf-8")
            return content.strip()

        version_filepath = root_obj / "version.py"
        if version_filepath.exists():
            try:
                return load_version_from_file(version_filepath)
            except Exception as exc:
                print(
                    f"Failed to load version file {version_filepath}. {exc}"
                )

    return None

def store_current_executable_info():
    """Store information about current executable to a file for future usage.

    Use information about current executable and version of executable and
    add it to a list of available executables.

    The function won't do anything if the application is not built or if
    version is not set or the executable is already available.

    Todos:
        Don't store executable if is located inside 'ayon-launcher' codebase?

    """
    if not IS_BUILT_APPLICATION:
        return

    store_executables([sys.executable])

def store_executables(executables: Iterable[str]):
    """Store information about executables.

    Args:
        executables (Iterable[str]): Paths to executables.

    """
    info = get_executables_info(check_cleanup=False)
    info.setdefault("available_versions", [])

    for executable in executables:
        if not executable or not os.path.exists(executable):
            continue

        root, filename = os.path.split(executable)
        # Store only 'ayon.exe' executable
        if filename == "ayon_console.exe":
            filename = "ayon.exe"
            executable = os.path.join(root, filename)

        version = load_version_from_root(root)

        match_item = None
        item_is_new = True
        for item in info["available_versions"]:
            # 'executable' is unique identifier if available versions
            item_executable = item.get("executable")
            if not item_executable or item_executable != executable:
                continue

            # Version has changed, update it
            if item.get("version") != version:
                match_item = item
            item_is_new = False
            break

        if match_item is None:
            if not item_is_new:
                continue
            match_item = {}
            info["available_versions"].append(match_item)

        match_item.update({
            "version": version,
            "executable": executable,
            "added": datetime.datetime.now().strftime("%y-%m-%d-%H%M"),
        })
    store_executables_info(info)

def store_executables_info(info: ExecutablesInfo):
    """Store information about executables.

    This will override existing information so use it wisely.

    """
    filepath = get_executables_info_filepath()
    os.makedirs(os.path.dirname(filepath), exist_ok=True)

    with open(filepath, "w") as stream:
        json.dump(info, stream, indent=4)

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)