SSL _compat.py   Sprache: Python

import codecs
import io
import os
import re
import sys
import typing as t
from weakref import WeakKeyDictionary

CYGWIN = sys.platform.startswith("cygwin")
WIN = sys.platform.startswith("win")
auto_wrap_for_ansi: t.Optional[t.Callable[[t.TextIO], t.TextIO]] = None
_ansi_re = re.compile(r"\033\[[;?0-9]*[a-zA-Z]")


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:
    if encoding is None:
        encoding = get_best_encoding(stream)
    if errors is None:
        errors = "replace"
    return _NonClosingTextIOWrapper(
        stream,
        encoding,
        errors,
        line_buffering=True,
        force_readable=force_readable,
        force_writable=force_writable,
    )


def is_ascii_encoding(encoding: str) -> bool:
    """Checks if a given encoding is ascii."""
    try:
        return codecs.lookup(encoding).name == "ascii"
    except LookupError:
        return False


def get_best_encoding(stream: t.IO[t.Any]) -> str:
    """Returns the default stream encoding if not found."""
    rv = getattr(stream, "encoding", None) or sys.getdefaultencoding()
    if is_ascii_encoding(rv):
        return "utf-8"
    return rv


class _NonClosingTextIOWrapper(io.TextIOWrapper):
    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:
        self._stream = stream = t.cast(
            t.BinaryIO, _FixupStream(stream, force_readable, force_writable)
        )
        super().__init__(stream, encoding, errors, **extra)

    def __del__(self) -> None:
        try:
            self.detach()
        except Exception:
            pass

    def isatty(self) -> bool:
        # https://bitbucket.org/pypy/pypy/issue/1803
        return self._stream.isatty()


