replace print statements with clibella functions

This commit is contained in:
ulinja
2022-04-21 01:06:29 +02:00
parent 058051f1e9
commit 21ff40f75b
4 changed files with 370 additions and 26 deletions

333
udib/clibella.py Normal file
View File

@@ -0,0 +1,333 @@
"""This module provides classes and methods for pretty console output."""
import sys
import colorama
from colorama import Fore, Style
class _Prefix:
"""The Prefix class represents text which is prepended to text output.
Prefixes have an associated colorama color which the prefix text will have.
Attributes
----------
text : str
Text displayed as a prefix for a message.
color : colorama color constant
Color in which the prefix text will be highlighted.
"""
# This class variable holds all created Prefix objects.
_all_prefixes = []
def __init__(self, prefix_text, colorama_color):
self.text = prefix_text
self.color = colorama_color
_Prefix._all_prefixes.append(self)
def get_max_length():
"""Returns the number of characters of the longest existing prefix.
If no prefixes exist, returns 0.
"""
longest_prefix_length = 0
for prefix in _Prefix._all_prefixes:
if len(prefix.text) > longest_prefix_length:
longest_prefix_length = len(prefix.text)
return longest_prefix_length
_prefix_info = _Prefix("INFO", Fore.WHITE)
_prefix_ok = _Prefix("OK", Fore.GREEN)
_prefix_success = _Prefix("SUCCESS", Fore.GREEN)
_prefix_debug = _Prefix("DEBUG", Fore.BLUE)
_prefix_input = _Prefix("PROMPT", Fore.CYAN)
_prefix_warning = _Prefix("WARNING", Fore.YELLOW)
_prefix_error = _Prefix("ERROR", Fore.RED)
_prefix_failure = _Prefix("FAILURE", Fore.RED)
class Printer:
"""Printer objects print prefixed output to the specified output stream."""
# These class variables are needed to keep track of whether colorama has
# been initialized.
_num_of_active_printers = 0
_colorama_previously_initialized = False
def __init__(self, file=sys.stdout):
"""Constructs a Printer object and sets its default behaviour.
If the created Printer is the first printer that has been created,
colorama is initialized.
If the newly created Printer is the only printer in existence, but
other Printers have existed previously (as indicated by the
'_colorama_previously_initialized' class variable), colorama is
re-initialized.
Parameters
----------
file : File object
The file to which the Printer prints text by default (defaults to
stdout).
"""
# check if colorama needs to be initialized
if not Printer._colorama_previously_initialized:
colorama.init()
Printer._colorama_previously_initialized = True
else:
if Printer._num_of_active_printers == 0:
colorama.reinit()
Printer._num_of_active_printers += 1
self.file = file
def __del__(self):
"""Deconstructs a Printer object.
If the deleted Printer is the last one in existence, colorama is
de-initialized.
"""
# check if colorama should be de-initialized
if Printer._num_of_active_printers == 1:
colorama.deinit()
Printer._num_of_active_printers -= 1
def _print_prefixed_output(self, prefix, *args, color_enabled=True, **kwargs):
"""Prints the output with the specified prefix prepended.
This method works the same as the standard library print() function,
but with the colored text specified by the input prefix object
prepended to the printed output.
The output stream to which the output is printed is the one specified
in this Printer object's file attribute.
Parameters
----------
prefix : Prefix object
The prefix prepended to the output.
*args : various
The printable object(s) to be printed.
color_enabled : bool
If True, the printed output prefix is colored.
If False, the printed output prefix is not colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
prefix_str = ""
prefix_text = prefix.text.center(_Prefix.get_max_length())
if color_enabled:
prefix_str = f"[{prefix.color}{prefix_text}{Style.RESET_ALL}] "
else:
prefix_str = f"[{prefix_text}] "
print(prefix_str, end='', sep='', file=self.file)
print(*args, **kwargs, file=self.file)
def _get_prefixed_input(self, prefix, *args, color_enabled=True, **kwargs):
"""Prints the input prompt with the specified prefix prepended.
This method works the same as the standard library input() function,
but with the colored text specified by the input prefix object
prepended to the printed output.
The output stream to which the output is printed is always stdout.
Parameters
----------
prefix : Prefix object
The prefix prepended to the output.
*args : various
The printable object(s) to be printed.
color_enabled : bool
If True, the printed output prefix is colored.
If False, the printed output prefix is not colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
prefix_str = ""
prefix_text = prefix.text.center(_Prefix.get_max_length())
if color_enabled:
prefix_str = f"[{prefix.color}{prefix_text}{Style.RESET_ALL}] "
else:
prefix_str = f"[{prefix_text}] "
print(prefix_str, end='', sep='', file=sys.stdout)
return input(*args, **kwargs)
def info(self, *args, color_enabled=True, **kwargs):
"""Prints the specified input with an "[INFO] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
self._print_prefixed_output(
_prefix_info,
*args,
color_enabled=color_enabled,
**kwargs
)
def ok(self, *args, color_enabled=True, **kwargs):
"""Prints the specified input with an "[OK] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
self._print_prefixed_output(
_prefix_ok,
*args,
color_enabled=color_enabled,
**kwargs
)
def success(self, *args, color_enabled=True, **kwargs):
"""Prints the specified input with a "[SUCCESS] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
self._print_prefixed_output(
_prefix_success,
*args,
color_enabled=color_enabled,
**kwargs
)
def debug(self, *args, color_enabled=True, **kwargs):
"""Prints the specified input with a "[DEBUG] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
self._print_prefixed_output(
_prefix_debug,
*args,
color_enabled=color_enabled,
**kwargs
)
def warning(self, *args, color_enabled=True, **kwargs):
"""Prints the specified output with a "[WARNING] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
self._print_prefixed_output(
_prefix_warning,
*args,
color_enabled=color_enabled,
**kwargs
)
def error(self, *args, color_enabled=True, **kwargs):
"""Prints the specified output with an "[ERROR] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
self._print_prefixed_output(
_prefix_error,
*args,
color_enabled=color_enabled,
**kwargs
)
def failure(self, *args, color_enabled=True, **kwargs):
"""Prints the specified output with a "[FAILURE] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin print() function accepts, with
the exception of the "file" argument.
"""
self._print_prefixed_output(
_prefix_failure,
*args,
color_enabled=color_enabled,
**kwargs
)
def input(self, *args, color_enabled=True, **kwargs):
"""Prompts the user for input with the "[PROMPT] " prefix prepended.
Parameters
----------
*args : various
The printable object(s) to be printed.
color_enabled : Bool
Whether or not the prefix text is colored.
**kwargs : various
The same keywords which the builtin input() function accepts.
"""
return self._get_prefixed_input(
_prefix_input,
*args,
color_enabled=color_enabled,
**kwargs
)

