CYGWIN = sys.platform.startswith("cygwin")
MSYS2 = sys.platform.startswith("win") and ("GCC" in sys.version)
APP_ENGINE = "APPENGINE_RUNTIME" in os.environ and "Development/" in os.environ.get(
    "SERVER_SOFTWARE", ""
)
WIN = sys.platform.startswith("win") and not APP_ENGINE and not MSYS2
def get_filesystem_encoding() -> str:
def _make_text_stream(
    stream: t.BinaryIO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def is_ascii_encoding(encoding: str) -> bool:
def get_best_encoding(stream: t.IO) -> str:
    def __init__(
        self,
        stream: t.BinaryIO,
        encoding: t.Optional[str],
        errors: t.Optional[str],
        force_readable: bool = False,
        force_writable: bool = False,
        **extra: t.Any,
    ) -> None:
    def __del__(self) -> None:
    def isatty(self) -> bool:
    def __init__(
        self,
        stream: t.BinaryIO,
        force_readable: bool = False,
        force_writable: bool = False,
    ):
    def __getattr__(self, name: str) -> t.Any:
    def read1(self, size: int) -> bytes:
    def readable(self) -> bool:
    def writable(self) -> bool:
    def seekable(self) -> bool:
def _is_binary_reader(stream: t.IO, default: bool = False) -> bool:
def _is_binary_writer(stream: t.IO, default: bool = False) -> bool:
def _find_binary_reader(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _find_binary_writer(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _stream_is_misconfigured(stream: t.TextIO) -> bool:
def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: t.Optional[str]) -> bool:
def _is_compatible_text_stream(
    stream: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
) -> bool:
def _force_correct_text_stream(
    text_stream: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    is_binary: t.Callable[[t.IO, bool], bool],
    find_binary: t.Callable[[t.IO], t.Optional[t.BinaryIO]],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def _force_correct_text_reader(
    text_reader: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
) -> t.TextIO:
def _force_correct_text_writer(
    text_writer: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_writable: bool = False,
) -> t.TextIO:
def get_binary_stdin() -> t.BinaryIO:
def get_binary_stdout() -> t.BinaryIO:
def get_binary_stderr() -> t.BinaryIO:
def get_text_stdin(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stdout(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stderr(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def _wrap_io_open(
    file: t.Union[str, os.PathLike, int],
    mode: str,
    encoding: t.Optional[str],
    errors: t.Optional[str],
) -> t.IO:
def open_stream(
    filename: str,
    mode: str = "r",
    encoding: t.Optional[str] = None,
    errors: t.Optional[str] = "strict",
    atomic: bool = False,
) -> t.Tuple[t.IO, bool]:
    def __init__(self, f: t.IO, tmp_filename: str, real_filename: str) -> None:
    def name(self) -> str:
    def close(self, delete: bool = False) -> None:
    def __getattr__(self, name: str) -> t.Any:
    def __enter__(self) -> "_AtomicFile":
    def __exit__(self, exc_type, exc_value, tb):  # type: ignore
    def __repr__(self) -> str:
def strip_ansi(value: str) -> str:
def _is_jupyter_kernel_output(stream: t.IO) -> bool:
def should_strip_ansi(
    stream: t.Optional[t.IO] = None, color: t.Optional[bool] = None
) -> bool:
    def _get_argv_encoding() -> str:
    def auto_wrap_for_ansi(
        stream: t.TextIO, color: t.Optional[bool] = None
    ) -> t.TextIO:
        def _safe_write(s):
    def _get_argv_encoding() -> str:
    def _get_windows_console_stream(
        f: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
    ) -> t.Optional[t.TextIO]:
def term_len(x: str) -> int:
def isatty(stream: t.IO) -> bool:
def _make_cached_stream_func(
    src_func: t.Callable[[], t.TextIO], wrapper_func: t.Callable[[], t.TextIO]
) -> t.Callable[[], t.TextIO]:
    def func() -> t.TextIO:
class _NonClosingTextIOWrapper(io.TextIOWrapper):
class _FixupStream:
class _AtomicFile: CYGWIN = sys.platform.startswith("cygwin")
MSYS2 = sys.platform.startswith("win") and ("GCC" in sys.version)
APP_ENGINE = "APPENGINE_RUNTIME" in os.environ and "Development/" in os.environ.get(
    "SERVER_SOFTWARE", ""
)
WIN = sys.platform.startswith("win") and not APP_ENGINE and not MSYS2
def get_filesystem_encoding() -> str:
def _make_text_stream(
    stream: t.BinaryIO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def is_ascii_encoding(encoding: str) -> bool:
def get_best_encoding(stream: t.IO) -> str:
    def __init__(
        self,
        stream: t.BinaryIO,
        encoding: t.Optional[str],
        errors: t.Optional[str],
        force_readable: bool = False,
        force_writable: bool = False,
        **extra: t.Any,
    ) -> None:
    def __del__(self) -> None:
    def isatty(self) -> bool:
    def __init__(
        self,
        stream: t.BinaryIO,
        force_readable: bool = False,
        force_writable: bool = False,
    ):
    def __getattr__(self, name: str) -> t.Any:
    def read1(self, size: int) -> bytes:
    def readable(self) -> bool:
    def writable(self) -> bool:
    def seekable(self) -> bool:
def _is_binary_reader(stream: t.IO, default: bool = False) -> bool:
def _is_binary_writer(stream: t.IO, default: bool = False) -> bool:
def _find_binary_reader(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _find_binary_writer(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _stream_is_misconfigured(stream: t.TextIO) -> bool:
def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: t.Optional[str]) -> bool:
def _is_compatible_text_stream(
    stream: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
) -> bool:
def _force_correct_text_stream(
    text_stream: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    is_binary: t.Callable[[t.IO, bool], bool],
    find_binary: t.Callable[[t.IO], t.Optional[t.BinaryIO]],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def _force_correct_text_reader(
    text_reader: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
) -> t.TextIO:
def _force_correct_text_writer(
    text_writer: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_writable: bool = False,
) -> t.TextIO:
def get_binary_stdin() -> t.BinaryIO:
def get_binary_stdout() -> t.BinaryIO:
def get_binary_stderr() -> t.BinaryIO:
def get_text_stdin(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stdout(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stderr(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def _wrap_io_open(
    file: t.Union[str, os.PathLike, int],
    mode: str,
    encoding: t.Optional[str],
    errors: t.Optional[str],
) -> t.IO:
def open_stream(
    filename: str,
    mode: str = "r",
    encoding: t.Optional[str] = None,
    errors: t.Optional[str] = "strict",
    atomic: bool = False,
) -> t.Tuple[t.IO, bool]:
    def __init__(self, f: t.IO, tmp_filename: str, real_filename: str) -> None:
    def name(self) -> str:
    def close(self, delete: bool = False) -> None:
    def __getattr__(self, name: str) -> t.Any:
    def __enter__(self) -> "_AtomicFile":
    def __exit__(self, exc_type, exc_value, tb):  # type: ignore
    def __repr__(self) -> str:
def strip_ansi(value: str) -> str:
def _is_jupyter_kernel_output(stream: t.IO) -> bool:
def should_strip_ansi(
    stream: t.Optional[t.IO] = None, color: t.Optional[bool] = None
) -> bool:
    def _get_argv_encoding() -> str:
    def auto_wrap_for_ansi(
        stream: t.TextIO, color: t.Optional[bool] = None
    ) -> t.TextIO:
        def _safe_write(s):
    def _get_argv_encoding() -> str:
    def _get_windows_console_stream(
        f: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
    ) -> t.Optional[t.TextIO]:
def term_len(x: str) -> int:
def isatty(stream: t.IO) -> bool:
def _make_cached_stream_func(
    src_func: t.Callable[[], t.TextIO], wrapper_func: t.Callable[[], t.TextIO]
) -> t.Callable[[], t.TextIO]:
    def func() -> t.TextIO:
class _NonClosingTextIOWrapper(io.TextIOWrapper):
class _FixupStream:
class _AtomicFile: def _find_binary_writer(stream: t.IO) -> t.Optional[t.BinaryIO]:
    # We need to figure out if the given stream is already binary.
    # This can happen because the official docs recommend detaching
    # the streams to get binary streams.  Some code might do this, so
    # we need to deal with this case explicitly.
    if _is_binary_writer(stream, False):
        return t.cast(t.BinaryIO, stream)

    buf = getattr(stream, "buffer", None)

    # Same situation here; this time we assume that the buffer is
    # actually binary in case it's closed.
    if buf is not None and _is_binary_writer(buf, True):
        return t.cast(t.BinaryIO, buf)

    return None def auto_wrap_for_ansi(
    stream: t.TextIO, color: t.Optional[bool] = None
) -> t.TextIO:
    """Support ANSI color and style codes on Windows by wrapping a
    stream with colorama.
    """
    try:
        cached = _ansi_stream_wrappers.get(stream)
    except Exception:
        cached = None

    if cached is not None:
        return cached

    import colorama

    strip = should_strip_ansi(stream, color)
    ansi_wrapper = colorama.AnsiToWin32(stream, strip=strip)
    rv = t.cast(t.TextIO, ansi_wrapper.stream)
    _write = rv.write

    def _safe_write(s):
        try:
            return _write(s)
        except BaseException:
            ansi_wrapper.reset_all()
            raise

    rv.write = _safe_write

    try:
        _ansi_stream_wrappers[stream] = rv
    except Exception:
        pass

    return rv CYGWIN = sys.platform.startswith("cygwin")
MSYS2 = sys.platform.startswith("win") and ("GCC" in sys.version)
APP_ENGINE = "APPENGINE_RUNTIME" in os.environ and "Development/" in os.environ.get(
    "SERVER_SOFTWARE", ""
)
WIN = sys.platform.startswith("win") and not APP_ENGINE and not MSYS2
def get_filesystem_encoding() -> str:
def _make_text_stream(
    stream: t.BinaryIO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def is_ascii_encoding(encoding: str) -> bool:
def get_best_encoding(stream: t.IO) -> str:
    def __init__(
        self,
        stream: t.BinaryIO,
        encoding: t.Optional[str],
        errors: t.Optional[str],
        force_readable: bool = False,
        force_writable: bool = False,
        **extra: t.Any,
    ) -> None:
    def __del__(self) -> None:
    def isatty(self) -> bool:
    def __init__(
        self,
        stream: t.BinaryIO,
        force_readable: bool = False,
        force_writable: bool = False,
    ):
    def __getattr__(self, name: str) -> t.Any:
    def read1(self, size: int) -> bytes:
    def readable(self) -> bool:
    def writable(self) -> bool:
    def seekable(self) -> bool:
def _is_binary_reader(stream: t.IO, default: bool = False) -> bool:
def _is_binary_writer(stream: t.IO, default: bool = False) -> bool:
def _find_binary_reader(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _find_binary_writer(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _stream_is_misconfigured(stream: t.TextIO) -> bool:
def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: t.Optional[str]) -> bool:
def _is_compatible_text_stream(
    stream: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
) -> bool:
def _force_correct_text_stream(
    text_stream: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    is_binary: t.Callable[[t.IO, bool], bool],
    find_binary: t.Callable[[t.IO], t.Optional[t.BinaryIO]],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def _force_correct_text_reader(
    text_reader: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
) -> t.TextIO:
def _force_correct_text_writer(
    text_writer: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_writable: bool = False,
) -> t.TextIO:
def get_binary_stdin() -> t.BinaryIO:
def get_binary_stdout() -> t.BinaryIO:
def get_binary_stderr() -> t.BinaryIO:
def get_text_stdin(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stdout(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stderr(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def _wrap_io_open(
    file: t.Union[str, os.PathLike, int],
    mode: str,
    encoding: t.Optional[str],
    errors: t.Optional[str],
) -> t.IO:
def open_stream(
    filename: str,
    mode: str = "r",
    encoding: t.Optional[str] = None,
    errors: t.Optional[str] = "strict",
    atomic: bool = False,
) -> t.Tuple[t.IO, bool]:
    def __init__(self, f: t.IO, tmp_filename: str, real_filename: str) -> None:
    def name(self) -> str:
    def close(self, delete: bool = False) -> None:
    def __getattr__(self, name: str) -> t.Any:
    def __enter__(self) -> "_AtomicFile":
    def __exit__(self, exc_type, exc_value, tb):  # type: ignore
    def __repr__(self) -> str:
def strip_ansi(value: str) -> str:
def _is_jupyter_kernel_output(stream: t.IO) -> bool:
def should_strip_ansi(
    stream: t.Optional[t.IO] = None, color: t.Optional[bool] = None
) -> bool:
    def _get_argv_encoding() -> str:
    def auto_wrap_for_ansi(
        stream: t.TextIO, color: t.Optional[bool] = None
    ) -> t.TextIO:
        def _safe_write(s):
    def _get_argv_encoding() -> str:
    def _get_windows_console_stream(
        f: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
    ) -> t.Optional[t.TextIO]:
def term_len(x: str) -> int:
def isatty(stream: t.IO) -> bool:
def _make_cached_stream_func(
    src_func: t.Callable[[], t.TextIO], wrapper_func: t.Callable[[], t.TextIO]
) -> t.Callable[[], t.TextIO]:
    def func() -> t.TextIO:
class _NonClosingTextIOWrapper(io.TextIOWrapper):
class _FixupStream:
class _AtomicFile: def get_filesystem_encoding() -> str:
    return sys.getfilesystemencoding() or sys.getdefaultencoding() def open_stream(
    filename: str,
    mode: str = "r",
    encoding: t.Optional[str] = None,
    errors: t.Optional[str] = "strict",
    atomic: bool = False,
) -> t.Tuple[t.IO, bool]:
    binary = "b" in mode

    # Standard streams first. These are simple because they ignore the
    # atomic flag. Use fsdecode to handle Path("-").
    if os.fsdecode(filename) == "-":
        if any(m in mode for m in ["w", "a", "x"]):
            if binary:
                return get_binary_stdout(), False
            return get_text_stdout(encoding=encoding, errors=errors), False
        if binary:
            return get_binary_stdin(), False
        return get_text_stdin(encoding=encoding, errors=errors), False

    # Non-atomic writes directly go out through the regular open functions.
    if not atomic:
        return _wrap_io_open(filename, mode, encoding, errors), True

    # Some usability stuff for atomic writes
    if "a" in mode:
        raise ValueError(
            "Appending to an existing file is not supported, because that"
            " would involve an expensive `copy`-operation to a temporary"
            " file. Open the file in normal `w`-mode and copy explicitly"
            " if that's what you're after."
        )
    if "x" in mode:
        raise ValueError("Use the `overwrite`-parameter instead.")
    if "w" not in mode:
        raise ValueError("Atomic writes only make sense with `w`-mode.")

    # Atomic writes are more complicated.  They work by opening a file
    # as a proxy in the same folder and then using the fdopen
    # functionality to wrap it in a Python file.  Then we wrap it in an
    # atomic file that moves the file over on close.
    import errno
    import random

    try:
        perm: t.Optional[int] = os.stat(filename).st_mode
    except OSError:
        perm = None

    flags = os.O_RDWR | os.O_CREAT | os.O_EXCL

    if binary:
        flags |= getattr(os, "O_BINARY", 0)

    while True:
        tmp_filename = os.path.join(
            os.path.dirname(filename),
            f".__atomic-write{random.randrange(1 << 32):08x}",
        )
        try:
            fd = os.open(tmp_filename, flags, 0o666 if perm is None else perm)
            break
        except OSError as e:
            if e.errno == errno.EEXIST or (
                os.name == "nt"
                and e.errno == errno.EACCES
                and os.path.isdir(e.filename)
                and os.access(e.filename, os.W_OK)
            ):
                continue
            raise

    if perm is not None:
        os.chmod(tmp_filename, perm)  # in case perm includes bits in umask

    f = _wrap_io_open(fd, mode, encoding, errors)
    af = _AtomicFile(f, tmp_filename, os.path.realpath(filename))
    return t.cast(t.IO, af), True def should_strip_ansi(
    stream: t.Optional[t.IO] = None, color: t.Optional[bool] = None
) -> bool:
    if color is None:
        if stream is None:
            stream = sys.stdin
        return not isatty(stream) and not _is_jupyter_kernel_output(stream)
    return not color def strip_ansi(value: str) -> str:
    return _ansi_re.sub("", value) CYGWIN = sys.platform.startswith("cygwin")
MSYS2 = sys.platform.startswith("win") and ("GCC" in sys.version)
APP_ENGINE = "APPENGINE_RUNTIME" in os.environ and "Development/" in os.environ.get(
    "SERVER_SOFTWARE", ""
)
WIN = sys.platform.startswith("win") and not APP_ENGINE and not MSYS2
def get_filesystem_encoding() -> str:
def _make_text_stream(
    stream: t.BinaryIO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def is_ascii_encoding(encoding: str) -> bool:
def get_best_encoding(stream: t.IO) -> str:
    def __init__(
        self,
        stream: t.BinaryIO,
        encoding: t.Optional[str],
        errors: t.Optional[str],
        force_readable: bool = False,
        force_writable: bool = False,
        **extra: t.Any,
    ) -> None:
    def __del__(self) -> None:
    def isatty(self) -> bool:
    def __init__(
        self,
        stream: t.BinaryIO,
        force_readable: bool = False,
        force_writable: bool = False,
    ):
    def __getattr__(self, name: str) -> t.Any:
    def read1(self, size: int) -> bytes:
    def readable(self) -> bool:
    def writable(self) -> bool:
    def seekable(self) -> bool:
def _is_binary_reader(stream: t.IO, default: bool = False) -> bool:
def _is_binary_writer(stream: t.IO, default: bool = False) -> bool:
def _find_binary_reader(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _find_binary_writer(stream: t.IO) -> t.Optional[t.BinaryIO]:
def _stream_is_misconfigured(stream: t.TextIO) -> bool:
def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: t.Optional[str]) -> bool:
def _is_compatible_text_stream(
    stream: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
) -> bool:
def _force_correct_text_stream(
    text_stream: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    is_binary: t.Callable[[t.IO, bool], bool],
    find_binary: t.Callable[[t.IO], t.Optional[t.BinaryIO]],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
def _force_correct_text_reader(
    text_reader: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_readable: bool = False,
) -> t.TextIO:
def _force_correct_text_writer(
    text_writer: t.IO,
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_writable: bool = False,
) -> t.TextIO:
def get_binary_stdin() -> t.BinaryIO:
def get_binary_stdout() -> t.BinaryIO:
def get_binary_stderr() -> t.BinaryIO:
def get_text_stdin(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stdout(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def get_text_stderr(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
def _wrap_io_open(
    file: t.Union[str, os.PathLike, int],
    mode: str,
    encoding: t.Optional[str],
    errors: t.Optional[str],
) -> t.IO:
def open_stream(
    filename: str,
    mode: str = "r",
    encoding: t.Optional[str] = None,
    errors: t.Optional[str] = "strict",
    atomic: bool = False,
) -> t.Tuple[t.IO, bool]:
    def __init__(self, f: t.IO, tmp_filename: str, real_filename: str) -> None:
    def name(self) -> str:
    def close(self, delete: bool = False) -> None:
    def __getattr__(self, name: str) -> t.Any:
    def __enter__(self) -> "_AtomicFile":
    def __exit__(self, exc_type, exc_value, tb):  # type: ignore
    def __repr__(self) -> str:
def strip_ansi(value: str) -> str:
def _is_jupyter_kernel_output(stream: t.IO) -> bool:
def should_strip_ansi(
    stream: t.Optional[t.IO] = None, color: t.Optional[bool] = None
) -> bool:
    def _get_argv_encoding() -> str:
    def auto_wrap_for_ansi(
        stream: t.TextIO, color: t.Optional[bool] = None
    ) -> t.TextIO:
        def _safe_write(s):
    def _get_argv_encoding() -> str:
    def _get_windows_console_stream(
        f: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
    ) -> t.Optional[t.TextIO]:
def term_len(x: str) -> int:
def isatty(stream: t.IO) -> bool:
def _make_cached_stream_func(
    src_func: t.Callable[[], t.TextIO], wrapper_func: t.Callable[[], t.TextIO]
) -> t.Callable[[], t.TextIO]:
    def func() -> t.TextIO:
class _NonClosingTextIOWrapper(io.TextIOWrapper):
class _FixupStream:
class _AtomicFile: WIN = sys.platform.startswith("win") and not APP_ENGINE and not MSYS2 def resolve_color_default(color: t.Optional[bool] = None) -> t.Optional[bool]:
    """Internal helper to get the default value of the color flag.  If a
    value is passed it's returned unchanged, otherwise it's looked up from
    the current context.
    """
    if color is not None:
        return color

    ctx = get_current_context(silent=True)

    if ctx is not None:
        return ctx.color

    return None
import os
import re
import sys
import typing as t
    import typing_extensions as te
            import errno
from functools import update_wrapper
from types import ModuleType
from ._compat import _default_text_stderr
from ._compat import _default_text_stdout
from ._compat import _find_binary_writer
from ._compat import auto_wrap_for_ansi
from ._compat import binary_streams
from ._compat import get_filesystem_encoding
from ._compat import open_stream
from ._compat import should_strip_ansi
from ._compat import strip_ansi
from ._compat import text_streams
from ._compat import WIN
from .globals import resolve_color_default
            from .exceptions import FileError
    from glob import glob
    def __getattr__(self, name: str) -> t.Any:
        return getattr(self._file, name)

    def __enter__(self) -> "KeepOpenFile":
        return self

    def __exit__(self, exc_type, exc_value, tb):  # type: ignore
        pass

    def __repr__(self) -> str:
        return repr(self._file)

    def __iter__(self) -> t.Iterator[t.AnyStr]:
        return iter(self._file)


def echo(
    message: t.Optional[t.Any] = None,
    file: t.Optional[t.IO[t.Any]] = None,
    nl: bool = True,
    err: bool = False,
    color: t.Optional[bool] = None,
) -> None:
    """Print a message and newline to stdout or a file. This should be
    used instead of :func:`print` because it provides better support
    for different data, files, and environments.

    Compared to :func:`print`, this does the following:

    -   Ensures that the output encoding is not misconfigured on Linux.
    -   Supports Unicode in the Windows console.
    -   Supports writing to binary outputs, and supports writing bytes
        to text outputs.
    -   Supports colors and styles on Windows.
    -   Removes ANSI color and style codes if the output does not look
        like an interactive terminal.
    -   Always flushes the output.

    :param message: The string or bytes to output. Other objects are
        converted to strings.
    :param file: The file to write to. Defaults to ``stdout``.
    :param err: Write to ``stderr`` instead of ``stdout``.
    :param nl: Print a newline after the message. Enabled by default.
    :param color: Force showing or hiding colors and other styles. By
        default Click will remove color if the output does not look like
        an interactive terminal.

    .. versionchanged:: 6.0
        Support Unicode output on the Windows console. Click does not
        modify ``sys.stdout``, so ``sys.stdout.write()`` and ``print()``
        will still not support Unicode.

    .. versionchanged:: 4.0
        Added the ``color`` parameter.

    .. versionadded:: 3.0
        Added the ``err`` parameter.

    .. versionchanged:: 2.0
        Support colors on Windows if colorama is installed.
    """
    if file is None:
        if err:
            file = _default_text_stderr()
        else:
            file = _default_text_stdout()

    # Convert non bytes/text into the native string type.
    if message is not None and not isinstance(message, (str, bytes, bytearray)):
        out: t.Optional[t.Union[str, bytes]] = str(message)
    else:
        out = message

    if nl:
        out = out or ""
        if isinstance(out, str):
            out += "\n"
        else:
            out += b"\n"

    if not out:
        file.flush()
        return

    # If there is a message and the value looks like bytes, we manually
    # need to find the binary stream and write the message in there.
    # This is done separately so that most stream types will work as you
    # would expect. Eg: you can write to StringIO for other cases.
    if isinstance(out, (bytes, bytearray)):
        binary_file = _find_binary_writer(file)

        if binary_file is not None:
            file.flush()
            binary_file.write(out)
            binary_file.flush()
            return

    # ANSI style code support. For no message or bytes, nothing happens.
    # When outputting to a file instead of a terminal, strip codes.
    else:
        color = resolve_color_default(color)

        if should_strip_ansi(file, color):
            out = strip_ansi(out)
        elif WIN:
            if auto_wrap_for_ansi is not None:
                file = auto_wrap_for_ansi(file)  # type: ignore
            elif not color:
                out = strip_ansi(out)

    file.write(out)  # type: ignore
    file.flush()


def get_binary_stream(name: "te.Literal['stdin', 'stdout', 'stderr']") -> t.BinaryIO:
    """Returns a system stream for byte processing.

    :param name: the name of the stream to open.  Valid names are ``'stdin'``,
                 ``'stdout'`` and ``'stderr'``
    """

    opener = binary_streams.get(name)