feat: add errh_json JSON error visitor (issue #50) (#163)

Adds pyx12/errh_json.py — a visitor over the err_handler tree that
emits a nested JSON document mirroring the ISA / GS / ST / segment /
element hierarchy. Inherits from pyx12.error_visitor.error_visitor
(same shape as error_999_visitor) and accumulates a Python dict during
the walk, dumped once via json.dump in visit_root_post.

This is the first slice of the issue #50 work — no callers wired up
yet; pyx12/x12n_document.py and pyx12/scripts/x12valid.py integration
follows in subsequent PRs per the plan.

JSON schema:
- interchanges[] -> groups[] -> transactions[] -> segments[] -> elements[]
- each level carries its identifying fields plus an "errors" array
- errors at ISA/GS/ST level are (cde, str) tuples; seg/ele errors are
  (cde, str, val) tuples
- empty branches still emitted (schema is structural, not pruned)

Tests build a synthetic err_handler tree exercising every level
without invoking the document orchestrator. Coverage: 9 tests,
all levels, empty-tree case, and indent argument pass-through.

Verification:
- pytest pyx12/test/: 548 passed (539 + 9 new)
- mypy --strict pyx12: clean (85 files)
- ruff check / ruff format --check: clean

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: azoner

pyx12/errh_json.py

@@ -0,0 +1,161 @@
+######################################################################
+# Copyright
+#   John Holland <john@zoner.org>
+# All rights reserved.
+#
+# This software is licensed as described in the file LICENSE.txt, which
+# you should have received as part of this distribution.
+#
+######################################################################
+
+"""
+Generates a JSON document of validation errors.
+Visitor - Visits an error_handler composite (issue #50).
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, TextIO
+
+import pyx12.error_visitor
+
+
+class errh_json_visitor(pyx12.error_visitor.error_visitor):
+    """
+    Visit an error_handler composite. Accumulate a nested dict mirroring
+    the ISA / GS / ST / segment / element hierarchy and dump it once as
+    JSON in visit_root_post.
+
+    Streaming JSON during the walk would require closing arrays mid-walk;
+    the err_handler tree always fits in memory for any X12 document, so
+    accumulate-then-dump is the right shape.
+    """
+
+    fd: TextIO
+    indent: int | None
+    doc: dict[str, Any]
+    _stack: list[dict[str, Any]]
+    _cur_seg: dict[str, Any] | None
+
+    def __init__(self, fd: TextIO, indent: int | None = 2) -> None:
+        """
+        :param fd: target file
+        :type fd: file descriptor
+        :param indent: json.dump indent argument; pass None for compact output
+        :type indent: int | None
+        """
+        self.fd = fd
+        self.indent = indent
+        self.doc = {"interchanges": []}
+        self._stack = []
+        self._cur_seg = None
+
+    def visit_root_post(self, errh: Any) -> None:
+        """
+        :param errh: Error handler
+        :type errh: L{error_handler.err_handler}
+        """
+        json.dump(self.doc, self.fd, indent=self.indent, default=str)
+
+    def visit_isa_pre(self, err_isa: Any) -> None:
+        """
+        :param err_isa: ISA error node
+        :type err_isa: L{error_handler.err_isa}
+        """
+        isa_dict: dict[str, Any] = {
+            "isa_trn_set_id": err_isa.isa_trn_set_id,
+            "ta1_req": err_isa.ta1_req,
+            "orig_date": err_isa.orig_date,
+            "orig_time": err_isa.orig_time,
+            "cur_line": err_isa.get_cur_line(),
+            "errors": [],
+            "groups": [],
+        }
+        self.doc["interchanges"].append(isa_dict)
+        self._stack.append(isa_dict)
+
+    def visit_isa_post(self, err_isa: Any) -> None:
+        isa_dict = self._stack[-1]
+        isa_dict["errors"] = [
+            {"err_cde": cde, "err_str": err_str} for (cde, err_str) in err_isa.errors
+        ]
+        self._stack.pop()
+
+    def visit_gs_pre(self, err_gs: Any) -> None:
+        gs_dict: dict[str, Any] = {
+            "gs_control_num": err_gs.gs_control_num,
+            "fic": err_gs.fic,
+            "vriic": err_gs.vriic,
+            "ack_code": err_gs.ack_code,
+            "st_count_orig": err_gs.st_count_orig,
+            "st_count_recv": err_gs.st_count_recv,
+            "cur_line": err_gs.get_cur_line(),
+            "errors": [],
+            "transactions": [],
+        }
+        self._stack[-1]["groups"].append(gs_dict)
+        self._stack.append(gs_dict)
+
+    def visit_gs_post(self, err_gs: Any) -> None:
+        gs_dict = self._stack[-1]
+        gs_dict["errors"] = [
+            {"err_cde": cde, "err_str": err_str} for (cde, err_str) in err_gs.errors
+        ]
+        self._stack.pop()
+
+    def visit_st_pre(self, err_st: Any) -> None:
+        st_dict: dict[str, Any] = {
+            "trn_set_id": err_st.trn_set_id,
+            "trn_set_control_num": err_st.trn_set_control_num,
+            "vriic": err_st.vriic,
+            "ack_code": err_st.ack_code,
+            "cur_line": err_st.get_cur_line(),
+            "errors": [],
+            "segments": [],
+        }
+        self._stack[-1]["transactions"].append(st_dict)
+        self._stack.append(st_dict)
+
+    def visit_st_post(self, err_st: Any) -> None:
+        st_dict = self._stack[-1]
+        st_dict["errors"] = [
+            {"err_cde": cde, "err_str": err_str} for (cde, err_str) in err_st.errors
+        ]
+        self._stack.pop()
+
+    def visit_seg(self, err_seg: Any) -> None:
+        seg_dict: dict[str, Any] = {
+            "seg_id": err_seg.seg_id,
+            "seg_count": err_seg.seg_count,
+            "pos": err_seg.pos,
+            "name": err_seg.name,
+            "ls_id": err_seg.ls_id,
+            "cur_line": err_seg.get_cur_line(),
+            "errors": [
+                {"err_cde": cde, "err_str": err_str, "err_val": err_val}
+                for (cde, err_str, err_val) in err_seg.errors
+            ],
+            "elements": [],
+        }
+        self._stack[-1]["segments"].append(seg_dict)
+        self._cur_seg = seg_dict
+
+    def visit_ele(self, err_ele: Any) -> None:
+        # err_ele.accept is only invoked from err_seg.accept (ISA / GS / ST
+        # accept methods do not walk their .elements list), so a non-None
+        # _cur_seg is guaranteed at this point.
+        if self._cur_seg is None:
+            return
+        ele_dict: dict[str, Any] = {
+            "ele_pos": err_ele.ele_pos,
+            "subele_pos": err_ele.subele_pos,
+            "repeat_pos": err_ele.repeat_pos,
+            "ele_ref_num": err_ele.ele_ref_num,
+            "name": err_ele.name,
+            "errors": [
+                {"err_cde": cde, "err_str": err_str, "err_val": err_val}
+                for (cde, err_str, err_val) in err_ele.errors
+            ],
+        }
+        self._cur_seg["elements"].append(ele_dict)

pyx12/test/test_errh_json.py

@@ -0,0 +1,191 @@
+import json
+import unittest
+from io import StringIO
+from unittest.mock import MagicMock
+
+import pyx12.errh_json
+import pyx12.error_handler
+import pyx12.segment
+
+
+def _fake_src(isa_id="000000001", gs_id="1", st_id="0001", cur_line=1, st_count=1):
+    src = MagicMock()
+    src.get_isa_id.return_value = isa_id
+    src.get_gs_id.return_value = gs_id
+    src.get_st_id.return_value = st_id
+    src.get_cur_line.return_value = cur_line
+    src.st_count = st_count
+    return src
+
+
+def _isa_segment():
+    raw = (
+        "ISA*00*          *00*          *ZZ*SENDER         *ZZ*RECEIVER       "
+        "*200101*1200*U*00501*000000001*0*P*:~"
+    )
+    return pyx12.segment.Segment(raw, "~", "*", ":")
+
+
+def _gs_segment():
+    return pyx12.segment.Segment(
+        "GS*HC*SENDER*RECEIVER*20200101*1200*1*X*005010X222A1", "~", "*", ":"
+    )
+
+
+def _st_segment():
+    return pyx12.segment.Segment("ST*837*0001*005010X222A1", "~", "*", ":")
+
+
+def _build_tree_with_errors() -> pyx12.error_handler.err_handler:
+    """Build a small but complete err_handler tree exercising every level
+    that the JSON visitor cares about: an ISA-level error, a GS-level
+    error, an ST-level error, a segment with seg-level errors plus two
+    element children carrying their own errors."""
+    eh = pyx12.error_handler.err_handler()
+    src = _fake_src()
+    eh.add_isa_loop(_isa_segment(), src)
+    eh.cur_isa_node.add_error("100", "ISA-level error")
+
+    eh.add_gs_loop(_gs_segment(), src)
+    eh.cur_gs_node.add_error("4", "GS-level error")
+    eh.cur_gs_node.ack_code = "R"
+    eh.cur_gs_node.st_count_orig = 1
+    eh.cur_gs_node.st_count_recv = 1
+
+    eh.add_st_loop(_st_segment(), src)
+    eh.cur_st_node.add_error("23", "ST-level error")
+    eh.cur_st_node.ack_code = "R"
+
+    seg_map = MagicMock()
+    seg_map.name = "Claim Information"
+    seg_map.pos = 130
+    eh.add_seg(seg_map, _seg_segment_clm(), seg_count=5, cur_line=42, ls_id="2300")
+    eh.seg_error("8", "Segment has data element errors", None, 42)
+
+    ele_map_a = MagicMock()
+    ele_map_a.data_ele = "1028"
+    ele_map_a.name = "Claim Submitter's Identifier"
+    ele_map_a.seq = 1
+    ele_map_a.parent.is_composite.return_value = False
+    eh.add_ele(ele_map_a)
+    eh.ele_error("7", "Invalid Code Value", "BAD", "CLM01")
+
+    ele_map_b = MagicMock()
+    ele_map_b.data_ele = "782"
+    ele_map_b.name = "Monetary Amount"
+    ele_map_b.seq = 2
+    ele_map_b.parent.is_composite.return_value = False
+    eh.add_ele(ele_map_b)
+    eh.ele_error("6", "Invalid character in data element", "X", "CLM02")
+
+    return eh
+
+
+def _seg_segment_clm():
+    return pyx12.segment.Segment("CLM*ABC*100***11:B:1", "~", "*", ":")
+
+
+class ErrhJsonVisitorOutput(unittest.TestCase):
+    """The JSON visitor walks the err_handler tree and emits a nested
+    document with one entry per ISA / GS / ST / segment / element level."""
+
+    def setUp(self):
+        eh = _build_tree_with_errors()
+        out = StringIO()
+        visitor = pyx12.errh_json.errh_json_visitor(out, indent=None)
+        eh.accept(visitor)
+        out.seek(0)
+        self.doc = json.loads(out.read())
+
+    def test_top_level_shape(self):
+        self.assertEqual(list(self.doc.keys()), ["interchanges"])
+        self.assertEqual(len(self.doc["interchanges"]), 1)
+
+    def test_isa_level_error_captured(self):
+        isa = self.doc["interchanges"][0]
+        self.assertEqual(isa["isa_trn_set_id"], "000000001")
+        self.assertEqual(isa["errors"], [{"err_cde": "100", "err_str": "ISA-level error"}])
+        self.assertEqual(len(isa["groups"]), 1)
+
+    def test_gs_level_error_captured(self):
+        gs = self.doc["interchanges"][0]["groups"][0]
+        self.assertEqual(gs["fic"], "HC")
+        self.assertEqual(gs["vriic"], "005010X222A1")
+        self.assertEqual(gs["ack_code"], "R")
+        self.assertEqual(gs["errors"], [{"err_cde": "4", "err_str": "GS-level error"}])
+        self.assertEqual(len(gs["transactions"]), 1)
+
+    def test_st_level_error_captured(self):
+        st = self.doc["interchanges"][0]["groups"][0]["transactions"][0]
+        self.assertEqual(st["trn_set_id"], "837")
+        self.assertEqual(st["trn_set_control_num"], "0001")
+        self.assertEqual(st["ack_code"], "R")
+        self.assertEqual(st["errors"], [{"err_cde": "23", "err_str": "ST-level error"}])
+        self.assertEqual(len(st["segments"]), 1)
+
+    def test_seg_level_error_captured(self):
+        seg = self.doc["interchanges"][0]["groups"][0]["transactions"][0]["segments"][0]
+        self.assertEqual(seg["seg_id"], "CLM")
+        self.assertEqual(seg["seg_count"], 5)
+        self.assertEqual(seg["pos"], 130)
+        self.assertEqual(seg["name"], "Claim Information")
+        self.assertEqual(seg["ls_id"], "2300")
+        self.assertEqual(
+            seg["errors"],
+            [{"err_cde": "8", "err_str": "Segment has data element errors", "err_val": None}],
+        )
+        self.assertEqual(len(seg["elements"]), 2)
+
+    def test_element_level_errors_captured(self):
+        elements = self.doc["interchanges"][0]["groups"][0]["transactions"][0]["segments"][0][
+            "elements"
+        ]
+        self.assertEqual(elements[0]["ele_pos"], 1)
+        self.assertEqual(elements[0]["ele_ref_num"], "1028")
+        self.assertEqual(elements[0]["name"], "Claim Submitter's Identifier")
+        self.assertEqual(
+            elements[0]["errors"],
+            [{"err_cde": "7", "err_str": "Invalid Code Value", "err_val": "BAD"}],
+        )
+        self.assertEqual(elements[1]["ele_pos"], 2)
+        self.assertEqual(elements[1]["ele_ref_num"], "782")
+
+
+class ErrhJsonVisitorEmptyTree(unittest.TestCase):
+    """A fresh err_handler with no children should produce a well-formed
+    JSON document with an empty interchanges list — no exceptions."""
+
+    def test_empty_tree_emits_empty_document(self):
+        eh = pyx12.error_handler.err_handler()
+        out = StringIO()
+        visitor = pyx12.errh_json.errh_json_visitor(out, indent=None)
+        eh.accept(visitor)
+        out.seek(0)
+        doc = json.loads(out.read())
+        self.assertEqual(doc, {"interchanges": []})
+
+
+class ErrhJsonVisitorIndent(unittest.TestCase):
+    """The indent argument is passed through to json.dump."""
+
+    def test_default_indent_produces_pretty_output(self):
+        eh = pyx12.error_handler.err_handler()
+        out = StringIO()
+        visitor = pyx12.errh_json.errh_json_visitor(out)  # default indent=2
+        eh.accept(visitor)
+        out.seek(0)
+        text = out.read()
+        self.assertIn("\n", text)
+
+    def test_indent_none_produces_compact_output(self):
+        eh = pyx12.error_handler.err_handler()
+        out = StringIO()
+        visitor = pyx12.errh_json.errh_json_visitor(out, indent=None)
+        eh.accept(visitor)
+        out.seek(0)
+        text = out.read()
+        self.assertNotIn("\n", text)
+
+
+if __name__ == "__main__":
+    unittest.main()