codekingpro/portable-devtools

1 2from __future__ import annotations3 4# This module contains abstractions for the input stream. You don't have to5# looks further, there are no pretty code.6#7# We define two classes here.8#9#   Mark(source, line, column)10# It's just a record and its only use is producing nice error messages.11# Parser does not use it for any other purposes.12#13#   Reader(source, data)14# Reader determines the encoding of `data` and converts it to unicode.15# Reader provides the following methods and attributes:16#   reader.peek(length=1) - return the next `length` characters17#   reader.forward(length=1) - move the current position to `length`18#      characters.19#   reader.index - the number of the current character.20#   reader.line, stream.column - the line and the column of the current21#      character.22 23import codecs24 25from ruamel.yaml.error import YAMLError, FileMark, StringMark, YAMLStreamError26from ruamel.yaml.util import RegExp27 28if False:  # MYPY29    from typing import Any, Dict, Optional, List, Union, Text, Tuple, Optional  # NOQA30# from ruamel.yaml.compat import StreamTextType  # NOQA31 32__all__ = ['Reader', 'ReaderError']33 34 35class ReaderError(YAMLError):36    def __init__(  # NOQA37        self, name: Any, position: Any, character: Any, encoding: Any, reason: Any,38    ) -> None:39        self.name = name40        self.character = character41        self.position = position42        self.encoding = encoding43        self.reason = reason44 45    def __str__(self) -> Any:46        if isinstance(self.character, bytes):47            return (48                f"'{self.encoding!s}' codec can't decode byte #x{ord(self.character):02x}: "49                f'{self.reason!s}\n'50                f'  in "{self.name!s}", position {self.position:d}'51            )52        else:53            return (54                f'unacceptable character #x{self.character:04x}: {self.reason!s}\n'55                f'  in "{self.name!s}", position {self.position:d}'56            )57 58 59class Reader:60    # Reader:61    # - determines the data encoding and converts it to a unicode string,62    # - checks if characters are in allowed range,63    # - adds '\0' to the end.64 65    # Reader accepts66    #  - a `bytes` object,67    #  - a `str` object,68    #  - a file-like object with its `read` method returning `str`,69    #  - a file-like object with its `read` method returning `unicode`.70 71    # Yeah, it's ugly and slow.72 73    def __init__(self, stream: Any, loader: Any = None) -> None:74        self.loader = loader75        if self.loader is not None and getattr(self.loader, '_reader', None) is None:76            self.loader._reader = self77        self.reset_reader()78        self.stream: Any = stream  # as .read is called79 80    def reset_reader(self) -> None:81        self.name: Any = None82        self.stream_pointer = 083        self.eof = True84        self.buffer = ""85        self.pointer = 086        self.raw_buffer: Any = None87        self.raw_decode = None88        self.encoding: Optional[Text] = None89        self.index = 090        self.line = 091        self.column = 092 93    @property94    def stream(self) -> Any:95        try:96            return self._stream97        except AttributeError:98            raise YAMLStreamError('input stream needs to be specified')99 100    @stream.setter101    def stream(self, val: Any) -> None:102        if val is None:103            return104        self._stream = None105        if isinstance(val, str):106            self.name = '<unicode string>'107            self.check_printable(val)108            self.buffer = val + '\0'109        elif isinstance(val, bytes):110            self.name = '<byte string>'111            self.raw_buffer = val112            self.determine_encoding()113        else:114            if not hasattr(val, 'read'):115                raise YAMLStreamError('stream argument needs to have a read() method')116            self._stream = val117            self.name = getattr(self.stream, 'name', '<file>')118            self.eof = False119            self.raw_buffer = None120            self.determine_encoding()121 122    def peek(self, index: int = 0) -> Text:123        try:124            return self.buffer[self.pointer + index]125        except IndexError:126            self.update(index + 1)127            return self.buffer[self.pointer + index]128 129    def prefix(self, length: int = 1) -> Any:130        if self.pointer + length >= len(self.buffer):131            self.update(length)132        return self.buffer[self.pointer : self.pointer + length]133 134    def forward_1_1(self, length: int = 1) -> None:135        if self.pointer + length + 1 >= len(self.buffer):136            self.update(length + 1)137        while length != 0:138            ch = self.buffer[self.pointer]139            self.pointer += 1140            self.index += 1141            if ch in '\n\x85\u2028\u2029' or (142                ch == '\r' and self.buffer[self.pointer] != '\n'143            ):144                self.line += 1145                self.column = 0146            elif ch != '\uFEFF':147                self.column += 1148            length -= 1149 150    def forward(self, length: int = 1) -> None:151        if self.pointer + length + 1 >= len(self.buffer):152            self.update(length + 1)153        while length != 0:154            ch = self.buffer[self.pointer]155            self.pointer += 1156            self.index += 1157            if ch == '\n' or (ch == '\r' and self.buffer[self.pointer] != '\n'):158                self.line += 1159                self.column = 0160            elif ch != '\uFEFF':161                self.column += 1162            length -= 1163 164    def get_mark(self) -> Any:165        if self.stream is None:166            return StringMark(167                self.name, self.index, self.line, self.column, self.buffer, self.pointer,168            )169        else:170            return FileMark(self.name, self.index, self.line, self.column)171 172    def determine_encoding(self) -> None:173        while not self.eof and (self.raw_buffer is None or len(self.raw_buffer) < 2):174            self.update_raw()175        if isinstance(self.raw_buffer, bytes):176            if self.raw_buffer.startswith(codecs.BOM_UTF16_LE):177                self.raw_decode = codecs.utf_16_le_decode  # type: ignore178                self.encoding = 'utf-16-le'179            elif self.raw_buffer.startswith(codecs.BOM_UTF16_BE):180                self.raw_decode = codecs.utf_16_be_decode  # type: ignore181                self.encoding = 'utf-16-be'182            else:183                self.raw_decode = codecs.utf_8_decode  # type: ignore184                self.encoding = 'utf-8'185        self.update(1)186 187    NON_PRINTABLE = RegExp(188        '[^\x09\x0A\x0D\x20-\x7E\x85' '\xA0-\uD7FF' '\uE000-\uFFFD' '\U00010000-\U0010FFFF' ']'  # NOQA189    )190 191    _printable_ascii = ('\x09\x0A\x0D' + "".join(map(chr, range(0x20, 0x7F)))).encode('ascii')192 193    @classmethod194    def _get_non_printable_ascii(cls: Text, data: bytes) -> Optional[Tuple[int, Text]]:  # type: ignore # NOQA195        ascii_bytes = data.encode('ascii')  # type: ignore196        non_printables = ascii_bytes.translate(None, cls._printable_ascii)  # type: ignore197        if not non_printables:198            return None199        non_printable = non_printables[:1]200        return ascii_bytes.index(non_printable), non_printable.decode('ascii')201 202    @classmethod203    def _get_non_printable_regex(cls, data: Text) -> Optional[Tuple[int, Text]]:204        match = cls.NON_PRINTABLE.search(data)205        if not bool(match):206            return None207        return match.start(), match.group()208 209    @classmethod210    def _get_non_printable(cls, data: Text) -> Optional[Tuple[int, Text]]:211        try:212            return cls._get_non_printable_ascii(data)  # type: ignore213        except UnicodeEncodeError:214            return cls._get_non_printable_regex(data)215 216    def check_printable(self, data: Any) -> None:217        non_printable_match = self._get_non_printable(data)218        if non_printable_match is not None:219            start, character = non_printable_match220            position = self.index + (len(self.buffer) - self.pointer) + start221            raise ReaderError(222                self.name,223                position,224                ord(character),225                'unicode',226                'special characters are not allowed',227            )228 229    def update(self, length: int) -> None:230        if self.raw_buffer is None:231            return232        self.buffer = self.buffer[self.pointer :]233        self.pointer = 0234        while len(self.buffer) < length:235            if not self.eof:236                self.update_raw()237            if self.raw_decode is not None:238                try:239                    data, converted = self.raw_decode(self.raw_buffer, 'strict', self.eof)240                except UnicodeDecodeError as exc:241                    character = self.raw_buffer[exc.start]242                    if self.stream is not None:243                        position = self.stream_pointer - len(self.raw_buffer) + exc.start244                    elif self.stream is not None:245                        position = self.stream_pointer - len(self.raw_buffer) + exc.start246                    else:247                        position = exc.start248                    raise ReaderError(self.name, position, character, exc.encoding, exc.reason)249            else:250                data = self.raw_buffer251                converted = len(data)252            self.check_printable(data)253            self.buffer += data254            self.raw_buffer = self.raw_buffer[converted:]255            if self.eof:256                self.buffer += '\0'257                self.raw_buffer = None258                break259 260    def update_raw(self, size: Optional[int] = None) -> None:261        if size is None:262            size = 4096263        data = self.stream.read(size)264        if self.raw_buffer is None:265            self.raw_buffer = data266        else:267            self.raw_buffer += data268        self.stream_pointer += len(data)269        if not data:270            self.eof = True271 272 273# try:274#     import psyco275#     psyco.bind(Reader)276# except ImportError:277#     pass278