mirror of
https://github.com/morpheus65535/bazarr
synced 2024-12-28 10:38:26 +00:00
54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
import io
|
|
from typing import List, Any, IO, TYPE_CHECKING
|
|
|
|
from .ansi import AnsiDecoder
|
|
from .text import Text
|
|
|
|
if TYPE_CHECKING:
|
|
from .console import Console
|
|
|
|
|
|
class FileProxy(io.TextIOBase):
|
|
"""Wraps a file (e.g. sys.stdout) and redirects writes to a console."""
|
|
|
|
def __init__(self, console: "Console", file: IO[str]) -> None:
|
|
self.__console = console
|
|
self.__file = file
|
|
self.__buffer: List[str] = []
|
|
self.__ansi_decoder = AnsiDecoder()
|
|
|
|
@property
|
|
def rich_proxied_file(self) -> IO[str]:
|
|
"""Get proxied file."""
|
|
return self.__file
|
|
|
|
def __getattr__(self, name: str) -> Any:
|
|
return getattr(self.__file, name)
|
|
|
|
def write(self, text: str) -> int:
|
|
if not isinstance(text, str):
|
|
raise TypeError(f"write() argument must be str, not {type(text).__name__}")
|
|
buffer = self.__buffer
|
|
lines: List[str] = []
|
|
while text:
|
|
line, new_line, text = text.partition("\n")
|
|
if new_line:
|
|
lines.append("".join(buffer) + line)
|
|
del buffer[:]
|
|
else:
|
|
buffer.append(line)
|
|
break
|
|
if lines:
|
|
console = self.__console
|
|
with console:
|
|
output = Text("\n").join(
|
|
self.__ansi_decoder.decode_line(line) for line in lines
|
|
)
|
|
console.print(output, markup=False, emoji=False, highlight=False)
|
|
return len(text)
|
|
|
|
def flush(self) -> None:
|
|
buffer = self.__buffer
|
|
if buffer:
|
|
self.__console.print("".join(buffer))
|
|
del buffer[:]
|