405 lines
12 KiB
Python
405 lines
12 KiB
Python
|
# coding: utf-8
|
||
|
|
|
||
|
|
from ruamel.yaml.compat import _F, nprintf # NOQA
|
||
|
|
|
||
|
|
if False: # MYPY
|
||
|
|
from typing import Text, Any, Dict, Optional, List # NOQA
|
||
|
|
from .error import StreamMark # NOQA
|
||
|
|
|
||
|
|
SHOW_LINES = True
|
||
|
|
|
||
|
|
|
||
|
|
class Token:
|
||
|
|
__slots__ = 'start_mark', 'end_mark', '_comment'
|
||
|
|
|
||
|
|
def __init__(self, start_mark, end_mark):
|
||
|
|
# type: (StreamMark, StreamMark) -> None
|
||
|
|
self.start_mark = start_mark
|
||
|
|
self.end_mark = end_mark
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
# type: () -> Any
|
||
|
|
# attributes = [key for key in self.__slots__ if not key.endswith('_mark') and
|
||
|
|
# hasattr('self', key)]
|
||
|
|
attributes = [key for key in self.__slots__ if not key.endswith('_mark')]
|
||
|
|
attributes.sort()
|
||
|
|
# arguments = ', '.join(
|
||
|
|
# [_F('{key!s}={gattr!r})', key=key, gattr=getattr(self, key)) for key in attributes]
|
||
|
|
# )
|
||
|
|
arguments = [
|
||
|
|
_F('{key!s}={gattr!r}', key=key, gattr=getattr(self, key)) for key in attributes
|
||
|
|
]
|
||
|
|
if SHOW_LINES:
|
||
|
|
try:
|
||
|
|
arguments.append('line: ' + str(self.start_mark.line))
|
||
|
|
except: # NOQA
|
||
|
|
pass
|
||
|
|
try:
|
||
|
|
arguments.append('comment: ' + str(self._comment))
|
||
|
|
except: # NOQA
|
||
|
|
pass
|
||
|
|
return '{}({})'.format(self.__class__.__name__, ', '.join(arguments))
|
||
|
|
|
||
|
|
@property
|
||
|
|
def column(self):
|
||
|
|
# type: () -> int
|
||
|
|
return self.start_mark.column
|
||
|
|
|
||
|
|
@column.setter
|
||
|
|
def column(self, pos):
|
||
|
|
# type: (Any) -> None
|
||
|
|
self.start_mark.column = pos
|
||
|
|
|
||
|
|
# old style ( <= 0.17) is a TWO element list with first being the EOL
|
||
|
|
# comment concatenated with following FLC/BLNK; and second being a list of FLC/BLNK
|
||
|
|
# preceding the token
|
||
|
|
# new style ( >= 0.17 ) is a THREE element list with the first being a list of
|
||
|
|
# preceding FLC/BLNK, the second EOL and the third following FLC/BLNK
|
||
|
|
# note that new style has differing order, and does not consist of CommentToken(s)
|
||
|
|
# but of CommentInfo instances
|
||
|
|
# any non-assigned values in new style are None, but first and last can be empty list
|
||
|
|
# new style routines add one comment at a time
|
||
|
|
|
||
|
|
# going to be deprecated in favour of add_comment_eol/post
|
||
|
|
def add_post_comment(self, comment):
|
||
|
|
# type: (Any) -> None
|
||
|
|
if not hasattr(self, '_comment'):
|
||
|
|
self._comment = [None, None]
|
||
|
|
else:
|
||
|
|
assert len(self._comment) in [2, 5] # make sure it is version 0
|
||
|
|
# if isinstance(comment, CommentToken):
|
||
|
|
# if comment.value.startswith('# C09'):
|
||
|
|
# raise
|
||
|
|
self._comment[0] = comment
|
||
|
|
|
||
|
|
# going to be deprecated in favour of add_comment_pre
|
||
|
|
def add_pre_comments(self, comments):
|
||
|
|
# type: (Any) -> None
|
||
|
|
if not hasattr(self, '_comment'):
|
||
|
|
self._comment = [None, None]
|
||
|
|
else:
|
||
|
|
assert len(self._comment) == 2 # make sure it is version 0
|
||
|
|
assert self._comment[1] is None
|
||
|
|
self._comment[1] = comments
|
||
|
|
return
|
||
|
|
|
||
|
|
# new style
|
||
|
|
def add_comment_pre(self, comment):
|
||
|
|
# type: (Any) -> None
|
||
|
|
if not hasattr(self, '_comment'):
|
||
|
|
self._comment = [[], None, None] # type: ignore
|
||
|
|
else:
|
||
|
|
assert len(self._comment) == 3
|
||
|
|
if self._comment[0] is None:
|
||
|
|
self._comment[0] = [] # type: ignore
|
||
|
|
self._comment[0].append(comment) # type: ignore
|
||
|
|
|
||
|
|
def add_comment_eol(self, comment, comment_type):
|
||
|
|
# type: (Any, Any) -> None
|
||
|
|
if not hasattr(self, '_comment'):
|
||
|
|
self._comment = [None, None, None]
|
||
|
|
else:
|
||
|
|
assert len(self._comment) == 3
|
||
|
|
assert self._comment[1] is None
|
||
|
|
if self.comment[1] is None:
|
||
|
|
self._comment[1] = [] # type: ignore
|
||
|
|
self._comment[1].extend([None] * (comment_type + 1 - len(self.comment[1]))) # type: ignore # NOQA
|
||
|
|
# nprintf('commy', self.comment, comment_type)
|
||
|
|
self._comment[1][comment_type] = comment # type: ignore
|
||
|
|
|
||
|
|
def add_comment_post(self, comment):
|
||
|
|
# type: (Any) -> None
|
||
|
|
if not hasattr(self, '_comment'):
|
||
|
|
self._comment = [None, None, []] # type: ignore
|
||
|
|
else:
|
||
|
|
assert len(self._comment) == 3
|
||
|
|
if self._comment[2] is None:
|
||
|
|
self._comment[2] = [] # type: ignore
|
||
|
|
self._comment[2].append(comment) # type: ignore
|
||
|
|
|
||
|
|
# def get_comment(self):
|
||
|
|
# # type: () -> Any
|
||
|
|
# return getattr(self, '_comment', None)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def comment(self):
|
||
|
|
# type: () -> Any
|
||
|
|
return getattr(self, '_comment', None)
|
||
|
|
|
||
|
|
def move_old_comment(self, target, empty=False):
|
||
|
|
# type: (Any, bool) -> Any
|
||
|
|
"""move a comment from this token to target (normally next token)
|
||
|
|
used to combine e.g. comments before a BlockEntryToken to the
|
||
|
|
ScalarToken that follows it
|
||
|
|
empty is a special for empty values -> comment after key
|
||
|
|
"""
|
||
|
|
c = self.comment
|
||
|
|
if c is None:
|
||
|
|
return
|
||
|
|
# don't push beyond last element
|
||
|
|
if isinstance(target, (StreamEndToken, DocumentStartToken)):
|
||
|
|
return
|
||
|
|
delattr(self, '_comment')
|
||
|
|
tc = target.comment
|
||
|
|
if not tc: # target comment, just insert
|
||
|
|
# special for empty value in key: value issue 25
|
||
|
|
if empty:
|
||
|
|
c = [c[0], c[1], None, None, c[0]]
|
||
|
|
target._comment = c
|
||
|
|
# nprint('mco2:', self, target, target.comment, empty)
|
||
|
|
return self
|
||
|
|
if c[0] and tc[0] or c[1] and tc[1]:
|
||
|
|
raise NotImplementedError(_F('overlap in comment {c!r} {tc!r}', c=c, tc=tc))
|
||
|
|
if c[0]:
|
||
|
|
tc[0] = c[0]
|
||
|
|
if c[1]:
|
||
|
|
tc[1] = c[1]
|
||
|
|
return self
|
||
|
|
|
||
|
|
def split_old_comment(self):
|
||
|
|
# type: () -> Any
|
||
|
|
""" split the post part of a comment, and return it
|
||
|
|
as comment to be added. Delete second part if [None, None]
|
||
|
|
abc: # this goes to sequence
|
||
|
|
# this goes to first element
|
||
|
|
- first element
|
||
|
|
"""
|
||
|
|
comment = self.comment
|
||
|
|
if comment is None or comment[0] is None:
|
||
|
|
return None # nothing to do
|
||
|
|
ret_val = [comment[0], None]
|
||
|
|
if comment[1] is None:
|
||
|
|
delattr(self, '_comment')
|
||
|
|
return ret_val
|
||
|
|
|
||
|
|
def move_new_comment(self, target, empty=False):
|
||
|
|
# type: (Any, bool) -> Any
|
||
|
|
"""move a comment from this token to target (normally next token)
|
||
|
|
used to combine e.g. comments before a BlockEntryToken to the
|
||
|
|
ScalarToken that follows it
|
||
|
|
empty is a special for empty values -> comment after key
|
||
|
|
"""
|
||
|
|
c = self.comment
|
||
|
|
if c is None:
|
||
|
|
return
|
||
|
|
# don't push beyond last element
|
||
|
|
if isinstance(target, (StreamEndToken, DocumentStartToken)):
|
||
|
|
return
|
||
|
|
delattr(self, '_comment')
|
||
|
|
tc = target.comment
|
||
|
|
if not tc: # target comment, just insert
|
||
|
|
# special for empty value in key: value issue 25
|
||
|
|
if empty:
|
||
|
|
c = [c[0], c[1], c[2]]
|
||
|
|
target._comment = c
|
||
|
|
# nprint('mco2:', self, target, target.comment, empty)
|
||
|
|
return self
|
||
|
|
# if self and target have both pre, eol or post comments, something seems wrong
|
||
|
|
for idx in range(3):
|
||
|
|
if c[idx] is not None and tc[idx] is not None:
|
||
|
|
raise NotImplementedError(_F('overlap in comment {c!r} {tc!r}', c=c, tc=tc))
|
||
|
|
# move the comment parts
|
||
|
|
for idx in range(3):
|
||
|
|
if c[idx]:
|
||
|
|
tc[idx] = c[idx]
|
||
|
|
return self
|
||
|
|
|
||
|
|
|
||
|
|
# class BOMToken(Token):
|
||
|
|
# id = '<byte order mark>'
|
||
|
|
|
||
|
|
|
||
|
|
class DirectiveToken(Token):
|
||
|
|
__slots__ = 'name', 'value'
|
||
|
|
id = '<directive>'
|
||
|
|
|
||
|
|
def __init__(self, name, value, start_mark, end_mark):
|
||
|
|
# type: (Any, Any, Any, Any) -> None
|
||
|
|
Token.__init__(self, start_mark, end_mark)
|
||
|
|
self.name = name
|
||
|
|
self.value = value
|
||
|
|
|
||
|
|
|
||
|
|
class DocumentStartToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '<document start>'
|
||
|
|
|
||
|
|
|
||
|
|
class DocumentEndToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '<document end>'
|
||
|
|
|
||
|
|
|
||
|
|
class StreamStartToken(Token):
|
||
|
|
__slots__ = ('encoding',)
|
||
|
|
id = '<stream start>'
|
||
|
|
|
||
|
|
def __init__(self, start_mark=None, end_mark=None, encoding=None):
|
||
|
|
# type: (Any, Any, Any) -> None
|
||
|
|
Token.__init__(self, start_mark, end_mark)
|
||
|
|
self.encoding = encoding
|
||
|
|
|
||
|
|
|
||
|
|
class StreamEndToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '<stream end>'
|
||
|
|
|
||
|
|
|
||
|
|
class BlockSequenceStartToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '<block sequence start>'
|
||
|
|
|
||
|
|
|
||
|
|
class BlockMappingStartToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '<block mapping start>'
|
||
|
|
|
||
|
|
|
||
|
|
class BlockEndToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '<block end>'
|
||
|
|
|
||
|
|
|
||
|
|
class FlowSequenceStartToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '['
|
||
|
|
|
||
|
|
|
||
|
|
class FlowMappingStartToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '{'
|
||
|
|
|
||
|
|
|
||
|
|
class FlowSequenceEndToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = ']'
|
||
|
|
|
||
|
|
|
||
|
|
class FlowMappingEndToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '}'
|
||
|
|
|
||
|
|
|
||
|
|
class KeyToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '?'
|
||
|
|
|
||
|
|
# def x__repr__(self):
|
||
|
|
# return 'KeyToken({})'.format(
|
||
|
|
# self.start_mark.buffer[self.start_mark.index:].split(None, 1)[0])
|
||
|
|
|
||
|
|
|
||
|
|
class ValueToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = ':'
|
||
|
|
|
||
|
|
|
||
|
|
class BlockEntryToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = '-'
|
||
|
|
|
||
|
|
|
||
|
|
class FlowEntryToken(Token):
|
||
|
|
__slots__ = ()
|
||
|
|
id = ','
|
||
|
|
|
||
|
|
|
||
|
|
class AliasToken(Token):
|
||
|
|
__slots__ = ('value',)
|
||
|
|
id = '<alias>'
|
||
|
|
|
||
|
|
def __init__(self, value, start_mark, end_mark):
|
||
|
|
# type: (Any, Any, Any) -> None
|
||
|
|
Token.__init__(self, start_mark, end_mark)
|
||
|
|
self.value = value
|
||
|
|
|
||
|
|
|
||
|
|
class AnchorToken(Token):
|
||
|
|
__slots__ = ('value',)
|
||
|
|
id = '<anchor>'
|
||
|
|
|
||
|
|
def __init__(self, value, start_mark, end_mark):
|
||
|
|
# type: (Any, Any, Any) -> None
|
||
|
|
Token.__init__(self, start_mark, end_mark)
|
||
|
|
self.value = value
|
||
|
|
|
||
|
|
|
||
|
|
class TagToken(Token):
|
||
|
|
__slots__ = ('value',)
|
||
|
|
id = '<tag>'
|
||
|
|
|
||
|
|
def __init__(self, value, start_mark, end_mark):
|
||
|
|
# type: (Any, Any, Any) -> None
|
||
|
|
Token.__init__(self, start_mark, end_mark)
|
||
|
|
self.value = value
|
||
|
|
|
||
|
|
|
||
|
|
class ScalarToken(Token):
|
||
|
|
__slots__ = 'value', 'plain', 'style'
|
||
|
|
id = '<scalar>'
|
||
|
|
|
||
|
|
def __init__(self, value, plain, start_mark, end_mark, style=None):
|
||
|
|
# type: (Any, Any, Any, Any, Any) -> None
|
||
|
|
Token.__init__(self, start_mark, end_mark)
|
||
|
|
self.value = value
|
||
|
|
self.plain = plain
|
||
|
|
self.style = style
|
||
|
|
|
||
|
|
|
||
|
|
class CommentToken(Token):
|
||
|
|
__slots__ = '_value', 'pre_done'
|
||
|
|
id = '<comment>'
|
||
|
|
|
||
|
|
def __init__(self, value, start_mark=None, end_mark=None, column=None):
|
||
|
|
# type: (Any, Any, Any, Any) -> None
|
||
|
|
if start_mark is None:
|
||
|
|
assert column is not None
|
||
|
|
self._column = column
|
||
|
|
Token.__init__(self, start_mark, None) # type: ignore
|
||
|
|
self._value = value
|
||
|
|
|
||
|
|
@property
|
||
|
|
def value(self):
|
||
|
|
# type: () -> str
|
||
|
|
if isinstance(self._value, str):
|
||
|
|
return self._value
|
||
|
|
return "".join(self._value)
|
||
|
|
|
||
|
|
@value.setter
|
||
|
|
def value(self, val):
|
||
|
|
# type: (Any) -> None
|
||
|
|
self._value = val
|
||
|
|
|
||
|
|
def reset(self):
|
||
|
|
# type: () -> None
|
||
|
|
if hasattr(self, 'pre_done'):
|
||
|
|
delattr(self, 'pre_done')
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
# type: () -> Any
|
||
|
|
v = '{!r}'.format(self.value)
|
||
|
|
if SHOW_LINES:
|
||
|
|
try:
|
||
|
|
v += ', line: ' + str(self.start_mark.line)
|
||
|
|
except: # NOQA
|
||
|
|
pass
|
||
|
|
try:
|
||
|
|
v += ', col: ' + str(self.start_mark.column)
|
||
|
|
except: # NOQA
|
||
|
|
pass
|
||
|
|
return 'CommentToken({})'.format(v)
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
# type: (Any) -> bool
|
||
|
|
if self.start_mark != other.start_mark:
|
||
|
|
return False
|
||
|
|
if self.end_mark != other.end_mark:
|
||
|
|
return False
|
||
|
|
if self.value != other.value:
|
||
|
|
return False
|
||
|
|
return True
|
||
|
|
|
||
|
|
def __ne__(self, other):
|
||
|
|
# type: (Any) -> bool
|
||
|
|
return not self.__eq__(other)
|