janware/jw-pkg
1
Code Pull requests Activity Actions

lib: More result log beautification

Browse source 
This commit adds more tweaks to shell command output in order to make it nicer. The biggest patch is in Result.__summarize(), which makes it more versatile, and allows removal of some code in SSHClient.
App sees some independent, minor result format beautification.
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-06-18 09:48:48 +02:00
commit 1e0dee5908
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
3 changed files with 60 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/python/jw/pkg/lib/App.py
View file

@ -9,9 +9,9 @@ from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace
from typing import TYPE_CHECKING, Any
from .AsyncRunner import AsyncRunner
from .log import DEBUG, ERR, NOTICE, log, set_log_flags, set_log_level
from .util import pretty_cmd
from .log import DEBUG, ERR, NOTICE, log, log_m, set_log_flags, set_log_level
from .Types import LoadTypes
from .util import pretty_cmd
if TYPE_CHECKING:
from collections.abc import Awaitable
@ -177,7 +177,9 @@ class App: # export
try:
# Import argcomplete only here to not require it to be compatible
# with minimal environments
from argcomplete.completers import BaseCompleter # type: ignore[import-not-found]
from argcomplete.completers import ( # type: ignore[import-not-found]
BaseCompleter
)
class NoopCompleter(BaseCompleter):
@ -208,7 +210,7 @@ class App: # export
if isinstance(ret, int) and ret >= 0 and ret <= 0xFF:
exit_status = ret
except Exception as e:
log(ERR, 'Failed: {}'.format(repr(e) if self.__back_trace else str(e)))
log_m(ERR, 'Failed: {}'.format(repr(e) if self.__back_trace else str(e)))
exit_status = 1
# AssertionErrors are programming errors, hence a programmer should
# get a chance to figure it out

60
src/python/jw/pkg/lib/Result.py
View file

@ -49,7 +49,7 @@ class Result:
ret = ''.join(chr(b) if 32 <= b <= 126 else '.' for b in chunk)
if (not annotate) or truncate is None or len(stdxxx) <= truncate:
return f'{label}"{ret}"' if quote else label + ret
ret = '{labe}"{ret} ..."' if quote else '{labe}{ret} ...'
ret = f'{label}"{ret} ..."' if quote else f'{label}{ret} ...'
ret += f' + {len(stdxxx) - truncate} more bytes'
return ret
@ -59,9 +59,10 @@ class Result:
wd: str | None = None,
verbose = True
) -> str:
if not verbose:
ret = f'{self.__status}: '
else:
def __status_str(cmd: list[str] | None, wd: str | None, verbose: bool) -> str:
if not verbose:
return str(self.__status)
from .util import pretty_cmd
if cmd is None:
cmd = self.__cmd
@ -70,17 +71,46 @@ class Result:
if wd is None:
wd = self.__wd
call = f'"{pretty_cmd(cmd, wd)}" '
ret = (
f'Command {call}has not yet exited' if self.__status is None else
f'Command {call}has exited with status {self.__status}'
)
label, stdxxx, truncate = (
('stdout', self.__stdout, 40) if self.status == 0
else ('stderr', self.__stderr, None)
)
ret += self.__try_decode(
stdxxx, quote = True, truncate = truncate, annotate = True, label = label
)
if self.__status is None:
return f'Command {call}has not yet exited'
return f'Command {call}has exited with status {self.__status}'
def __out(truncate: int) -> list[str]:
ret: list[str] = []
for label, stdxxx, tr in [
('stdout', self.__stdout, truncate),
('stderr', self.__stderr, None),
]:
if stdxxx is None:
continue
ret.append(
self.__try_decode(
stdxxx,
quote = True,
truncate = tr,
annotate = True,
label = label,
)
)
return ret
max_width = 120
# Try a one-liner first
status_str = __status_str(cmd, wd, verbose)
out = __out(40)
ret = f'{status_str}'
if out:
ret += ', ' + ', '.join(out)
if len(ret) > max_width:
ret = f'{status_str}'
if out:
# Now that we're reaking into multiple lines anyway, they may
# just as well be longer
out = __out(max_width)
ret += ':\n' + '\n'.join([' ' + line for line in out])
return ret
def __repr__(self) -> str:

27
src/python/jw/pkg/lib/ec/SSHClient.py
View file

@ -3,13 +3,12 @@ from __future__ import annotations
import abc
import os
import pwd
import sys
from enum import Flag, auto
from typing import TYPE_CHECKING
from ..ExecContext import ExecContext
from ..log import DEBUG, ERR, INFO, NOTICE, WARNING, log
from ..log import DEBUG, ERR, INFO, NOTICE, get_caller_pos, log, log_m
from ..Uri import Uri
if TYPE_CHECKING:
@ -54,24 +53,18 @@ class SSHClient(ExecContext):
log_prefix: str,
) -> Result:
def __log(prio: int, *args):
log(prio, log_prefix, *args)
def __log(prio: int, *args, **kwargs):
caller = kwargs.get('caller')
if caller is None:
kwargs['caller'] = get_caller_pos(1)
log(prio, log_prefix, *args, **kwargs)
def __log_block(prio: int, title: str, block: bytes | str | None):
def __log_block(prio: int, title: str, block: str | None):
if self.__caps & self.Caps.LogOutput:
return
if not block:
if block is None:
return
if isinstance(block, bytes):
encoding = sys.stdout.encoding or 'utf-8'
block = block.decode(encoding).strip()
# Needed to pacify pyright: block can't be anything else at this point
assert isinstance(block, str)
delim = f'---- {title} ----'
__log(prio, f',{delim}')
for line in block.splitlines():
__log(prio, '|', line)
__log(prio, f'`{delim}')
log_m(prio, f'---- {title} ----\n{block}', caller = get_caller_pos(1))
if wd is not None and not self.__caps & self.Caps.Wd:
cmd = ['cd', wd, '&&', *cmd]
@ -97,8 +90,6 @@ class SSHClient(ExecContext):
if verbose:
__log_block(NOTICE, 'stdout', ret.stdout_str_or_none)
__log_block(NOTICE, 'stderr', ret.stderr_str_or_none)
if ret.status != 0:
__log(WARNING, f'Exit code {ret.status}')
return ret