codekingpro/portable-devtools

1"""2This module contains implementations for the termui module. To keep the3import time of Click down, some infrequently used functionality is4placed in this module and only imported as needed.5"""6 7from __future__ import annotations8 9import collections.abc as cabc10import contextlib11import math12import os13import shlex14import sys15import time16import typing as t17from gettext import gettext as _18from io import StringIO19from pathlib import Path20from types import TracebackType21 22from ._compat import _default_text_stdout23from ._compat import CYGWIN24from ._compat import get_best_encoding25from ._compat import isatty26from ._compat import open_stream27from ._compat import strip_ansi28from ._compat import term_len29from ._compat import WIN30from .exceptions import ClickException31from .utils import echo32 33V = t.TypeVar("V")34 35if os.name == "nt":36    BEFORE_BAR = "\r"37    AFTER_BAR = "\n"38else:39    BEFORE_BAR = "\r\033[?25l"40    AFTER_BAR = "\033[?25h\n"41 42 43class ProgressBar(t.Generic[V]):44    def __init__(45        self,46        iterable: cabc.Iterable[V] | None,47        length: int | None = None,48        fill_char: str = "#",49        empty_char: str = " ",50        bar_template: str = "%(bar)s",51        info_sep: str = "  ",52        hidden: bool = False,53        show_eta: bool = True,54        show_percent: bool | None = None,55        show_pos: bool = False,56        item_show_func: t.Callable[[V | None], str | None] | None = None,57        label: str | None = None,58        file: t.TextIO | None = None,59        color: bool | None = None,60        update_min_steps: int = 1,61        width: int = 30,62    ) -> None:63        self.fill_char = fill_char64        self.empty_char = empty_char65        self.bar_template = bar_template66        self.info_sep = info_sep67        self.hidden = hidden68        self.show_eta = show_eta69        self.show_percent = show_percent70        self.show_pos = show_pos71        self.item_show_func = item_show_func72        self.label: str = label or ""73 74        if file is None:75            file = _default_text_stdout()76 77            # There are no standard streams attached to write to. For example,78            # pythonw on Windows.79            if file is None:80                file = StringIO()81 82        self.file = file83        self.color = color84        self.update_min_steps = update_min_steps85        self._completed_intervals = 086        self.width: int = width87        self.autowidth: bool = width == 088 89        if length is None:90            from operator import length_hint91 92            length = length_hint(iterable, -1)93 94            if length == -1:95                length = None96        if iterable is None:97            if length is None:98                raise TypeError("iterable or length is required")99            iterable = t.cast("cabc.Iterable[V]", range(length))100        self.iter: cabc.Iterable[V] = iter(iterable)101        self.length = length102        self.pos: int = 0103        self.avg: list[float] = []104        self.last_eta: float105        self.start: float106        self.start = self.last_eta = time.time()107        self.eta_known: bool = False108        self.finished: bool = False109        self.max_width: int | None = None110        self.entered: bool = False111        self.current_item: V | None = None112        self._is_atty = isatty(self.file)113        self._last_line: str | None = None114 115    def __enter__(self) -> ProgressBar[V]:116        self.entered = True117        self.render_progress()118        return self119 120    def __exit__(121        self,122        exc_type: type[BaseException] | None,123        exc_value: BaseException | None,124        tb: TracebackType | None,125    ) -> None:126        self.render_finish()127 128    def __iter__(self) -> cabc.Iterator[V]:129        if not self.entered:130            raise RuntimeError("You need to use progress bars in a with block.")131        self.render_progress()132        return self.generator()133 134    def __next__(self) -> V:135        # Iteration is defined in terms of a generator function,136        # returned by iter(self); use that to define next(). This works137        # because `self.iter` is an iterable consumed by that generator,138        # so it is re-entry safe. Calling `next(self.generator())`139        # twice works and does "what you want".140        return next(iter(self))141 142    def render_finish(self) -> None:143        if self.hidden or not self._is_atty:144            return145        self.file.write(AFTER_BAR)146        self.file.flush()147 148    @property149    def pct(self) -> float:150        if self.finished:151            return 1.0152        return min(self.pos / (float(self.length or 1) or 1), 1.0)153 154    @property155    def time_per_iteration(self) -> float:156        if not self.avg:157            return 0.0158        return sum(self.avg) / float(len(self.avg))159 160    @property161    def eta(self) -> float:162        if self.length is not None and not self.finished:163            return self.time_per_iteration * (self.length - self.pos)164        return 0.0165 166    def format_eta(self) -> str:167        if self.eta_known:168            t = int(self.eta)169            seconds = t % 60170            t //= 60171            minutes = t % 60172            t //= 60173            hours = t % 24174            t //= 24175            if t > 0:176                return f"{t}d {hours:02}:{minutes:02}:{seconds:02}"177            else:178                return f"{hours:02}:{minutes:02}:{seconds:02}"179        return ""180 181    def format_pos(self) -> str:182        pos = str(self.pos)183        if self.length is not None:184            pos += f"/{self.length}"185        return pos186 187    def format_pct(self) -> str:188        return f"{int(self.pct * 100): 4}%"[1:]189 190    def format_bar(self) -> str:191        if self.length is not None:192            bar_length = int(self.pct * self.width)193            bar = self.fill_char * bar_length194            bar += self.empty_char * (self.width - bar_length)195        elif self.finished:196            bar = self.fill_char * self.width197        else:198            chars = list(self.empty_char * (self.width or 1))199            if self.time_per_iteration != 0:200                chars[201                    int(202                        (math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5)203                        * self.width204                    )205                ] = self.fill_char206            bar = "".join(chars)207        return bar208 209    def format_progress_line(self) -> str:210        show_percent = self.show_percent211 212        info_bits = []213        if self.length is not None and show_percent is None:214            show_percent = not self.show_pos215 216        if self.show_pos:217            info_bits.append(self.format_pos())218        if show_percent:219            info_bits.append(self.format_pct())220        if self.show_eta and self.eta_known and not self.finished:221            info_bits.append(self.format_eta())222        if self.item_show_func is not None:223            item_info = self.item_show_func(self.current_item)224            if item_info is not None:225                info_bits.append(item_info)226 227        return (228            self.bar_template229            % {230                "label": self.label,231                "bar": self.format_bar(),232                "info": self.info_sep.join(info_bits),233            }234        ).rstrip()235 236    def render_progress(self) -> None:237        if self.hidden:238            return239 240        if not self._is_atty:241            # Only output the label once if the output is not a TTY.242            if self._last_line != self.label:243                self._last_line = self.label244                echo(self.label, file=self.file, color=self.color)245            return246 247        buf = []248        # Update width in case the terminal has been resized249        if self.autowidth:250            import shutil251 252            old_width = self.width253            self.width = 0254            clutter_length = term_len(self.format_progress_line())255            new_width = max(0, shutil.get_terminal_size().columns - clutter_length)256            if new_width < old_width and self.max_width is not None:257                buf.append(BEFORE_BAR)258                buf.append(" " * self.max_width)259                self.max_width = new_width260            self.width = new_width261 262        clear_width = self.width263        if self.max_width is not None:264            clear_width = self.max_width265 266        buf.append(BEFORE_BAR)267        line = self.format_progress_line()268        line_len = term_len(line)269        if self.max_width is None or self.max_width < line_len:270            self.max_width = line_len271 272        buf.append(line)273        buf.append(" " * (clear_width - line_len))274        line = "".join(buf)275        # Render the line only if it changed.276 277        if line != self._last_line:278            self._last_line = line279            echo(line, file=self.file, color=self.color, nl=False)280            self.file.flush()281 282    def make_step(self, n_steps: int) -> None:283        self.pos += n_steps284        if self.length is not None and self.pos >= self.length:285            self.finished = True286 287        if (time.time() - self.last_eta) < 1.0:288            return289 290        self.last_eta = time.time()291 292        # self.avg is a rolling list of length <= 7 of steps where steps are293        # defined as time elapsed divided by the total progress through294        # self.length.295        if self.pos:296            step = (time.time() - self.start) / self.pos297        else:298            step = time.time() - self.start299 300        self.avg = self.avg[-6:] + [step]301 302        self.eta_known = self.length is not None303 304    def update(self, n_steps: int, current_item: V | None = None) -> None:305        """Update the progress bar by advancing a specified number of306        steps, and optionally set the ``current_item`` for this new307        position.308 309        :param n_steps: Number of steps to advance.310        :param current_item: Optional item to set as ``current_item``311            for the updated position.312 313        .. versionchanged:: 8.0314            Added the ``current_item`` optional parameter.315 316        .. versionchanged:: 8.0317            Only render when the number of steps meets the318            ``update_min_steps`` threshold.319        """320        if current_item is not None:321            self.current_item = current_item322 323        self._completed_intervals += n_steps324 325        if self._completed_intervals >= self.update_min_steps:326            self.make_step(self._completed_intervals)327            self.render_progress()328            self._completed_intervals = 0329 330    def finish(self) -> None:331        self.eta_known = False332        self.current_item = None333        self.finished = True334 335    def generator(self) -> cabc.Iterator[V]:336        """Return a generator which yields the items added to the bar337        during construction, and updates the progress bar *after* the338        yielded block returns.339        """340        # WARNING: the iterator interface for `ProgressBar` relies on341        # this and only works because this is a simple generator which342        # doesn't create or manage additional state. If this function343        # changes, the impact should be evaluated both against344        # `iter(bar)` and `next(bar)`. `next()` in particular may call345        # `self.generator()` repeatedly, and this must remain safe in346        # order for that interface to work.347        if not self.entered:348            raise RuntimeError("You need to use progress bars in a with block.")349 350        if not self._is_atty:351            yield from self.iter352        else:353            for rv in self.iter:354                self.current_item = rv355 356                # This allows show_item_func to be updated before the357                # item is processed. Only trigger at the beginning of358                # the update interval.359                if self._completed_intervals == 0:360                    self.render_progress()361 362                yield rv363                self.update(1)364 365            self.finish()366            self.render_progress()367 368 369def pager(generator: cabc.Iterable[str], color: bool | None = None) -> None:370    """Decide what method to use for paging through text."""371    stdout = _default_text_stdout()372 373    # There are no standard streams attached to write to. For example,374    # pythonw on Windows.375    if stdout is None:376        stdout = StringIO()377 378    if not isatty(sys.stdin) or not isatty(stdout):379        return _nullpager(stdout, generator, color)380 381    # Split using POSIX mode (the default) so that quote characters are382    # stripped from tokens and quoted Windows paths are preserved.383    # Non-POSIX mode retains quotes in tokens, and wrapping tokens384    # with shlex.quote re-introduces quoting issues on Windows.385    pager_cmd_parts = shlex.split(os.environ.get("PAGER", ""))386    if pager_cmd_parts:387        if WIN:388            if _tempfilepager(generator, pager_cmd_parts, color):389                return390        elif _pipepager(generator, pager_cmd_parts, color):391            return392 393    if os.environ.get("TERM") in ("dumb", "emacs"):394        return _nullpager(stdout, generator, color)395    if (WIN or sys.platform.startswith("os2")) and _tempfilepager(396        generator, ["more"], color397    ):398        return399    if _pipepager(generator, ["less"], color):400        return401 402    import tempfile403 404    fd, filename = tempfile.mkstemp()405    os.close(fd)406    try:407        if _pipepager(generator, ["more"], color):408            return409        return _nullpager(stdout, generator, color)410    finally:411        os.unlink(filename)412 413 414def _pipepager(415    generator: cabc.Iterable[str], cmd_parts: list[str], color: bool | None416) -> bool:417    """Page through text by feeding it to another program.418 419    Invokes the pager via :class:`subprocess.Popen` with an ``argv`` list420    produced by :func:`shlex.split`. The command is resolved to an absolute421    path with :func:`shutil.which` as recommended by the422    :mod:`subprocess` docs for Windows compatibility.423 424    Invoking a pager through this might support colors: if piping to425    ``less`` and the user hasn't decided on colors, ``LESS=-R`` is set426    automatically.427 428    Returns ``True`` if the command was found and executed, ``False``429    otherwise so another pager can be attempted.430    """431    # Split the command into the invoked CLI and its parameters.432    if not cmd_parts:433        return False434 435    import shutil436 437    cmd = cmd_parts[0]438    cmd_params = cmd_parts[1:]439 440    cmd_filepath = shutil.which(cmd)441    if not cmd_filepath:442        return False443 444    # Produces a normalized absolute path string.445    # multi-call binaries such as busybox derive their identity from the symlink446    # less -> busybox. resolve() causes them to misbehave. (eg. less becomes busybox)447    cmd_path = Path(cmd_filepath).absolute()448    cmd_name = cmd_path.name449 450    import subprocess451 452    # Make a local copy of the environment to not affect the global one.453    env = dict(os.environ)454 455    # If we're piping to less and the user hasn't decided on colors, we enable456    # them by default we find the -R flag in the command line arguments.457    if color is None and cmd_name == "less":458        less_flags = f"{os.environ.get('LESS', '')}{' '.join(cmd_params)}"459        if not less_flags:460            env["LESS"] = "-R"461            color = True462        elif "r" in less_flags or "R" in less_flags:463            color = True464 465    c = subprocess.Popen(466        [str(cmd_path)] + cmd_params,467        shell=False,468        stdin=subprocess.PIPE,469        env=env,470        errors="replace",471        text=True,472    )473    assert c.stdin is not None474    try:475        for text in generator:476            if not color:477                text = strip_ansi(text)478 479            c.stdin.write(text)480    except BrokenPipeError:481        # In case the pager exited unexpectedly, ignore the broken pipe error.482        pass483    except Exception as e:484        # In case there is an exception we want to close the pager immediately485        # and let the caller handle it.486        # Otherwise the pager will keep running, and the user may not notice487        # the error message, or worse yet it may leave the terminal in a broken state.488        c.terminate()489        raise e490    finally:491        # We must close stdin and wait for the pager to exit before we continue492        try:493            c.stdin.close()494        # Close implies flush, so it might throw a BrokenPipeError if the pager495        # process exited already.496        except BrokenPipeError:497            pass498 499        # Less doesn't respect ^C, but catches it for its own UI purposes (aborting500        # search or other commands inside less).501        #502        # That means when the user hits ^C, the parent process (click) terminates,503        # but less is still alive, paging the output and messing up the terminal.504        #505        # If the user wants to make the pager exit on ^C, they should set506        # `LESS='-K'`. It's not our decision to make.507        while True:508            try:509                c.wait()510            except KeyboardInterrupt:511                pass512            else:513                break514 515    return True516 517 518def _tempfilepager(519    generator: cabc.Iterable[str], cmd_parts: list[str], color: bool | None520) -> bool:521    """Page through text by invoking a program on a temporary file.522 523    Used as the primary pager strategy on Windows (where piping to524    ``more`` adds spurious ``\\r\\n``), and as a fallback on other525    platforms. The command is resolved to an absolute path with526    :func:`shutil.which`.527 528    Returns ``True`` if the command was found and executed, ``False``529    otherwise so another pager can be attempted.530    """531    # Split the command into the invoked CLI and its parameters.532    if not cmd_parts:533        return False534 535    import shutil536 537    cmd = cmd_parts[0]538 539    cmd_filepath = shutil.which(cmd)540    if not cmd_filepath:541        return False542    # Produces a normalized absolute path string.543    # multi-call binaries such as busybox derive their identity from the symlink544    # less -> busybox. resolve() causes them to misbehave. (eg. less becomes busybox)545    cmd_path = Path(cmd_filepath).absolute()546 547    import subprocess548    import tempfile549 550    fd, filename = tempfile.mkstemp()551    # TODO: This never terminates if the passed generator never terminates.552    text = "".join(generator)553    if not color:554        text = strip_ansi(text)555    encoding = get_best_encoding(sys.stdout)556    with open_stream(filename, "wb")[0] as f:557        f.write(text.encode(encoding))558    try:559        subprocess.call([str(cmd_path), filename])560    except OSError:561        # Command not found562        pass563    finally:564        os.close(fd)565        os.unlink(filename)566 567    return True568 569 570def _nullpager(571    stream: t.TextIO, generator: cabc.Iterable[str], color: bool | None572) -> None:573    """Simply print unformatted text.  This is the ultimate fallback."""574    for text in generator:575        if not color:576            text = strip_ansi(text)577        stream.write(text)578 579 580class Editor:581    def __init__(582        self,583        editor: str | None = None,584        env: cabc.Mapping[str, str] | None = None,585        require_save: bool = True,586        extension: str = ".txt",587    ) -> None:588        self.editor = editor589        self.env = env590        self.require_save = require_save591        self.extension = extension592 593    def get_editor(self) -> str:594        if self.editor is not None:595            return self.editor596        for key in "VISUAL", "EDITOR":597            rv = os.environ.get(key)598            if rv:599                return rv600        if WIN:601            return "notepad"602 603        from shutil import which604 605        for editor in "sensible-editor", "vim", "nano":606            if which(editor) is not None:607                return editor608        return "vi"609 610    def edit_files(self, filenames: cabc.Iterable[str]) -> None:611        """Open files in the user's editor."""612        import shlex613        import subprocess614 615        editor = self.get_editor()616        environ: dict[str, str] | None = None617 618        if self.env:619            environ = os.environ.copy()620            environ.update(self.env)621 622        try:623            # Split in POSIX mode (the default) for the same reasons as624            # in pager(): strips quotes from tokens and preserves quoted625            # Windows paths.626            c = subprocess.Popen(627                args=shlex.split(editor) + list(filenames),628                env=environ,629            )630            exit_code = c.wait()631            if exit_code != 0:632                raise ClickException(633                    _("{editor}: Editing failed").format(editor=editor)634                )635        except OSError as e:636            raise ClickException(637                _("{editor}: Editing failed: {e}").format(editor=editor, e=e)638            ) from e639 640    @t.overload641    def edit(self, text: bytes | bytearray) -> bytes | None: ...642 643    # We cannot know whether or not the type expected is str or bytes when None644    # is passed, so str is returned as that was what was done before.645    @t.overload646    def edit(self, text: str | None) -> str | None: ...647 648    def edit(self, text: str | bytes | bytearray | None) -> str | bytes | None:649        import tempfile650 651        if text is None:652            data: bytes | bytearray = b""653        elif isinstance(text, (bytes, bytearray)):654            data = text655        else:656            if text and not text.endswith("\n"):657                text += "\n"658 659            if WIN:660                data = text.replace("\n", "\r\n").encode("utf-8-sig")661            else:662                data = text.encode("utf-8")663 664        fd, name = tempfile.mkstemp(prefix="editor-", suffix=self.extension)665        f: t.BinaryIO666 667        try:668            with os.fdopen(fd, "wb") as f:669                f.write(data)670 671            # If the filesystem resolution is 1 second, like Mac OS672            # 10.12 Extended, or 2 seconds, like FAT32, and the editor673            # closes very fast, require_save can fail. Set the modified674            # time to be 2 seconds in the past to work around this.675            os.utime(name, (os.path.getatime(name), os.path.getmtime(name) - 2))676            # Depending on the resolution, the exact value might not be677            # recorded, so get the new recorded value.678            timestamp = os.path.getmtime(name)679 680            self.edit_files((name,))681 682            if self.require_save and os.path.getmtime(name) == timestamp:683                return None684 685            with open(name, "rb") as f:686                rv = f.read()687 688            if isinstance(text, (bytes, bytearray)):689                return rv690 691            return rv.decode("utf-8-sig").replace("\r\n", "\n")692        finally:693            os.unlink(name)694 695 696def open_url(url: str, wait: bool = False, locate: bool = False) -> int:697    import subprocess698 699    def _unquote_file(url: str) -> str:700        from urllib.parse import unquote701 702        if url.startswith("file://"):703            url = unquote(url[7:])704 705        return url706 707    if sys.platform == "darwin":708        args = ["open"]709        if wait:710            args.append("-W")711        if locate:712            args.append("-R")713        args.append(_unquote_file(url))714        null = open("/dev/null", "w")715        try:716            return subprocess.Popen(args, stderr=null).wait()717        finally:718            null.close()719    elif WIN:720        if locate:721            url = _unquote_file(url)722            args = ["explorer", f"/select,{url}"]723        else:724            args = ["start"]725            if wait:726                args.append("/WAIT")727            args.append("")728            args.append(url)729        try:730            return subprocess.call(args)731        except OSError:732            # Command not found733            return 127734    elif CYGWIN:735        if locate:736            url = _unquote_file(url)737            args = ["cygstart", os.path.dirname(url)]738        else:739            args = ["cygstart"]740            if wait:741                args.append("-w")742            args.append(url)743        try:744            return subprocess.call(args)745        except OSError:746            # Command not found747            return 127748 749    try:750        if locate:751            url = os.path.dirname(_unquote_file(url)) or "."752        else:753            url = _unquote_file(url)754        c = subprocess.Popen(["xdg-open", url])755        if wait:756            return c.wait()757        return 0758    except OSError:759        if url.startswith(("http://", "https://")) and not locate and not wait:760            import webbrowser761 762            webbrowser.open(url)763            return 0764        return 1765 766 767def _translate_ch_to_exc(ch: str) -> None:768    if ch == "\x03":769        raise KeyboardInterrupt()770 771    if ch == "\x04" and not WIN:  # Unix-like, Ctrl+D772        raise EOFError()773 774    if ch == "\x1a" and WIN:  # Windows, Ctrl+Z775        raise EOFError()776 777 778if sys.platform == "win32":779    import msvcrt780 781    @contextlib.contextmanager782    def raw_terminal() -> cabc.Iterator[int]:783        yield -1784 785    def getchar(echo: bool) -> str:786        # The function `getch` will return a bytes object corresponding to787        # the pressed character. Since Windows 10 build 1803, it will also788        # return \x00 when called a second time after pressing a regular key.789        #790        # `getwch` does not share this probably-bugged behavior. Moreover, it791        # returns a Unicode object by default, which is what we want.792        #793        # Either of these functions will return \x00 or \xe0 to indicate794        # a special key, and you need to call the same function again to get795        # the "rest" of the code. The fun part is that \u00e0 is796        # "latin small letter a with grave", so if you type that on a French797        # keyboard, you _also_ get a \xe0.798        # E.g., consider the Up arrow. This returns \xe0 and then \x48. The799        # resulting Unicode string reads as "a with grave" + "capital H".800        # This is indistinguishable from when the user actually types801        # "a with grave" and then "capital H".802        #803        # When \xe0 is returned, we assume it's part of a special-key sequence804        # and call `getwch` again, but that means that when the user types805        # the \u00e0 character, `getchar` doesn't return until a second806        # character is typed.807        # The alternative is returning immediately, but that would mess up808        # cross-platform handling of arrow keys and others that start with809        # \xe0. Another option is using `getch`, but then we can't reliably810        # read non-ASCII characters, because return values of `getch` are811        # limited to the current 8-bit codepage.812        #813        # Anyway, Click doesn't claim to do this Right(tm), and using `getwch`814        # is doing the right thing in more situations than with `getch`.815 816        if echo:817            func = t.cast(t.Callable[[], str], msvcrt.getwche)818        else:819            func = t.cast(t.Callable[[], str], msvcrt.getwch)820 821        rv = func()822 823        if rv in ("\x00", "\xe0"):824            # \x00 and \xe0 are control characters that indicate special key,825            # see above.826            rv += func()827 828        _translate_ch_to_exc(rv)829        return rv830 831else:832    import termios833    import tty834 835    @contextlib.contextmanager836    def raw_terminal() -> cabc.Iterator[int]:837        f: t.TextIO | None838        fd: int839 840        if not isatty(sys.stdin):841            f = open("/dev/tty")842            fd = f.fileno()843        else:844            fd = sys.stdin.fileno()845            f = None846 847        try:848            old_settings = termios.tcgetattr(fd)849 850            try:851                tty.setraw(fd)852                yield fd853            finally:854                termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)855                sys.stdout.flush()856 857                if f is not None:858                    f.close()859        except termios.error:860            pass861 862    def getchar(echo: bool) -> str:863        with raw_terminal() as fd:864            ch = os.read(fd, 32).decode(get_best_encoding(sys.stdin), "replace")865 866            if echo and isatty(sys.stdout):867                sys.stdout.write(ch)868 869            _translate_ch_to_exc(ch)870            return ch871