feat(ack): add project knowledge guardrails
This commit is contained in:
@@ -0,0 +1,686 @@
|
||||
#!/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
|
||||
Reference in New Issue
Block a user