bin/get_blobs.py

#!/usr/bin/env python3
"""Download proprietary blobs for Fairphone devices.

This tool helps to download and automatically extract device needed blobs in the
correct path into source tree.
"""
import argparse
from pathlib import Path
import logging
import sys
import subprocess
import shutil
import tarfile
import tempfile
import urllib.request
import urllib.error

BLOBS_BASE_URL = (
    "https://storage.googleapis.com/fairphone-updates/"
    "14812f31-efc8-4f57-a419-e14bb878e6dc/"
)


__log__ = logging.getLogger(__name__)
__sh__ = logging.StreamHandler()
__sh__.setFormatter(logging.Formatter("%(levelname)-6s - %(message)s"))
__log__.addHandler(__sh__)


class GetBlobsError(Exception):
    """Generic error that leads to aborting of this script."""


def parse_cmdline_arguments() -> argparse.Namespace:
    """Parse the command line arguments.

    Returns:
        argparse Namespace containing the parsed command line arguments of this
        script.

    """
    parser = argparse.ArgumentParser(
        description=(
            "Download and extract required proprietary blobs in the source "
            "tree."
        )
    )
    parser.add_argument(
        "--device",
        required=True,
        help="Name of the device to download blobs for.",
    )
    parser.add_argument(
        "--build-id",
        required=True,
        help="ID of the build to download blobs for.",
    )
    parser.add_argument(
        "--blobs-dir",
        type=Path,
        required=True,
        help="Directory to store the blobs in.",
    )
    parser.add_argument(
        "--quiet", action="store_true", help="Print only fatal errors."
    )
    return parser.parse_args()


def progress_bar(count: int, total: int, status: str = "") -> None:
    """Show a simple progress bar on command line.

    Arguments:
        count: The current counter.
        total: The expected total to complete.
        status: String that is displayed on the left of the progress bar.
    """
    bar_len = 60
    filled_len = int(round(bar_len * count / float(total)))

    percents = round(100.0 * count / float(total), 1)
    _bar = "=" * filled_len + "-" * (bar_len - filled_len)

    sys.stdout.write("[%s] %s%s ... %s\r" % (_bar, percents, "%", status))
    sys.stdout.flush()


def download_with_progress(url: str, target_file_name: Path) -> None:
    """Download a file from a URL.

    Arguments:
        url: URL to download the file from.
        target_file_name: Path object pointing to target file.

    Throws:
        GetBlobsError: In case of errors, after logging the reason of the error.
    """
    try:
        with urllib.request.urlopen(url) as response, open(
            target_file_name, "wb"
        ) as out_file:
            file_size = int(  # type: ignore
                response.getheader("content-length")  # type: ignore
            )
            chunk_size = int(max(4096, file_size / 100))
            downloaded_data = 0
            while downloaded_data < file_size:
                data = response.read(chunk_size)
                out_file.write(data)
                downloaded_data += len(data)
                progress_bar(
                    downloaded_data,
                    file_size,
                    "[ {} MB / {} MB ]".format(
                        round((downloaded_data / 1024) / 1024, 2),
                        round((file_size / 1024) / 1024, 2),
                    ),
                )
            print("")  # To properly begin a new line
            __log__.info("Download completed.")
    except urllib.error.HTTPError as e:
        __log__.error("HTTP response: %d %s", e.code, e.reason)
        raise GetBlobsError()
    except urllib.error.URLError:
        __log__.error("Malformed download URL")
        raise GetBlobsError()


def extract_archive(
    archive_file_name: str,
    script_file_name: str,
    target_path: Path,
    tmp_path: Path,
) -> None:
    """Extract a self-extracting archive into a target folder.

    Arguments:
        archive_file_name: File name of the archive to extract.
        script_file_name:
            File name of the self-extracting script inside the archive.
        target_path: Destination for the extracted files.
        tmp_path:
            Temporary directory used during the extraction process. This must
            exist and should be removed after calling this function.

    Throws:
        GetBlobsError: In case of errors, after logging the reason of the error.

    """
    __log__.info("Extracting the archive...")
    try:
        archive = tarfile.open(archive_file_name)
        archive.extractall(tmp_path)
    except (tarfile.ReadError, tarfile.CompressionError):
        __log__.error("Archive file is corrupted.")
        raise GetBlobsError()
    except tarfile.ExtractError:
        __log__.error("Extraction error.")
        raise GetBlobsError()
    __log__.info("Extraction complete.")

    # Executing makeself archive
    try:
        subprocess.run(
            "{} --noexec --target {}".format(script_file_name, target_path),
            shell=True,
        )
    except OSError:
        __log__.error("Cannot execute self-extracting archive.")
        raise GetBlobsError()


def main() -> None:
    """Entry point into the configure_lava command line tool."""
    args = parse_cmdline_arguments()
    if args.quiet:
        __log__.setLevel(logging.ERROR)
    else:
        __log__.setLevel(logging.INFO)

    base_name = args.device + "-" + args.build_id + "-blobs"
    archive_name = base_name + ".tgz"
    tmp_path = Path(tempfile.mkdtemp())
    archive_file_name = tmp_path / archive_name

    extracted_successfully = False

    try:
        # Download archive in temporary directory
        __log__.info("Downloading %s...", archive_name)
        download_with_progress(
            url=BLOBS_BASE_URL + archive_name,
            target_file_name=archive_file_name,
        )

        # Extract the self-extracting archive
        extract_archive(
            archive_file_name=archive_file_name,
            script_file_name="{}.sh".format(tmp_path / base_name),
            target_path=args.blobs_dir,
            tmp_path=tmp_path,
        )

        extracted_successfully = True
    finally:
        # Clean the temporary directory
        try:
            shutil.rmtree(tmp_path)
        except OSError:
            __log__.warning("Cannot remove temporary folder.")

    if extracted_successfully:
        __log__.info("Done.")


if __name__ == "__main__":
    main()