REPL command /save-parse

Author: telmomenezes

CHANGELOG.md

@@ -11,7 +11,7 @@
 - readers provide source information.
 - `hyperbase.parsers.badness` module for parser-agnostic combined structural + token-matching validation.
 - built-in REPL setting `check_badness` to render a badness panel after each parse, available regardless of the active parser plugin.
-- more REPL commands: /load, /unload, /save, /search, /count, /count-csv, /types, /transform, /classify and /unload-parser.
+- more REPL commands: /load, /unload, /save, /save-parse, /search, /count, /count-csv, /types, /transform, /classify and /unload-parser.
 
 ### Changed
 

src/hyperbase/cli/repl.py

@@ -26,6 +26,7 @@
 from hyperbase.hyperedge import Atom, Hyperedge
 from hyperbase.parsers import Parser, get_parser, list_parsers
 from hyperbase.parsers.badness import badness_check
+from hyperbase.parsers.parse_result import ParseResult
 from hyperbase.parsers.repl_api import (
     CommandHandler,
     PostResultHook,
@@ -70,6 +71,12 @@
         "default": 20,
         "description": "Number of results shown per page in paginated commands.",
     },
+    "save_parses_to": {
+        "type": str,
+        "default": None,
+        "description": "Path to a .jsonl file where /save-parse appends parse results.",
+        "is_path": True,
+    },
 }
 
 
@@ -328,6 +335,9 @@ def __init__(self, parser_name: str | None, settings: dict[str, Any]) -> None:
         self.edges: list[Hyperedge] = []
         self.edges_source: Path | None = None
 
+        # Last result of parse_text(), exposed for /save-parse.
+        self.last_parse_result: list[ParseResult] | None = None
+
         # Per-parser registrations -- everything in these collections is
         # cleared and re-installed whenever the active parser changes.
         self._extra_settings: dict[str, dict[str, Any]] = {}
@@ -383,6 +393,10 @@ def __init__(self, parser_name: str | None, settings: dict[str, Any]) -> None:
                 "help": "Save in-memory edges to a .jsonl file",
                 "handler": self.cmd_save,
             },
+            "save-parse": {
+                "help": "Append the last parse results to the file in save_parses_to",
+                "handler": self.cmd_save_parse,
+            },
             "edges": {
                 "help": "Show in-memory edges (count and source file)",
                 "handler": self.cmd_edges,
@@ -597,6 +611,9 @@ def _all_commands(self) -> dict[str, dict[str, Any]]:
 
     def _path_settings(self) -> frozenset[str]:
         names: set[str] = set()
+        for name, info in BUILTIN_REPL_SETTINGS.items():
+            if info.get("is_path"):
+                names.add(name)
         if self.parser is not None:
             for name, info in type(self.parser).accepted_params().items():
                 if info.get("is_path"):
@@ -909,6 +926,37 @@ def cmd_save(self, args: list) -> bool:
         )
         return False
 
+    def cmd_save_parse(self, args: list) -> bool:
+        path_str = self.settings.get("save_parses_to")
+        if not path_str:
+            self.console.print(
+                "[red]Error:[/red] no destination file set. "
+                "Use [cyan]/set save_parses_to <path>[/cyan] first."
+            )
+            return False
+        if not self.last_parse_result:
+            self.console.print(
+                "[yellow]No parse to save.[/yellow] "
+                "[dim]Parse some text first, then run /save-parse.[/dim]"
+            )
+            return False
+
+        path = Path(path_str).expanduser()
+        try:
+            with open(path, "a") as f:
+                for result in self.last_parse_result:
+                    f.write(result.to_json() + "\n")
+        except OSError as e:
+            self.console.print(f"[red]Error:[/red] failed to write {path}: {e}")
+            return False
+
+        n = len(self.last_parse_result)
+        self.console.print(
+            f"[green]✓[/green] Appended [cyan]{n}[/cyan] parse result(s) "
+            f"to [cyan]{path}[/cyan]"
+        )
+        return False
+
     def _build_tok_pos_tree(self, edge: Hyperedge) -> Hyperedge:
         """Build a tok_pos mirror tree where each atom is replaced by its
         ``tok_pos`` (or ``-1`` for synthetic atoms with no position)."""
@@ -1595,8 +1643,6 @@ def _render_count_row(self, n: int, key: object, count: int) -> None:
         self.console.print()
 
     def _load_edges_from_jsonl(self, path: Path) -> tuple[list[Hyperedge], int]:
-        from hyperbase.parsers.parse_result import ParseResult
-
         edges: list[Hyperedge] = []
         skipped = 0
         with open(path) as f:
@@ -1681,6 +1727,7 @@ def parse_text(self, text: str) -> None:
             start_time = time.perf_counter()
             parse_result = self.parser.parse(text)
             elapsed_time = time.perf_counter() - start_time
+            self.last_parse_result = parse_result
 
             if not parse_result:
                 self.console.print()