Linux ip-148-66-134-25.ip.secureserver.net 3.10.0-1160.119.1.el7.tuxcare.els10.x86_64 #1 SMP Fri Oct 11 21:40:41 UTC 2024 x86_64
Apache
: 148.66.134.25 | : 18.221.175.48
66 Domain
8.0.30
amvm
www.github.com/MadExploits
Terminal
AUTO ROOT
Adminer
Backdoor Destroyer
Linux Exploit
Lock Shell
Lock File
Create User
CREATE RDP
PHP Mailer
BACKCONNECT
UNLOCK SHELL
HASH IDENTIFIER
CPANEL RESET
BLACK DEFEND!
README
+ Create Folder
+ Create File
/
usr /
lib64 /
python2.7 /
site-packages /
yaml /
[ HOME SHELL ]
Name
Size
Permission
Action
__init__.py
9.55
KB
-rw-r--r--
__init__.pyc
11.14
KB
-rw-r--r--
__init__.pyo
11.14
KB
-rw-r--r--
composer.py
4.81
KB
-rw-r--r--
composer.pyc
4.42
KB
-rw-r--r--
composer.pyo
4.42
KB
-rw-r--r--
constructor.py
24.56
KB
-rw-r--r--
constructor.pyc
21.39
KB
-rw-r--r--
constructor.pyo
21.39
KB
-rw-r--r--
cyaml.py
3.21
KB
-rw-r--r--
cyaml.pyc
3.72
KB
-rw-r--r--
cyaml.pyo
3.72
KB
-rw-r--r--
dumper.py
2.66
KB
-rw-r--r--
dumper.pyc
2.51
KB
-rw-r--r--
dumper.pyo
2.51
KB
-rw-r--r--
emitter.py
42.28
KB
-rw-r--r--
emitter.pyc
31.2
KB
-rw-r--r--
emitter.pyo
31.2
KB
-rw-r--r--
error.py
2.5
KB
-rw-r--r--
error.pyc
2.89
KB
-rw-r--r--
error.pyo
2.89
KB
-rw-r--r--
events.py
2.39
KB
-rw-r--r--
events.pyc
4.82
KB
-rw-r--r--
events.pyo
4.82
KB
-rw-r--r--
loader.py
1.11
KB
-rw-r--r--
loader.pyc
1.85
KB
-rw-r--r--
loader.pyo
1.85
KB
-rw-r--r--
nodes.py
1.41
KB
-rw-r--r--
nodes.pyc
2.13
KB
-rw-r--r--
nodes.pyo
2.13
KB
-rw-r--r--
parser.py
24.94
KB
-rw-r--r--
parser.pyc
14.37
KB
-rw-r--r--
parser.pyo
14.31
KB
-rw-r--r--
reader.py
6.59
KB
-rw-r--r--
reader.pyc
5.46
KB
-rw-r--r--
reader.pyo
5.46
KB
-rw-r--r--
representer.py
17.23
KB
-rw-r--r--
representer.pyc
14.32
KB
-rw-r--r--
representer.pyo
14.32
KB
-rw-r--r--
resolver.py
8.76
KB
-rw-r--r--
resolver.pyc
6.48
KB
-rw-r--r--
resolver.pyo
6.48
KB
-rw-r--r--
scanner.py
51.4
KB
-rw-r--r--
scanner.pyc
32.28
KB
-rw-r--r--
scanner.pyo
32.24
KB
-rw-r--r--
serializer.py
4.07
KB
-rw-r--r--
serializer.pyc
4.23
KB
-rw-r--r--
serializer.pyo
4.23
KB
-rw-r--r--
tokens.py
2.51
KB
-rw-r--r--
tokens.pyc
6.34
KB
-rw-r--r--
tokens.pyo
6.34
KB
-rw-r--r--
Delete
Unzip
Zip
${this.title}
Close
Code Editor : emitter.py
# Emitter expects events obeying the following grammar: # stream ::= STREAM-START document* STREAM-END # document ::= DOCUMENT-START node DOCUMENT-END # node ::= SCALAR | sequence | mapping # sequence ::= SEQUENCE-START node* SEQUENCE-END # mapping ::= MAPPING-START (node node)* MAPPING-END __all__ = ['Emitter', 'EmitterError'] from error import YAMLError from events import * class EmitterError(YAMLError): pass class ScalarAnalysis(object): def __init__(self, scalar, empty, multiline, allow_flow_plain, allow_block_plain, allow_single_quoted, allow_double_quoted, allow_block): self.scalar = scalar self.empty = empty self.multiline = multiline self.allow_flow_plain = allow_flow_plain self.allow_block_plain = allow_block_plain self.allow_single_quoted = allow_single_quoted self.allow_double_quoted = allow_double_quoted self.allow_block = allow_block class Emitter(object): DEFAULT_TAG_PREFIXES = { u'!' : u'!', u'tag:yaml.org,2002:' : u'!!', } def __init__(self, stream, canonical=None, indent=None, width=None, allow_unicode=None, line_break=None): # The stream should have the methods `write` and possibly `flush`. self.stream = stream # Encoding can be overriden by STREAM-START. self.encoding = None # Emitter is a state machine with a stack of states to handle nested # structures. self.states = [] self.state = self.expect_stream_start # Current event and the event queue. self.events = [] self.event = None # The current indentation level and the stack of previous indents. self.indents = [] self.indent = None # Flow level. self.flow_level = 0 # Contexts. self.root_context = False self.sequence_context = False self.mapping_context = False self.simple_key_context = False # Characteristics of the last emitted character: # - current position. # - is it a whitespace? # - is it an indention character # (indentation space, '-', '?', or ':')? self.line = 0 self.column = 0 self.whitespace = True self.indention = True # Whether the document requires an explicit document indicator self.open_ended = False # Formatting details. self.canonical = canonical self.allow_unicode = allow_unicode self.best_indent = 2 if indent and 1 < indent < 10: self.best_indent = indent self.best_width = 80 if width and width > self.best_indent*2: self.best_width = width self.best_line_break = u'\n' if line_break in [u'\r', u'\n', u'\r\n']: self.best_line_break = line_break # Tag prefixes. self.tag_prefixes = None # Prepared anchor and tag. self.prepared_anchor = None self.prepared_tag = None # Scalar analysis and style. self.analysis = None self.style = None def dispose(self): # Reset the state attributes (to clear self-references) self.states = [] self.state = None def emit(self, event): self.events.append(event) while not self.need_more_events(): self.event = self.events.pop(0) self.state() self.event = None # In some cases, we wait for a few next events before emitting. def need_more_events(self): if not self.events: return True event = self.events[0] if isinstance(event, DocumentStartEvent): return self.need_events(1) elif isinstance(event, SequenceStartEvent): return self.need_events(2) elif isinstance(event, MappingStartEvent): return self.need_events(3) else: return False def need_events(self, count): level = 0 for event in self.events[1:]: if isinstance(event, (DocumentStartEvent, CollectionStartEvent)): level += 1 elif isinstance(event, (DocumentEndEvent, CollectionEndEvent)): level -= 1 elif isinstance(event, StreamEndEvent): level = -1 if level < 0: return False return (len(self.events) < count+1) def increase_indent(self, flow=False, indentless=False): self.indents.append(self.indent) if self.indent is None: if flow: self.indent = self.best_indent else: self.indent = 0 elif not indentless: self.indent += self.best_indent # States. # Stream handlers. def expect_stream_start(self): if isinstance(self.event, StreamStartEvent): if self.event.encoding and not getattr(self.stream, 'encoding', None): self.encoding = self.event.encoding self.write_stream_start() self.state = self.expect_first_document_start else: raise EmitterError("expected StreamStartEvent, but got %s" % self.event) def expect_nothing(self): raise EmitterError("expected nothing, but got %s" % self.event) # Document handlers. def expect_first_document_start(self): return self.expect_document_start(first=True) def expect_document_start(self, first=False): if isinstance(self.event, DocumentStartEvent): if (self.event.version or self.event.tags) and self.open_ended: self.write_indicator(u'...', True) self.write_indent() if self.event.version: version_text = self.prepare_version(self.event.version) self.write_version_directive(version_text) self.tag_prefixes = self.DEFAULT_TAG_PREFIXES.copy() if self.event.tags: handles = self.event.tags.keys() handles.sort() for handle in handles: prefix = self.event.tags[handle] self.tag_prefixes[prefix] = handle handle_text = self.prepare_tag_handle(handle) prefix_text = self.prepare_tag_prefix(prefix) self.write_tag_directive(handle_text, prefix_text) implicit = (first and not self.event.explicit and not self.canonical and not self.event.version and not self.event.tags and not self.check_empty_document()) if not implicit: self.write_indent() self.write_indicator(u'---', True) if self.canonical: self.write_indent() self.state = self.expect_document_root elif isinstance(self.event, StreamEndEvent): if self.open_ended: self.write_indicator(u'...', True) self.write_indent() self.write_stream_end() self.state = self.expect_nothing else: raise EmitterError("expected DocumentStartEvent, but got %s" % self.event) def expect_document_end(self): if isinstance(self.event, DocumentEndEvent): self.write_indent() if self.event.explicit: self.write_indicator(u'...', True) self.write_indent() self.flush_stream() self.state = self.expect_document_start else: raise EmitterError("expected DocumentEndEvent, but got %s" % self.event) def expect_document_root(self): self.states.append(self.expect_document_end) self.expect_node(root=True) # Node handlers. def expect_node(self, root=False, sequence=False, mapping=False, simple_key=False): self.root_context = root self.sequence_context = sequence self.mapping_context = mapping self.simple_key_context = simple_key if isinstance(self.event, AliasEvent): self.expect_alias() elif isinstance(self.event, (ScalarEvent, CollectionStartEvent)): self.process_anchor(u'&') self.process_tag() if isinstance(self.event, ScalarEvent): self.expect_scalar() elif isinstance(self.event, SequenceStartEvent): if self.flow_level or self.canonical or self.event.flow_style \ or self.check_empty_sequence(): self.expect_flow_sequence() else: self.expect_block_sequence() elif isinstance(self.event, MappingStartEvent): if self.flow_level or self.canonical or self.event.flow_style \ or self.check_empty_mapping(): self.expect_flow_mapping() else: self.expect_block_mapping() else: raise EmitterError("expected NodeEvent, but got %s" % self.event) def expect_alias(self): if self.event.anchor is None: raise EmitterError("anchor is not specified for alias") self.process_anchor(u'*') self.state = self.states.pop() def expect_scalar(self): self.increase_indent(flow=True) self.process_scalar() self.indent = self.indents.pop() self.state = self.states.pop() # Flow sequence handlers. def expect_flow_sequence(self): self.write_indicator(u'[', True, whitespace=True) self.flow_level += 1 self.increase_indent(flow=True) self.state = self.expect_first_flow_sequence_item def expect_first_flow_sequence_item(self): if isinstance(self.event, SequenceEndEvent): self.indent = self.indents.pop() self.flow_level -= 1 self.write_indicator(u']', False) self.state = self.states.pop() else: if self.canonical or self.column > self.best_width: self.write_indent() self.states.append(self.expect_flow_sequence_item) self.expect_node(sequence=True) def expect_flow_sequence_item(self): if isinstance(self.event, SequenceEndEvent): self.indent = self.indents.pop() self.flow_level -= 1 if self.canonical: self.write_indicator(u',', False) self.write_indent() self.write_indicator(u']', False) self.state = self.states.pop() else: self.write_indicator(u',', False) if self.canonical or self.column > self.best_width: self.write_indent() self.states.append(self.expect_flow_sequence_item) self.expect_node(sequence=True) # Flow mapping handlers. def expect_flow_mapping(self): self.write_indicator(u'{', True, whitespace=True) self.flow_level += 1 self.increase_indent(flow=True) self.state = self.expect_first_flow_mapping_key def expect_first_flow_mapping_key(self): if isinstance(self.event, MappingEndEvent): self.indent = self.indents.pop() self.flow_level -= 1 self.write_indicator(u'}', False) self.state = self.states.pop() else: if self.canonical or self.column > self.best_width: self.write_indent() if not self.canonical and self.check_simple_key(): self.states.append(self.expect_flow_mapping_simple_value) self.expect_node(mapping=True, simple_key=True) else: self.write_indicator(u'?', True) self.states.append(self.expect_flow_mapping_value) self.expect_node(mapping=True) def expect_flow_mapping_key(self): if isinstance(self.event, MappingEndEvent): self.indent = self.indents.pop() self.flow_level -= 1 if self.canonical: self.write_indicator(u',', False) self.write_indent() self.write_indicator(u'}', False) self.state = self.states.pop() else: self.write_indicator(u',', False) if self.canonical or self.column > self.best_width: self.write_indent() if not self.canonical and self.check_simple_key(): self.states.append(self.expect_flow_mapping_simple_value) self.expect_node(mapping=True, simple_key=True) else: self.write_indicator(u'?', True) self.states.append(self.expect_flow_mapping_value) self.expect_node(mapping=True) def expect_flow_mapping_simple_value(self): self.write_indicator(u':', False) self.states.append(self.expect_flow_mapping_key) self.expect_node(mapping=True) def expect_flow_mapping_value(self): if self.canonical or self.column > self.best_width: self.write_indent() self.write_indicator(u':', True) self.states.append(self.expect_flow_mapping_key) self.expect_node(mapping=True) # Block sequence handlers. def expect_block_sequence(self): indentless = (self.mapping_context and not self.indention) self.increase_indent(flow=False, indentless=indentless) self.state = self.expect_first_block_sequence_item def expect_first_block_sequence_item(self): return self.expect_block_sequence_item(first=True) def expect_block_sequence_item(self, first=False): if not first and isinstance(self.event, SequenceEndEvent): self.indent = self.indents.pop() self.state = self.states.pop() else: self.write_indent() self.write_indicator(u'-', True, indention=True) self.states.append(self.expect_block_sequence_item) self.expect_node(sequence=True) # Block mapping handlers. def expect_block_mapping(self): self.increase_indent(flow=False) self.state = self.expect_first_block_mapping_key def expect_first_block_mapping_key(self): return self.expect_block_mapping_key(first=True) def expect_block_mapping_key(self, first=False): if not first and isinstance(self.event, MappingEndEvent): self.indent = self.indents.pop() self.state = self.states.pop() else: self.write_indent() if self.check_simple_key(): self.states.append(self.expect_block_mapping_simple_value) self.expect_node(mapping=True, simple_key=True) else: self.write_indicator(u'?', True, indention=True) self.states.append(self.expect_block_mapping_value) self.expect_node(mapping=True) def expect_block_mapping_simple_value(self): self.write_indicator(u':', False) self.states.append(self.expect_block_mapping_key) self.expect_node(mapping=True) def expect_block_mapping_value(self): self.write_indent() self.write_indicator(u':', True, indention=True) self.states.append(self.expect_block_mapping_key) self.expect_node(mapping=True) # Checkers. def check_empty_sequence(self): return (isinstance(self.event, SequenceStartEvent) and self.events and isinstance(self.events[0], SequenceEndEvent)) def check_empty_mapping(self): return (isinstance(self.event, MappingStartEvent) and self.events and isinstance(self.events[0], MappingEndEvent)) def check_empty_document(self): if not isinstance(self.event, DocumentStartEvent) or not self.events: return False event = self.events[0] return (isinstance(event, ScalarEvent) and event.anchor is None and event.tag is None and event.implicit and event.value == u'') def check_simple_key(self): length = 0 if isinstance(self.event, NodeEvent) and self.event.anchor is not None: if self.prepared_anchor is None: self.prepared_anchor = self.prepare_anchor(self.event.anchor) length += len(self.prepared_anchor) if isinstance(self.event, (ScalarEvent, CollectionStartEvent)) \ and self.event.tag is not None: if self.prepared_tag is None: self.prepared_tag = self.prepare_tag(self.event.tag) length += len(self.prepared_tag) if isinstance(self.event, ScalarEvent): if self.analysis is None: self.analysis = self.analyze_scalar(self.event.value) length += len(self.analysis.scalar) return (length < 128 and (isinstance(self.event, AliasEvent) or (isinstance(self.event, ScalarEvent) and not self.analysis.empty and not self.analysis.multiline) or self.check_empty_sequence() or self.check_empty_mapping())) # Anchor, Tag, and Scalar processors. def process_anchor(self, indicator): if self.event.anchor is None: self.prepared_anchor = None return if self.prepared_anchor is None: self.prepared_anchor = self.prepare_anchor(self.event.anchor) if self.prepared_anchor: self.write_indicator(indicator+self.prepared_anchor, True) self.prepared_anchor = None def process_tag(self): tag = self.event.tag if isinstance(self.event, ScalarEvent): if self.style is None: self.style = self.choose_scalar_style() if ((not self.canonical or tag is None) and ((self.style == '' and self.event.implicit[0]) or (self.style != '' and self.event.implicit[1]))): self.prepared_tag = None return if self.event.implicit[0] and tag is None: tag = u'!' self.prepared_tag = None else: if (not self.canonical or tag is None) and self.event.implicit: self.prepared_tag = None return if tag is None: raise EmitterError("tag is not specified") if self.prepared_tag is None: self.prepared_tag = self.prepare_tag(tag) if self.prepared_tag: self.write_indicator(self.prepared_tag, True) self.prepared_tag = None def choose_scalar_style(self): if self.analysis is None: self.analysis = self.analyze_scalar(self.event.value) if self.event.style == '"' or self.canonical: return '"' if not self.event.style and self.event.implicit[0]: if (not (self.simple_key_context and (self.analysis.empty or self.analysis.multiline)) and (self.flow_level and self.analysis.allow_flow_plain or (not self.flow_level and self.analysis.allow_block_plain))): return '' if self.event.style and self.event.style in '|>': if (not self.flow_level and not self.simple_key_context and self.analysis.allow_block): return self.event.style if not self.event.style or self.event.style == '\'': if (self.analysis.allow_single_quoted and not (self.simple_key_context and self.analysis.multiline)): return '\'' return '"' def process_scalar(self): if self.analysis is None: self.analysis = self.analyze_scalar(self.event.value) if self.style is None: self.style = self.choose_scalar_style() split = (not self.simple_key_context) #if self.analysis.multiline and split \ # and (not self.style or self.style in '\'\"'): # self.write_indent() if self.style == '"': self.write_double_quoted(self.analysis.scalar, split) elif self.style == '\'': self.write_single_quoted(self.analysis.scalar, split) elif self.style == '>': self.write_folded(self.analysis.scalar) elif self.style == '|': self.write_literal(self.analysis.scalar) else: self.write_plain(self.analysis.scalar, split) self.analysis = None self.style = None # Analyzers. def prepare_version(self, version): major, minor = version if major != 1: raise EmitterError("unsupported YAML version: %d.%d" % (major, minor)) return u'%d.%d' % (major, minor) def prepare_tag_handle(self, handle): if not handle: raise EmitterError("tag handle must not be empty") if handle[0] != u'!' or handle[-1] != u'!': raise EmitterError("tag handle must start and end with '!': %r" % (handle.encode('utf-8'))) for ch in handle[1:-1]: if not (u'0' <= ch <= u'9' or u'A' <= ch <= u'Z' or u'a' <= ch <= u'z' \ or ch in u'-_'): raise EmitterError("invalid character %r in the tag handle: %r" % (ch.encode('utf-8'), handle.encode('utf-8'))) return handle def prepare_tag_prefix(self, prefix): if not prefix: raise EmitterError("tag prefix must not be empty") chunks = [] start = end = 0 if prefix[0] == u'!': end = 1 while end < len(prefix): ch = prefix[end] if u'0' <= ch <= u'9' or u'A' <= ch <= u'Z' or u'a' <= ch <= u'z' \ or ch in u'-;/?!:@&=+$,_.~*\'()[]': end += 1 else: if start < end: chunks.append(prefix[start:end]) start = end = end+1 data = ch.encode('utf-8') for ch in data: chunks.append(u'%%%02X' % ord(ch)) if start < end: chunks.append(prefix[start:end]) return u''.join(chunks) def prepare_tag(self, tag): if not tag: raise EmitterError("tag must not be empty") if tag == u'!': return tag handle = None suffix = tag prefixes = self.tag_prefixes.keys() prefixes.sort() for prefix in prefixes: if tag.startswith(prefix) \ and (prefix == u'!' or len(prefix) < len(tag)): handle = self.tag_prefixes[prefix] suffix = tag[len(prefix):] chunks = [] start = end = 0 while end < len(suffix): ch = suffix[end] if u'0' <= ch <= u'9' or u'A' <= ch <= u'Z' or u'a' <= ch <= u'z' \ or ch in u'-;/?:@&=+$,_.~*\'()[]' \ or (ch == u'!' and handle != u'!'): end += 1 else: if start < end: chunks.append(suffix[start:end]) start = end = end+1 data = ch.encode('utf-8') for ch in data: chunks.append(u'%%%02X' % ord(ch)) if start < end: chunks.append(suffix[start:end]) suffix_text = u''.join(chunks) if handle: return u'%s%s' % (handle, suffix_text) else: return u'!<%s>' % suffix_text def prepare_anchor(self, anchor): if not anchor: raise EmitterError("anchor must not be empty") for ch in anchor: if not (u'0' <= ch <= u'9' or u'A' <= ch <= u'Z' or u'a' <= ch <= u'z' \ or ch in u'-_'): raise EmitterError("invalid character %r in the anchor: %r" % (ch.encode('utf-8'), anchor.encode('utf-8'))) return anchor def analyze_scalar(self, scalar): # Empty scalar is a special case. if not scalar: return ScalarAnalysis(scalar=scalar, empty=True, multiline=False, allow_flow_plain=False, allow_block_plain=True, allow_single_quoted=True, allow_double_quoted=True, allow_block=False) # Indicators and special characters. block_indicators = False flow_indicators = False line_breaks = False special_characters = False # Important whitespace combinations. leading_space = False leading_break = False trailing_space = False trailing_break = False break_space = False space_break = False # Check document indicators. if scalar.startswith(u'---') or scalar.startswith(u'...'): block_indicators = True flow_indicators = True # First character or preceded by a whitespace. preceeded_by_whitespace = True # Last character or followed by a whitespace. followed_by_whitespace = (len(scalar) == 1 or scalar[1] in u'\0 \t\r\n\x85\u2028\u2029') # The previous character is a space. previous_space = False # The previous character is a break. previous_break = False index = 0 while index < len(scalar): ch = scalar[index] # Check for indicators. if index == 0: # Leading indicators are special characters. if ch in u'#,[]{}&*!|>\'\"%@`': flow_indicators = True block_indicators = True if ch in u'?:': flow_indicators = True if followed_by_whitespace: block_indicators = True if ch == u'-' and followed_by_whitespace: flow_indicators = True block_indicators = True else: # Some indicators cannot appear within a scalar as well. if ch in u',?[]{}': flow_indicators = True if ch == u':': flow_indicators = True if followed_by_whitespace: block_indicators = True if ch == u'#' and preceeded_by_whitespace: flow_indicators = True block_indicators = True # Check for line breaks, special, and unicode characters. if ch in u'\n\x85\u2028\u2029': line_breaks = True if not (ch == u'\n' or u'\x20' <= ch <= u'\x7E'): if (ch == u'\x85' or u'\xA0' <= ch <= u'\uD7FF' or u'\uE000' <= ch <= u'\uFFFD') and ch != u'\uFEFF': unicode_characters = True if not self.allow_unicode: special_characters = True else: special_characters = True # Detect important whitespace combinations. if ch == u' ': if index == 0: leading_space = True if index == len(scalar)-1: trailing_space = True if previous_break: break_space = True previous_space = True previous_break = False elif ch in u'\n\x85\u2028\u2029': if index == 0: leading_break = True if index == len(scalar)-1: trailing_break = True if previous_space: space_break = True previous_space = False previous_break = True else: previous_space = False previous_break = False # Prepare for the next character. index += 1 preceeded_by_whitespace = (ch in u'\0 \t\r\n\x85\u2028\u2029') followed_by_whitespace = (index+1 >= len(scalar) or scalar[index+1] in u'\0 \t\r\n\x85\u2028\u2029') # Let's decide what styles are allowed. allow_flow_plain = True allow_block_plain = True allow_single_quoted = True allow_double_quoted = True allow_block = True # Leading and trailing whitespaces are bad for plain scalars. if (leading_space or leading_break or trailing_space or trailing_break): allow_flow_plain = allow_block_plain = False # We do not permit trailing spaces for block scalars. if trailing_space: allow_block = False # Spaces at the beginning of a new line are only acceptable for block # scalars. if break_space: allow_flow_plain = allow_block_plain = allow_single_quoted = False # Spaces followed by breaks, as well as special character are only # allowed for double quoted scalars. if space_break or special_characters: allow_flow_plain = allow_block_plain = \ allow_single_quoted = allow_block = False # Although the plain scalar writer supports breaks, we never emit # multiline plain scalars. if line_breaks: allow_flow_plain = allow_block_plain = False # Flow indicators are forbidden for flow plain scalars. if flow_indicators: allow_flow_plain = False # Block indicators are forbidden for block plain scalars. if block_indicators: allow_block_plain = False return ScalarAnalysis(scalar=scalar, empty=False, multiline=line_breaks, allow_flow_plain=allow_flow_plain, allow_block_plain=allow_block_plain, allow_single_quoted=allow_single_quoted, allow_double_quoted=allow_double_quoted, allow_block=allow_block) # Writers. def flush_stream(self): if hasattr(self.stream, 'flush'): self.stream.flush() def write_stream_start(self): # Write BOM if needed. if self.encoding and self.encoding.startswith('utf-16'): self.stream.write(u'\uFEFF'.encode(self.encoding)) def write_stream_end(self): self.flush_stream() def write_indicator(self, indicator, need_whitespace, whitespace=False, indention=False): if self.whitespace or not need_whitespace: data = indicator else: data = u' '+indicator self.whitespace = whitespace self.indention = self.indention and indention self.column += len(data) self.open_ended = False if self.encoding: data = data.encode(self.encoding) self.stream.write(data) def write_indent(self): indent = self.indent or 0 if not self.indention or self.column > indent \ or (self.column == indent and not self.whitespace): self.write_line_break() if self.column < indent: self.whitespace = True data = u' '*(indent-self.column) self.column = indent if self.encoding: data = data.encode(self.encoding) self.stream.write(data) def write_line_break(self, data=None): if data is None: data = self.best_line_break self.whitespace = True self.indention = True self.line += 1 self.column = 0 if self.encoding: data = data.encode(self.encoding) self.stream.write(data) def write_version_directive(self, version_text): data = u'%%YAML %s' % version_text if self.encoding: data = data.encode(self.encoding) self.stream.write(data) self.write_line_break() def write_tag_directive(self, handle_text, prefix_text): data = u'%%TAG %s %s' % (handle_text, prefix_text) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) self.write_line_break() # Scalar streams. def write_single_quoted(self, text, split=True): self.write_indicator(u'\'', True) spaces = False breaks = False start = end = 0 while end <= len(text): ch = None if end < len(text): ch = text[end] if spaces: if ch is None or ch != u' ': if start+1 == end and self.column > self.best_width and split \ and start != 0 and end != len(text): self.write_indent() else: data = text[start:end] self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end elif breaks: if ch is None or ch not in u'\n\x85\u2028\u2029': if text[start] == u'\n': self.write_line_break() for br in text[start:end]: if br == u'\n': self.write_line_break() else: self.write_line_break(br) self.write_indent() start = end else: if ch is None or ch in u' \n\x85\u2028\u2029' or ch == u'\'': if start < end: data = text[start:end] self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end if ch == u'\'': data = u'\'\'' self.column += 2 if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end + 1 if ch is not None: spaces = (ch == u' ') breaks = (ch in u'\n\x85\u2028\u2029') end += 1 self.write_indicator(u'\'', False) ESCAPE_REPLACEMENTS = { u'\0': u'0', u'\x07': u'a', u'\x08': u'b', u'\x09': u't', u'\x0A': u'n', u'\x0B': u'v', u'\x0C': u'f', u'\x0D': u'r', u'\x1B': u'e', u'\"': u'\"', u'\\': u'\\', u'\x85': u'N', u'\xA0': u'_', u'\u2028': u'L', u'\u2029': u'P', } def write_double_quoted(self, text, split=True): self.write_indicator(u'"', True) start = end = 0 while end <= len(text): ch = None if end < len(text): ch = text[end] if ch is None or ch in u'"\\\x85\u2028\u2029\uFEFF' \ or not (u'\x20' <= ch <= u'\x7E' or (self.allow_unicode and (u'\xA0' <= ch <= u'\uD7FF' or u'\uE000' <= ch <= u'\uFFFD'))): if start < end: data = text[start:end] self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end if ch is not None: if ch in self.ESCAPE_REPLACEMENTS: data = u'\\'+self.ESCAPE_REPLACEMENTS[ch] elif ch <= u'\xFF': data = u'\\x%02X' % ord(ch) elif ch <= u'\uFFFF': data = u'\\u%04X' % ord(ch) else: data = u'\\U%08X' % ord(ch) self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end+1 if 0 < end < len(text)-1 and (ch == u' ' or start >= end) \ and self.column+(end-start) > self.best_width and split: data = text[start:end]+u'\\' if start < end: start = end self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) self.write_indent() self.whitespace = False self.indention = False if text[start] == u' ': data = u'\\' self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) end += 1 self.write_indicator(u'"', False) def determine_block_hints(self, text): hints = u'' if text: if text[0] in u' \n\x85\u2028\u2029': hints += unicode(self.best_indent) if text[-1] not in u'\n\x85\u2028\u2029': hints += u'-' elif len(text) == 1 or text[-2] in u'\n\x85\u2028\u2029': hints += u'+' return hints def write_folded(self, text): hints = self.determine_block_hints(text) self.write_indicator(u'>'+hints, True) if hints[-1:] == u'+': self.open_ended = True self.write_line_break() leading_space = True spaces = False breaks = True start = end = 0 while end <= len(text): ch = None if end < len(text): ch = text[end] if breaks: if ch is None or ch not in u'\n\x85\u2028\u2029': if not leading_space and ch is not None and ch != u' ' \ and text[start] == u'\n': self.write_line_break() leading_space = (ch == u' ') for br in text[start:end]: if br == u'\n': self.write_line_break() else: self.write_line_break(br) if ch is not None: self.write_indent() start = end elif spaces: if ch != u' ': if start+1 == end and self.column > self.best_width: self.write_indent() else: data = text[start:end] self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end else: if ch is None or ch in u' \n\x85\u2028\u2029': data = text[start:end] self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) if ch is None: self.write_line_break() start = end if ch is not None: breaks = (ch in u'\n\x85\u2028\u2029') spaces = (ch == u' ') end += 1 def write_literal(self, text): hints = self.determine_block_hints(text) self.write_indicator(u'|'+hints, True) if hints[-1:] == u'+': self.open_ended = True self.write_line_break() breaks = True start = end = 0 while end <= len(text): ch = None if end < len(text): ch = text[end] if breaks: if ch is None or ch not in u'\n\x85\u2028\u2029': for br in text[start:end]: if br == u'\n': self.write_line_break() else: self.write_line_break(br) if ch is not None: self.write_indent() start = end else: if ch is None or ch in u'\n\x85\u2028\u2029': data = text[start:end] if self.encoding: data = data.encode(self.encoding) self.stream.write(data) if ch is None: self.write_line_break() start = end if ch is not None: breaks = (ch in u'\n\x85\u2028\u2029') end += 1 def write_plain(self, text, split=True): if self.root_context: self.open_ended = True if not text: return if not self.whitespace: data = u' ' self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) self.whitespace = False self.indention = False spaces = False breaks = False start = end = 0 while end <= len(text): ch = None if end < len(text): ch = text[end] if spaces: if ch != u' ': if start+1 == end and self.column > self.best_width and split: self.write_indent() self.whitespace = False self.indention = False else: data = text[start:end] self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end elif breaks: if ch not in u'\n\x85\u2028\u2029': if text[start] == u'\n': self.write_line_break() for br in text[start:end]: if br == u'\n': self.write_line_break() else: self.write_line_break(br) self.write_indent() self.whitespace = False self.indention = False start = end else: if ch is None or ch in u' \n\x85\u2028\u2029': data = text[start:end] self.column += len(data) if self.encoding: data = data.encode(self.encoding) self.stream.write(data) start = end if ch is not None: spaces = (ch == u' ') breaks = (ch in u'\n\x85\u2028\u2029') end += 1
Close