import io import sys def init_streams(log_write, stdout_level, stderr_level): # Redirect stdout and stderr to the Apple system log. This method is # invoked by init_apple_streams() (initconfig.c) if config->use_system_logger # is enabled. sys.stdout = SystemLog(log_write, stdout_level, errors=sys.stderr.errors) sys.stderr = SystemLog(log_write, stderr_level, errors=sys.stderr.errors) class SystemLog(io.TextIOWrapper): def __init__(self, log_write, level, **kwargs): kwargs.setdefault("encoding", "UTF-8") kwargs.setdefault("line_buffering", True) super().__init__(LogStream(log_write, level), **kwargs) def __repr__(self): return f"" def write(self, s): if not isinstance(s, str): raise TypeError( f"write() argument must be str, not {type(s).__name__}") # In case `s` is a str subclass that writes itself to stdout or stderr # when we call its methods, convert it to an actual str. s = str.__str__(s) # We want to emit one log message per line, so split # the string before sending it to the superclass. for line in s.splitlines(keepends=True): super().write(line) return len(s) class LogStream(io.RawIOBase): def __init__(self, log_write, level): self.log_write = log_write self.level = level def __repr__(self): return f"" def writable(self): return True def write(self, b): if type(b) is not bytes: try: b = bytes(memoryview(b)) except TypeError: raise TypeError( f"write() argument must be bytes-like, not {type(b).__name__}" ) from None # Writing an empty string to the stream should have no effect. if b: # Encode null bytes using "modified UTF-8" to avoid truncating the # message. This should not affect the return value, as the caller # may be expecting it to match the length of the input. self.log_write(self.level, b.replace(b"\x00", b"\xc0\x80")) return len(b)