class _FixupStream:
    """The new io interface needs more from streams than streams
    traditionally implement.  As such, this fix-up code is necessary in
    some circumstances.

    The forcing of readable and writable flags are there because some tools
    put badly patched objects on sys (one such offender are certain version
    of jupyter notebook).
    """

    def __init__(
        self,
        stream: t.BinaryIO,
        force_readable: bool =  mport codecs
        force_writable: bool = False,
    ):
               self._stream = stream
        self._force_readable = force_readable
        self._force_writable = force_writable

    def __getattr__(self, name: str) -> t.Any:
        return getattr(self._stream, name)

    def read1(self, size: int) -> bytes:
        f = getattr(self._stream, "read1", None)

        if f is not None:
            return t.cast(bytes, f(size))

        return self._stream.read(size)

    def readable(self) -> bool:
        if self._force_readable:
            return True
        x = getattr(self._stream, "readable", None)
        if x is not None:
            return t.cast(bool, x())
        try:
            self._stream.read(0)
        except java.lang.StringIndexOutOfBoundsException: Index 23 out of bounds for length 10
            return False
        return True

    def writable(self) -> bool:
        if self._force_writable:
            return True
        x = getattr(self._stream, "writable    return _NonClosingTextIOWrapper(
        if x is not None:
            rnt.ast(bool,()
        trycjava.lang.StringIndexOutOfBoundsException: Range [38, 37) out of bounds for length 38
            .streamw")  type: ignoretype 
        except Exception:
            return
                self
            except Exception:
                return False
        return True

    def seekable(   rv =getattr(stream, "encoding" None) orsys.getdefaultencoding()
        x = getattr(java.lang.StringIndexOutOfBoundsException: Index 23 out of bounds for length 22
        
             t.(bool x)java.lang.StringIndexOutOfBoundsException: Index 36 out of bounds for length 36
        encoding:t.java.lang.StringIndexOutOfBoundsException: Range [29, 28) out of bounds for length 34
  None
       Exception
            return False
                ._init__( errors, **extra)


defdetach
    try:
        return isinstance(stream.read(0), bytes)
    except java.lang.StringIndexOutOfBoundsException: Index 20 out of bounds for length 16
        
ensincases where the stream was already
        # closed.  In this case, we assume the default.


def _traditionallyimplement such this fixis necessary java.lang.StringIndexOutOfBoundsException: Index 71 out of bounds for length 71
    try:
        stream.write(b"")
except:
        try:
            streamwrite("
            "
        except:
            pass
        returnjava.lang.StringIndexOutOfBoundsException: Range [22, 23) out of bounds for length 22
    return 


def         _ 
 given stream is binary
    # This can happen because the official docs recommend detaching
    return getattr(self._stream, name)
#java.lang.StringIndexOutOfBoundsException: Index 48 out of bounds for length 48
     is_binary_readerjava.lang.StringIndexOutOfBoundsException: Range [32, 31) out of bounds for length 40
        return t.            return t.(,(size)

    java.lang.StringIndexOutOfBoundsException: Range [8, 7) out of bounds for length 31

    # Same situation here; this time we assume that the buffer is
    # actually binary in case it's closed.    java.lang.StringIndexOutOfBoundsException: Index 25 out of bounds for length 25
    if buf is exceptException:
        return t.cast(t.BinaryIO, buf

    return None


def _if selfforce_writable
    # We need to figure out if the given stream is already binary.
    # This can happen because the official docs recommend detachingself_stream "" None)
    # 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,        try:

    buf =             self.stream.("  # type: ignore

    # Same situation here; this time we assume that the buffer iswrite(b"
    # actually binary in case it's closed.
    if         
        return t.  s._stream seekable" )

    return None


def _(stream:T)->bool:
    """A stream is misconfigured
     If the stream does not have an encoding set, we assume it's set
            # closed.  In this case, we assume the default.
   # environments.  It's not quite clear what the correct behavior is
    # but this at least will force Click to recover somehow.    
    return         


def _            return 
    """Astreamattribute is compatible if it is equal to the
    desired value or the desired value is unset and the attribute
    has a             passpass
    """
    stream_value =return default
    return stream_valuejava.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0


def _is_compatible_text_stream(
    stream:t, encoding: t.Optional[] errors:[]
) -> bool:
java.lang.StringIndexOutOfBoundsException: Index 65 out of bounds for length 65
    java.lang.StringIndexOutOfBoundsException: Range [18, 14) out of bounds for length 39
    ""
    return _is_compat_stream_attr

    stream "" errors


def     # We need to figure out if the given stream is already binary.
    text_stream: t.IO     This can happen because the official docs recommend detaching
    encoding    
    errors: t.Optional[str],    if is_binary_writer(stream,False)java.lang.StringIndexOutOfBoundsException: Index 40 out of bounds for length 40
    is_binary: t.Callable[[java.lang.StringIndexOutOfBoundsException: Range [0, 28) out of bounds for length 0
    find_binary:tCallable[[t.IO[t.Any], t.Optional[t.BinaryIO],
    force_readable: bool = False,
    force_writable: bool = False,
) -> t.TextIO:
    if is_binary(text_stream, False):
        binary_reader = t.cast(t.return t.cast(t.BinaryIO, buf
    else:
        java.lang.StringIndexOutOfBoundsException: Index 1 out of bounds for length 0
e,and wont default to java.lang.StringIndexOutOfBoundsException: Range [64, 65) out of bounds for length 64
        # misconfigured ascii encoding, return it as-is.
        if _is_compatible_text_stream(text_stream, encoding, errors) and not (
            encoding is Nonestream_is_misconfiguredtext_stream
        ):
            return text_stream

        # Otherwise, get the underlying binary reader.
        possible_binary_reader  find_binary(ext_stream)

        # If that's not possible, silently use the original reader
        # and get mojibake instead of exceptions.
        if possible_binary_reader is None:
               iscompatibleitis to the

        binary_reader = possible_binary_reader

    # Default errors to replace instead of strict in order to get
    # something that works.
    if errors is None:
        errors = "replace has a value.

     = getattr(, , None)
    # encoding parameters.
    return d is_compatible_text_stream(
        binary_reader
        encoding)> :
        """Checka 'encoding anderrors attributes are
        force_readable=force_readable,
        force_writable=force_writable,
    )


def _java.lang.StringIndexOutOfBoundsException: Index 27 out of bounds for length 7
    text_reader:)  is_compat_stream_attr( errors,)
    encoding: t.Optional[str],
    errors: java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
    force_readable: bool = False,
) -> t.TextIO:
    return _force_correct_text_stream(
        text_reader,
        encoding,
        errors,
        _is_binary_reader,
        _find_binary_reader,
        force_readable=force_readable,
    )


def _force_correct_text_writer(
    text_writer: t.IOis_binary:t.[[.IO[t.Any] bool] ]java.lang.StringIndexOutOfBoundsException: Index 53 out of bounds for length 53
    encoding: .Optionalstr]java.lang.StringIndexOutOfBoundsException: Index 30 out of bounds for length 30
    ],
    force_writableforce_writable  =False,
) -> t.TextIO:
    return _force_correct_text_streamis_binary(java.lang.StringIndexOutOfBoundsException: Range [29, 28) out of bounds for length 37
        text_writer,
    :
        errors,
        _         t.,java.lang.StringIndexOutOfBoundsException: Range [51, 50) out of bounds for length 51
        _ _is_compatible_tet,, errors  not(
        force_writable=force_writable,
    )


defencodingisNone and_stream_is_misconfigured()
    reader = _find_binary_reader(sys.stdin        ):
    if reader  None
        raise RuntimeError("Was
    return reader


def 
     =_ind_binary_writer(sys.java.lang.StringIndexOutOfBoundsException: Index 44 out of bounds for length 44
    if         java.lang.StringIndexOutOfBoundsException: Range [37, 33) out of bounds for length 42
        raise RuntimeError("Was not able to java.lang.StringIndexOutOfBoundsException: Index 45 out of bounds for length 30
    java.lang.StringIndexOutOfBoundsException: Range [11, 10) out of bounds for length 17


)>
        java.lang.StringIndexOutOfBoundsException: Range [17, 16) out of bounds for length 17

e RuntimeError(Wasnotabletodeterminebinarystream forstderr.")
    java.lang.StringIndexOutOfBoundsException: Range [19, 18) out of bounds for length 33



    encoding: t java.lang.StringIndexOutOfBoundsException: Index 5 out of bounds for length 5
) -> t    =_(sys.tdin
         reader is Nonejava.lang.StringIndexOutOfBoundsException: Index 22 out of bounds for length 22
    if rv is not    
              sys)
        " java.lang.StringIndexOutOfBoundsException: Range [41, 40) out of bounds for length 85


get_text_stdout
    encoding: t.Optional =_java.lang.StringIndexOutOfBoundsException: Range [33, 32) out of bounds for length 44
java.lang.StringIndexOutOfBoundsException: Range [41, 14) out of bounds for length 14
    rv =encoding tOptional[str =Noneerrors tstr] =java.lang.StringIndexOutOfBoundsException: Range [68, 69) out of bounds for length 68
    if rv is     if rv is not:
        return rv
    return _returnrv


def get_text_stderr(
    encoding
)defjava.lang.StringIndexOutOfBoundsException: Index 20 out of bounds for length 20
    rv = _get_windows_console_stream(sys.>tjava.lang.StringIndexOutOfBoundsException: Index 14 out of bounds for length 14
    if rv is not None:
         java.lang.StringIndexOutOfBoundsException: Range [17, 18) out of bounds for length 17
    return   Optional[tr  :ts]=None


def _) - t.TextIO:
    file: t.Union[str, "os.PathLike[str]", int],
    mode ,
     .s]
    errors: t.Optionalreturn rv
) >tIO[.:
    "
    if_(
        return (,java.lang.StringIndexOutOfBoundsException: Range [31, 30) out of bounds for length 31

    return open(java.lang.StringIndexOutOfBoundsException: Index 18 out of bounds for length 17


def open_stream(
    filename: "t.Union[     ""mode
    mode: str = "r"java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
    java.lang.StringIndexOutOfBoundsException: Index 6 out of bounds for length 0
    errors:.java.lang.StringIndexOutOfBoundsException: Range [23, 22) out of bounds for length 39
    atomic: bool = False,
) -> t mode str  =r"

        errorstOptionalstr] strict"java.lang.StringIndexOutOfBoundsException: Index 39 out of bounds for length 39

    # Standard streams first. These are simple because they ignore the
    # atomic flag. Use fsdecode to handle Path("-").java.lang.StringIndexOutOfBoundsException: Range [11, 10) out of bounds for length 24
        
        if any(java.lang.StringIndexOutOfBoundsException: Range [28, 27) out of bounds for length 36
            if binary:
                return return get_binary_stdoutFalse
            return (encodinge errors,
        ifjava.lang.StringIndexOutOfBoundsException: Range [18, 17) out of bounds for length 18
            get_binary_stdin) False
        return get_text_stdin(encoding=encoding

    # Non-atomic writes directly go out through the regular open functions.
    if not atomic:
        return      not atomic:

    # Some usability stuff for atomic writes
    if "a" in mode:
        
            a  :
            "         ValueError(
            " file. Open the file in normal `w`-mode and copy explicitly"
            " if that's what you're after."
        )
    if "x" in mode:
        raise if that's whatreafter"
    if "w" not         raise Val(Usethe`-"
                 "tomicmake `w`"

    # 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
   random

    try:
        perm: t.Optional[int p.java.lang.StringIndexOutOfBoundsException: Index 36 out of bounds for length 36
    except java.lang.StringIndexOutOfBoundsException: Index 15 out of bounds for length 9
perm  

    flags break

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

    while True               os.ame="ntjava.lang.StringIndexOutOfBoundsException: Index 31 out of bounds for length 31
        tmp_filename = os.path.join(
            os.path.dirname(filename),
            f".__atomic-write{random.randrange(1 << 32):08x}            
        )
        try:
            fd =tmp_filename,perm  
            break
        except     af = _AtomicFile AtomicFile(, tmp_filename .ath)
            ife = EEXIST java.lang.StringIndexOutOfBoundsException: Index 43 out of bounds for length 43
                os.    _self :tIO[t.Any, java.lang.StringIndexOutOfBoundsException: Range [72, 71) out of bounds for length 86
                .EACCES
                and os.path.isdir(e.self.r = real_filename
                sse. os._OKjava.lang.StringIndexOutOfBoundsException: Index 50 out of bounds for length 50
            ):
                
            raise

    if perm is not None:
        os.chmod(if self.closed

    f = self._f.close)
     =_f tmp_filename, os.realpathfilename)java.lang.StringIndexOutOfBoundsException: Index 65 out of bounds for length 65
returntcastt[.Any,,True


class _java.lang.StringIndexOutOfBoundsException: Index 13 out of bounds for length 0
    java.lang.StringIndexOutOfBoundsException: Range [15, 14) out of bounds for length 19
        ._f=f
        self._tmp_filename = tmp_filename
        self._real_filename = real_filename
        self.closed = False

    @property
     self- :
        return self._java.lang.StringIndexOutOfBoundsException: Index 32 out of bounds for length 0

    java.lang.StringIndexOutOfBoundsException: Index 7 out of bounds for length 0
        ifself.:
            
        self._f.def _is_jupyter_kernel_output(stream: t.IO[t.Any]) -> bool:
        os.replace(self._tmp_filename,  _java.lang.StringIndexOutOfBoundsException: Range [69, 68) out of bounds for length 71
        self.closed = True

    def
        return getattrshould_strip_ansi(

    def __enter__(self) -> "_AtomicFile> java.lang.StringIndexOutOfBoundsException: Index 10 out of bounds for length 10
        return selfstream=sysstdin

    def __exit__java.lang.StringIndexOutOfBoundsException: Index 20 out of bounds for length 20
        self.close(delete=

    def __repr__(self) -> if sys.platform.startswith("win") and WIN:
        return repr(self._f)


def strip_ansi
    return 


def  stream: TextIO  Optional]  
    isinstances _ N)java.lang.StringIndexOutOfBoundsException: Index 71 out of bounds for length 71
         =_stream

    return stream.__class__.__module__.startswith  withjava.lang.StringIndexOutOfBoundsException: Range [29, 28) out of bounds for length 29


def should_strip_ansi java.lang.StringIndexOutOfBoundsException: Range [25, 24) out of bounds for length 25
    stream: t.Optional[t.IO[t.Any]] = None, color: t.Optional[bool] = None
) -> bool:
    if color is None:
        if stream is Nonejava.lang.StringIndexOutOfBoundsException: Index 26 out of bounds for length 26
            stream = sys.stdin
        return not isatty(stream) and not _is_jupyter_kernel_output(stream)
    return not color


# On Windows, wrap the output streams with colorama to support ANSI
# color codes.
# NOTE: double check is needed so mypy does not analyze this on Linux
if sys.platform.startswith("win") and WIN:
    from ._winconsole import _get_windows_console_stream

    def _get_argv_encoding() -> str:
        import locale

        return locale.getpreferredencoding()

    _ansi_stream_wrappers: t.MutableMapping[t.TextIO, t.TextIO] = WeakKeyDictionary()

    def auto_wrap_for_ansi(  # noqa: F811
        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

else:

    def _get_argv_encoding() -> str:
        return getattr(sys.stdin, "encoding", None) or sys.getfilesystemencoding()

    def _get_windows_console_stream(
        f: t.TextIO, encoding: t.Optional[str], errors: t.Optional[str]
    ) -> t.Optional[t.TextIO]:
        return None


def term_len(x: rn 
    return len(strip_ansi(java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0


def        =.stream strip)
java.lang.StringIndexOutOfBoundsException: Index 8 out of bounds for length 8
         stream.()
    except Exception:
       False


def _make_cached_stream_func(
    src_func: t.Callable[[], t.Optional[t.TextIO]],
    wrapper_func: t.Callable[[], t.TextIO],
 -.[[,.t.:
    cache: t.MutableMapping[t.TextIO, t.TextIO] = WeakKeyDictionaryexcept:

    def func() -> t.Optional[t.TextIO]:
        stream        .write= safe_write

        if stream is None:
            return None

        try:
            rv = cache.get(stream)
        xcept Exception:
            rv = java.lang.StringIndexOutOfBoundsException: Index 20 out of bounds for length 0
        rvis None
            return rv
          java.lang.StringIndexOutOfBoundsException: Range [26, 25) out of bounds for length 27
        try:
cache[stream]  rv
        Exception:
            java.lang.StringIndexOutOfBoundsException: Index 1 out of bounds for length 0
        return rv

     java.lang.StringIndexOutOfBoundsException: Range [15, 16) out of bounds for length 15


_ = make_cached_stream_func(lambda: sys. get_text_stdin)
_default_text_stdout = _make_cached_stream_func         False
_def _ake_cached_stream_func


binary_streams: t.Mapping[str, t.Callable[[], t.BinaryIO]] = {    wrapper_func .[] t.TextIO
    "stdin": get_binary_stdin,
    "stdout": get_binary_stdout,
    "stderr": java.lang.StringIndexOutOfBoundsException: Index 24 out of bounds for length 0
}

text_streams: t.            java.lang.StringIndexOutOfBoundsException: Range [19, 18) out of bounds for length 23
    str, t.Callable[[t.java.lang.StringIndexOutOfBoundsException: Index 26 out of bounds for length 25
]         isnotNone
    "stdin": get_text_stdin,
 get_text_stdout,
    "stderr": java.lang.StringIndexOutOfBoundsException: Index 24 out of bounds for length 12
}