"""轻量 YAML 读写(覆盖 skiff 使用的子集,无第三方依赖)。""" from __future__ import annotations import json from typing import Any def safe_load(text: str) -> Any: if not text or not text.strip(): return None lines = text.splitlines() # 跳过纯注释/空行,定位首个有效块 start = 0 while start < len(lines): line = _strip_comment(lines[start]) if line.strip(): break start += 1 if start >= len(lines): return None result, _ = _parse_block(lines, start, _indent_of(_strip_comment(lines[start]))) return result def safe_dump(data: Any, *, allow_unicode: bool = True, sort_keys: bool = False) -> str: del allow_unicode, sort_keys return _dump(data).rstrip() + "\n" def _indent_of(line: str) -> int: return len(line) - len(line.lstrip(" ")) def _strip_comment(line: str) -> str: if "#" in line: in_single = False in_double = False for i, ch in enumerate(line): if ch == "'" and not in_double: in_single = not in_single elif ch == '"' and not in_single: in_double = not in_double elif ch == "#" and not in_single and not in_double: return line[:i].rstrip() return line.rstrip() def _parse_scalar(raw: str) -> Any: raw = raw.strip() if not raw: return "" if (raw.startswith('"') and raw.endswith('"')) or (raw.startswith("'") and raw.endswith("'")): return raw[1:-1] lower = raw.lower() if lower in ("null", "~"): return None if lower == "true": return True if lower == "false": return False if raw.isdigit() or (raw.startswith("-") and raw[1:].isdigit()): return int(raw) return raw def _parse_block(lines: list[str], start: int, base_indent: int) -> tuple[Any, int]: if start >= len(lines): return None, start line = _strip_comment(lines[start]) if not line.strip(): return _parse_block(lines, start + 1, base_indent) stripped = line.lstrip(" ") indent = _indent_of(line) if stripped.startswith("- "): return _parse_list(lines, start, indent) if ":" in stripped: return _parse_mapping(lines, start, indent) raise ValueError(f"无法解析 YAML 行: {line}") def _parse_list(lines: list[str], start: int, list_indent: int) -> tuple[list[Any], int]: items: list[Any] = [] i = start while i < len(lines): line = _strip_comment(lines[i]) if not line.strip(): i += 1 continue if _indent_of(line) < list_indent: break if _indent_of(line) > list_indent or not line.lstrip().startswith("- "): break content = line.lstrip()[2:].strip() if not content: i += 1 continue if ":" in content and not content.startswith(("http://", "https://")): key, rest = content.split(":", 1) key = key.strip() rest = rest.strip() if rest: item = {key: _parse_scalar(rest)} i += 1 else: nested, i = _parse_mapping(lines, i, list_indent + 2) item = {key: nested} while i < len(lines): nxt = _strip_comment(lines[i]) if not nxt.strip(): i += 1 continue if _indent_of(nxt) <= list_indent: break if not nxt.lstrip().startswith("- "): extra, i = _parse_mapping(lines, i, list_indent + 2) if isinstance(item, dict): item.update(extra) break i += 1 items.append(item) continue items.append(_parse_scalar(content)) i += 1 return items, i def _parse_mapping(lines: list[str], start: int, map_indent: int) -> tuple[dict[str, Any], int]: result: dict[str, Any] = {} i = start while i < len(lines): line = _strip_comment(lines[i]) if not line.strip(): i += 1 continue indent = _indent_of(line) if indent < map_indent: break if indent > map_indent: raise ValueError(f"缩进不一致: {line}") stripped = line.lstrip() if stripped.startswith("- "): break key, rest = stripped.split(":", 1) key = key.strip() rest = rest.strip() i += 1 if rest: result[key] = _parse_scalar(rest) continue if i >= len(lines): result[key] = None break peek = _strip_comment(lines[i]) while peek == "" and i < len(lines): i += 1 peek = _strip_comment(lines[i]) if i < len(lines) else "" if i >= len(lines): result[key] = None break child_indent = _indent_of(peek) if child_indent <= map_indent: result[key] = None continue if peek.lstrip().startswith("- "): value, i = _parse_list(lines, i, child_indent) result[key] = value else: value, i = _parse_mapping(lines, i, child_indent) result[key] = value return result, i def _dump(data: Any, indent: int = 0) -> str: pad = " " * indent if isinstance(data, dict): if not data: return f"{pad}{{}}\n" lines: list[str] = [] for key, value in data.items(): if isinstance(value, (dict, list)): if isinstance(value, list) and value and all(isinstance(x, str) for x in value): lines.append(f"{pad}{key}:") for item in value: lines.append(f"{pad} - {_scalar(item)}") elif isinstance(value, list) and value and all(isinstance(x, dict) for x in value): lines.append(f"{pad}{key}:") for item in value: lines.extend(_dump_list_dict_item(item, indent + 2)) elif isinstance(value, dict) and value: lines.append(f"{pad}{key}:") lines.append(_dump(value, indent + 2).rstrip()) elif isinstance(value, list): lines.append(f"{pad}{key}:") for item in value: if isinstance(item, dict): lines.extend(_dump_list_dict_item(item, indent + 2)) else: lines.append(f"{pad} - {_scalar(item)}") else: lines.append(f"{pad}{key}: {_scalar(value)}") else: lines.append(f"{pad}{key}: {_scalar(value)}") return "\n".join(lines) + "\n" if isinstance(data, list): lines = [] for item in data: if isinstance(item, dict): lines.extend(_dump_list_dict_item(item, indent)) else: lines.append(f"{pad}- {_scalar(item)}") return "\n".join(lines) + "\n" return f"{pad}{_scalar(data)}\n" def _dump_list_dict_item(item: dict[str, Any], indent: int) -> list[str]: pad = " " * indent lines: list[str] = [] first = True for key, value in item.items(): is_first = first prefix = f"{pad}- " if is_first else f"{pad} " first = False if isinstance(value, (dict, list)): lines.append(f"{prefix}{key}:") nested = _dump(value, indent + 2 if is_first else indent + 4) lines.append(nested.rstrip()) else: lines.append(f"{prefix}{key}: {_scalar(value)}") return lines def _scalar(value: Any) -> str: if value is None: return "null" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float)): return str(value) text = str(value) if text == "" or any(ch in text for ch in ":#{}[],&*?|>-%@`") or text.startswith((" ", "-")): return json.dumps(text, ensure_ascii=False) return text