ayon_common

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 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_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 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 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)