"""Edits to node.toml made as text, so the operator's comments survive them.""" from __future__ import annotations import re from pathlib import Path from meshbay_node.ops.core import OpError 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).""" id_re = re.compile(r'^\s*id\s*=\s*"([^"]*)"') 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): m = id_re.match(lines[k]) if m and m.group(1) == 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(f'"{v}"' for v in value) return f"{key} = [{items}]" 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 path_re = re.compile(r'^\s*path\s*=\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 for k in range(rs, rs_end): m = path_re.match(lines[k]) if m: try: p = str(Path(m.group(1)).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 path_re = re.compile(r'^\s*path\s*=\s*"([^"]*)"') 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): m = path_re.match(lines[k]) if m: try: p = str(Path(m.group(1)).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)