pyrtan / src / pyrtan / grow.py
19,108 bytes · last changed 2026-10-26 in Watches and growing from a gap: trees that run when something speaks, and the branch a gap asks for drafted by rules; release 0.9.0
| 1 | """Where the swarm grows: sentences nobody answered, and the tree that would. |
| 2 | |
| 3 | A ``decide`` that meets a value no branch covers records a *gap*: a branch |
| 4 | waiting to be written. A sentence no tree answers is the same thing one |
| 5 | level up: a tree waiting to be written. This module gives that second case |
| 6 | the same treatment the first already has. |
| 7 | |
| 8 | - :func:`propose` turns a sentence into a :class:`Proposal`: a tree name |
| 9 | drawn from its words, a trigger pattern with slots where the sentence had |
| 10 | numbers or hosts, and the inputs those slots need. |
| 11 | - :meth:`Proposal.scaffold` renders the proposal as a tree file an author |
| 12 | fills in, the same skeleton ``pyrtan new`` writes. |
| 13 | - :class:`Unanswered` is the log a service or hub keeps of sentences it |
| 14 | could not answer, counted, each with its proposal. ``GET /unanswered`` |
| 15 | serves it; ``pyrtan unanswered --at URL --scaffold DIR`` turns it into |
| 16 | files. |
| 17 | |
| 18 | No model is involved. The proposal is derived from the words by rules, and |
| 19 | it is a starting point, not an answer: the author decides what the tree |
| 20 | observes and how it decides. |
| 21 | """ |
| 22 | |
| 23 | from __future__ import annotations |
| 24 | |
| 25 | import json |
| 26 | import re |
| 27 | import shlex |
| 28 | import threading |
| 29 | import time |
| 30 | from dataclasses import dataclass, field |
| 31 | from pathlib import Path |
| 32 | from typing import Any |
| 33 | |
| 34 | from . import conditions |
| 35 | from .intent import Trigger, TriggerError, normalise |
| 36 | from .tree import Tree, TreeError, check, load_tree |
| 37 | |
| 38 | # Words that carry no meaning for a tree's name: dropped from the name, |
| 39 | # kept in the trigger (where "what's the" is worth matching literally). |
| 40 | FILLER = {"a", "an", "the", "what", "what's", "whats", "how", "please", "me", "my", "tell", "show", "give", "is", "are", "of", "to", "for", "in", "on", "at", "do", "does", "it", "and", "with"} |
| 41 | INT_RE = re.compile(r"^-?\d+$") |
| 42 | FLOAT_RE = re.compile(r"^-?\d+\.\d+$") |
| 43 | HOST_RE = re.compile(r"^[a-z0-9-]+(\.[a-z0-9-]+)+(:\d+)?$", re.IGNORECASE) |
| 44 | PATH_RE = re.compile(r"^[\w.\-]+/[\w.\-/]+$") |
| 45 | NAME_MAX = 40 |
| 46 | UNANSWERED_LIMIT = 200 |
| 47 | |
| 48 | SCAFFOLD = '''# {name}: say what this tree does, in one line, below. |
| 49 | # Proposed from the sentence: {text} |
| 50 | |
| 51 | name = "{name}" |
| 52 | version = 1 |
| 53 | description = "" |
| 54 | tags = [] |
| 55 | license = "" # an SPDX identifier such as "MIT", if you share it |
| 56 | hosts = {hosts} |
| 57 | start = "fetch" |
| 58 | |
| 59 | [[triggers]] |
| 60 | pattern = "{trigger}" |
| 61 | {inputs} |
| 62 | [nodes.fetch] |
| 63 | kind = "observe" |
| 64 | http = {{ method = "GET", url = "{url}" }} |
| 65 | into = "resp" |
| 66 | extract = [{{ into = "title", css = "title" }}] |
| 67 | next = "status" |
| 68 | |
| 69 | [nodes.status] |
| 70 | kind = "decide" |
| 71 | on = "resp.status" |
| 72 | branches = [ |
| 73 | {{ when = {{ eq = 200 }}, then = "reply" }}, |
| 74 | {{ else = "unexpected" }}, |
| 75 | ] |
| 76 | |
| 77 | [nodes.reply] |
| 78 | kind = "done" |
| 79 | say = "{{{{ state.title }}}}" |
| 80 | output = {{ title = "{{{{ state.title }}}}" }} |
| 81 | |
| 82 | [nodes.unexpected] |
| 83 | kind = "fail" |
| 84 | reason = "unexpected-response" |
| 85 | say = "{host_word} answered {{{{ state.resp.status }}}}." |
| 86 | ''' |
| 87 | |
| 88 | INPUT_BLOCK = ''' |
| 89 | [inputs.{slot}] |
| 90 | type = "{type}" |
| 91 | description = "{description}" |
| 92 | ''' |
| 93 | |
| 94 | |
| 95 | @dataclass |
| 96 | class Proposal: |
| 97 | """A tree that would answer ``text``, before anyone has written it.""" |
| 98 | |
| 99 | text: str |
| 100 | name: str |
| 101 | trigger: str |
| 102 | inputs: dict[str, dict[str, Any]] = field(default_factory=dict) |
| 103 | |
| 104 | def to_dict(self) -> dict[str, Any]: |
| 105 | return {"text": self.text, "name": self.name, "trigger": self.trigger, "inputs": self.inputs, "scaffold": self.command()} |
| 106 | |
| 107 | def command(self, directory: str | None = None) -> str: |
| 108 | """The ``pyrtan new`` invocation that writes this tree's file.""" |
| 109 | parts = ["pyrtan", "new", self.name, "--from", _quote(self.text)] |
| 110 | if directory: |
| 111 | parts += ["--dir", _quote(directory)] |
| 112 | return " ".join(parts) |
| 113 | |
| 114 | def scaffold(self) -> str: |
| 115 | """The tree file, ready for ``pyrtan check`` and for an author.""" |
| 116 | host_slot = next((s for s, spec in self.inputs.items() if spec.get("role") == "host"), None) |
| 117 | if host_slot: |
| 118 | hosts, url, host_word = '["*"]', f"https://{{{{ inputs.{host_slot} | lower }}}}/", f"{{{{ inputs.{host_slot} }}}}" |
| 119 | else: |
| 120 | hosts, url, host_word = '["example.com"]', "https://example.com/", "example.com" |
| 121 | inputs = "".join(INPUT_BLOCK.format(slot=slot, type=spec["type"], description=spec.get("description", "")) for slot, spec in self.inputs.items()) |
| 122 | return SCAFFOLD.format(name=self.name, text=_toml_safe(self.text), hosts=hosts, trigger=_toml_safe(self.trigger), inputs=inputs, url=url, host_word=host_word) |
| 123 | |
| 124 | |
| 125 | def propose(text: str, name: str | None = None) -> Proposal: |
| 126 | """A proposal for the sentence ``text``. |
| 127 | |
| 128 | Numbers in the sentence become ``{count:int}`` / ``{value:float}`` |
| 129 | slots, host names ``{host:host}``, ``owner/repo``-shaped words |
| 130 | ``{target:word}``; every other word is literal. The name is the literal |
| 131 | words with filler dropped, or ``name`` if given. |
| 132 | """ |
| 133 | clean = normalise(text) or text.strip() or "new tree" |
| 134 | tokens = clean.split(" ") |
| 135 | pieces: list[str] = [] |
| 136 | inputs: dict[str, dict[str, Any]] = {} |
| 137 | literal_words: list[str] = [] |
| 138 | for tok in tokens: |
| 139 | slot = _slot_for(tok, inputs) |
| 140 | if slot is not None: |
| 141 | pieces.append(slot) |
| 142 | else: |
| 143 | pieces.append(_literal(tok)) |
| 144 | literal_words.append(tok) |
| 145 | trigger = " ".join(p for p in pieces if p) |
| 146 | try: |
| 147 | Trigger(trigger) |
| 148 | except TriggerError: |
| 149 | # Everything became a slot, or the words left nothing matchable: |
| 150 | # fall back to the sentence itself, literally. |
| 151 | trigger = " ".join(_literal(t) for t in tokens if _literal(t)) or "new tree" |
| 152 | inputs = {} |
| 153 | literal_words = tokens |
| 154 | return Proposal(text=clean, name=_name_for(name, literal_words, tokens, list(inputs)), trigger=trigger, inputs=inputs) |
| 155 | |
| 156 | |
| 157 | def _slot_for(tok: str, inputs: dict[str, dict[str, Any]]) -> str | None: |
| 158 | bare = tok.strip("\"'(),;:!?") |
| 159 | if INT_RE.match(bare): |
| 160 | slot = _unique("count", inputs) |
| 161 | inputs[slot] = {"type": "int", "description": f"as in: {bare}"} |
| 162 | return f"{{{slot}:int}}" |
| 163 | if FLOAT_RE.match(bare): |
| 164 | slot = _unique("value", inputs) |
| 165 | inputs[slot] = {"type": "float", "description": f"as in: {bare}"} |
| 166 | return f"{{{slot}:float}}" |
| 167 | if HOST_RE.match(bare) and not bare.endswith(("...", ".")): |
| 168 | slot = _unique("host", inputs) |
| 169 | inputs[slot] = {"type": "str", "description": f"a host name, as in: {bare}", "role": "host"} |
| 170 | return f"{{{slot}:host}}" |
| 171 | if PATH_RE.match(bare): |
| 172 | slot = _unique("target", inputs) |
| 173 | inputs[slot] = {"type": "str", "description": f"as in: {bare}"} |
| 174 | return f"{{{slot}:word}}" |
| 175 | return None |
| 176 | |
| 177 | |
| 178 | def _unique(base: str, taken: dict[str, Any]) -> str: |
| 179 | if base not in taken: |
| 180 | return base |
| 181 | n = 2 |
| 182 | while f"{base}{n}" in taken: |
| 183 | n += 1 |
| 184 | return f"{base}{n}" |
| 185 | |
| 186 | |
| 187 | def _literal(tok: str) -> str: |
| 188 | # Trigger syntax characters would be read as slots or groups; a literal |
| 189 | # word must not contain them. |
| 190 | return re.sub(r"[{}\[\]()|]", "", tok).strip() |
| 191 | |
| 192 | |
| 193 | def _name_for(given: str | None, literal_words: list[str], tokens: list[str], slots: list[str] = ()) -> str: |
| 194 | if given: |
| 195 | return given |
| 196 | words = [re.sub(r"[^a-z0-9]+", "-", w.lower()).strip("-") for w in literal_words if w.lower() not in FILLER] |
| 197 | words = [w for w in words if w] |
| 198 | if len(words) < 2 and slots: |
| 199 | # "is example.com up" would be called "up"; "host-up" says more. |
| 200 | words = [re.sub(r"\d+$", "", s) for s in slots] + words |
| 201 | if not words: |
| 202 | words = [re.sub(r"[^a-z0-9]+", "-", w.lower()).strip("-") for w in tokens] |
| 203 | words = [w for w in words if w] |
| 204 | name = "-".join(words)[:NAME_MAX].strip("-") |
| 205 | name = re.sub(r"-{2,}", "-", name) |
| 206 | if not name or not name[0].isalnum(): |
| 207 | name = "new-tree" |
| 208 | return name |
| 209 | |
| 210 | |
| 211 | def _quote(s: str) -> str: |
| 212 | """Shell-quote ``s``, preferring double quotes so an apostrophe in a |
| 213 | sentence does not turn into ``'what'"'"'s'``.""" |
| 214 | if re.fullmatch(r"[\w./:@%+=,-]+", s): |
| 215 | return s |
| 216 | if not any(c in s for c in '"$`\\'): |
| 217 | return f'"{s}"' |
| 218 | return shlex.quote(s) |
| 219 | |
| 220 | |
| 221 | def _toml_safe(s: str) -> str: |
| 222 | return s.replace("\\", "\\\\").replace('"', '\\"') |
| 223 | |
| 224 | |
| 225 | def no_match_outcome(text: str, where: str) -> dict[str, Any]: |
| 226 | """The ``say`` and ``output`` an unanswered sentence gets back: what |
| 227 | happened, and the one command that would change it.""" |
| 228 | p = propose(text) |
| 229 | say = f"{where} answers {text!r}. To write the tree that would: {p.command()}" |
| 230 | return {"say": say, "output": {"unanswered": text, "proposal": p.to_dict()}} |
| 231 | |
| 232 | |
| 233 | class Unanswered: |
| 234 | """Sentences nobody answered, counted, with a proposal for each. |
| 235 | |
| 236 | Bounded: beyond ``limit`` distinct sentences the least recently seen is |
| 237 | dropped. Lives in memory, like gaps; a restart forgets it. |
| 238 | """ |
| 239 | |
| 240 | def __init__(self, limit: int = UNANSWERED_LIMIT, path: "str | Path | None" = None) -> None: |
| 241 | self.limit = limit |
| 242 | self.path = Path(path) if path else None |
| 243 | self.error: str | None = None |
| 244 | self._entries: dict[str, dict[str, Any]] = {} |
| 245 | self._lock = threading.Lock() |
| 246 | if self.path is not None and self.path.exists(): |
| 247 | try: |
| 248 | for line in self.path.read_text(encoding="utf-8").splitlines(): |
| 249 | try: |
| 250 | rec = json.loads(line) |
| 251 | except json.JSONDecodeError: |
| 252 | continue |
| 253 | if isinstance(rec, dict) and isinstance(rec.get("text"), str): |
| 254 | self._take(rec["text"], rec.get("peer"), int(rec.get("count", 1)), float(rec.get("last", time.time())), persist=False) |
| 255 | except OSError as e: |
| 256 | self.error = str(e) |
| 257 | |
| 258 | def record(self, text: str, peer: str | None = None) -> dict[str, Any]: |
| 259 | return self._take(text, peer, 1, time.time(), persist=True) |
| 260 | |
| 261 | def _take(self, text: str, peer: str | None, count: int, when: float, *, persist: bool) -> dict[str, Any]: |
| 262 | key = normalise(text).lower() or text |
| 263 | with self._lock: |
| 264 | entry = self._entries.get(key) |
| 265 | if entry is None: |
| 266 | entry = {"text": normalise(text) or text, "count": 0, "first": when, "last": when, "proposal": propose(text).to_dict()} |
| 267 | if peer: |
| 268 | entry["peer"] = peer |
| 269 | self._entries[key] = entry |
| 270 | entry["count"] += count |
| 271 | entry["last"] = max(entry["last"], when) |
| 272 | while len(self._entries) > self.limit: |
| 273 | oldest = min(self._entries, key=lambda k: self._entries[k]["last"]) |
| 274 | del self._entries[oldest] |
| 275 | if persist and self.path is not None: |
| 276 | try: |
| 277 | self.path.parent.mkdir(parents=True, exist_ok=True) |
| 278 | with self.path.open("a", encoding="utf-8") as f: |
| 279 | f.write(json.dumps({"text": entry["text"], "peer": peer, "count": 1, "last": when}, ensure_ascii=False) + "\n") |
| 280 | self.error = None |
| 281 | except OSError as e: |
| 282 | self.error = str(e) |
| 283 | return dict(entry) |
| 284 | |
| 285 | def to_list(self) -> list[dict[str, Any]]: |
| 286 | with self._lock: |
| 287 | return sorted((dict(e) for e in self._entries.values()), key=lambda e: (-e["count"], -e["last"])) |
| 288 | |
| 289 | def __len__(self) -> int: |
| 290 | return len(self._entries) |
| 291 | |
| 292 | |
| 293 | # ---------------------------------------------------------------- growing a tree from a gap |
| 294 | |
| 295 | |
| 296 | @dataclass |
| 297 | class Draft: |
| 298 | """A branch a gap asks for, drafted by rules: the condition that would |
| 299 | have taken the observed value, a new terminal node for it, and the |
| 300 | text of both. A person reads it, changes the node's kind and words to |
| 301 | what the value means, and keeps it. Nothing here decides anything.""" |
| 302 | |
| 303 | tree: str |
| 304 | node: str |
| 305 | on: str |
| 306 | observed: Any |
| 307 | when: dict[str, Any] |
| 308 | then: str |
| 309 | branch: str |
| 310 | node_text: str |
| 311 | reason: str = "" |
| 312 | |
| 313 | def to_dict(self) -> dict[str, Any]: |
| 314 | return {"tree": self.tree, "node": self.node, "on": self.on, "observed": self.observed, "when": self.when, "then": self.then, "branch": self.branch, "node_text": self.node_text, "reason": self.reason} |
| 315 | |
| 316 | def render(self) -> str: |
| 317 | return f"# {self.tree}: in [nodes.{self.node}], add to branches (before the else, if any):\n {self.branch}\n\n{self.node_text}" |
| 318 | |
| 319 | |
| 320 | def draft_branch(tree: Tree, gap: dict[str, Any]) -> Draft | None: |
| 321 | """The branch a gap on ``tree`` asks for, or None when the gap names |
| 322 | no decide node of this tree or a branch already covers the value |
| 323 | (the file is newer than the gap). |
| 324 | |
| 325 | Rules: a missing value asks for ``{ exists = false }``; a number that |
| 326 | falls in a hole between the ``between`` ranges or ``eq`` points the |
| 327 | node already has asks for the range of that hole; anything else asks |
| 328 | for ``{ eq = value }`` (``{ type = ... }`` for a value a report could |
| 329 | only carry as its type). The new node is a ``fail`` that says what |
| 330 | was seen; the author decides what it means.""" |
| 331 | node = tree.nodes.get(str(gap.get("node", ""))) |
| 332 | if node is None or node.kind != "decide": |
| 333 | return None |
| 334 | on = str(gap.get("on") or node.get("on") or "") |
| 335 | observed = gap.get("observed") |
| 336 | branches = [b for b in (node.get("branches") or []) if isinstance(b, dict)] |
| 337 | if any("when" in b and conditions.holds(b["when"], observed) for b in branches): |
| 338 | return None |
| 339 | when, reason = _condition_for(observed, branches) |
| 340 | then = _unique_node(tree, f"{node.id}-{_slug(observed)}") |
| 341 | slug = _slug(observed) |
| 342 | branch = "{ when = " + _toml_inline(when) + f', then = "{then}" }},' |
| 343 | node_text = ( |
| 344 | f"# Drafted by pyrtan grow from a gap: {on} = {json.dumps(observed, ensure_ascii=False)} had no branch.\n" |
| 345 | f"# Decide what that value means: keep this as a fail, or make it a done, a set, an observe...\n" |
| 346 | f"[nodes.{then}]\n" |
| 347 | f'kind = "fail"\n' |
| 348 | f'reason = "{_slug(observed, 40) or "unhandled"}"\n' |
| 349 | f'say = "{_toml_str(on)} was {_toml_str(json.dumps(observed, ensure_ascii=False))}; nobody has said what that means yet."\n' |
| 350 | ) |
| 351 | return Draft(tree.name, node.id, on, observed, when, then, branch, node_text, reason) |
| 352 | |
| 353 | |
| 354 | def _condition_for(observed: Any, branches: list[dict[str, Any]]) -> tuple[dict[str, Any], str]: |
| 355 | if observed is None: |
| 356 | return {"exists": False}, "the value was missing" |
| 357 | if isinstance(observed, str) and observed.startswith("<") and observed.endswith(">") and observed[1:-1] in ("str", "dict", "list", "int", "float", "bool"): |
| 358 | return {"type": observed[1:-1]}, "a report carries a long or structured value as its type" |
| 359 | if isinstance(observed, (int, float)) and not isinstance(observed, bool): |
| 360 | covered: list[tuple[float, float]] = [] |
| 361 | for b in branches: |
| 362 | w = b.get("when") if isinstance(b, dict) else None |
| 363 | if not isinstance(w, dict): |
| 364 | continue |
| 365 | if "between" in w and isinstance(w["between"], list) and len(w["between"]) == 2: |
| 366 | try: |
| 367 | covered.append((float(w["between"][0]), float(w["between"][1]))) |
| 368 | except (TypeError, ValueError): |
| 369 | pass |
| 370 | elif "eq" in w and isinstance(w["eq"], (int, float)) and not isinstance(w["eq"], bool): |
| 371 | covered.append((float(w["eq"]), float(w["eq"]))) |
| 372 | if covered and isinstance(observed, int): |
| 373 | below = [hi for lo, hi in covered if hi < observed] |
| 374 | above = [lo for lo, hi in covered if lo > observed] |
| 375 | if below and above and all(float(x).is_integer() for x in below + above): |
| 376 | lo, hi = int(max(below)) + 1, int(min(above)) - 1 |
| 377 | if lo <= hi: |
| 378 | return {"between": [lo, hi]}, f"the value falls in the hole between {int(max(below))} and {int(min(above))}, which the ranges around it leave open" |
| 379 | return {"eq": observed}, "the value itself" |
| 380 | |
| 381 | |
| 382 | def _unique_node(tree: Tree, base: str) -> str: |
| 383 | name = base |
| 384 | n = 2 |
| 385 | while name in tree.nodes: |
| 386 | name = f"{base}-{n}" |
| 387 | n += 1 |
| 388 | return name |
| 389 | |
| 390 | |
| 391 | def _slug(value: Any, limit: int = 24) -> str: |
| 392 | text = "missing" if value is None else str(value).lower() |
| 393 | text = re.sub(r"[^a-z0-9]+", "-", text).strip("-") |
| 394 | return (text[:limit].rstrip("-") or "value") |
| 395 | |
| 396 | |
| 397 | def _toml_inline(cond: dict[str, Any]) -> str: |
| 398 | return "{ " + ", ".join(f"{k} = {_toml_value(v)}" for k, v in cond.items()) + " }" |
| 399 | |
| 400 | |
| 401 | def _toml_value(v: Any) -> str: |
| 402 | if isinstance(v, bool): |
| 403 | return "true" if v else "false" |
| 404 | if isinstance(v, (int, float)): |
| 405 | return repr(v) |
| 406 | if isinstance(v, list): |
| 407 | return "[" + ", ".join(_toml_value(x) for x in v) + "]" |
| 408 | return '"' + _toml_str(str(v)) + '"' |
| 409 | |
| 410 | |
| 411 | def _toml_str(s: str) -> str: |
| 412 | return s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") |
| 413 | |
| 414 | |
| 415 | def apply_draft(path: str | Path, draft: Draft) -> str: |
| 416 | """Write a draft into the tree file: the branch into the node's |
| 417 | ``branches`` list (before its ``else``), the node at the end. The file |
| 418 | must load clean afterwards, else it is put back and :class:`TreeError` |
| 419 | raised. Returns the new text.""" |
| 420 | path = Path(path) |
| 421 | original = path.read_text(encoding="utf-8") |
| 422 | lines = original.splitlines(keepends=True) |
| 423 | header = f"[nodes.{draft.node}]" |
| 424 | start = next((i for i, l in enumerate(lines) if l.strip() == header), None) |
| 425 | if start is None: |
| 426 | raise TreeError(f"{path}: no {header} table") |
| 427 | end = next((i for i in range(start + 1, len(lines)) if lines[i].lstrip().startswith("[")), len(lines)) |
| 428 | b = next((i for i in range(start, end) if re.match(r"^\s*branches\s*=\s*\[", lines[i])), None) |
| 429 | if b is None: |
| 430 | raise TreeError(f"{path}: {header} has no branches list") |
| 431 | indent = " " |
| 432 | if "]" in lines[b].split("[", 1)[1]: |
| 433 | # A one-line list. |
| 434 | head, tail = lines[b].rsplit("]", 1) |
| 435 | m = re.search(r",?\s*\{\s*else\s*=", head) |
| 436 | if m: |
| 437 | head = head[: m.start()] + ", " + draft.branch.rstrip(",") + head[m.start():] |
| 438 | else: |
| 439 | head = head.rstrip().rstrip(",") + ", " + draft.branch.rstrip(",") |
| 440 | lines[b] = head + "]" + tail |
| 441 | else: |
| 442 | close = next(i for i in range(b + 1, end) if lines[i].strip().startswith("]")) |
| 443 | else_line = next((i for i in range(b + 1, close) if re.match(r"^\s*\{\s*else\s*=", lines[i])), None) |
| 444 | at = else_line if else_line is not None else close |
| 445 | if at > b + 1: |
| 446 | indent = re.match(r"^\s*", lines[at - 1 if else_line is None else at]).group(0) or indent |
| 447 | lines.insert(at, f"{indent}{draft.branch}\n") |
| 448 | text = "".join(lines) |
| 449 | if not text.endswith("\n"): |
| 450 | text += "\n" |
| 451 | text += "\n" + draft.node_text |
| 452 | path.write_text(text, encoding="utf-8") |
| 453 | try: |
| 454 | tree = load_tree(path) |
| 455 | errors = [p for p in check(tree) if p.level == "error"] |
| 456 | if errors: |
| 457 | raise TreeError("; ".join(f"{p.where}: {p.message}" for p in errors)) |
| 458 | except TreeError as e: |
| 459 | path.write_text(original, encoding="utf-8") |
| 460 | raise TreeError(f"{path}: the draft did not fit the file ({e}); nothing changed") from e |
| 461 | return text |