View File

@@ -7,7 +7,11 @@ from pathlib import Path
import gnupg
import userinput
from printmsg import perror, pfailure, pinfo, pok, psuccess, pwarning
import clibella
p = clibella.Printer()
def gpg_signature_is_valid(
path_to_signature_file,
@@ -47,7 +51,7 @@ def gpg_signature_is_valid(
gpg = gnupg.GPG()
gpg.encoding = "utf-8"
pinfo("Validating signature...")
p.info("Validating signature...")
with open(path_to_signature_file, "rb") as signature_file:
verification = gpg.verify_file(
signature_file,
@@ -61,20 +65,20 @@ def gpg_signature_is_valid(
f"Not a valid PGP signature file: '{path_to_signature_file}'."
)
else:
pinfo(f"Signature mentions a key with ID "\
p.info(f"Signature mentions a key with ID "\
f"{verification.key_id} and fingerprint "\
f"{verification.fingerprint}."
)
if verification.valid:
pok(f"GPG signature is valid with trustlevel "\
p.ok(f"GPG signature is valid with trustlevel "\
f"'{verification.trust_level}'."
)
return True
# this case commonly occurrs when the GPG key has not been imported
if verification.status == "no public key":
pwarning("Could not find the public GPG key locally!")
p.warning("Could not find the public GPG key locally!")
# prompt user until answer is unambiguous
key_will_be_imported = None
@@ -85,30 +89,30 @@ def gpg_signature_is_valid(
)
if key_will_be_imported is None:
perror("Unrecognized input. Please try again.")
p.error("Unrecognized input. Please try again.")
if not key_will_be_imported:
pwarning("Aborting without importing key.")
p.warning("Aborting without importing key.")
return False
# import missing key
pinfo("Importing key...")
p.info("Importing key...")
import_result = gpg.recv_keys(
fallback_keyserver_name, verification.key_id
)
if import_result.count < 1:
perror("Failed to import key.")
p.error("Failed to import key.")
return False
# display some of gpg's output
gpg_output = import_result.stderr.split('\n')
for line in gpg_output:
if line.startswith("gpg: "):
pinfo(f"{line}")
p.info(f"{line}")
# validate signature again
pinfo("Validating signature...")
p.info("Validating signature...")
with open(path_to_signature_file, "rb") as signature_file:
verification = gpg.verify_file(
signature_file,
@@ -117,10 +121,10 @@ def gpg_signature_is_valid(
)
if verification.valid:
pok(f"GPG signature is valid with trustlevel "\
p.ok(f"GPG signature is valid with trustlevel "\
f"'{verification.trust_level}'."
)
return True
else:
perror("GPG signature is not valid!!!")
p.error("GPG signature is not valid!!!")
return False

View File

@@ -4,7 +4,11 @@
import re
from printmsg import pinput
import clibella
p = clibella.Printer()
def prompt_yes_or_no(question, ask_until_valid = False):
""" Prompts the user with the specified yes/no question.
@@ -37,7 +41,7 @@ def prompt_yes_or_no(question, ask_until_valid = False):
regex_no = re.compile(r"^(n)$|^(N)$|^(NO)$|^(No)$|^(no)$")
while(not user_input_is_valid):
user_input = pinput(f"{question} (Yes/No): ")
user_input = p.input(f"{question} (Yes/No): ")
if (regex_yes.match(user_input)):
return True
@@ -45,4 +49,3 @@ def prompt_yes_or_no(question, ask_until_valid = False):
return False
elif (not ask_until_valid):
return None

View File

@@ -10,9 +10,13 @@ import requests
from bs4 import BeautifulSoup
from tqdm import tqdm
from printmsg import perror, pfailure, pinfo, pok, psuccess, pwarning
import clibella
import gpgverify
p = clibella.Printer()
def download_file(path_to_output_file, url_to_file, show_progress = False):
""" Downloads the file at the specified URL via HTTP and saves it as the
specified output file. Optionally, displays a nice status bar.
@@ -40,7 +44,7 @@ def download_file(path_to_output_file, url_to_file, show_progress = False):
output_file_name = path_to_output_file.name
with open(path_to_output_file, "wb") as output_file:
pinfo(f"Downloading '{output_file_name}'...")
p.info(f"Downloading '{output_file_name}'...")
file_response = requests.get(url_to_file, stream=True)
total_length = file_response.headers.get('content-length')
@@ -64,7 +68,7 @@ def download_file(path_to_output_file, url_to_file, show_progress = False):
if (show_progress):
progress_bar.close()
pok(f"Received '{output_file_name}'.")
p.ok(f"Received '{output_file_name}'.")
def debian_obtain_image(path_to_output_dir):
""" Obtains the latest official debian installation image, the SHA512SUMS
@@ -90,7 +94,7 @@ def debian_obtain_image(path_to_output_dir):
Full path to the obtained and validated image file.
"""
pinfo("Obtaining and verifying the latest Debian stable image...")
p.info("Obtaining and verifying the latest Debian stable image...")
path_to_output_dir = Path(path_to_output_dir)
@@ -147,7 +151,7 @@ def debian_obtain_image(path_to_output_dir):
hash_file.writelines(hash_file_lines_to_keep)
# validate SHA512 checksum
pinfo("Validating file integrity...")
p.info("Validating file integrity...")
hash_check_result = subprocess.run(
["sha512sum", "--check", path_to_output_dir / hash_file_name],
capture_output = True,
@@ -160,22 +164,22 @@ def debian_obtain_image(path_to_output_dir):
if len(stdout_lines) > 0:
for line in stdout_lines:
if len(line) > 0:
pinfo(f"{line}")
p.info(f"{line}")
if hash_check_result.returncode != 0:
if len(stderr_lines) > 0:
for line in stderr_lines:
if len(line) > 0:
perror(f"{line}")
p.error(f"{line}")
raise RuntimeError("SHA512 validation failed.")
pok("File integrity checks passed.")
p.ok("File integrity checks passed.")
# clean up obsolete files
pinfo("Cleaning up files...")
p.info("Cleaning up files...")
os.remove(path_to_output_dir / hash_file_name)
os.remove(path_to_output_dir / signature_file_name)
psuccess("Debian image obtained.")
p.success("Debian image obtained.")
return str(path_to_output_dir / image_file_name)