from __future__ import annotations # Scanner produces tokens of the following types: # STREAM-START # STREAM-END # DIRECTIVE(name, value) # DOCUMENT-START # DOCUMENT-END # BLOCK-SEQUENCE-START # BLOCK-MAPPING-START # BLOCK-END # FLOW-SEQUENCE-START # FLOW-MAPPING-START # FLOW-SEQUENCE-END # FLOW-MAPPING-END # BLOCK-ENTRY # FLOW-ENTRY # KEY # VALUE # ALIAS(value) # ANCHOR(value) # TAG(value) # SCALAR(value, plain, style) # # RoundTripScanner # COMMENT(value) # # Read comments in the Scanner code for more details. # from ruamel.yaml.error import MarkedYAMLError, CommentMark # NOQA from ruamel.yaml.tokens import * # NOQA from ruamel.yaml.docinfo import Version, Tag # NOQA from ruamel.yaml.compat import check_anchorname_char, _debug, nprint, nprintf # NOQA if False: # MYPY from typing import Any, Dict, Optional, List, Union, Text, Tuple # NOQA __all__ = ['Scanner', 'RoundTripScanner', 'ScannerError'] _THE_END = '\n\0\r\x85\u2028\u2029' _THE_END_SPACE_TAB = ' \n\0\t\r\x85\u2028\u2029' _SPACE_TAB = ' \t' if _debug != 0: def xprintf(*args: Any, **kw: Any) -> Any: return nprintf(*args, **kw) class ScannerError(MarkedYAMLError): pass class SimpleKey: # See below simple keys treatment. def __init__( self, token_number: Any, required: Any, index: int, line: int, column: int, mark: Any, ) -> None: self.token_number = token_number self.required = required self.index = index self.line = line self.column = column self.mark = mark class Scanner: def __init__(self, loader: Any = None) -> None: """Initialize the scanner.""" # It is assumed that Scanner and Reader will have a common descendant. # Reader do the dirty work of checking for BOM and converting the # input data to Unicode. It also adds NUL to the end. # # Reader supports the following methods # self.peek(i=0) # peek the next i-th character # self.prefix(l=1) # peek the next l characters # self.forward(l=1) # read the next l characters and move the pointer self.loader = loader if self.loader is not None and getattr(self.loader, '_scanner', None) is None: self.loader._scanner = self self.reset_scanner() self.first_time = False @property def flow_level(self) -> int: return len(self.flow_context) def reset_scanner(self) -> None: # Had we reached the end of the stream? self.done = False # flow_context is an expanding/shrinking list consisting of '{' and '[' # for each unclosed flow context. If empty list that means block context self.flow_context: List[Text] = [] # List of processed tokens that are not yet emitted. self.tokens: List[Any] = [] # Add the STREAM-START token. self.fetch_stream_start() # Number of tokens that were emitted through the `get_token` method. self.tokens_taken = 0 # The current indentation level. self.indent = -1 # Past indentation levels. self.indents: List[int] = [] # Variables related to simple keys treatment. # A simple key is a key that is not denoted by the '?' indicator. # Example of simple keys: # --- # block simple key: value # ? not a simple key: # : { flow simple key: value } # We emit the KEY token before all keys, so when we find a potential # simple key, we try to locate the corresponding ':' indicator. # Simple keys should be limited to a single line and 1024 characters. # Can a simple key start at the current position? A simple key may # start: # - at the beginning of the line, not counting indentation spaces # (in block context), # - after '{', '[', ',' (in the flow context), # - after '?', ':', '-' (in the block context). # In the block context, this flag also signifies if a block collection # may start at the current position. self.allow_simple_key = True # Keep track of possible simple keys. This is a dictionary. The key # is `flow_level`; there can be no more that one possible simple key # for each level. The value is a SimpleKey record: # (token_number, required, index, line, column, mark) # A simple key may start with ALIAS, ANCHOR, TAG, SCALAR(flow), # '[', or '{' tokens. self.possible_simple_keys: Dict[Any, Any] = {} self.yaml_version: Any = None self.tag_directives: List[Tuple[Any, Any]] = [] @property def reader(self) -> Any: try: return self._scanner_reader # type: ignore except AttributeError: if hasattr(self.loader, 'typ'): self._scanner_reader = self.loader.reader else: self._scanner_reader = self.loader._reader return self._scanner_reader @property def scanner_processing_version(self) -> Any: # prefix until un-composited if hasattr(self.loader, 'typ'): return self.loader.resolver.processing_version return self.loader.processing_version # Public methods. def check_token(self, *choices: Any) -> bool: # Check if the next token is one of the given types. while self.need_more_tokens(): self.fetch_more_tokens() if len(self.tokens) > 0: if not choices: return True for choice in choices: if isinstance(self.tokens[0], choice): return True return False def peek_token(self) -> Any: # Return the next token, but do not delete if from the queue. while self.need_more_tokens(): self.fetch_more_tokens() if len(self.tokens) > 0: return self.tokens[0] def get_token(self) -> Any: # Return the next token. while self.need_more_tokens(): self.fetch_more_tokens() if len(self.tokens) > 0: self.tokens_taken += 1 return self.tokens.pop(0) # Private methods. def need_more_tokens(self) -> bool: if self.done: return False if len(self.tokens) == 0: return True # The current token may be a potential simple key, so we # need to look further. self.stale_possible_simple_keys() if self.next_possible_simple_key() == self.tokens_taken: return True return False def fetch_comment(self, comment: Any) -> None: raise NotImplementedError def fetch_more_tokens(self) -> Any: # Eat whitespaces and comments until we reach the next token. comment = self.scan_to_next_token() if comment is not None: # never happens for base scanner return self.fetch_comment(comment) # Remove obsolete possible simple keys. self.stale_possible_simple_keys() # Compare the current indentation and column. It may add some tokens # and decrease the current indentation level. self.unwind_indent(self.reader.column) # Peek the next character. ch = self.reader.peek() # Is it the end of stream? if ch == '\0': return self.fetch_stream_end() # Is it a directive? if ch == '%' and self.check_directive(): return self.fetch_directive() # Is it the document start? if ch == '-' and self.check_document_start(): return self.fetch_document_start() # Is it the document end? if ch == '.' and self.check_document_end(): return self.fetch_document_end() # TODO: support for BOM within a stream. # if ch == '\uFEFF': # return self.fetch_bom() <-- issue BOMToken # Note: the order of the following checks is NOT significant. # Is it the flow sequence start indicator? if ch == '[': return self.fetch_flow_sequence_start() # Is it the flow mapping start indicator? if ch == '{': return self.fetch_flow_mapping_start() # Is it the flow sequence end indicator? if ch == ']': return self.fetch_flow_sequence_end() # Is it the flow mapping end indicator? if ch == '}': return self.fetch_flow_mapping_end() # Is it the flow entry indicator? if ch == ',': return self.fetch_flow_entry() # Is it the block entry indicator? if ch == '-' and self.check_block_entry(): return self.fetch_block_entry() # Is it the key indicator? if ch == '?' and self.check_key(): return self.fetch_key() # Is it the value indicator? if ch == ':' and self.check_value(): return self.fetch_value() # Is it an alias? if ch == '*': return self.fetch_alias() # Is it an anchor? if ch == '&': return self.fetch_anchor() # Is it a tag? if ch == '!': return self.fetch_tag() # Is it a literal scalar? if ch == '|' and not self.flow_level: return self.fetch_literal() # Is it a folded scalar? if ch == '>' and not self.flow_level: return self.fetch_folded() # Is it a single quoted scalar? if ch == "'": return self.fetch_single() # Is it a double quoted scalar? if ch == '"': return self.fetch_double() # It must be a plain scalar then. if self.check_plain(): return self.fetch_plain() # No? It's an error. Let's produce a nice error message. raise ScannerError( 'while scanning for the next token', None, f'found character {ch!r} that cannot start any token', self.reader.get_mark(), ) # Simple keys treatment. def next_possible_simple_key(self) -> Any: # Return the number of the nearest possible simple key. Actually we # don't need to loop through the whole dictionary. We may replace it # with the following code: # if not self.possible_simple_keys: # return None # return self.possible_simple_keys[ # min(self.possible_simple_keys.keys())].token_number min_token_number = None for level in self.possible_simple_keys: key = self.possible_simple_keys[level] if min_token_number is None or key.token_number < min_token_number: min_token_number = key.token_number return min_token_number def stale_possible_simple_keys(self) -> None: # Remove entries that are no longer possible simple keys. According to # the YAML specification, simple keys # - should be limited to a single line, # - should be no longer than 1024 characters. # Disabling this procedure will allow simple keys of any length and # height (may cause problems if indentation is broken though). for level in list(self.possible_simple_keys): key = self.possible_simple_keys[level] if key.line != self.reader.line or self.reader.index - key.index > 1024: if key.required: raise ScannerError( 'while scanning a simple key', key.mark, "could not find expected ':'", self.reader.get_mark(), ) del self.possible_simple_keys[level] def save_possible_simple_key(self) -> None: # The next token may start a simple key. We check if it's possible # and save its position. This function is called for # ALIAS, ANCHOR, TAG, SCALAR(flow), '[', and '{'. # Check if a simple key is required at the current position. required = not self.flow_level and self.indent == self.reader.column # The next token might be a simple key. Let's save it's number and # position. if self.allow_simple_key: self.remove_possible_simple_key() token_number = self.tokens_taken + len(self.tokens) key = SimpleKey( token_number, required, self.reader.index, self.reader.line, self.reader.column, self.reader.get_mark(), ) self.possible_simple_keys[self.flow_level] = key def remove_possible_simple_key(self) -> None: # Remove the saved possible key position at the current flow level. if self.flow_level in self.possible_simple_keys: key = self.possible_simple_keys[self.flow_level] if key.required: raise ScannerError( 'while scanning a simple key', key.mark, "could not find expected ':'", self.reader.get_mark(), ) del self.possible_simple_keys[self.flow_level] # Indentation functions. def unwind_indent(self, column: Any) -> None: # In flow context, tokens should respect indentation. # Actually the condition should be `self.indent >= column` according to # the spec. But this condition will prohibit intuitively correct # constructions such as # key : { # } # #### # if self.flow_level and self.indent > column: # raise ScannerError(None, None, # "invalid intendation or unclosed '[' or '{'", # self.reader.get_mark()) # In the flow context, indentation is ignored. We make the scanner less # restrictive then specification requires. if bool(self.flow_level): return # In block context, we may need to issue the BLOCK-END tokens. while self.indent > column: mark = self.reader.get_mark() self.indent = self.indents.pop() self.tokens.append(BlockEndToken(mark, mark)) def add_indent(self, column: int) -> bool: # Check if we need to increase indentation. if self.indent < column: self.indents.append(self.indent) self.indent = column return True return False # Fetchers. def fetch_stream_start(self) -> None: # We always add STREAM-START as the first token and STREAM-END as the # last token. # Read the token. mark = self.reader.get_mark() # Add STREAM-START. self.tokens.append(StreamStartToken(mark, mark, encoding=self.reader.encoding)) def fetch_stream_end(self) -> None: # Set the current intendation to -1. self.unwind_indent(-1) # Reset simple keys. self.remove_possible_simple_key() self.allow_simple_key = False self.possible_simple_keys = {} # Read the token. mark = self.reader.get_mark() # Add STREAM-END. self.tokens.append(StreamEndToken(mark, mark)) # The steam is finished. self.done = True def fetch_directive(self) -> None: # Set the current intendation to -1. self.unwind_indent(-1) # Reset simple keys. self.remove_possible_simple_key() self.allow_simple_key = False # Scan and add DIRECTIVE. self.tokens.append(self.scan_directive()) def fetch_document_start(self) -> None: self.fetch_document_indicator(DocumentStartToken) def fetch_document_end(self) -> None: self.fetch_document_indicator(DocumentEndToken) def fetch_document_indicator(self, TokenClass: Any) -> None: # Set the current intendation to -1. self.unwind_indent(-1) # Reset simple keys. Note that there could not be a block collection # after '---'. self.remove_possible_simple_key() self.allow_simple_key = False # Add DOCUMENT-START or DOCUMENT-END. start_mark = self.reader.get_mark() self.reader.forward(3) end_mark = self.reader.get_mark() self.tokens.append(TokenClass(start_mark, end_mark)) def fetch_flow_sequence_start(self) -> None: self.fetch_flow_collection_start(FlowSequenceStartToken, to_push='[') def fetch_flow_mapping_start(self) -> None: self.fetch_flow_collection_start(FlowMappingStartToken, to_push='{') def fetch_flow_collection_start(self, TokenClass: Any, to_push: Text) -> None: # '[' and '{' may start a simple key. self.save_possible_simple_key() # Increase the flow level. self.flow_context.append(to_push) # Simple keys are allowed after '[' and '{'. self.allow_simple_key = True # Add FLOW-SEQUENCE-START or FLOW-MAPPING-START. start_mark = self.reader.get_mark() self.reader.forward() end_mark = self.reader.get_mark() self.tokens.append(TokenClass(start_mark, end_mark)) def fetch_flow_sequence_end(self) -> None: self.fetch_flow_collection_end(FlowSequenceEndToken) def fetch_flow_mapping_end(self) -> None: self.fetch_flow_collection_end(FlowMappingEndToken) def fetch_flow_collection_end(self, TokenClass: Any) -> None: # Reset possible simple key on the current level. self.remove_possible_simple_key() # Decrease the flow level. try: popped = self.flow_context.pop() # NOQA except IndexError: # We must not be in a list or object. # Defer error handling to the parser. pass # No simple keys after ']' or '}'. self.allow_simple_key = False # Add FLOW-SEQUENCE-END or FLOW-MAPPING-END. start_mark = self.reader.get_mark() self.reader.forward() end_mark = self.reader.get_mark() self.tokens.append(TokenClass(start_mark, end_mark)) def fetch_flow_entry(self) -> None: # Simple keys are allowed after ','. self.allow_simple_key = True # Reset possible simple key on the current level. self.remove_possible_simple_key() # Add FLOW-ENTRY. start_mark = self.reader.get_mark() self.reader.forward() end_mark = self.reader.get_mark() self.tokens.append(FlowEntryToken(start_mark, end_mark)) def fetch_block_entry(self) -> None: # Block context needs additional checks. if not self.flow_level: # Are we allowed to start a new entry? if not self.allow_simple_key: raise ScannerError( None, None, 'sequence entries are not allowed here', self.reader.get_mark(), ) # We may need to add BLOCK-SEQUENCE-START. if self.add_indent(self.reader.column): mark = self.reader.get_mark() self.tokens.append(BlockSequenceStartToken(mark, mark)) # It's an error for the block entry to occur in the flow context, # but we let the parser detect this. else: pass # Simple keys are allowed after '-'. self.allow_simple_key = True # Reset possible simple key on the current level. self.remove_possible_simple_key() # Add BLOCK-ENTRY. start_mark = self.reader.get_mark() self.reader.forward() end_mark = self.reader.get_mark() self.tokens.append(BlockEntryToken(start_mark, end_mark)) def fetch_key(self) -> None: # Block context needs additional checks. if not self.flow_level: # Are we allowed to start a key (not nessesary a simple)? if not self.allow_simple_key: raise ScannerError( None, None, 'mapping keys are not allowed here', self.reader.get_mark(), ) # We may need to add BLOCK-MAPPING-START. if self.add_indent(self.reader.column): mark = self.reader.get_mark() self.tokens.append(BlockMappingStartToken(mark, mark)) # Simple keys are allowed after '?' in the block context. self.allow_simple_key = not self.flow_level # Reset possible simple key on the current level. self.remove_possible_simple_key() # Add KEY. start_mark = self.reader.get_mark() self.reader.forward() end_mark = self.reader.get_mark() self.tokens.append(KeyToken(start_mark, end_mark)) def fetch_value(self) -> None: # Do we determine a simple key? if self.flow_level in self.possible_simple_keys: # Add KEY. key = self.possible_simple_keys[self.flow_level] del self.possible_simple_keys[self.flow_level] self.tokens.insert( key.token_number - self.tokens_taken, KeyToken(key.mark, key.mark), ) # If this key starts a new block mapping, we need to add # BLOCK-MAPPING-START. if not self.flow_level: if self.add_indent(key.column): self.tokens.insert( key.token_number - self.tokens_taken, BlockMappingStartToken(key.mark, key.mark), ) # There cannot be two simple keys one after another. self.allow_simple_key = False # It must be a part of a complex key. else: # Block context needs additional checks. # (Do we really need them? They will be caught by the parser # anyway.) if not self.flow_level: # We are allowed to start a complex value if and only if # we can start a simple key. if not self.allow_simple_key: raise ScannerError( None, None, 'mapping values are not allowed here', self.reader.get_mark(), ) # If this value starts a new block mapping, we need to add # BLOCK-MAPPING-START. It will be detected as an error later by # the parser. if not self.flow_level: if self.add_indent(self.reader.column): mark = self.reader.get_mark() self.tokens.append(BlockMappingStartToken(mark, mark)) # Simple keys are allowed after ':' in the block context. self.allow_simple_key = not self.flow_level # Reset possible simple key on the current level. self.remove_possible_simple_key() # Add VALUE. start_mark = self.reader.get_mark() self.reader.forward() end_mark = self.reader.get_mark() self.tokens.append(ValueToken(start_mark, end_mark)) def fetch_alias(self) -> None: # ALIAS could be a simple key. self.save_possible_simple_key() # No simple keys after ALIAS. self.allow_simple_key = False # Scan and add ALIAS. self.tokens.append(self.scan_anchor(AliasToken)) def fetch_anchor(self) -> None: # ANCHOR could start a simple key. self.save_possible_simple_key() # No simple keys after ANCHOR. self.allow_simple_key = False # Scan and add ANCHOR. self.tokens.append(self.scan_anchor(AnchorToken)) def fetch_tag(self) -> None: # TAG could start a simple key. self.save_possible_simple_key() # No simple keys after TAG. self.allow_simple_key = False # Scan and add TAG. self.tokens.append(self.scan_tag()) def fetch_literal(self) -> None: self.fetch_block_scalar(style='|') def fetch_folded(self) -> None: self.fetch_block_scalar(style='>') def fetch_block_scalar(self, style: Any) -> None: # A simple key may follow a block scalar. self.allow_simple_key = True # Reset possible simple key on the current level. self.remove_possible_simple_key() # Scan and add SCALAR. self.tokens.append(self.scan_block_scalar(style)) def fetch_single(self) -> None: self.fetch_flow_scalar(style="'") def fetch_double(self) -> None: self.fetch_flow_scalar(style='"') def fetch_flow_scalar(self, style: Any) -> None: # A flow scalar could be a simple key. self.save_possible_simple_key() # No simple keys after flow scalars. self.allow_simple_key = False # Scan and add SCALAR. self.tokens.append(self.scan_flow_scalar(style)) def fetch_plain(self) -> None: # A plain scalar could be a simple key. self.save_possible_simple_key() # No simple keys after plain scalars. But note that `scan_plain` will # change this flag if the scan is finished at the beginning of the # line. self.allow_simple_key = False # Scan and add SCALAR. May change `allow_simple_key`. self.tokens.append(self.scan_plain()) # Checkers. def check_directive(self) -> Any: # DIRECTIVE: ^ '%' ... # The '%' indicator is already checked. if self.reader.column == 0: return True return None def check_document_start(self) -> Any: # DOCUMENT-START: ^ '---' (' '|'\n') if self.reader.column == 0: if self.reader.prefix(3) == '---' and self.reader.peek(3) in _THE_END_SPACE_TAB: return True return None def check_document_end(self) -> Any: # DOCUMENT-END: ^ '...' (' '|'\n') if self.reader.column == 0: if self.reader.prefix(3) == '...' and self.reader.peek(3) in _THE_END_SPACE_TAB: return True return None def check_block_entry(self) -> Any: # BLOCK-ENTRY: '-' (' '|'\n') return self.reader.peek(1) in _THE_END_SPACE_TAB def check_key(self) -> Any: # KEY(flow context): '?' if bool(self.flow_level): return True # KEY(block context): '?' (' '|'\n') return self.reader.peek(1) in _THE_END_SPACE_TAB def check_value(self) -> Any: # VALUE(flow context): ':' if self.scanner_processing_version == (1, 1): if bool(self.flow_level): return True else: if bool(self.flow_level): if self.flow_context[-1] == '[': if self.reader.peek(1) not in _THE_END_SPACE_TAB: return False elif self.tokens and isinstance(self.tokens[-1], ValueToken): # mapping flow context scanning a value token if self.reader.peek(1) not in _THE_END_SPACE_TAB: return False return True # VALUE(block context): ':' (' '|'\n') return self.reader.peek(1) in _THE_END_SPACE_TAB def check_plain(self) -> Any: # A plain scalar may start with any non-space character except: # '-', '?', ':', ',', '[', ']', '{', '}', # '#', '&', '*', '!', '|', '>', '\'', '\"', # '%', '@', '`'. # # It may also start with # '-', '?', ':' # if it is followed by a non-space character. # # Note that we limit the last rule to the block context (except the # '-' character) because we want the flow context to be space # independent. srp = self.reader.peek ch = srp() if self.scanner_processing_version == (1, 1): return ch not in '\0 \t\r\n\x85\u2028\u2029-?:,[]{}#&*!|>\'"%@`' or ( srp(1) not in _THE_END_SPACE_TAB and (ch == '-' or (not self.flow_level and ch in '?:')) ) # YAML 1.2 if ch not in '\0 \t\r\n\x85\u2028\u2029-?:,[]{}#&*!|>\'"%@`': # ################### ^ ??? return True ch1 = srp(1) if ch == '-' and ch1 not in _THE_END_SPACE_TAB: return True if ch == ':' and bool(self.flow_level) and ch1 not in _SPACE_TAB: return True return srp(1) not in _THE_END_SPACE_TAB and ( ch == '-' or (not self.flow_level and ch in '?:') ) # Scanners. def scan_to_next_token(self) -> Any: # We ignore spaces, line breaks and comments. # If we find a line break in the block context, we set the flag # `allow_simple_key` on. # The byte order mark is stripped if it's the first character in the # stream. We do not yet support BOM inside the stream as the # specification requires. Any such mark will be considered as a part # of the document. # # TODO: We need to make tab handling rules more sane. A good rule is # Tabs cannot precede tokens # BLOCK-SEQUENCE-START, BLOCK-MAPPING-START, BLOCK-END, # KEY(block), VALUE(block), BLOCK-ENTRY # So the checking code is # if : # self.allow_simple_keys = False # We also need to add the check for `allow_simple_keys == True` to # `unwind_indent` before issuing BLOCK-END. # Scanners for block, flow, and plain scalars need to be modified. srp = self.reader.peek srf = self.reader.forward if self.reader.index == 0 and srp() == '\uFEFF': srf() found = False _the_end = _THE_END white_space = ' \t' if self.flow_level > 0 else ' ' while not found: while srp() in white_space: srf() if srp() == '#': while srp() not in _the_end: srf() if self.scan_line_break(): if not self.flow_level: self.allow_simple_key = True else: found = True return None def scan_directive(self) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward start_mark = self.reader.get_mark() srf() name = self.scan_directive_name(start_mark) value = None if name == 'YAML': value = self.scan_yaml_directive_value(start_mark) end_mark = self.reader.get_mark() elif name == 'TAG': value = self.scan_tag_directive_value(start_mark) end_mark = self.reader.get_mark() else: end_mark = self.reader.get_mark() while srp() not in _THE_END: srf() self.scan_directive_ignored_line(start_mark) return DirectiveToken(name, value, start_mark, end_mark) def scan_directive_name(self, start_mark: Any) -> Any: # See the specification for details. length = 0 srp = self.reader.peek ch = srp(length) while '0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' or ch in '-_:.': length += 1 ch = srp(length) if not length: raise ScannerError( 'while scanning a directive', start_mark, f'expected alphabetic or numeric character, but found {ch!r}', self.reader.get_mark(), ) value = self.reader.prefix(length) self.reader.forward(length) ch = srp() if ch not in '\0 \r\n\x85\u2028\u2029': raise ScannerError( 'while scanning a directive', start_mark, f'expected alphabetic or numeric character, but found {ch!r}', self.reader.get_mark(), ) return value def scan_yaml_directive_value(self, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward while srp() == ' ': srf() major = self.scan_yaml_directive_number(start_mark) if srp() != '.': raise ScannerError( 'while scanning a directive', start_mark, f"expected a digit or '.', but found {srp()!r}", self.reader.get_mark(), ) srf() minor = self.scan_yaml_directive_number(start_mark) if srp() not in '\0 \r\n\x85\u2028\u2029': raise ScannerError( 'while scanning a directive', start_mark, f"expected a digit or '.', but found {srp()!r}", self.reader.get_mark(), ) self.yaml_version = (major, minor) self.loader.doc_infos[-1].doc_version = Version(major, minor) return self.yaml_version def scan_yaml_directive_number(self, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward ch = srp() if not ('0' <= ch <= '9'): raise ScannerError( 'while scanning a directive', start_mark, f'expected a digit, but found {ch!r}', self.reader.get_mark(), ) length = 0 while '0' <= srp(length) <= '9': length += 1 value = int(self.reader.prefix(length)) srf(length) return value def scan_tag_directive_value(self, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward while srp() == ' ': srf() handle = self.scan_tag_directive_handle(start_mark) while srp() == ' ': srf() prefix = self.scan_tag_directive_prefix(start_mark) ret_val = (handle, prefix) self.tag_directives.append(ret_val) return ret_val def scan_tag_directive_handle(self, start_mark: Any) -> Any: # See the specification for details. value = self.scan_tag_handle('directive', start_mark) ch = self.reader.peek() if ch != ' ': raise ScannerError( 'while scanning a directive', start_mark, f"expected ' ', but found {ch!r}", self.reader.get_mark(), ) return value def scan_tag_directive_prefix(self, start_mark: Any) -> Any: # See the specification for details. value = self.scan_tag_uri('directive', start_mark) ch = self.reader.peek() if ch not in '\0 \r\n\x85\u2028\u2029': raise ScannerError( 'while scanning a directive', start_mark, f"expected ' ', but found {ch!r}", self.reader.get_mark(), ) return value def scan_directive_ignored_line(self, start_mark: Any) -> None: # See the specification for details. srp = self.reader.peek srf = self.reader.forward while srp() == ' ': srf() if srp() == '#': while srp() not in _THE_END: srf() ch = srp() if ch not in _THE_END: raise ScannerError( 'while scanning a directive', start_mark, f'expected a comment or a line break, but found {ch!r}', self.reader.get_mark(), ) self.scan_line_break() def scan_anchor(self, TokenClass: Any) -> Any: # The specification does not restrict characters for anchors and # aliases. This may lead to problems, for instance, the document: # [ *alias, value ] # can be interpteted in two ways, as # [ "value" ] # and # [ *alias , "value" ] # Therefore we restrict aliases to numbers and ASCII letters. srp = self.reader.peek start_mark = self.reader.get_mark() indicator = srp() if indicator == '*': name = 'alias' else: name = 'anchor' self.reader.forward() length = 0 ch = srp(length) # while '0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' \ # or ch in '-_': while check_anchorname_char(ch): length += 1 ch = srp(length) if not length: raise ScannerError( f'while scanning an {name!s}', start_mark, f'expected alphabetic or numeric character, but found {ch!r}', self.reader.get_mark(), ) value = self.reader.prefix(length) self.reader.forward(length) # ch1 = ch # ch = srp() # no need to peek, ch is already set # assert ch1 == ch if ch not in '\0 \t\r\n\x85\u2028\u2029?:,[]{}%@`': raise ScannerError( f'while scanning an {name!s}', start_mark, f'expected alphabetic or numeric character, but found {ch!r}', self.reader.get_mark(), ) end_mark = self.reader.get_mark() return TokenClass(value, start_mark, end_mark) def scan_tag(self) -> Any: # See the specification for details. srp = self.reader.peek start_mark = self.reader.get_mark() ch = srp(1) short_handle = '!' if ch == '!': short_handle = '!!' self.reader.forward() srp = self.reader.peek ch = srp(1) if ch == '<': handle = None self.reader.forward(2) suffix = self.scan_tag_uri('tag', start_mark) if srp() != '>': raise ScannerError( 'while parsing a tag', start_mark, f"expected '>' but found {srp()!r}", self.reader.get_mark(), ) self.reader.forward() elif ch in _THE_END_SPACE_TAB: handle = None suffix = short_handle self.reader.forward() else: length = 1 use_handle = False while ch not in '\0 \r\n\x85\u2028\u2029': if ch == '!': use_handle = True break length += 1 ch = srp(length) handle = short_handle if use_handle: handle = self.scan_tag_handle('tag', start_mark) else: handle = short_handle self.reader.forward() suffix = self.scan_tag_uri('tag', start_mark) ch = srp() if ch not in '\0 \r\n\x85\u2028\u2029': raise ScannerError( 'while scanning a tag', start_mark, f"expected ' ', but found {ch!r}", self.reader.get_mark(), ) value = (handle, suffix) end_mark = self.reader.get_mark() return TagToken(value, start_mark, end_mark) def scan_block_scalar(self, style: Any, rt: Optional[bool] = False) -> Any: # See the specification for details. srp = self.reader.peek if style == '>': folded = True else: folded = False chunks: List[Any] = [] start_mark = self.reader.get_mark() # Scan the header. self.reader.forward() chomping, increment = self.scan_block_scalar_indicators(start_mark) # block scalar comment e.g. : |+ # comment text block_scalar_comment = self.scan_block_scalar_ignored_line(start_mark) # Determine the indentation level and go to the first non-empty line. min_indent = self.indent + 1 if increment is None: # no increment and top level, min_indent could be 0 if min_indent < 1 and ( style not in '|>' or (self.scanner_processing_version == (1, 1)) and getattr( self.loader, 'top_level_block_style_scalar_no_indent_error_1_1', False, ) ): min_indent = 1 breaks, max_indent, end_mark = self.scan_block_scalar_indentation() indent = max(min_indent, max_indent) else: if min_indent < 1: min_indent = 1 indent = min_indent + increment - 1 breaks, end_mark = self.scan_block_scalar_breaks(indent) line_break = "" # Scan the inner part of the block scalar. while self.reader.column == indent and srp() != '\0': chunks.extend(breaks) leading_non_space = srp() not in ' \t' length = 0 while srp(length) not in _THE_END: length += 1 chunks.append(self.reader.prefix(length)) self.reader.forward(length) line_break = self.scan_line_break() breaks, end_mark = self.scan_block_scalar_breaks(indent) if style in '|>' and min_indent == 0: # at the beginning of a line, if in block style see if # end of document/start_new_document if self.check_document_start() or self.check_document_end(): break if self.reader.column == indent and srp() != '\0': # Unfortunately, folding rules are ambiguous. # # This is the folding according to the specification: if rt and folded and line_break == '\n': chunks.append('\a') if folded and line_break == '\n' and leading_non_space and srp() not in ' \t': if not breaks: chunks.append(' ') else: chunks.append(line_break) # This is Clark Evans's interpretation (also in the spec # examples): # # if folded and line_break == '\n': # if not breaks: # if srp() not in ' \t': # chunks.append(' ') # else: # chunks.append(line_break) # else: # chunks.append(line_break) else: break # Process trailing line breaks. The 'chomping' setting determines # whether they are included in the value. trailing: List[Any] = [] if chomping in [None, True]: chunks.append(line_break) if chomping is True: chunks.extend(breaks) elif chomping in [None, False]: trailing.extend(breaks) # We are done. token = ScalarToken("".join(chunks), False, start_mark, end_mark, style) if self.loader is not None: comment_handler = getattr(self.loader, 'comment_handling', False) if comment_handler is None: if block_scalar_comment is not None: token.add_pre_comments([block_scalar_comment]) if len(trailing) > 0: # Eat whitespaces and comments until we reach the next token. if self.loader is not None: comment_handler = getattr(self.loader, 'comment_handling', None) if comment_handler is not None: line = end_mark.line - len(trailing) for x in trailing: assert x[-1] == '\n' self.comments.add_blank_line(x, 0, line) # type: ignore line += 1 comment = self.scan_to_next_token() while comment: trailing.append(' ' * comment[1].column + comment[0]) comment = self.scan_to_next_token() if self.loader is not None: comment_handler = getattr(self.loader, 'comment_handling', False) if comment_handler is None: # Keep track of the trailing whitespace and following comments # as a comment token, if isn't all included in the actual value. comment_end_mark = self.reader.get_mark() comment = CommentToken("".join(trailing), end_mark, comment_end_mark) token.add_post_comment(comment) return token def scan_block_scalar_indicators(self, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek chomping = None increment = None ch = srp() if ch in '+-': if ch == '+': chomping = True else: chomping = False self.reader.forward() ch = srp() if ch in '0123456789': increment = int(ch) if increment == 0: raise ScannerError( 'while scanning a block scalar', start_mark, 'expected indentation indicator in the range 1-9, ' 'but found 0', self.reader.get_mark(), ) self.reader.forward() elif ch in '0123456789': increment = int(ch) if increment == 0: raise ScannerError( 'while scanning a block scalar', start_mark, 'expected indentation indicator in the range 1-9, ' 'but found 0', self.reader.get_mark(), ) self.reader.forward() ch = srp() if ch in '+-': if ch == '+': chomping = True else: chomping = False self.reader.forward() ch = srp() if ch not in '\0 \r\n\x85\u2028\u2029': raise ScannerError( 'while scanning a block scalar', start_mark, f'expected chomping or indentation indicators, but found {ch!r}', self.reader.get_mark(), ) return chomping, increment def scan_block_scalar_ignored_line(self, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward prefix = '' comment = None while srp() == ' ': prefix += srp() srf() if srp() == '#': comment = prefix while srp() not in _THE_END: comment += srp() srf() ch = srp() if ch not in _THE_END: raise ScannerError( 'while scanning a block scalar', start_mark, f'expected a comment or a line break, but found {ch!r}', self.reader.get_mark(), ) self.scan_line_break() return comment def scan_block_scalar_indentation(self) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward chunks = [] first_indent = -1 max_indent = 0 end_mark = self.reader.get_mark() while srp() in ' \r\n\x85\u2028\u2029': if srp() != ' ': if first_indent < 0: first_indent = self.reader.column chunks.append(self.scan_line_break()) end_mark = self.reader.get_mark() else: srf() if self.reader.column > max_indent: max_indent = self.reader.column if first_indent > 0 and max_indent > first_indent: start_mark = self.reader.get_mark() raise ScannerError( 'more indented follow up line than first in a block scalar', start_mark, ) return chunks, max_indent, end_mark def scan_block_scalar_breaks(self, indent: int) -> Any: # See the specification for details. chunks = [] srp = self.reader.peek srf = self.reader.forward end_mark = self.reader.get_mark() while self.reader.column < indent and srp() == ' ': srf() while srp() in '\r\n\x85\u2028\u2029': chunks.append(self.scan_line_break()) end_mark = self.reader.get_mark() while self.reader.column < indent and srp() == ' ': srf() return chunks, end_mark def scan_flow_scalar(self, style: Any) -> Any: # See the specification for details. # Note that we loose indentation rules for quoted scalars. Quoted # scalars don't need to adhere indentation because " and ' clearly # mark the beginning and the end of them. Therefore we are less # restrictive then the specification requires. We only need to check # that document separators are not included in scalars. if style == '"': double = True else: double = False srp = self.reader.peek chunks: List[Any] = [] start_mark = self.reader.get_mark() quote = srp() self.reader.forward() chunks.extend(self.scan_flow_scalar_non_spaces(double, start_mark)) while srp() != quote: chunks.extend(self.scan_flow_scalar_spaces(double, start_mark)) chunks.extend(self.scan_flow_scalar_non_spaces(double, start_mark)) self.reader.forward() end_mark = self.reader.get_mark() return ScalarToken("".join(chunks), False, start_mark, end_mark, style) ESCAPE_REPLACEMENTS = { '0': '\0', 'a': '\x07', 'b': '\x08', 't': '\x09', '\t': '\x09', 'n': '\x0A', 'v': '\x0B', 'f': '\x0C', 'r': '\x0D', 'e': '\x1B', ' ': '\x20', '"': '"', '/': '/', # as per http://www.json.org/ '\\': '\\', 'N': '\x85', '_': '\xA0', 'L': '\u2028', 'P': '\u2029', } ESCAPE_CODES = {'x': 2, 'u': 4, 'U': 8} def scan_flow_scalar_non_spaces(self, double: Any, start_mark: Any) -> Any: # See the specification for details. chunks: List[Any] = [] srp = self.reader.peek srf = self.reader.forward while True: length = 0 while srp(length) not in ' \n\'"\\\0\t\r\x85\u2028\u2029': length += 1 if length != 0: chunks.append(self.reader.prefix(length)) srf(length) ch = srp() if not double and ch == "'" and srp(1) == "'": chunks.append("'") srf(2) elif (double and ch == "'") or (not double and ch in '"\\'): chunks.append(ch) srf() elif double and ch == '\\': srf() ch = srp() if ch in self.ESCAPE_REPLACEMENTS: chunks.append(self.ESCAPE_REPLACEMENTS[ch]) srf() elif ch in self.ESCAPE_CODES: length = self.ESCAPE_CODES[ch] srf() for k in range(length): if srp(k) not in '0123456789ABCDEFabcdef': raise ScannerError( 'while scanning a double-quoted scalar', start_mark, f'expected escape sequence of {length:d} ' f'hexdecimal numbers, but found {srp(k)!r}', self.reader.get_mark(), ) code = int(self.reader.prefix(length), 16) chunks.append(chr(code)) srf(length) elif ch in '\n\r\x85\u2028\u2029': self.scan_line_break() chunks.extend(self.scan_flow_scalar_breaks(double, start_mark)) else: raise ScannerError( 'while scanning a double-quoted scalar', start_mark, f'found unknown escape character {ch!r}', self.reader.get_mark(), ) else: return chunks def scan_flow_scalar_spaces(self, double: Any, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek chunks = [] length = 0 while srp(length) in ' \t': length += 1 whitespaces = self.reader.prefix(length) self.reader.forward(length) ch = srp() if ch == '\0': raise ScannerError( 'while scanning a quoted scalar', start_mark, 'found unexpected end of stream', self.reader.get_mark(), ) elif ch in '\r\n\x85\u2028\u2029': line_break = self.scan_line_break() breaks = self.scan_flow_scalar_breaks(double, start_mark) if line_break != '\n': chunks.append(line_break) elif not breaks: chunks.append(' ') chunks.extend(breaks) else: chunks.append(whitespaces) return chunks def scan_flow_scalar_breaks(self, double: Any, start_mark: Any) -> Any: # See the specification for details. chunks: List[Any] = [] srp = self.reader.peek srf = self.reader.forward while True: # Instead of checking indentation, we check for document # separators. prefix = self.reader.prefix(3) if (prefix == '---' or prefix == '...') and srp(3) in _THE_END_SPACE_TAB: raise ScannerError( 'while scanning a quoted scalar', start_mark, 'found unexpected document separator', self.reader.get_mark(), ) while srp() in ' \t': srf() if srp() in '\r\n\x85\u2028\u2029': chunks.append(self.scan_line_break()) else: return chunks def scan_plain(self) -> Any: # See the specification for details. # We add an additional restriction for the flow context: # plain scalars in the flow context cannot contain ',', ': ' and '?'. # We also keep track of the `allow_simple_key` flag here. # Indentation rules are loosed for the flow context. srp = self.reader.peek srf = self.reader.forward chunks: List[Any] = [] start_mark = self.reader.get_mark() end_mark = start_mark indent = self.indent + 1 # We allow zero indentation for scalars, but then we need to check for # document separators at the beginning of the line. # if indent == 0: # indent = 1 spaces: List[Any] = [] while True: length = 0 if srp() == '#': break while True: ch = srp(length) if False and ch == ':' and srp(length + 1) == ',': break elif ch == ':' and srp(length + 1) not in _THE_END_SPACE_TAB: pass elif ch == '?' and self.scanner_processing_version != (1, 1): pass elif ( ch in _THE_END_SPACE_TAB or ( not self.flow_level and ch == ':' and srp(length + 1) in _THE_END_SPACE_TAB ) or (self.flow_level and ch in ',:?[]{}') ): break length += 1 # It's not clear what we should do with ':' in the flow context. if ( self.flow_level and ch == ':' and srp(length + 1) not in '\0 \t\r\n\x85\u2028\u2029,[]{}' ): srf(length) raise ScannerError( 'while scanning a plain scalar', start_mark, "found unexpected ':'", self.reader.get_mark(), 'Please check ' 'http://pyyaml.org/wiki/YAMLColonInFlowContext ' 'for details.', ) if length == 0: break self.allow_simple_key = False chunks.extend(spaces) chunks.append(self.reader.prefix(length)) srf(length) end_mark = self.reader.get_mark() spaces = self.scan_plain_spaces(indent, start_mark) if ( not spaces or srp() == '#' or (not self.flow_level and self.reader.column < indent) ): break token = ScalarToken("".join(chunks), True, start_mark, end_mark) # getattr provides True so C type loader, which cannot handle comment, # will not make CommentToken if self.loader is not None: comment_handler = getattr(self.loader, 'comment_handling', False) if comment_handler is None: if spaces and spaces[0] == '\n': # Create a comment token to preserve the trailing line breaks. comment = CommentToken("".join(spaces) + '\n', start_mark, end_mark) token.add_post_comment(comment) elif comment_handler is not False: line = start_mark.line + 1 for ch in spaces: if ch == '\n': self.comments.add_blank_line('\n', 0, line) # type: ignore line += 1 return token def scan_plain_spaces(self, indent: Any, start_mark: Any) -> Any: # See the specification for details. # The specification is really confusing about tabs in plain scalars. # We just forbid them completely. Do not use tabs in YAML! srp = self.reader.peek srf = self.reader.forward chunks = [] length = 0 while srp(length) in ' ': length += 1 whitespaces = self.reader.prefix(length) self.reader.forward(length) ch = srp() if ch in '\r\n\x85\u2028\u2029': line_break = self.scan_line_break() self.allow_simple_key = True prefix = self.reader.prefix(3) if (prefix == '---' or prefix == '...') and srp(3) in _THE_END_SPACE_TAB: return breaks = [] while srp() in ' \r\n\x85\u2028\u2029': if srp() == ' ': srf() else: breaks.append(self.scan_line_break()) prefix = self.reader.prefix(3) if (prefix == '---' or prefix == '...') and srp(3) in _THE_END_SPACE_TAB: return if line_break != '\n': chunks.append(line_break) elif not breaks: chunks.append(' ') chunks.extend(breaks) elif whitespaces: chunks.append(whitespaces) return chunks def scan_tag_handle(self, name: Any, start_mark: Any) -> Any: # See the specification for details. # For some strange reasons, the specification does not allow '_' in # tag handles. I have allowed it anyway. srp = self.reader.peek ch = srp() if ch != '!': raise ScannerError( f'while scanning an {name!s}', start_mark, f"expected '!', but found {ch!r}", self.reader.get_mark(), ) length = 1 ch = srp(length) if ch != ' ': while '0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' or ch in '-_': length += 1 ch = srp(length) if ch != '!': self.reader.forward(length) raise ScannerError( f'while scanning an {name!s}', start_mark, f"expected '!' but found {ch!r}", self.reader.get_mark(), ) length += 1 value = self.reader.prefix(length) self.reader.forward(length) return value def scan_tag_uri(self, name: Any, start_mark: Any) -> Any: # See the specification for details. # Note: we do not check if URI is well-formed. srp = self.reader.peek chunks = [] length = 0 ch = srp(length) while ( '0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' or ch in "-;/?:@&=+$,_.!~*'()[]%" or ((self.scanner_processing_version > (1, 1)) and ch == '#') ): if ch == '%': chunks.append(self.reader.prefix(length)) self.reader.forward(length) length = 0 chunks.append(self.scan_uri_escapes(name, start_mark)) else: length += 1 ch = srp(length) if length != 0: chunks.append(self.reader.prefix(length)) self.reader.forward(length) length = 0 if not chunks: raise ScannerError( f'while parsing an {name!s}', start_mark, f'expected URI, but found {ch!r}', self.reader.get_mark(), ) return "".join(chunks) def scan_uri_escapes(self, name: Any, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward code_bytes: List[Any] = [] mark = self.reader.get_mark() while srp() == '%': srf() for k in range(2): if srp(k) not in '0123456789ABCDEFabcdef': raise ScannerError( f'while scanning an {name!s}', start_mark, f'expected URI escape sequence of 2 hexdecimal numbers, ' f'but found {srp(k)!r}', self.reader.get_mark(), ) code_bytes.append(int(self.reader.prefix(2), 16)) srf(2) try: value = bytes(code_bytes).decode('utf-8') except UnicodeDecodeError as exc: raise ScannerError(f'while scanning an {name!s}', start_mark, str(exc), mark) return value def scan_line_break(self) -> Any: # Transforms: # '\r\n' : '\n' # '\r' : '\n' # '\n' : '\n' # '\x85' : '\n' # '\u2028' : '\u2028' # '\u2029 : '\u2029' # default : '' ch = self.reader.peek() if ch in '\r\n\x85': if self.reader.prefix(2) == '\r\n': self.reader.forward(2) else: self.reader.forward() return '\n' elif ch in '\u2028\u2029': self.reader.forward() return ch return "" class RoundTripScanner(Scanner): def check_token(self, *choices: Any) -> bool: # Check if the next token is one of the given types. while self.need_more_tokens(): self.fetch_more_tokens() self._gather_comments() if len(self.tokens) > 0: if not choices: return True for choice in choices: if isinstance(self.tokens[0], choice): return True return False def peek_token(self) -> Any: # Return the next token, but do not delete if from the queue. while self.need_more_tokens(): self.fetch_more_tokens() self._gather_comments() if len(self.tokens) > 0: return self.tokens[0] return None def _gather_comments(self) -> Any: """combine multiple comment lines and assign to next non-comment-token""" comments: List[Any] = [] if not self.tokens: return comments if isinstance(self.tokens[0], CommentToken): comment = self.tokens.pop(0) self.tokens_taken += 1 comments.append(comment) while self.need_more_tokens(): self.fetch_more_tokens() if not self.tokens: return comments if isinstance(self.tokens[0], CommentToken): self.tokens_taken += 1 comment = self.tokens.pop(0) # nprint('dropping2', comment) comments.append(comment) if len(comments) >= 1: self.tokens[0].add_pre_comments(comments) # pull in post comment on e.g. ':' if not self.done and len(self.tokens) < 2: self.fetch_more_tokens() def get_token(self) -> Any: # Return the next token. while self.need_more_tokens(): self.fetch_more_tokens() self._gather_comments() if len(self.tokens) > 0: # nprint('tk', self.tokens) # only add post comment to single line tokens: # scalar, value token. FlowXEndToken, otherwise # hidden streamtokens could get them (leave them and they will be # pre comments for the next map/seq if ( len(self.tokens) > 1 and isinstance( self.tokens[0], (ScalarToken, ValueToken, FlowSequenceEndToken, FlowMappingEndToken), ) and isinstance(self.tokens[1], CommentToken) and self.tokens[0].end_mark.line == self.tokens[1].start_mark.line ): self.tokens_taken += 1 c = self.tokens.pop(1) self.fetch_more_tokens() while len(self.tokens) > 1 and isinstance(self.tokens[1], CommentToken): self.tokens_taken += 1 c1 = self.tokens.pop(1) c.value = c.value + (' ' * c1.start_mark.column) + c1.value self.fetch_more_tokens() self.tokens[0].add_post_comment(c) elif ( len(self.tokens) > 1 and isinstance(self.tokens[0], ScalarToken) and isinstance(self.tokens[1], CommentToken) and self.tokens[0].end_mark.line != self.tokens[1].start_mark.line ): self.tokens_taken += 1 c = self.tokens.pop(1) c.value = ( '\n' * (c.start_mark.line - self.tokens[0].end_mark.line) + (' ' * c.start_mark.column) + c.value ) self.tokens[0].add_post_comment(c) self.fetch_more_tokens() while len(self.tokens) > 1 and isinstance(self.tokens[1], CommentToken): self.tokens_taken += 1 c1 = self.tokens.pop(1) c.value = c.value + (' ' * c1.start_mark.column) + c1.value self.fetch_more_tokens() self.tokens_taken += 1 return self.tokens.pop(0) return None def fetch_comment(self, comment: Any) -> None: value, start_mark, end_mark = comment while value and value[-1] == ' ': # empty line within indented key context # no need to update end-mark, that is not used value = value[:-1] self.tokens.append(CommentToken(value, start_mark, end_mark)) # scanner def scan_to_next_token(self) -> Any: # We ignore spaces, line breaks and comments. # If we find a line break in the block context, we set the flag # `allow_simple_key` on. # The byte order mark is stripped if it's the first character in the # stream. We do not yet support BOM inside the stream as the # specification requires. Any such mark will be considered as a part # of the document. # # TODO: We need to make tab handling rules more sane. A good rule is # Tabs cannot precede tokens # BLOCK-SEQUENCE-START, BLOCK-MAPPING-START, BLOCK-END, # KEY(block), VALUE(block), BLOCK-ENTRY # So the checking code is # if : # self.allow_simple_keys = False # We also need to add the check for `allow_simple_keys == True` to # `unwind_indent` before issuing BLOCK-END. # Scanners for block, flow, and plain scalars need to be modified. srp = self.reader.peek srf = self.reader.forward if self.reader.index == 0 and srp() == '\uFEFF': srf() found = False white_space = ' \t' if self.flow_level > 0 else ' ' while not found: while srp() in white_space: srf() ch = srp() if ch == '#': start_mark = self.reader.get_mark() comment = ch srf() while ch not in _THE_END: ch = srp() if ch == '\0': # don't gobble the end-of-stream character # but add an explicit newline as "YAML processors should terminate # the stream with an explicit line break # https://yaml.org/spec/1.2/spec.html#id2780069 comment += '\n' break comment += ch srf() # gather any blank lines following the comment ch = self.scan_line_break() while len(ch) > 0: comment += ch ch = self.scan_line_break() end_mark = self.reader.get_mark() if not self.flow_level: self.allow_simple_key = True return comment, start_mark, end_mark if self.scan_line_break() != '': start_mark = self.reader.get_mark() if not self.flow_level: self.allow_simple_key = True ch = srp() if ch == '\n': # empty toplevel lines start_mark = self.reader.get_mark() comment = "" while ch: ch = self.scan_line_break(empty_line=True) comment += ch if srp() == '#': # empty line followed by indented real comment comment = comment.rsplit('\n', 1)[0] + '\n' end_mark = self.reader.get_mark() return comment, start_mark, end_mark else: found = True return None def scan_line_break(self, empty_line: bool = False) -> Text: # Transforms: # '\r\n' : '\n' # '\r' : '\n' # '\n' : '\n' # '\x85' : '\n' # '\u2028' : '\u2028' # '\u2029 : '\u2029' # default : '' ch: Text = self.reader.peek() if ch in '\r\n\x85': if self.reader.prefix(2) == '\r\n': self.reader.forward(2) else: self.reader.forward() return '\n' elif ch in '\u2028\u2029': self.reader.forward() return ch elif empty_line and ch in '\t ': self.reader.forward() return ch return "" def scan_block_scalar(self, style: Any, rt: Optional[bool] = True) -> Any: return Scanner.scan_block_scalar(self, style, rt=rt) def scan_uri_escapes(self, name: Any, start_mark: Any) -> Any: """ The roundtripscanner doesn't do URI escaping """ # See the specification for details. srp = self.reader.peek srf = self.reader.forward code_bytes: List[Any] = [] chunk = '' mark = self.reader.get_mark() while srp() == '%': chunk += '%' srf() for k in range(2): if srp(k) not in '0123456789ABCDEFabcdef': raise ScannerError( f'while scanning an {name!s}', start_mark, f'expected URI escape sequence of 2 hexdecimal numbers, ' f'but found {srp(k)!r}', self.reader.get_mark(), ) code_bytes.append(int(self.reader.prefix(2), 16)) chunk += self.reader.prefix(2) srf(2) try: _ = bytes(code_bytes).decode('utf-8') except UnicodeDecodeError as exc: raise ScannerError(f'while scanning an {name!s}', start_mark, str(exc), mark) return chunk # commenthandling 2021, differentiatiation not needed VALUECMNT = 0 KEYCMNT = 0 # 1 # TAGCMNT = 2 # ANCHORCMNT = 3 class CommentBase: __slots__ = ('value', 'line', 'column', 'used', 'function', 'fline', 'ufun', 'uline') def __init__(self, value: Any, line: Any, column: Any) -> None: self.value = value self.line = line self.column = column self.used = ' ' if _debug != 0: import inspect info = inspect.getframeinfo(inspect.stack()[3][0]) self.function = info.function self.fline = info.lineno self.ufun = None self.uline = None def set_used(self, v: Any = '+') -> None: self.used = v if _debug != 0: import inspect info = inspect.getframeinfo(inspect.stack()[1][0]) self.ufun = info.function # type: ignore self.uline = info.lineno # type: ignore def set_assigned(self) -> None: self.used = '|' def __str__(self) -> str: return f'{self.value}' def __repr__(self) -> str: return f'{self.value!r}' def info(self) -> str: xv = self.value + '"' name = self.name # type: ignore return ( f'{name}{self.used} {self.line:2}:{self.column:<2} "{xv:40s} ' f'{self.function}:{self.fline} {self.ufun}:{self.uline}' ) class EOLComment(CommentBase): name = 'EOLC' def __init__(self, value: Any, line: Any, column: Any) -> None: super().__init__(value, line, column) class FullLineComment(CommentBase): name = 'FULL' def __init__(self, value: Any, line: Any, column: Any) -> None: super().__init__(value, line, column) class BlankLineComment(CommentBase): name = 'BLNK' def __init__(self, value: Any, line: Any, column: Any) -> None: super().__init__(value, line, column) class ScannedComments: def __init__(self: Any) -> None: self.comments = {} # type: ignore self.unused = [] # type: ignore def add_eol_comment(self, comment: Any, column: Any, line: Any) -> Any: # info = inspect.getframeinfo(inspect.stack()[1][0]) if comment.count('\n') == 1: assert comment[-1] == '\n' else: assert '\n' not in comment self.comments[line] = retval = EOLComment(comment[:-1], line, column) self.unused.append(line) return retval def add_blank_line(self, comment: Any, column: Any, line: Any) -> Any: # info = inspect.getframeinfo(inspect.stack()[1][0]) assert comment.count('\n') == 1 and comment[-1] == '\n' assert line not in self.comments self.comments[line] = retval = BlankLineComment(comment[:-1], line, column) self.unused.append(line) return retval def add_full_line_comment(self, comment: Any, column: Any, line: Any) -> Any: # info = inspect.getframeinfo(inspect.stack()[1][0]) assert comment.count('\n') == 1 and comment[-1] == '\n' # if comment.startswith('# C12'): # raise # this raises in line 2127 fro 330 self.comments[line] = retval = FullLineComment(comment[:-1], line, column) self.unused.append(line) return retval def __getitem__(self, idx: Any) -> Any: return self.comments[idx] def __str__(self) -> Any: return ( 'ParsedComments:\n ' + '\n '.join((f'{lineno:2} {x.info()}' for lineno, x in self.comments.items())) + '\n' ) def last(self) -> str: lineno, x = list(self.comments.items())[-1] return f'{lineno:2} {x.info()}\n' def any_unprocessed(self) -> bool: # ToDo: might want to differentiate based on lineno return len(self.unused) > 0 # for lno, comment in reversed(self.comments.items()): # if comment.used == ' ': # return True # return False def unprocessed(self, use: Any = False) -> Any: while len(self.unused) > 0: if _debug != 0: import inspect first = self.unused.pop(0) if use else self.unused[0] info = inspect.getframeinfo(inspect.stack()[1][0]) xprintf('using', first, self.comments[first].value, info.function, info.lineno) yield first, self.comments[first] if use: self.comments[first].set_used() def assign_pre(self, token: Any) -> Any: token_line = token.start_mark.line if _debug != 0: import inspect info = inspect.getframeinfo(inspect.stack()[1][0]) xprintf('assign_pre', token_line, self.unused, info.function, info.lineno) gobbled = False while self.unused and self.unused[0] < token_line: gobbled = True first = self.unused.pop(0) if _debug != 0: xprintf('assign_pre < ', first) self.comments[first].set_used() token.add_comment_pre(first) return gobbled def assign_eol(self, tokens: Any) -> Any: try: comment_line = self.unused[0] except IndexError: return if not isinstance(self.comments[comment_line], EOLComment): return idx = 1 while tokens[-idx].start_mark.line > comment_line or isinstance( tokens[-idx], ValueToken, ): idx += 1 if _debug != 0: xprintf('idx1', idx) if ( len(tokens) > idx and isinstance(tokens[-idx], ScalarToken) and isinstance(tokens[-(idx + 1)], ScalarToken) ): return try: if isinstance(tokens[-idx], ScalarToken) and isinstance( tokens[-(idx + 1)], KeyToken, ): try: eol_idx = self.unused.pop(0) self.comments[eol_idx].set_used() if _debug != 0: xprintf('>>>>>a', idx, eol_idx, KEYCMNT) tokens[-idx].add_comment_eol(eol_idx, KEYCMNT) except IndexError: raise NotImplementedError return except IndexError: if _debug != 0: xprintf('IndexError1') pass try: if isinstance(tokens[-idx], ScalarToken) and isinstance( tokens[-(idx + 1)], (ValueToken, BlockEntryToken), ): try: eol_idx = self.unused.pop(0) self.comments[eol_idx].set_used() tokens[-idx].add_comment_eol(eol_idx, VALUECMNT) except IndexError: raise NotImplementedError return except IndexError: if _debug != 0: xprintf('IndexError2') pass for t in tokens: xprintf('tt-', t) if _debug != 0: xprintf('not implemented EOL', type(tokens[-idx])) import sys sys.exit(0) def assign_post(self, token: Any) -> Any: token_line = token.start_mark.line if _debug != 0: import inspect info = inspect.getframeinfo(inspect.stack()[1][0]) xprintf('assign_post', token_line, self.unused, info.function, info.lineno) gobbled = False while self.unused and self.unused[0] < token_line: gobbled = True first = self.unused.pop(0) if _debug != 0: xprintf('assign_post < ', first) self.comments[first].set_used() token.add_comment_post(first) return gobbled def str_unprocessed(self) -> Any: return ''.join( (f' {ind:2} {x.info()}\n' for ind, x in self.comments.items() if x.used == ' '), ) class RoundTripScannerSC(Scanner): # RoundTripScanner Split Comments def __init__(self, *arg: Any, **kw: Any) -> None: super().__init__(*arg, **kw) assert self.loader is not None # comments isinitialised on .need_more_tokens and persist on # self.loader.parsed_comments self.comments = None def get_token(self) -> Any: # Return the next token. while self.need_more_tokens(): self.fetch_more_tokens() if len(self.tokens) > 0: if isinstance(self.tokens[0], BlockEndToken): self.comments.assign_post(self.tokens[0]) # type: ignore else: self.comments.assign_pre(self.tokens[0]) # type: ignore self.tokens_taken += 1 return self.tokens.pop(0) def need_more_tokens(self) -> bool: if self.comments is None: self.loader.parsed_comments = self.comments = ScannedComments() # type: ignore if self.done: return False if len(self.tokens) == 0: return True # The current token may be a potential simple key, so we # need to look further. self.stale_possible_simple_keys() if self.next_possible_simple_key() == self.tokens_taken: return True if len(self.tokens) < 2: return True if self.tokens[0].start_mark.line == self.tokens[-1].start_mark.line: return True if True: if _debug != 0: xprintf('-x--', len(self.tokens)) for t in self.tokens: xprintf(t) # xprintf(self.comments.last()) xprintf(self.comments.str_unprocessed()) # type: ignore self.comments.assign_pre(self.tokens[0]) # type: ignore self.comments.assign_eol(self.tokens) # type: ignore return False def scan_to_next_token(self) -> None: srp = self.reader.peek srf = self.reader.forward if self.reader.index == 0 and srp() == '\uFEFF': srf() start_mark = self.reader.get_mark() # xprintf('current_mark', start_mark.line, start_mark.column) found = False while not found: while srp() == ' ': srf() ch = srp() if ch == '#': comment_start_mark = self.reader.get_mark() comment = ch srf() # skipt the '#' while ch not in _THE_END: ch = srp() if ch == '\0': # don't gobble the end-of-stream character # but add an explicit newline as "YAML processors should terminate # the stream with an explicit line break # https://yaml.org/spec/1.2/spec.html#id2780069 comment += '\n' break comment += ch srf() # we have a comment if start_mark.column == 0: self.comments.add_full_line_comment( # type: ignore comment, comment_start_mark.column, comment_start_mark.line, ) else: self.comments.add_eol_comment( # type: ignore comment, comment_start_mark.column, comment_start_mark.line, ) comment = "" # gather any blank lines or full line comments following the comment as well self.scan_empty_or_full_line_comments() if not self.flow_level: self.allow_simple_key = True return if bool(self.scan_line_break()): # start_mark = self.reader.get_mark() if not self.flow_level: self.allow_simple_key = True self.scan_empty_or_full_line_comments() return None ch = srp() if ch == '\n': # empty toplevel lines start_mark = self.reader.get_mark() comment = "" while ch: ch = self.scan_line_break(empty_line=True) comment += ch if srp() == '#': # empty line followed by indented real comment comment = comment.rsplit('\n', 1)[0] + '\n' _ = self.reader.get_mark() # gobble end_mark return None else: found = True return None def scan_empty_or_full_line_comments(self) -> None: blmark = self.reader.get_mark() assert blmark.column == 0 blanks = "" comment = None mark = None ch = self.reader.peek() while True: # nprint('ch', repr(ch), self.reader.get_mark().column) if ch in '\r\n\x85\u2028\u2029': if self.reader.prefix(2) == '\r\n': self.reader.forward(2) else: self.reader.forward() if comment is not None: comment += '\n' self.comments.add_full_line_comment(comment, mark.column, mark.line) comment = None else: blanks += '\n' self.comments.add_blank_line(blanks, blmark.column, blmark.line) # type: ignore # NOQA blanks = "" blmark = self.reader.get_mark() ch = self.reader.peek() continue if comment is None: if ch in ' \t': blanks += ch elif ch == '#': mark = self.reader.get_mark() comment = '#' else: # xprintf('breaking on', repr(ch)) break else: comment += ch self.reader.forward() ch = self.reader.peek() def scan_block_scalar_ignored_line(self, start_mark: Any) -> Any: # See the specification for details. srp = self.reader.peek srf = self.reader.forward prefix = '' comment = None while srp() == ' ': prefix += srp() srf() if srp() == '#': comment = '' mark = self.reader.get_mark() while srp() not in _THE_END: comment += srp() srf() comment += '\n' # type: ignore ch = srp() if ch not in _THE_END: raise ScannerError( 'while scanning a block scalar', start_mark, f'expected a comment or a line break, but found {ch!r}', self.reader.get_mark(), ) if comment is not None: self.comments.add_eol_comment(comment, mark.column, mark.line) # type: ignore self.scan_line_break() return None