jw-pkg/src/python/jw/pkg/lib/Result.py

from __future__ import annotations

from typing import override

class Result:

    def __init__(
        self,
        stdout: bytes | None = None,
        stderr: bytes | None = None,
        status: int | None = None,  # Command has not yet exited
        encoding: str = 'UTF-8',
        strip_output: bool = False,
        cmd: list[str] | None = None,
        wd: str | None = None,
    ) -> None:
        self.__stdout = stdout
        self.__stderr = stderr
        self.__status = status
        self.__encoding = encoding
        self.__strip_output = strip_output
        self.__cmd = cmd
        self.__wd = wd

    def __decode(self, stdxxx: bytes | None) -> str | None:
        if stdxxx is None:
            return None
        ret = stdxxx.decode(self.encoding)
        if self.strip_output:
            return ret.strip()
        return ret

    def __try_decode(
        self,
        stdxxx: bytes | None,
        quote = False,
        truncate: int | None = None,
        annotate: bool = True,
        label: str | None = None,
    ) -> str:
        if label is None:
            label = ''
        else:
            label = f'{label}: '
        if stdxxx is None:
            return f'{label}None'
        try:
            ret = stdxxx.decode()[:truncate].strip()
        except UnicodeDecodeError:
            chunk = stdxxx[:truncate]
            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 = f'{label}"{ret} ..."' if quote else f'{label}{ret} ...'
        ret += f' + {len(stdxxx) - truncate} more bytes'
        return ret

    def __summarize(
        self,
        cmd: list[str] | None = None,
        wd: str | None = None,
        verbose = True
    ) -> str:

        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
            call = ''
            if cmd is not None:
                if wd is None:
                    wd = self.__wd
                call = f'"{pretty_cmd(cmd, wd)}" '
            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

    @override
    def __repr__(self) -> str:
        return self.__summarize(verbose = False)

    @property
    def status(self) -> int | None:
        return self.__status

    @property
    def encoding(self) -> str:
        return self.__encoding

    @encoding.setter
    def encoding(self, value: str) -> None:
        self.__encoding = value

    @property
    def strip_output(self) -> bool:
        return self.__strip_output

    @strip_output.setter
    def strip_output(self, value: bool) -> None:
        self.__strip_output = value

    @property
    def cmd(self) -> list[str] | None:
        return self.__cmd

    @cmd.setter
    def cmd(self, value: list[str]) -> None:
        self.__cmd = value

    @property
    def wd(self) -> str | None:
        return self.__wd

    @wd.setter
    def wd(self, value: str) -> None:
        self.__wd = value

    def matches_error(self, pattern: str) -> bool:
        if self.status == 0:
            return False
        err = self.stderr_str
        import re
        return re.search(pattern, err) is not None

    def summarize(self, cmd: list[str] | None = None, wd: str | None = None) -> str:
        return self.__summarize(cmd, wd)

    @property
    def summary(self) -> str:
        return self.__summarize(None, None)

    @property
    def stdout(self) -> bytes:
        if self.__stdout is None:
            if self.__status == 0:
                return b''
            raise Exception(f'Result has no standard output stream: {self.summary}')
        return self.__stdout

    @property
    def stdout_or_none(self) -> bytes | None:
        return self.__stdout

    @property
    def stdout_str_or_none(self) -> str | None:
        return self.__decode(self.__stdout)

    @property
    def stdout_str(self) -> str:
        return self.stdout.decode(self.__encoding)

    @property
    def stderr(self) -> bytes:
        if self.__stderr is None:
            if isinstance(self.__status, int) and self.__status != 0:
                return b''
            raise Exception(f'Result has no standard error stream: {self.summary}')
        return self.__stderr

    @property
    def stderr_or_none(self) -> bytes | None:
        return self.__stderr

    @property
    def stderr_str_or_none(self) -> str | None:
        return self.__decode(self.__stderr)

    @property
    def stderr_str(self) -> str:
        return self.stderr.decode(self.__encoding)