feat: put_json_file / put_json_object to feed JSON into an analyzer's KB

NLPPlus/__init__.py

@@ -10,7 +10,7 @@
 
 import json
 import logging
-from shutil import copytree, rmtree
+from shutil import copytree, rmtree, copyfile
 from tempfile import TemporaryDirectory
 from os import PathLike, getcwd
 from pathlib import Path
@@ -299,6 +299,67 @@ def input_text(self, analyzer_name: str, file_name: str) -> str:
             text = file.read()
         return text
 
+    def _kb_user_dir(self, analyzer_name: str) -> Path:
+        """The analyzer's ``kb/user`` directory, created if missing.
+
+        This is where JSON data is dropped for the analyzer's ``json2kbb``
+        python pass to convert into a ``.kbb`` knowledge base on the next run.
+        """
+        if self.analyzer_path:
+            analyzer_dir = Path(self.analyzer_path) / analyzer_name
+        else:
+            analyzer_dir = self.working_folder / "analyzers" / analyzer_name
+        kbdir = analyzer_dir / "kb" / "user"
+        kbdir.mkdir(parents=True, exist_ok=True)
+        return kbdir
+
+    def put_json_file(self, analyzer_name: str, json_path: PathLike,
+                      name: Optional[str] = None) -> Path:
+        """Place a JSON file in the analyzer's ``kb/user`` directory.
+
+        The file is copied to ``<analyzer>/kb/user/<name>.json`` (defaulting
+        the name to the source file's own name). The analyzer's ``json2kbb``
+        python pass then converts it to ``<name>.kbb`` on the next run, so the
+        data is loaded into the knowledge base. Returns the destination path.
+
+        Args:
+          analyzer_name: analyzer under the working / analyzers folder.
+          json_path: path to a JSON file to copy in.
+          name: optional output file name (``.json`` is appended if missing).
+        """
+        src = Path(json_path)
+        if not src.is_file():
+            raise EngineException(f"JSON file not found: {src}")
+        with open(src, "r", encoding="utf-8") as fh:
+            json.load(fh)  # validate it is JSON before copying
+        target = name if name else src.name
+        if not target.lower().endswith(".json"):
+            target += ".json"
+        dest = self._kb_user_dir(analyzer_name) / target
+        copyfile(src, dest)
+        LOGGER.info("Wrote JSON to %s", dest)
+        return dest
+
+    def put_json_object(self, analyzer_name: str, obj: Any,
+                        name: str) -> Path:
+        """Write a JSON-serializable object to the analyzer's ``kb/user`` dir.
+
+        The object is serialized to ``<analyzer>/kb/user/<name>.json``. The
+        analyzer's ``json2kbb`` python pass then converts it to ``<name>.kbb``
+        on the next run. Returns the destination path.
+
+        Args:
+          analyzer_name: analyzer under the working / analyzers folder.
+          obj: any JSON-serializable object (dict, list, str, number, ...).
+          name: output file name (``.json`` is appended if missing).
+        """
+        target = name if name.lower().endswith(".json") else name + ".json"
+        dest = self._kb_user_dir(analyzer_name) / target
+        with open(dest, "w", encoding="utf-8") as fh:
+            json.dump(obj, fh, ensure_ascii=False, indent=2)
+        LOGGER.info("Wrote JSON to %s", dest)
+        return dest
+
     def set_analyzers_folder(self, analyzer_name: str):
         """Set analyzers directory path."""
         self.analyzer_path = analyzer_name
@@ -349,6 +410,20 @@ def set_analyzers_folder(analyzer_folder_path: str):
     engine.set_analyzers_folder(analyzer_folder_path)
 
 
