1import re
  2from pathlib import Path
  3from typing import Any, NamedTuple
  4
  5
  6class Message(NamedTuple):
  7    type: str
  8    data: bytes | dict[str, Any] | str
  9    time: Any
 10    name: str | None
 11    color: str | None
 12    stream: str = "stdout"
 13
 14
 15class Printer:
 16    """
 17    Printer is where Poncho's user-visible output is defined. A Printer
 18    instance receives typed messages and prints them to its output (usually
 19    STDOUT) in the Poncho format.
 20    """
 21
 22    def __init__(
 23        self,
 24        print_func: Any,
 25        time_format: str = "%H:%M:%S",
 26        width: int = 0,
 27        color: bool = True,
 28        prefix: bool = True,
 29        log_file: Path | str | None = None,
 30    ) -> None:
 31        self.print_func = print_func
 32        self.time_format = time_format
 33        self.width = width
 34        self.color = color
 35        self.prefix = prefix
 36        if log_file is not None:
 37            log_path = Path(log_file)
 38            log_path.parent.mkdir(parents=True, exist_ok=True)
 39            self.log_file = log_path.open("w", encoding="utf-8")
 40        else:
 41            self.log_file = None
 42
 43    def write(self, message: Message) -> None:
 44        if message.type != "line":
 45            raise RuntimeError('Printer can only process messages of type "line"')
 46
 47        name = message.name if message.name is not None else ""
 48        name = name.ljust(self.width)
 49        if name:
 50            name += " "
 51
 52        # When encountering data that cannot be interpreted as UTF-8 encoded
 53        # Unicode, Printer will replace the unrecognisable bytes with the
 54        # Unicode replacement character (U+FFFD).
 55        # message.data is always bytes for type="line" messages (the only type this handles)
 56        if isinstance(message.data, bytes):
 57            string = message.data.decode("utf-8", "replace")
 58        else:
 59            string = str(message.data)
 60
 61        for line in string.splitlines():
 62            prefix = ""
 63            if self.prefix:
 64                time_formatted = message.time.strftime(self.time_format)
 65                prefix_base = f"{time_formatted} {name}"
 66
 67                # Color the timestamp and name with process color
 68                if self.color and message.color:
 69                    prefix_base = _color_string(message.color, prefix_base)
 70
 71                # Use fat red pipe for stderr, dim pipe for stdout
 72                if message.stream == "stderr" and self.color:
 73                    pipe = _color_string("31", "┃")
 74                elif self.color:
 75                    pipe = _color_string("2", "|")
 76                else:
 77                    pipe = "|"
 78
 79                prefix = prefix_base + pipe + " "
 80
 81            # Dim the line content for system messages (color="2")
 82            if self.color and message.color == "2":
 83                line = _color_string("2", line)
 84
 85            formatted = prefix + line
 86
 87            # Send original (possibly ANSI-coloured) string to stdout.
 88            self.print_func(formatted)
 89
 90            # Strip ANSI escape sequences before persisting to disk so the log
 91            # file contains plain text only.  This avoids leftover control
 92            # codes (e.g. hidden-cursor) that can confuse terminals when the
 93            # log is displayed later via `plain dev logs`.
 94            if self.log_file is not None:
 95                plain = _ANSI_RE.sub("", formatted)
 96                self.log_file.write(plain + "\n")
 97                self.log_file.flush()
 98
 99    def close(self) -> None:
100        if self.log_file and hasattr(self.log_file, "close"):
101            self.log_file.close()
102
103
104def _color_string(color: str, s: str) -> str:
105    def _ansi(code: str | int) -> str:
106        return f"\033[{code}m"
107
108    return f"{_ansi(0)}{_ansi(color)}{s}{_ansi(0)}"
109
110
111# Regex that matches ANSI escape sequences (e.g. colour codes, cursor control
112# sequences, etc.).  Adapted from ECMA-48 / VT100 patterns.
113_ANSI_RE = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")