"""Edits to node.toml made as text, so the operator's comments survive them.""" from __future__ import annotations import re import tomllib from pathlib import Path from meshbay_node.ops.core import OpError def toml_string(value: str) -> str: """A TOML basic string holding `value` exactly, quotes included. Every string written into node.toml goes through here. A value with a quote or a newline in it — a group name, a folder name, any of them chosen by someone else — would otherwise end the string and write lines of its own. """ out = ['"'] for ch in str(value): if ch == '"': out.append('\\"') elif ch == "\\": out.append("\\\\") elif ord(ch) < 0x20 or ord(ch) == 0x7F: out.append(f"\\u{ord(ch):04x}") else: out.append(ch) out.append('"') return "".join(out) def _string_value(line: str, key: str) -> str | None: """The string `key` holds on this line, unescaped — or None. Read as TOML, not by pattern: a value written by `toml_string` may carry an escaped quote or backslash, which a `"([^"]*)"` pattern would cut short. """ if not re.match(r"^\s*" + re.escape(key) + r"\s*=", line): return None try: value = tomllib.loads(line.strip()).get(key) except tomllib.TOMLDecodeError: return None return value if isinstance(value, str) else None def _find_group_range(lines: list[str], group_id: str) -> tuple[int, int] | None: """Line range of a [[groups]] block by id: (start, end_exclusive).""" block_starts: list[int] = [] for i, line in enumerate(lines): if line.strip() == "[[groups]]": block_starts.append(i) for j, start in enumerate(block_starts): boundary = block_starts[j + 1] if j + 1 < len(block_starts) else len(lines) for k in range(start + 1, boundary): s = lines[k].strip() if s.startswith("[") and s != "[[groups.roots]]": boundary = k break for k in range(start + 1, boundary): if _string_value(lines[k], "id") == group_id: return (start, boundary) return None def _update_node_toml(conf_path: Path, updates: dict) -> None: """Write changed [node] settings back to node.toml without disturbing comments. For each key, if the line exists (commented or not) it is replaced in place; otherwise the key is appended to the end of the [node] section. """ if not conf_path.exists(): return text = conf_path.read_text(encoding="utf-8") lines = text.split("\n") node_start = None node_end = len(lines) for i, line in enumerate(lines): stripped = line.strip() if stripped == "[node]": node_start = i elif node_start is not None and re.match(r'^\[', stripped): node_end = i break if node_start is None: lines.append("") lines.append("[node]") node_start = len(lines) - 1 node_end = len(lines) def _format_value(key, value): if isinstance(value, bool): return f"{key} = {'true' if value else 'false'}" if isinstance(value, list): items = ", ".join(toml_string(v) for v in value) return f"{key} = [{items}]" if isinstance(value, str): return f"{key} = {toml_string(value)}" return f"{key} = {value}" remaining = dict(updates) for i in range(node_start + 1, node_end): for key in list(remaining): pattern = re.compile( r'^(\s*#?\s*)' + re.escape(key) + r'\s*=\s*.*$') if pattern.match(lines[i]): value = remaining.pop(key) lines[i] = _format_value(key, value) break for key, value in remaining.items(): lines.insert(node_end, _format_value(key, value)) node_end += 1 conf_path.write_text("\n".join(lines), encoding="utf-8", newline="\n") def _insert_roots_block(conf_path: Path, group_id: str, root_block: str) -> None: """Append a [[groups.roots]] block inside the matching [[groups]] section.""" text = conf_path.read_text(encoding="utf-8") lines = text.split("\n") rng = _find_group_range(lines, group_id) if rng is None: raise OpError(f"Group {group_id[:8]} not found in {conf_path}") _start, end = rng insert_at = end while insert_at > _start + 1 and lines[insert_at - 1].strip() == "": insert_at -= 1 new_lines = (lines[:insert_at] + [""] + root_block.rstrip("\n").split("\n") + lines[insert_at:]) conf_path.write_text("\n".join(new_lines), encoding="utf-8", newline="\n") def _remove_roots_block(conf_path: Path, group_id: str, resolved_path: str) -> None: """Remove a [[groups.roots]] block whose resolved path matches.""" text = conf_path.read_text(encoding="utf-8") lines = text.split("\n") rng = _find_group_range(lines, group_id) if rng is None: raise OpError(f"Group {group_id[:8]} not found in {conf_path}") start, end = rng roots_starts: list[int] = [] for i in range(start + 1, end): if lines[i].strip() == "[[groups.roots]]": roots_starts.append(i) for j, rs in enumerate(roots_starts): rs_end = roots_starts[j + 1] if j + 1 < len(roots_starts) else end for k in range(rs, rs_end): raw = _string_value(lines[k], "path") if raw is not None: try: p = str(Path(raw).expanduser().resolve()) except OSError: continue if p == resolved_path: rm_start = rs if rm_start > 0 and lines[rm_start - 1].strip() == "": rm_start -= 1 new_lines = lines[:rm_start] + lines[rs_end:] conf_path.write_text("\n".join(new_lines), encoding="utf-8", newline="\n") return raise OpError("Root path not found in config", status=404) def _update_root_field(conf_path: Path, group_id: str, resolved_path: str, *, writable: bool, removable: bool) -> None: """Update writable/removable fields on a root in node.toml.""" text = conf_path.read_text(encoding="utf-8") lines = text.split("\n") rng = _find_group_range(lines, group_id) if rng is None: raise OpError(f"Group {group_id[:8]} not found in {conf_path}") start, end = rng writable_re = re.compile(r'^\s*(writable|upload)\s*=') removable_re = re.compile(r'^\s*removable\s*=') roots_starts: list[int] = [] for i in range(start + 1, end): if lines[i].strip() == "[[groups.roots]]": roots_starts.append(i) for j, rs in enumerate(roots_starts): rs_end = roots_starts[j + 1] if j + 1 < len(roots_starts) else end found_path = False for k in range(rs, rs_end): raw = _string_value(lines[k], "path") if raw is not None: try: p = str(Path(raw).expanduser().resolve()) except OSError: continue if p == resolved_path: found_path = True break if not found_path: continue writable_idx = None removable_idx = None for k in range(rs, rs_end): if writable_re.match(lines[k]): writable_idx = k if removable_re.match(lines[k]): removable_idx = k if writable_idx is not None: lines[writable_idx] = f" writable = {'true' if writable else 'false'}" else: lines.insert(rs_end, f" writable = {'true' if writable else 'false'}") if removable_idx is not None and removable_idx >= rs_end: removable_idx += 1 rs_end += 1 if removable_idx is not None: lines[removable_idx] = f" removable = {'true' if removable else 'false'}" else: lines.insert(rs_end, f" removable = {'true' if removable else 'false'}") conf_path.write_text("\n".join(lines), encoding="utf-8", newline="\n") return raise OpError("Root path not found in config", status=404)