+def put_json_file(analyzer_name: str, json_path: PathLike,
+                  name: Optional[str] = None) -> Path:
+    """Place a JSON file in the analyzer's ``kb/user`` directory so its
+    ``json2kbb`` pass converts it to a KBB (see :meth:`Engine.put_json_file`)."""
+    return engine.put_json_file(analyzer_name, json_path, name)
+
+
+def put_json_object(analyzer_name: str, obj: Any, name: str) -> Path:
+    """Write a JSON-serializable object to the analyzer's ``kb/user`` directory
+    so its ``json2kbb`` pass converts it to a KBB
+    (see :meth:`Engine.put_json_object`)."""
+    return engine.put_json_object(analyzer_name, obj, name)
+
+
 def analyze(text: str, parser: str = "parse-en-us", develop: bool = False,
             compiled: bool = False) -> str:
     """Run the analyzer named on the input string.

README.md

@@ -267,6 +267,44 @@ text from a file in the analyzer's input directory for easy
 access while developing your python code in conjunction with
 and NLP++ analyzer.
 
+#### Passing JSON data to an analyzer (used with the `json2kbb` pass)
+
+> **These functions work in conjunction with the `json2kbb.py` python pass in
+> the analyzer's sequence.** They only *place* the JSON in the analyzer's
+> `kb/user` directory — the actual JSON → KBB conversion happens **when the
+> analyzer runs**, via the `json2kbb` python pass. So the target analyzer's
+> **sequence must include the `json2kbb` pass, placed before the tokenizer**.
+> Add it in the VS Code NLP++ extension: Sequence view → right‑click the
+> tokenizer (or a pass) → **Insert Python Library Pass (Before Tokenizer)** →
+> choose **json2kbb**. Without that pass in the sequence the JSON is written
+> but never converted, so nothing is loaded into the knowledge base.
+
+#### put_json_file(analyzer_name: str, json_path, name: Optional[str] = None)
+Copies a JSON file into the analyzer's `kb/user` directory (as
+`<name>.json`, defaulting to the source file's name). The analyzer's
+`json2kbb` python pass then converts it to a `<name>.kbb` knowledge base
+on the next run, so the JSON data is loaded into the KB. This is the
+easy way to hand structured JSON data to an NLP++ analyzer. Returns the
+destination path. (Add the `json2kbb` pass to the analyzer sequence,
+before the tokenizer, via **Insert Python Library Pass** in the VS Code
+extension.)
+
+#### put_json_object(analyzer_name: str, obj, name: str)
+Same as `put_json_file`, but takes any JSON-serializable object (dict,
+list, etc.) directly and serializes it to `<analyzer>/kb/user/<name>.json`.
+The analyzer's `json2kbb` pass converts it to `<name>.kbb` on the next run.
+Returns the destination path.
+
+```python
+import NLPPlus
+# from a Python object
+NLPPlus.put_json_object("myanalyzer", {"company": {"name": "Acme"}}, "company")
+# or from an existing JSON file
+NLPPlus.put_json_file("myanalyzer", "data/company.json")
+# then analyze — the json2kbb pass builds company.kbb into the KB first
+NLPPlus.analyze("some text", "myanalyzer")
+```
+
 ### NLPPlus Engine Results
 
 #### output

tests/test_json.py

@@ -0,0 +1,61 @@
+"""Tests for placing JSON in an analyzer's kb/user directory.
+
+``put_json_file`` / ``put_json_object`` drop JSON into ``<analyzer>/kb/user``
+so the analyzer's ``json2kbb`` python pass can convert it to a ``.kbb``. These
+tests only exercise the file placement (no engine ``analyze`` run), so they do
+not hit the engine-teardown segfault the main suite is skipped for.
+"""
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    import NLPPlus
+    HAVE_NLPPLUS = True
+except Exception:  # native binding not built in this environment
+    HAVE_NLPPLUS = False
+
+ANALYZER = "emailaddress"  # a bundled library analyzer
+
+
+@unittest.skipUnless(HAVE_NLPPLUS, "NLPPlus native binding not built")
+class TestPutJson(unittest.TestCase):
+    def setUp(self):
+        self.wf = tempfile.mkdtemp(prefix="nlpjson-")
+        NLPPlus.set_working_folder(self.wf, initialize=True)
+        self.kbuser = Path(self.wf) / "analyzers" / ANALYZER / "kb" / "user"
+
+    def test_put_json_object(self):
+        dest = NLPPlus.put_json_object(ANALYZER, {"company": {"name": "Acme"}}, "company")
+        self.assertEqual(Path(dest), self.kbuser / "company.json")
+        self.assertTrue(Path(dest).is_file())
+        self.assertEqual(json.loads(Path(dest).read_text(encoding="utf-8")),
+                         {"company": {"name": "Acme"}})
+
+    def test_put_json_object_appends_extension(self):
+        dest = NLPPlus.put_json_object(ANALYZER, [1, 2, 3], "nums")
+        self.assertEqual(Path(dest).name, "nums.json")
+
+    def test_put_json_file_default_name(self):
+        src = Path(self.wf) / "src.json"
+        src.write_text('{"x": [1, 2]}', encoding="utf-8")
+        dest = NLPPlus.put_json_file(ANALYZER, src)
+        self.assertEqual(Path(dest), self.kbuser / "src.json")
+        self.assertEqual(json.loads(Path(dest).read_text(encoding="utf-8")),
+                         {"x": [1, 2]})
+
+    def test_put_json_file_named(self):
+        src = Path(self.wf) / "src.json"
+        src.write_text('{"x": 1}', encoding="utf-8")
+        dest = NLPPlus.put_json_file(ANALYZER, src, "renamed")
+        self.assertEqual(Path(dest).name, "renamed.json")
+
+    def test_put_json_file_missing_raises(self):
+        with self.assertRaises(NLPPlus.EngineException):
+            NLPPlus.put_json_file(ANALYZER, Path(self.wf) / "does-not-exist.json")
+
+
+if __name__ == "__main__":
+    unittest.main()