codekingpro/portable-devtools

1import io2import sys3 4 5def init_streams(log_write, stdout_level, stderr_level):6    # Redirect stdout and stderr to the Apple system log. This method is7    # invoked by init_apple_streams() (initconfig.c) if config->use_system_logger8    # is enabled.9    sys.stdout = SystemLog(log_write, stdout_level, errors=sys.stderr.errors)10    sys.stderr = SystemLog(log_write, stderr_level, errors=sys.stderr.errors)11 12 13class SystemLog(io.TextIOWrapper):14    def __init__(self, log_write, level, **kwargs):15        kwargs.setdefault("encoding", "UTF-8")16        kwargs.setdefault("line_buffering", True)17        super().__init__(LogStream(log_write, level), **kwargs)18 19    def __repr__(self):20        return f"<SystemLog (level {self.buffer.level})>"21 22    def write(self, s):23        if not isinstance(s, str):24            raise TypeError(25                f"write() argument must be str, not {type(s).__name__}")26 27        # In case `s` is a str subclass that writes itself to stdout or stderr28        # when we call its methods, convert it to an actual str.29        s = str.__str__(s)30 31        # We want to emit one log message per line, so split32        # the string before sending it to the superclass.33        for line in s.splitlines(keepends=True):34            super().write(line)35 36        return len(s)37 38 39class LogStream(io.RawIOBase):40    def __init__(self, log_write, level):41        self.log_write = log_write42        self.level = level43 44    def __repr__(self):45        return f"<LogStream (level {self.level!r})>"46 47    def writable(self):48        return True49 50    def write(self, b):51        if type(b) is not bytes:52            try:53                b = bytes(memoryview(b))54            except TypeError:55                raise TypeError(56                    f"write() argument must be bytes-like, not {type(b).__name__}"57                ) from None58 59        # Writing an empty string to the stream should have no effect.60        if b:61            # Encode null bytes using "modified UTF-8" to avoid truncating the62            # message. This should not affect the return value, as the caller63            # may be expecting it to match the length of the input.64            self.log_write(self.level, b.replace(b"\x00", b"\xc0\x80"))65 66        return len(b)67