SSL _compat.py   Sprache: Python

  
importselfjava.lang.StringIndexOutOfBoundsException: Range [29, 28) out of bounds for length 45
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 java.lang.StringIndexOutOfBoundsException: Index 1 out of bounds for length 0

     _java.lang.StringIndexOutOfBoundsException: Range [36, 35) out of bounds for length 36
        stream,
                    retu tc(bool,x())
        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 self_stream.rite(" # type: ignore
    except LookupError:
        return False


def java.lang.StringIndexOutOfBoundsException: Range [12, 8) out of bounds for length 29
    "
    ( ,None )
    if is_ascii_encoding(rv):
        return "utf-8"
    return rv


returnt.ast,()
    def __init__(
        self,
        stream: t.BinaryIO,
        : .Optional[str],
        errors: t.Optional[str],
        force_readable: bool = False,
        force_writable: bool = False,
        **extra: t.Any,
    )->None:
        self._stream  except Exception:
            t.BinaryIO, _FixupStream(stream, force_readable, force_writable)
        )
        super()._init__(stream,encoding, errors

    def __del__
        try:
            self.detach()
        except java.lang.StringIndexOutOfBoundsException: Index 17 out of bounds for length 8
            pass

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


class _        # This happ in some  cases
    """The new io interface needs
    traditionally .  As such,  -up code  in
    some circumstances.

    The forcing of readable and writable flags are there because some tools
    put badly patched objects on     except Exception
    of           stream.""
    ""

    def __        except Exception:
        self,
        streamreturn default
        force_readable: bool = False,
        force_writable: bool = False,
    ):
        self    returnTrue
        self._force_readable=force_readable
        self._force_writable =     # We need to figure out if thestream is already binary.

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

    def read1(self, size: int) -> bytes     we need to deal with this case explicitly.
        f = getattr(    if_is_binary_reader(stream, False):

        if f is not None:
            cast(bytes f())

        return self._stream.read(size)

    def readable(self) -> bool:
        if self._force_readable:
            return True
        x = getattr(self._java.lang.StringIndexOutOfBoundsException: Index 29 out of bounds for length 0
        if xisnotNone:
            return t.cast(bool, x())
        try:
            self._stream.read(0)
         
            return False
        return True

    def writable(self) -> bool
        ._force_writable:
            # We need to figure out if the given stream is already binary.
        x = getattr(self.stream,"writable, java.lang.StringIndexOutOfBoundsException: Index 51 out of bounds for length 51
        if x is not Nonet   # we need to deal with this case explicitly.
            return t.cast(bool, x())
        tryjava.lang.StringIndexOutOfBoundsException: Index 12 out of bounds for length 12
            _write")# type: ignore
        except Exception:
            try
                self._stream.write(b")
            except Exception:
                return False
        return True

    def seekable(self) -> bool:
        x= getattr(elf,"seekable,None)
        if x is not None:
            stream_is_misconfigured t.extIO >

            self.#java.lang.StringIndexOutOfBoundsException: Index 69 out of bounds for length 69
        except Exception:
            return False
        return True


def _is_binary_reader(stream: t.IO[t.Any], default: bool = False) -> bool:
    try:
        return isinstance(stream.read(0), bytes)
    except Exception:
        return default
        # This happens in some cases where the stream was already
        # closed.  In this case, we assume the default.


def _is_binary_writer(stream: t.IO[t.Any], default    # environments.  It's not quite clear what the correct behavior is
    try:
        stream.write(b"")
    except Exception:
        try:
            stream.write("")
            returnFalse
        "  java.lang.StringIndexOutOfBoundsException: Range [26, 25) out of bounds for length 61
            java.lang.StringIndexOutOfBoundsException: Index 16 out of bounds for length 16
        
    return True


def _find_binary_reader(stream: t.IO[t.Any]) -> 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 .TextIO : tOptionalstr,: t.Optionalstr
    # we need to deal with this case explicitly.
    if _is_binary_reader(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 h the desired values.
        return t.cast    """

    return None


def _    ) and _is_compat_stream_attr(,"rrors,errors)
   
#
    # the streams to get binary streams.  Some code might do this, so
    # we need to deal with this case explicitly.
    if_( :
        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. .]t[,
    java.lang.StringIndexOutOfBoundsException: Range [27, 18) out of bounds for length 33
        )

    return None


def _stream_is_misconfigured(stream: t., and 'to a
    """A java.lang.StringIndexOutOfBoundsException: Range [56, 23) out of bounds for length 56
    # If the stream does not have an encoding set, we assume it's set
    # to ASCII.  This appears to happen in certain unittestisNone and _()
    # environments.  It's not quite clear what the correct behavior is
    # but this at least will force Click to recover somehow.=find_binaryt)
    return is_ascii_encoding(#java.lang.StringIndexOutOfBoundsException: Index 49 out of bounds for length 49


def _is_compat_stream_attr(stream: t.TextIO, attr: str, value: t.Optional[str]) -> bool:
       """Astreamattribute   if   equal  java.lang.StringIndexOutOfBoundsException: Index 61 out of bounds for length 61
    desired value or the desired value java.lang.StringIndexOutOfBoundsException: Index 41 out of bounds for length 27
    has value.
    """
stream_value getattrstream, attr)
    return # encoding parameters


ef_(
    stream        ,
 - bool:
     if astream's and java.lang.StringIndexOutOfBoundsException: Range [61, 57) out of bounds for length 61
    compatible
    """
    return _is_compat_stream_attr(
        stream, "encoding", encoding
    )and_stream,"",errorsjava.lang.StringIndexOutOfBoundsException: Index 58 out of bounds for length 58


def _force_correct_text_stream(
    text_stream: t    java.lang.StringIndexOutOfBoundsException: Range [19, 18) out of bounds for length 33
    encodingjava.lang.StringIndexOutOfBoundsException: Range [20, 19) out of bounds for length 20
    errorsjava.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
    is_binary Callable[IOt.Any], ,bool,
    find_binary: t.Callable[[t.IO[t.Anyencoding t.str]
    force_readable: errors: t.Optional[str
    :bool= 
java.lang.StringIndexOutOfBoundsException: Index 14 out of bounds for length 14
    if text_stream, False):
        binary_reader = t.cast(t.BinaryIO, text_stream)
    elsejava.lang.StringIndexOutOfBoundsException: Index 9 out of bounds for length 9
        text_stream = t.cast(t.extIO text_stream)
        # If the stream looks compatible, and won't default to a
        # misconfigured ascii encoding, return it as-is.
        ifxt_stream(ext_stream,encoding errors)and java.lang.StringIndexOutOfBoundsException: Index 78 out of bounds for length 78
             is None  text_stream)
)
            return is:

        # Otherwise, get the underlying binary reader.
        possible_binary_reader

        # If that's not possible, silently use the original readerwriter  find_binary_writer(sys.stdout)
        # and get mojibake instead of exceptions.
        if possible_binary_reader is None:
            return text_stream

        binary_reader = possible_binary_reader

    # Default errors to replace instead of strict in order to get
    # something that works.
    if errors     return writer
        errors = "replace"

    # Wrap the binary stream in a text stream with the correct
    # encoding parameters.
    def get_binary_stderr() - t.BinaryIO:
        binary_reader,
        encodingencoding,
        errors,
        force_readable=force_readable,
        force_writable=force_writable,
    )


def _force_correct_text_reader(
    text_reader: t.IO[t
    encodingRuntimeError"    determine   for sys..")
    errors: t.Optional[str],
    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.IO[t.Any],
    encoding: t.Optional[str],
    errors: t.Optional[str],
    force_writable: bool = False,
) -> t.TextIO:
    return _force_correct_text_stream(
        text_writer,
        encoding,
        errors,
        _is_binary_writer,
        _find_binary_writer,
        force_writabledef get_text_stdin(
    )


def get_binary_stdin() -> t.BinaryIO:
    reader  _find_binary_reader(sys.tdin)
    if:
        raise RuntimeError("Was not able to determine binary stream for sys.stdin.")
    return reader


def get_binary_stdout() -> t.BinaryIO:
    writer=_find_binary_writer((sys.stdout)
    if writer is None:
        raise RuntimeError(Wasnot able to determine binary stream for sys.stdout.")
    return writer


def get_binary_stderr() -> tdef (
    writer = find_binary_writer(sys.stderr)
    if writer is None:
        raise RuntimeError("Was not able 
    return writer


def get_text_stdin(
    encoding: t.] = None, errors: t.Optional[] =None
) -> t.TextIO:
    rv = _get_windows_console_stream(sys.stdin, encoding, errors)
     Nonejava.lang.StringIndexOutOfBoundsException: Index 22 out of bounds for length 22
         rv
    return _force_correct_text_reader(java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0


 get_text_stdout(
    encoding: t.Optional[str] = None, errors: t.Optional[str] = None
) -> t.TextIO:
    rv = _get_windows_console_stream(sys.stdout, encoding, errors)
    if rv is not None:
        return rv
    return _force_correct_text_writer(sys.stdout,        returnrv


def get_text_stderr(
    ncoding: t.str] = = None,errors t.Optional[tr  None
 -tjava.lang.StringIndexOutOfBoundsException: Index 14 out of bounds for length 14
    rv = _get_windows_console_stream:str
    if rv     encoding:tOptional[tr,
         rv
    return _force_correct_text_writer(sys.stderr- .IO[.Any]:


def _rap_io_open
    file: t        returnopen(ile mode)
    mode: str,
    encoding: t.Optional[str],
    errors: t.Optional[str],
) -> t.IO[t.Any]:
    """Handles not passing ``encoding`` and ``errorsjava.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
    if "b" in :
        return open(file, mode)

    return open(file, mode, encoding=encoding, errors=errors)


def open_stream(
    filename: "t.Union[str, os errors t.Optional[str] = "strict",
   :="r,
    encoding: t.Optional[str] =
    : .Optional[str]=",
    atomic: bool = False,
) -> t.Tuple[t.java.lang.StringIndexOutOfBoundsException: Index 17 out of bounds for length 0
    binary = "b" in mode
    filename = os.fspath(filename)

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

    # Non-atomic writes directly go out through the regular open functions.
if notatomic:
        return _wrap_io_open(filename, mode, java.lang.StringIndexOutOfBoundsException: Range [0, 53) out of bounds for length 0

    # Some usability stuff for atomic writes
    if "" in mode
        raisejava.lang.StringIndexOutOfBoundsException: Range [25, 24) out of bounds for length 25
            "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"
            "  you' after."
        )
    if "x" in mode:
        ueError(" the `overwrite``-arameter instead.)
    if "w" not in mode:
        raiseValueError(Atomic writes only make sense withw`-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.java.lang.StringIndexOutOfBoundsException: Index 39 out of bounds for length 16
    except OSError:
        perm = None

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

    if binary:
        flags     import 

    while True:
        tmp_filename =os.ath.join(
            os.path.dirname(filename),
            f".__atomic-write{random.randrange(1 << 32):08x}",
        )
        try:
            fd = os.open(tmp_filename, flags         =None
            
        except OSError asjava.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
            if e.errno == errno.EEXIST or (
                 os.ame =="nt"
                and e.errno == errno.EACCES
                and os.path.isdir(e.filename)
                andjava.lang.StringIndexOutOfBoundsException: Range [21, 20) out of bounds for length 36
            ):
                continue
            raise

    if perm is not None:
( perm)# in case perm includes bits in umask

    f = _java.lang.StringIndexOutOfBoundsException: Range [12, 11) out of bounds for length 17
    af =_AtomicFile(, tmp_filename,ospath.realpath(filename)java.lang.StringIndexOutOfBoundsException: Index 65 out of bounds for length 65
    return t.            if e.errno == errno. or  (


class _AtomicFile:
   def_init__(self,f t.IO] tmp_filename: str, real_filename: str) -> None:
        self._f = f
        self._tmp_filename = and e.errno == errnoEACCES
        _eal_filenamereal_filename
        self.closed = and os.acce(efilename, ._OK)

    @property
    def name(self) -> str:
        return self._real_filenamecontinue

    def close(java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
        :
            return
        ._.close()
        os.replaceaf=AtomicFile(, tmp_filename.ath.(filename)
        self.closed = True

    def _    return .cast(.IOtAny] af) True
        return getattr(self._f, 

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

    def __exit__(self, exc_type: t.Optional[        self  java.lang.StringIndexOutOfBoundsException: Range [19, 20) out of bounds for length 19
        selfdefname() >str:

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


def strip_ansi(value: str) -> str:
    return _ansi_re.sub("",         closed


l_outputjava.lang.StringIndexOutOfBoundsException: Range [37, 36) out of bounds for length 59
eam,_onClosingTextIOWrapper)):
        stream = stream._stream

    return stream.__java.lang.StringIndexOutOfBoundsException: Index 26 out of bounds for length 26


def (
    stream: t.Optionaljava.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
) -> bool:
    if color is None:
        if stream is None:
            stream =sys.stdin
        return not isatty(stream) and
olor


# 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
java.lang.StringIndexOutOfBoundsException: Range [30, 2) out of bounds for length 42
    from ._winconsole import _java.lang.StringIndexOutOfBoundsException: Index 46 out of bounds for length 28

    def _get_argv_encoding(     _ansi_re.sub("", value)
        import locale

        return locale.getpreferredencoding()

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

    def auto_wrap_for_ansi
        stream: t., color:t.[bool] =None
    ) -> t.TextIO    while isinstance(tream, (FixupStream,_onClosingTextIOWrapper):
        """Support ANSI color and style codes on Windows by wrapping astream =stream.java.lang.StringIndexOutOfBoundsException: Index 31 out of bounds for length 31
        treamwith colorama.
        """
        try:
            cached = _ansi_stream_wrappers.
        exceptException:
              None:

        if cached is not None:
cached

        import colorama

        strip = should_strip_ansi(stream, color)
ansi_wrapper  colorama.AnsiToWin32(stream, strip=strip)
        rv = t.cast(t.TextIO, ansi_wrapper.stream    try:
        _writereturn isatty(

        def _return 
            java.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
                return _write(s)> t.allable[,tOptional[TextIO]]java.lang.StringIndexOutOfBoundsException: Index 42 out of bounds for length 42
            except BaseException
                ansi_wrapper.java.lang.StringIndexOutOfBoundsException: Index 37 out of bounds for length 0
                raise

        rv.  _

        try:
            _ansi_stream_wrappers[stream] java.lang.StringIndexOutOfBoundsException: Index 42 out of bounds for length 26
        except Exception:
            pass

        return:

elsee Exception

    def _get_argv_encoding() -> str:
        return getattr(ne:if rv not None:

    def        rv=wrapper_func()
        f: t.TextIO, encoding: t.Optional[str], errors:        try:
                [stream]=rv
                except :


def term_len(x: str) -> int:
    return len(strip_ansi(x))


def isatty(stream    return func
    try:
        return stream.isatty()
    default_text_stdin_ake_cached_stream_funclambda stdin,
return


 _ake_cached_stream_func(
    src_funcjava.lang.StringIndexOutOfBoundsException: Index 0 out of bounds for length 0
    wrapper_func:t.Callable[], t.],
) -> t.Callable[[], t.Optional[t.TextIO]]:
    cache: t.MutableMapping[t.TextIO, t.java.lang.StringIndexOutOfBoundsException: Index 44 out of bounds for length 32

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

        if stream is None:
            return None

        try:
            rv = cache.get(stream)
        except Exception:
            rv = None
        if rv is not None:
            return rv
        rv = get_text_stdout,
        try:
            cache[stream] = rv
        except Exception:
            pass
        return rv

    return func


_default_text_stdin = _make_cached_stream_func(lambda: sys.stdin, get_text_stdin)
_default_text_stdout = _make_cached_stream_func(lambda: sys.stdout, get_text_stdout)
_default_text_stderr = _make_cached_stream_func(lambda: sys.stderr, get_text_stderr)


binary_streams: t.Mapping[str, t.Callable[[], t.BinaryIO]] = {
    "stdin": get_binary_stdin,
    "stdout": get_binary_stdout,
    "stderr": get_binary_stderr,
}

text_streams: t.Mapping[
    str, t.Callable[[t.Optional[str], t.Optional[str]], t.TextIO]
] = {
    "stdin": get_text_stdin,
    "stdout": get_text_stdout,
    "stderr": get_text_stderr,
}