687 lines
25 KiB
Python
687 lines
25 KiB
Python
#!/usr/bin/env python3
|
||
"""ACK YAML 的零依赖、fail-closed 子集解析器。
|
||
|
||
这不是通用 YAML 实现。它只覆盖 ACK 状态文件所需的 mapping、
|
||
sequence、flow collection、标量和 ``>`` / ``|`` block scalar。锚点、
|
||
alias、tag、多文档和其它未实现语法会显式失败，不会猜测或静默误解析。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
|
||
class YamlSubsetError(ValueError):
|
||
"""YAML 超出 ACK 子集或语法无效。"""
|
||
|
||
|
||
class DuplicateKeyError(ValueError):
|
||
"""JSON/YAML mapping 包含重复键。"""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _Line:
|
||
number: int
|
||
indent: int
|
||
content: str
|
||
|
||
|
||
_DECIMAL_INT_RE = re.compile(r"[-+]?(?:0|[1-9][0-9]*)\Z")
|
||
_AMBIGUOUS_NUMBER_RE = re.compile(
|
||
r"[-+]?(?:"
|
||
r"[0-9][0-9_]*\.[0-9_]*(?:[eE][-+]?[0-9]+)?|"
|
||
r"[0-9][0-9_]*(?:[eE][-+]?[0-9]+)|"
|
||
r"0[xX][0-9a-fA-F_]+|0[oO][0-7_]+|0[bB][01_]+|"
|
||
r"0[0-9_]+|[0-9][0-9_]*:[0-9_:]+"
|
||
r")\Z"
|
||
)
|
||
_PLAIN_KEY_FORBIDDEN_RE = re.compile(r"[\[\]{},#]")
|
||
_ANCHOR_OR_ALIAS_RE = re.compile(
|
||
r"(?:^|\s)[&*][A-Za-z0-9_-]+(?:\s|$)"
|
||
)
|
||
|
||
|
||
def load_json_unique(content: str) -> Any:
|
||
"""Parse JSON while rejecting duplicate object keys at every depth."""
|
||
|
||
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
||
result: dict[str, Any] = {}
|
||
for key, value in pairs:
|
||
if key in result:
|
||
raise DuplicateKeyError(f"JSON 存在重复键 {key!r}")
|
||
result[key] = value
|
||
return result
|
||
|
||
return json.loads(content, object_pairs_hook=unique_object)
|
||
|
||
|
||
def make_unique_pyyaml_loader(yaml_module: Any) -> type:
|
||
"""Build a SafeLoader that rejects duplicate keys and graph features.
|
||
|
||
ACK documents are trees. Anchors/aliases can introduce shared identity or
|
||
cycles, which are unnecessary here and make recursive validation unsafe.
|
||
Explicit tags are also outside the fallback grammar, so both code paths
|
||
reject them consistently.
|
||
"""
|
||
|
||
class UniqueKeySafeLoader(yaml_module.SafeLoader): # type: ignore[misc]
|
||
def compose_node(self, parent: Any, index: Any) -> Any:
|
||
if self.check_event(yaml_module.events.AliasEvent):
|
||
event = self.peek_event()
|
||
raise yaml_module.YAMLError(
|
||
f"ACK YAML 不支持 alias: *{event.anchor}"
|
||
)
|
||
event = self.peek_event()
|
||
if getattr(event, "anchor", None) is not None:
|
||
raise yaml_module.YAMLError(
|
||
f"ACK YAML 不支持 anchor: &{event.anchor}"
|
||
)
|
||
if getattr(event, "tag", None) is not None:
|
||
raise yaml_module.YAMLError(
|
||
f"ACK YAML 不支持显式 tag: {event.tag}"
|
||
)
|
||
return super().compose_node(parent, index)
|
||
|
||
def construct_unique_mapping(
|
||
loader: Any,
|
||
node: Any,
|
||
deep: bool = False,
|
||
) -> dict[str, Any]:
|
||
for key_node, _ in node.value:
|
||
if (
|
||
getattr(key_node, "tag", None) == "tag:yaml.org,2002:merge"
|
||
or getattr(key_node, "value", None) == "<<"
|
||
):
|
||
raise yaml_module.constructor.ConstructorError(
|
||
"while constructing an ACK mapping",
|
||
node.start_mark,
|
||
"merge keys are not supported",
|
||
key_node.start_mark,
|
||
)
|
||
loader.flatten_mapping(node)
|
||
mapping: dict[str, Any] = {}
|
||
for key_node, value_node in node.value:
|
||
key = loader.construct_object(key_node, deep=deep)
|
||
if not isinstance(key, str):
|
||
raise yaml_module.constructor.ConstructorError(
|
||
"while constructing an ACK mapping",
|
||
node.start_mark,
|
||
"mapping key must be a string",
|
||
key_node.start_mark,
|
||
)
|
||
if key == "<<":
|
||
raise yaml_module.constructor.ConstructorError(
|
||
"while constructing an ACK mapping",
|
||
node.start_mark,
|
||
"merge keys are not supported",
|
||
key_node.start_mark,
|
||
)
|
||
if key in mapping:
|
||
raise yaml_module.constructor.ConstructorError(
|
||
"while constructing an ACK mapping",
|
||
node.start_mark,
|
||
f"found duplicate key {key!r}",
|
||
key_node.start_mark,
|
||
)
|
||
mapping[key] = loader.construct_object(value_node, deep=deep)
|
||
return mapping
|
||
|
||
UniqueKeySafeLoader.add_constructor(
|
||
yaml_module.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
|
||
construct_unique_mapping,
|
||
)
|
||
return UniqueKeySafeLoader
|
||
|
||
|
||
def load_yaml_subset(content: str) -> Any:
|
||
"""Parse the deliberately small YAML subset used by ACK files."""
|
||
|
||
return _SubsetParser(content).parse()
|
||
|
||
|
||
class _SubsetParser:
|
||
def __init__(self, content: str) -> None:
|
||
if content.startswith("\ufeff"):
|
||
content = content[1:]
|
||
if "\t" in content:
|
||
raise YamlSubsetError("ACK YAML 子集不支持 Tab，请使用空格")
|
||
self.lines = [
|
||
_Line(number, len(raw) - len(raw.lstrip(" ")), raw.lstrip(" "))
|
||
for number, raw in enumerate(content.splitlines(), start=1)
|
||
]
|
||
self.index = 0
|
||
|
||
def parse(self) -> Any:
|
||
self._skip_insignificant()
|
||
if self.index >= len(self.lines):
|
||
return None
|
||
first = self.lines[self.index]
|
||
if first.indent != 0:
|
||
self._error(first, "顶层不能缩进")
|
||
value = self._parse_node(0)
|
||
self._skip_insignificant()
|
||
if self.index != len(self.lines):
|
||
line = self.lines[self.index]
|
||
self._error(line, "文档尾部存在无法解析的内容")
|
||
return value
|
||
|
||
def _parse_node(self, indent: int) -> Any:
|
||
self._skip_insignificant()
|
||
if self.index >= len(self.lines):
|
||
return None
|
||
line = self.lines[self.index]
|
||
if line.indent != indent:
|
||
self._error(line, f"期望 {indent} 个空格的缩进")
|
||
content = self._without_comment(line.content).rstrip()
|
||
self._reject_document_syntax(content, line)
|
||
if self._is_sequence_marker(content):
|
||
return self._parse_sequence(indent)
|
||
return self._parse_mapping(indent)
|
||
|
||
def _parse_mapping(self, indent: int) -> dict[str, Any]:
|
||
result: dict[str, Any] = {}
|
||
while True:
|
||
self._skip_insignificant()
|
||
if self.index >= len(self.lines):
|
||
break
|
||
line = self.lines[self.index]
|
||
if line.indent < indent:
|
||
break
|
||
if line.indent > indent:
|
||
self._error(line, "mapping 存在意外缩进")
|
||
content = self._without_comment(line.content).rstrip()
|
||
self._reject_document_syntax(content, line)
|
||
if self._is_sequence_marker(content):
|
||
break
|
||
self.index += 1
|
||
self._consume_mapping_entry(
|
||
result,
|
||
content,
|
||
mapping_indent=indent,
|
||
line=line,
|
||
)
|
||
return result
|
||
|
||
def _parse_sequence(self, indent: int) -> list[Any]:
|
||
result: list[Any] = []
|
||
while True:
|
||
self._skip_insignificant()
|
||
if self.index >= len(self.lines):
|
||
break
|
||
line = self.lines[self.index]
|
||
if line.indent < indent:
|
||
break
|
||
if line.indent > indent:
|
||
self._error(line, "sequence 存在意外缩进")
|
||
content = self._without_comment(line.content).rstrip()
|
||
self._reject_document_syntax(content, line)
|
||
if not self._is_sequence_marker(content):
|
||
break
|
||
rest = content[1:].lstrip(" ")
|
||
self.index += 1
|
||
if not rest:
|
||
next_line = self._peek_significant()
|
||
if next_line is not None and next_line.indent > indent:
|
||
result.append(self._parse_node(next_line.indent))
|
||
else:
|
||
result.append(None)
|
||
continue
|
||
if self._is_sequence_marker(rest):
|
||
self._error(
|
||
line,
|
||
"不支持紧凑嵌套 sequence，请把内层 '-' 放到下一行",
|
||
)
|
||
if self._looks_like_mapping_entry(rest):
|
||
mapping_indent = indent + 2
|
||
item: dict[str, Any] = {}
|
||
self._consume_mapping_entry(
|
||
item,
|
||
rest,
|
||
mapping_indent=mapping_indent,
|
||
line=line,
|
||
)
|
||
while True:
|
||
self._skip_insignificant()
|
||
continuation = self._peek_significant()
|
||
if continuation is None or continuation.indent < mapping_indent:
|
||
break
|
||
if continuation.indent > mapping_indent:
|
||
self._error(
|
||
continuation,
|
||
"sequence mapping 存在意外缩进",
|
||
)
|
||
continuation_content = self._without_comment(
|
||
continuation.content
|
||
).rstrip()
|
||
if self._is_sequence_marker(continuation_content):
|
||
self._error(
|
||
continuation,
|
||
"sequence mapping 中需要 key: value",
|
||
)
|
||
self.index += 1
|
||
self._consume_mapping_entry(
|
||
item,
|
||
continuation_content,
|
||
mapping_indent=mapping_indent,
|
||
line=continuation,
|
||
)
|
||
result.append(item)
|
||
continue
|
||
if rest in {">", "|"}:
|
||
result.append(self._parse_block_scalar(indent, rest, line))
|
||
else:
|
||
result.append(self._parse_inline_value(rest, line))
|
||
next_line = self._peek_significant()
|
||
if next_line is not None and next_line.indent > indent:
|
||
self._error(next_line, "scalar sequence 项后存在意外缩进")
|
||
return result
|
||
|
||
def _consume_mapping_entry(
|
||
self,
|
||
result: dict[str, Any],
|
||
content: str,
|
||
*,
|
||
mapping_indent: int,
|
||
line: _Line,
|
||
) -> None:
|
||
key_text, value_text = self._split_mapping_entry(content, line)
|
||
key = self._parse_key(key_text, line)
|
||
if key in result:
|
||
self._error(line, f"mapping 存在重复键 {key!r}")
|
||
|
||
value_text = value_text.strip()
|
||
if value_text in {">", "|"}:
|
||
value = self._parse_block_scalar(mapping_indent, value_text, line)
|
||
elif value_text:
|
||
if value_text.startswith((">", "|")):
|
||
self._error(
|
||
line,
|
||
"block scalar 只支持 '>' 或 '|'，不支持 chomping/indent 指示符",
|
||
)
|
||
value = self._parse_inline_value(value_text, line)
|
||
next_line = self._peek_significant()
|
||
if next_line is not None and next_line.indent > mapping_indent:
|
||
self._error(next_line, f"{key!r} 的 scalar 后存在意外缩进")
|
||
else:
|
||
next_line = self._peek_significant()
|
||
if next_line is not None and next_line.indent > mapping_indent:
|
||
value = self._parse_node(next_line.indent)
|
||
else:
|
||
value = None
|
||
result[key] = value
|
||
|
||
def _parse_block_scalar(
|
||
self,
|
||
parent_indent: int,
|
||
style: str,
|
||
header: _Line,
|
||
) -> str:
|
||
probe = self.index
|
||
while probe < len(self.lines) and not self.lines[probe].content.strip():
|
||
probe += 1
|
||
if probe >= len(self.lines) or self.lines[probe].indent <= parent_indent:
|
||
return ""
|
||
block_indent = self.lines[probe].indent
|
||
if block_indent <= parent_indent:
|
||
self._error(header, "block scalar 内容必须比键更深缩进")
|
||
|
||
values: list[str] = []
|
||
while self.index < len(self.lines):
|
||
line = self.lines[self.index]
|
||
if not line.content.strip():
|
||
values.append("")
|
||
self.index += 1
|
||
continue
|
||
if line.indent < block_indent:
|
||
break
|
||
values.append(" " * (line.indent - block_indent) + line.content)
|
||
self.index += 1
|
||
|
||
while values and values[-1] == "":
|
||
values.pop()
|
||
if not values:
|
||
return ""
|
||
if style == "|":
|
||
return "\n".join(values) + "\n"
|
||
|
||
output = ""
|
||
previous: str | None = None
|
||
blank_count = 0
|
||
for value in values:
|
||
if value == "":
|
||
blank_count += 1
|
||
continue
|
||
if previous is None:
|
||
output = "\n" * blank_count + value
|
||
elif blank_count:
|
||
output += "\n" * blank_count + value
|
||
elif previous.startswith(" ") or value.startswith(" "):
|
||
output += "\n" + value
|
||
else:
|
||
output += " " + value
|
||
previous = value
|
||
blank_count = 0
|
||
return output + "\n"
|
||
|
||
def _parse_key(self, text: str, line: _Line) -> str:
|
||
text = text.strip()
|
||
if not text:
|
||
self._error(line, "mapping key 不能为空")
|
||
if text[0] in {'"', "'"}:
|
||
parsed = _FlowParser(text, line.number).parse_complete_value()
|
||
if not isinstance(parsed, str):
|
||
self._error(line, "mapping key 必须是字符串")
|
||
return parsed
|
||
if text == "<<":
|
||
self._error(line, "ACK YAML 子集不支持 merge key '<<'")
|
||
if text[0] in "-?:!&*%@`" or _PLAIN_KEY_FORBIDDEN_RE.search(text):
|
||
self._error(line, f"不支持的 plain mapping key {text!r}")
|
||
parsed = _plain_scalar(text, line.number)
|
||
if not isinstance(parsed, str):
|
||
self._error(line, "mapping key 必须是字符串，特殊标量请加引号")
|
||
return text
|
||
|
||
def _parse_inline_value(self, text: str, line: _Line) -> Any:
|
||
parser = _FlowParser(text, line.number)
|
||
return parser.parse_complete_value()
|
||
|
||
def _split_mapping_entry(self, content: str, line: _Line) -> tuple[str, str]:
|
||
quote: str | None = None
|
||
escaped = False
|
||
depth = 0
|
||
index = 0
|
||
while index < len(content):
|
||
char = content[index]
|
||
if quote == '"':
|
||
if escaped:
|
||
escaped = False
|
||
elif char == "\\":
|
||
escaped = True
|
||
elif char == quote:
|
||
quote = None
|
||
index += 1
|
||
continue
|
||
if quote == "'":
|
||
if char == "'":
|
||
if index + 1 < len(content) and content[index + 1] == "'":
|
||
index += 2
|
||
continue
|
||
quote = None
|
||
index += 1
|
||
continue
|
||
if char in {'"', "'"}:
|
||
quote = char
|
||
elif char in "[{":
|
||
depth += 1
|
||
elif char in "]}":
|
||
depth -= 1
|
||
if depth < 0:
|
||
self._error(line, "flow collection 括号不匹配")
|
||
elif (
|
||
char == ":"
|
||
and depth == 0
|
||
and (index + 1 == len(content) or content[index + 1].isspace())
|
||
):
|
||
return content[:index], content[index + 1 :]
|
||
index += 1
|
||
self._error(line, "mapping 项必须使用 'key: value'")
|
||
|
||
def _looks_like_mapping_entry(self, content: str) -> bool:
|
||
try:
|
||
self._split_mapping_entry(content, self.lines[self.index - 1])
|
||
except YamlSubsetError:
|
||
return False
|
||
return True
|
||
|
||
@staticmethod
|
||
def _is_sequence_marker(content: str) -> bool:
|
||
return content == "-" or content.startswith("- ")
|
||
|
||
def _peek_significant(self) -> _Line | None:
|
||
probe = self.index
|
||
while probe < len(self.lines):
|
||
line = self.lines[probe]
|
||
if line.content.strip() and not line.content.lstrip().startswith("#"):
|
||
return line
|
||
probe += 1
|
||
return None
|
||
|
||
def _skip_insignificant(self) -> None:
|
||
while self.index < len(self.lines):
|
||
content = self.lines[self.index].content
|
||
if content.strip() and not content.lstrip().startswith("#"):
|
||
break
|
||
self.index += 1
|
||
|
||
def _without_comment(self, content: str) -> str:
|
||
quote: str | None = None
|
||
escaped = False
|
||
index = 0
|
||
while index < len(content):
|
||
char = content[index]
|
||
if quote == '"':
|
||
if escaped:
|
||
escaped = False
|
||
elif char == "\\":
|
||
escaped = True
|
||
elif char == quote:
|
||
quote = None
|
||
elif quote == "'":
|
||
if char == "'":
|
||
if index + 1 < len(content) and content[index + 1] == "'":
|
||
index += 1
|
||
else:
|
||
quote = None
|
||
elif char in {'"', "'"}:
|
||
quote = char
|
||
elif char == "#" and (index == 0 or content[index - 1].isspace()):
|
||
return content[:index]
|
||
index += 1
|
||
if quote is not None:
|
||
raise YamlSubsetError("未结束的引号标量")
|
||
return content
|
||
|
||
def _reject_document_syntax(self, content: str, line: _Line) -> None:
|
||
if content in {"---", "..."} or content.startswith("%"):
|
||
self._error(line, "ACK YAML 子集只支持单文档，不支持 directive/marker")
|
||
if content.startswith(("!", "&", "*")):
|
||
self._error(line, "ACK YAML 子集不支持 tag/anchor/alias")
|
||
|
||
@staticmethod
|
||
def _error(line: _Line, message: str) -> None:
|
||
raise YamlSubsetError(f"第 {line.number} 行: {message}")
|
||
|
||
|
||
class _FlowParser:
|
||
def __init__(self, text: str, line_number: int) -> None:
|
||
self.text = text
|
||
self.line_number = line_number
|
||
self.index = 0
|
||
|
||
def parse_complete_value(self) -> Any:
|
||
value = self._parse_value()
|
||
self._skip_space()
|
||
if self.index != len(self.text):
|
||
self._error(f"标量后存在未支持内容: {self.text[self.index:]!r}")
|
||
return value
|
||
|
||
def _parse_value(self) -> Any:
|
||
self._skip_space()
|
||
if self.index >= len(self.text):
|
||
self._error("缺少标量")
|
||
char = self.text[self.index]
|
||
if char == "[":
|
||
return self._parse_list()
|
||
if char == "{":
|
||
return self._parse_map()
|
||
if char in {'"', "'"}:
|
||
return self._parse_quoted()
|
||
if char in "]},":
|
||
self._error(f"意外字符 {char!r}")
|
||
return self._parse_plain({",", "]", "}"})
|
||
|
||
def _parse_list(self) -> list[Any]:
|
||
self.index += 1
|
||
result: list[Any] = []
|
||
self._skip_space()
|
||
if self._consume("]"):
|
||
return result
|
||
while True:
|
||
result.append(self._parse_value())
|
||
self._skip_space()
|
||
if self._consume("]"):
|
||
return result
|
||
if not self._consume(","):
|
||
self._error("flow list 项之间必须用 ',' 分隔")
|
||
self._skip_space()
|
||
if self.index >= len(self.text) or self.text[self.index] == "]":
|
||
self._error("flow list 不支持尾随逗号")
|
||
|
||
def _parse_map(self) -> dict[str, Any]:
|
||
self.index += 1
|
||
result: dict[str, Any] = {}
|
||
self._skip_space()
|
||
if self._consume("}"):
|
||
return result
|
||
while True:
|
||
self._skip_space()
|
||
if self.index >= len(self.text):
|
||
self._error("flow mapping 未结束")
|
||
if self.text[self.index] in {'"', "'"}:
|
||
key = self._parse_quoted()
|
||
else:
|
||
key = self._parse_plain({":"}, convert=False)
|
||
if not isinstance(key, str) or not key:
|
||
self._error("flow mapping key 必须是非空字符串")
|
||
if key == "<<":
|
||
self._error("ACK YAML 子集不支持 merge key '<<'")
|
||
self._skip_space()
|
||
if not self._consume(":"):
|
||
self._error("flow mapping key 后必须是 ':'")
|
||
value = self._parse_value()
|
||
if key in result:
|
||
self._error(f"flow mapping 存在重复键 {key!r}")
|
||
result[key] = value
|
||
self._skip_space()
|
||
if self._consume("}"):
|
||
return result
|
||
if not self._consume(","):
|
||
self._error("flow mapping 项之间必须用 ',' 分隔")
|
||
self._skip_space()
|
||
if self.index >= len(self.text) or self.text[self.index] == "}":
|
||
self._error("flow mapping 不支持尾随逗号")
|
||
|
||
def _parse_quoted(self) -> str:
|
||
quote = self.text[self.index]
|
||
self.index += 1
|
||
result: list[str] = []
|
||
while self.index < len(self.text):
|
||
char = self.text[self.index]
|
||
self.index += 1
|
||
if char == quote:
|
||
if quote == "'" and self.index < len(self.text) and self.text[
|
||
self.index
|
||
] == "'":
|
||
result.append("'")
|
||
self.index += 1
|
||
continue
|
||
return "".join(result)
|
||
if quote == "'" or char != "\\":
|
||
result.append(char)
|
||
continue
|
||
if self.index >= len(self.text):
|
||
self._error("双引号标量以转义符结尾")
|
||
escape = self.text[self.index]
|
||
self.index += 1
|
||
simple = {
|
||
"0": "\0",
|
||
"a": "\a",
|
||
"b": "\b",
|
||
"t": "\t",
|
||
"n": "\n",
|
||
"v": "\v",
|
||
"f": "\f",
|
||
"r": "\r",
|
||
"e": "\x1b",
|
||
" ": " ",
|
||
'"': '"',
|
||
"/": "/",
|
||
"\\": "\\",
|
||
}
|
||
if escape in simple:
|
||
result.append(simple[escape])
|
||
continue
|
||
widths = {"x": 2, "u": 4, "U": 8}
|
||
if escape in widths:
|
||
width = widths[escape]
|
||
digits = self.text[self.index : self.index + width]
|
||
if len(digits) != width or not re.fullmatch(
|
||
rf"[0-9a-fA-F]{{{width}}}", digits
|
||
):
|
||
self._error(f"无效 Unicode 转义 \\{escape}{digits}")
|
||
codepoint = int(digits, 16)
|
||
try:
|
||
result.append(chr(codepoint))
|
||
except ValueError as exc:
|
||
raise YamlSubsetError(
|
||
f"第 {self.line_number} 行: 无效 Unicode 码点"
|
||
) from exc
|
||
self.index += width
|
||
continue
|
||
self._error(f"不支持的双引号转义 \\{escape}")
|
||
self._error("引号标量未结束")
|
||
|
||
def _parse_plain(
|
||
self,
|
||
delimiters: set[str],
|
||
*,
|
||
convert: bool = True,
|
||
) -> Any:
|
||
start = self.index
|
||
while self.index < len(self.text) and self.text[self.index] not in delimiters:
|
||
self.index += 1
|
||
token = self.text[start : self.index].strip()
|
||
if not token:
|
||
self._error("空 plain scalar")
|
||
if "#" in token:
|
||
self._error("flow collection 内的注释不受支持")
|
||
if ": " in token:
|
||
self._error("plain scalar 中的 ': ' 必须加引号")
|
||
if token[0] in "!&*%@`?" or _ANCHOR_OR_ALIAS_RE.search(token):
|
||
self._error("ACK YAML 子集不支持 tag/anchor/alias/directive")
|
||
return _plain_scalar(token, self.line_number) if convert else token
|
||
|
||
def _skip_space(self) -> None:
|
||
while self.index < len(self.text) and self.text[self.index] == " ":
|
||
self.index += 1
|
||
|
||
def _consume(self, expected: str) -> bool:
|
||
if self.index < len(self.text) and self.text[self.index] == expected:
|
||
self.index += 1
|
||
return True
|
||
return False
|
||
|
||
def _error(self, message: str) -> None:
|
||
raise YamlSubsetError(f"第 {self.line_number} 行: {message}")
|
||
|
||
|
||
def _plain_scalar(token: str, line_number: int) -> Any:
|
||
lowered = token.lower()
|
||
if lowered in {"null", "~"}:
|
||
return None
|
||
if lowered in {"true", "yes", "on"}:
|
||
return True
|
||
if lowered in {"false", "no", "off"}:
|
||
return False
|
||
if _DECIMAL_INT_RE.fullmatch(token):
|
||
return int(token, 10)
|
||
if _AMBIGUOUS_NUMBER_RE.fullmatch(token):
|
||
raise YamlSubsetError(
|
||
f"第 {line_number} 行: ACK YAML 子集不支持该数字格式 {token!r}，"
|
||
"如需字符串请加引号"
|
||
)
|
||
return token
|