json-doctor

JSON Doctor — understand, validate and compare JSON without opening it.

Four things you actually need when a JSON file is too big to read:

inspect   what shape is this? (inferred schema, key coverage, sample values)
validate  is it well-formed, and where exactly does it break?
diff      what changed between two of them, structurally?
query     pull a value out by dotted path, including through arrays

Works on .json and .jsonl. No network, no credentials, no dependencies.

WHY IT REPORTS COVERAGE, NOT JUST KEYS

A key that appears in 3% of records is a different fact from one that appears in 100%, and the difference is usually the bug. A schema that says {"id": "string"} hides that half the records have no id at all. So every inferred field carries how often it was actually present, and optional fields are named as optional.

WHY VALIDATION POINTS AT A LINE AND COLUMN

"Invalid JSON" is a useless error on a 40MB file. json.JSONDecodeError already knows the position; this surfaces it with the surrounding text so you can see the trailing comma rather than go hunting for it.

Parameters

The typed contract this capability answers to (JSON Schema — the deterministic layer):

{
  "type": "object",
  "properties": {
    "action": {
      "type": "string",
      "enum": [
        "inspect",
        "validate",
        "diff",
        "query"
      ],
      "description": "What to do."
    },
    "path": {
      "type": "string",
      "description": "Path to the JSON/JSONL file."
    },
    "other": {
      "type": "string",
      "description": "Second file, for action=diff."
    },
    "key": {
      "type": "string",
      "description": "Dotted path for action=query, e.g. users.0.name"
    }
  },
  "required": [
    "action",
    "path"
  ]
}

Deterministic implementation

Run this instead of improvising when the inputs are well-formed:

def perform(self, **kwargs):
        action = kwargs.get("action")
        path = kwargs.get("path")
        if not path or not os.path.isfile(path):
            return json.dumps({"status": "error",
                               "message": f"file not found: {path}"}, indent=2)
        try:
            if action == "validate":
                try:
                    data, mode = _load(path)
                except (json.JSONDecodeError, ValueError) as e:
                    detail = {"status": "ok", "valid": False, "error": str(e)}
                    if isinstance(e, json.JSONDecodeError):
                        text = open(path, encoding="utf-8").read()
                        lo = max(0, e.pos - 60)
                        detail.update({"line": e.lineno, "column": e.colno,
                                       "context": text[lo:e.pos + 60]})
                    return json.dumps(detail, indent=2)
                n = len(data) if isinstance(data, list) else 1
                return json.dumps({"status": "ok", "valid": True, "mode": mode,
                                   "records": n,
                                   "root_type": _typename(data)}, indent=2)

            data, mode = _load(path)

            if action == "inspect":
                n = len(data) if isinstance(data, list) else 1
                return json.dumps({
                    "status": "ok", "mode": mode, "root_type": _typename(data),
                    "records": n, "bytes": os.path.getsize(path),
                    "fields": _infer(data),
                    "note": "coverage is the share of records containing the "
                            "field; anything under 100% is marked optional",
                }, indent=2)

            if action == "query":
                key = kwargs.get("key") or ""
                val, err = _walk(data, key)
                if err:
                    return json.dumps({"status": "error", "path": key,
                                       "message": err}, indent=2)
                return json.dumps({"status": "ok", "path": key,
                                   "type": _typename(val), "value": val}, indent=2)

            if action == "diff":
                other = kwargs.get("other")
                if not other or not os.path.isfile(other):
                    return json.dumps({"status": "error",
                                       "message": f"second file not found: {other}"}, indent=2)
                b, _ = _load(other)
                fa, fb = _infer(data), _infer(b)
                added = sorted(set(fb) - set(fa))
                removed = sorted(set(fa) - set(fb))
                changed = []
                for k in sorted(set(fa) & set(fb)):
                    if fa[k]["types"] != fb[k]["types"]:
                        changed.append({"field": k, "from": fa[k]["types"],
                                        "to": fb[k]["types"]})
                    elif fa[k]["coverage"] != fb[k]["coverage"]:
                        changed.append({"field": k, "coverage":
                                        f"{fa[k]['coverage']} -> {fb[k]['coverage']}"})
                return json.dumps({
                    "status": "ok",
                    "identical_shape": not (added or removed or changed),
                    "fields_added": added, "fields_removed": removed,
                    "fields_changed": changed,
                    "records": {"a": len(data) if isinstance(data, list) else 1,
                                "b": len(b) if isinstance(b, list) else 1},
                }, indent=2)

            return json.dumps({"status": "error",
                               "message": f"unknown action {action!r}",
                               "valid": ["inspect", "validate", "diff", "query"]}, indent=2)
        except Exception as e:
            return json.dumps({"status": "error",
                               "message": f"{type(e).__name__}: {e}"}, indent=2)