JustAnyone/decman
1
0
Fork 0
mirror of https://github.com/kiviktnm/decman.git synced 2026-09-19 12:08:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve printed output

Browse Source
This commit is contained in:
Kivi Kaitaniemi
2024-05-19 03:05:46 +03:00
parent 0d8118dd0e
commit 06a0f573cd
4 changed files with 94 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/decman/app.py
+26 -15
View File
@@ -37,6 +37,10 @@ def main():
help=
"print what would happen as a result of running decman (doesn't print removed files)"
)
parser.add_argument("--debug",
action="store_true",
default=False,
help="show debug output")
parser.add_argument(
"--no-packages",
action="store_true",
@@ -88,6 +92,12 @@ def main():
try:
opts = _set_up(store, args)
# Override debug_output if cli option is used
if args.debug:
conf.debug_output = True
# When print cli option is used, show info output
if args.print:
conf.quiet_output = False
Core(store, opts).run()
except err.UserFacingError as error:
l.print_error(error.user_facing_msg)
@@ -207,20 +217,20 @@ class Core:
def _disable_units(self):
to_disable = self.source.units_to_disable(self.store)
l.print_list_summary("Disabling systemd units:", to_disable)
l.print_list("Disabling systemd units:", to_disable)
if not self.only_print:
self.systemctl.disable_units(to_disable)
user_units_to_disable = self.source.user_units_to_disable(self.store)
for user, units in user_units_to_disable.items():
l.print_list_summary(f"Disabling systemd units for {user}:", units)
l.print_list(f"Disabling systemd units for {user}:", units)
if not self.only_print:
self.systemctl.disable_user_units(units, user)
def _remove_pkgs(self):
currently_installed = self.pacman.get_installed()
to_remove = self.source.packages_to_remove(currently_installed)
l.print_list_summary("Removing packages:", to_remove)
l.print_list("Removing packages:", to_remove)
if not self.only_print:
self.pacman.remove(to_remove)
@@ -239,12 +249,11 @@ class Core:
to_install_fpm = self.source.foreign_packages_to_install(
currently_installed)
l.print_list_summary("Installing pacman packages:", to_install_pacman)
l.print_list("Installing pacman packages:", to_install_pacman)
# fpm prints a summary so no need to print it twice
if self.only_print:
l.print_list_summary("Installing foreign packages:",
to_install_fpm)
l.print_list("Installing foreign packages:", to_install_fpm)
if not self.only_print:
self.pacman.install(to_install_pacman)
@@ -252,12 +261,14 @@ class Core:
self.fpm.install(to_install_fpm, force=self.force_build)
def _create_and_remove_files(self):
l.print_list_summary("Installing files:",
self.source.all_file_targets(),
elements_per_line=1)
l.print_list_summary("Installing directories:",
self.source.all_directory_targets(),
elements_per_line=1)
l.print_list("Installing files:",
self.source.all_file_targets(),
elements_per_line=1,
level=l.INFO)
l.print_list("Installing directories:",
self.source.all_directory_targets(),
elements_per_line=1,
level=l.INFO)
if self.only_print:
return
@@ -265,7 +276,7 @@ class Core:
all_created = self.source.create_all_files()
to_remove = self.source.files_to_remove(self.store, all_created)
l.print_list_summary("Removing files:", to_remove, elements_per_line=1)
l.print_list("Removing files:", to_remove, elements_per_line=1)
for file in to_remove:
try:
@@ -278,13 +289,13 @@ class Core:
def _enable_units(self):
to_enable = self.source.units_to_enable(self.store)
l.print_list_summary("Enabling systemd units:", to_enable)
l.print_list("Enabling systemd units:", to_enable)
if not self.only_print:
self.systemctl.enable_units(to_enable)
user_units_to_enable = self.source.user_units_to_enable(self.store)
for user, units in user_units_to_enable.items():
l.print_list_summary(f"Enabling systemd units for {user}:", units)
l.print_list(f"Enabling systemd units for {user}:", units)
if not self.only_print:
self.systemctl.enable_user_units(units, user)
src/decman/config.py
+4 -6
View File
@@ -89,27 +89,25 @@ class Commands:
"""
Running this command enables the given systemd units.
"""
return ["systemctl", "enable", "--quiet"] + units
return ["systemctl", "enable"] + units
def disable_units(self, units: list[str]) -> list[str]:
"""
Running this command disables the given systemd units.
"""
return ["systemctl", "disable", "--quiet"] + units
return ["systemctl", "disable"] + units
def enable_user_units(self, units: list[str], user: str) -> list[str]:
"""
Running this command enables the given systemd units for the user.
"""
return ["systemctl", "--quiet", "--user", "-M", f"{user}@", "enable"
] + units
return ["systemctl", "--user", "-M", f"{user}@", "enable"] + units
def disable_user_units(self, units: list[str], user: str) -> list[str]:
"""
Running this command disables the given systemd units for the user.
"""
return ["systemctl", "--quiet", "--user", "-M", f"{user}@", "disable"
] + units
return ["systemctl", "--user", "-M", f"{user}@", "disable"] + units
def compare_versions(self, installed_version: str,
new_version: str) -> list[str]:
src/decman/lib/__init__.py
+46 -24
View File
@@ -23,12 +23,16 @@ _RESET_SUFFIX = "\033[m"
_SPACING = " "
_CONTINUATION_PREFIX = f"{_DECMAN_MSG_TAG}{_SPACING} "
INFO = 1
SUMMARY = 2
def print_continuation(msg: str):
def print_continuation(msg: str, level: int = SUMMARY):
"""
Prints a message without a prefix.
"""
print(f"{_CONTINUATION_PREFIX}{msg}")
if level == SUMMARY or conf.debug_output or not conf.quiet_output:
print(f"{_CONTINUATION_PREFIX}{msg}")
def print_error(error_msg: str):
@@ -55,11 +59,12 @@ def print_summary(msg: str):
print(f"{_DECMAN_MSG_TAG} {_CYAN_PREFIX}SUMMARY{_RESET_SUFFIX}: {msg}")
def print_list_summary(msg: str,
l: list[str],
elements_per_line: typing.Optional[int] = None,
max_line_width: typing.Optional[int] = None,
limit_to_term_size: bool = True):
def print_list(msg: str,
l: list[str],
elements_per_line: typing.Optional[int] = None,
max_line_width: typing.Optional[int] = None,
limit_to_term_size: bool = True,
level: int = SUMMARY):
"""
Prints a summary message to the user along with a list of elements.
@@ -69,8 +74,12 @@ def print_list_summary(msg: str,
return
l = l.copy()
print_summary(msg)
print_continuation("")
if level == SUMMARY:
print_summary(msg)
elif level == INFO:
print_info(msg)
print_continuation("", level=level)
if elements_per_line is None:
elements_per_line = len(l)
@@ -100,9 +109,9 @@ def print_list_summary(msg: str,
elements_in_current_line = 1
for line in lines:
print_continuation(line)
print_continuation(line, level=level)
print_continuation("")
print_continuation("", level=level)
def print_info(msg: str):
@@ -739,7 +748,8 @@ class Pacman:
try:
subprocess.run(conf.commands.install_pkgs(packages), check=True)
subprocess.run(conf.commands.set_as_explicitly_installed(packages),
check=True)
check=True,
capture_output=conf.suppress_command_output)
except subprocess.CalledProcessError as error:
raise err.UserFacingError(
"Failed to install packages using pacman.") from error
@@ -820,7 +830,9 @@ class Systemd:
return
try:
subprocess.run(conf.commands.enable_units(units), check=True)
subprocess.run(conf.commands.enable_units(units),
check=True,
capture_output=conf.suppress_command_output)
except subprocess.CalledProcessError as error:
raise err.UserFacingError(
f"Failed to enable systemd units: {units}") from error
@@ -834,7 +846,9 @@ class Systemd:
return
try:
subprocess.run(conf.commands.disable_units(units), check=True)
subprocess.run(conf.commands.disable_units(units),
check=True,
capture_output=conf.suppress_command_output)
except subprocess.CalledProcessError as error:
raise err.UserFacingError(
f"Failed to disable systemd units: {units}") from error
@@ -851,11 +865,15 @@ class Systemd:
if not units:
return
with subprocess.Popen(conf.commands.enable_user_units(
units, user)) as process:
if process.wait() != 0:
raise err.UserFacingError(
f"Failed to enable systemd units: {units} for {user}.")
try:
subprocess.run(conf.commands.enable_user_units(units, user),
check=True,
capture_output=conf.suppress_command_output)
except subprocess.CalledProcessError as error:
raise err.UserFacingError(
f"Failed to enable systemd units: {units} for {user}."
) from error
for unit in units:
self.state.add_enabled_user_systemd_unit(user, unit)
@@ -866,10 +884,14 @@ class Systemd:
if not units:
return
with subprocess.Popen(conf.commands.disable_user_units(
units, user)) as process:
if process.wait() != 0:
raise err.UserFacingError(
f"Failed to disable systemd units: {units} for {user}.")
try:
subprocess.run(conf.commands.disable_user_units(units, user),
check=True,
capture_output=conf.suppress_command_output)
except subprocess.CalledProcessError as error:
raise err.UserFacingError(
f"Failed to disable systemd units: {units} for {user}."
) from error
for unit in units:
self.state.remove_enabled_user_systemd_unit(user, unit)
src/decman/lib/fpm.py
+18 -15
View File
@@ -472,7 +472,7 @@ class ExtendedPackageSearch:
providers = "Providers: "
for index, name in enumerate(possible_providers):
providers += f"{index + 1}:{name} "
l.print_info(providers)
l.print_summary(providers)
selection = l.prompt_number(
f"Select a provider [{min_selection}-{max_selection}] (default: {min_selection}): ",
@@ -557,7 +557,7 @@ class ForeignPackageManager:
if ignored_pkgs is None:
ignored_pkgs = set()
l.print_summary("Determining packages to upgrade.")
l.print_summary("Determining foreign packages to upgrade.")
all_foreign_pkgs = self._pacman.get_versioned_foreign_packages()
all_explicit_pkgs = set(self._pacman.get_installed())
@@ -609,17 +609,20 @@ class ForeignPackageManager:
resolved_dependencies = self.resolve_dependencies(
foreign_pkgs, foreign_dep_pkgs)
l.print_list_summary(
l.print_list(
"The following foreign packages will be installed explicitly:",
list(resolved_dependencies.foreign_pkgs))
list(resolved_dependencies.foreign_pkgs),
level=l.SUMMARY)
l.print_list_summary(
l.print_list(
"The following foreign packages will be installed as dependencies:",
list(resolved_dependencies.foreign_dep_pkgs))
list(resolved_dependencies.foreign_dep_pkgs),
level=l.SUMMARY)
l.print_list_summary(
l.print_list(
"The following foreign packages will be built in order to install other packages. They will not be installed:",
list(resolved_dependencies.foreign_build_dep_pkgs))
list(resolved_dependencies.foreign_build_dep_pkgs),
level=l.SUMMARY)
if not l.prompt_confirm("Proceed?", default=True):
raise err.UserFacingError("Installing aborted.")
@@ -675,7 +678,7 @@ class ForeignPackageManager:
Resolves foreign dependencies of foreign packages.
"""
l.print_summary("Resolving foreign package dependencies.")
l.print_info("Resolving foreign package dependencies.")
l.print_debug(f"Packages: {foreign_pkgs}")
if foreign_dep_pkgs is None:
@@ -739,7 +742,7 @@ class ForeignPackageManager:
total_processed += 1
l.print_info(f"Progress: {total_processed}/{len(seen_packages)}.")
l.print_summary("Determining build order.")
l.print_info("Determining build order.")
while True:
to_add = graph.get_and_remove_outer_dep_pkgs()
@@ -831,7 +834,7 @@ class PackageBuilder:
"""
Creates a new chroot and clones all PKGBUILDS.
"""
l.print_summary("Creating a build environment..")
l.print_info("Creating a build environment..")
if os.path.exists(conf.build_dir):
l.print_info("Removing previous build directory.")
@@ -858,7 +861,7 @@ class PackageBuilder:
self._git_clone_and_review_pkgbuild(pkgbase, git_url)
shutil.chown(pkgbuild_dir, user=conf.makepkg_user)
l.print_summary("Creating a new chroot.")
l.print_info("Creating a new chroot.")
os.makedirs(self.chroot_wd_dir)
# Remove GNUPGHOME from mkarchroot environment variables since it may interfere with
@@ -896,12 +899,12 @@ class PackageBuilder:
# Rebuild is only needed if at least one package is not in the cache.
if self._are_all_pkgs_cached(packages) and not force:
l.print_summary(
l.print_info(
f"Skipped building '{' '.join(package_names)}'. Already up to date."
)
return
l.print_summary(f"Building '{' '.join(package_names)}'.")
l.print_info(f"Building '{' '.join(package_names)}'.")
chroot_new_pacman_pkgs, chroot_pkg_files = self._get_chroot_packages(
packages)
@@ -960,7 +963,7 @@ class PackageBuilder:
check=True,
capture_output=conf.suppress_command_output)
l.print_summary(f"Finished building: '{' '.join(package_names)}'.")
l.print_info(f"Finished building: '{' '.join(package_names)}'.")
def _are_all_pkgs_cached(self, pkgs: list[ForeignPackage]) -> bool:
for pkg in pkgs: