troupe-standup/wizard-work/beam/jsonl_validate.py

#!/usr/bin/env python3
"""jsonl_validate.py — streaming JSONL schema validator.

Usage:
    jsonl_validate.py SCHEMA.json INPUT.jsonl [--max-errors N]

SCHEMA is a small JSON document:
    {
      "required": ["id", "name"],
      "fields": {
        "id": "int",
        "name": "str",
        "tags": "list",
        "meta": "object",
        "score": "float",
        "active": "bool",
        "email": "str?"
      }
    }

Type names: str, int, float, bool, list, object, any. Append '?' for optional
(missing allowed). A field present with null is treated as missing for optional
fields and as an error for required fields -- one error per record per field, not
one for the absence and another for the null.

Input is read as UTF-8; a leading BOM is stripped, not reported as bad data.

Exit codes (pinned by test_jsonl_validate.py; a run that cannot read its input
returns 2 and prints NO summary line, so a partial count is never a verdict):
    0  all records valid
    1  one or more validation errors
    2  usage / IO / schema / decode error -- including an input, or a schema,
       that is not valid UTF-8 text
"""
import argparse
import json
import sys
from typing import Any, Dict, List, Tuple


TYPE_CHECKS = {
    "str": lambda v: isinstance(v, str),
    "int": lambda v: isinstance(v, int) and not isinstance(v, bool),
    "float": lambda v: (isinstance(v, (int, float)) and not isinstance(v, bool)),
    "bool": lambda v: isinstance(v, bool),
    "list": lambda v: isinstance(v, list),
    "object": lambda v: isinstance(v, dict),
    "any": lambda v: True,
}


def load_json(path: str) -> Any:
    # utf-8-sig on both inputs: a BOM is an encoding marker, not bad data.
    with open(path, "r", encoding="utf-8-sig") as f:
        return json.load(f)


def validate_schema(schema: Any) -> Tuple[bool, str]:
    if not isinstance(schema, dict):
        return False, "schema must be an object"
    if "required" in schema and not (isinstance(schema["required"], list)
                                    and all(isinstance(x, str) for x in schema["required"])):
        return False, "'required' must be a list of strings"
    if "fields" in schema and not isinstance(schema["fields"], dict):
        return False, "'fields' must be an object"
    fields = schema.get("fields", {})
    for name, typ in fields.items():
        if not isinstance(name, str):
            return False, "field names must be strings"
        base = typ[:-1] if isinstance(typ, str) and typ.endswith("?") else typ
        if base not in TYPE_CHECKS:
            return False, f"unknown type '{typ}' for field '{name}'"
    return True, ""


def check_value(typ: str, value: Any) -> bool:
    optional = typ.endswith("?")
    base = typ[:-1] if optional else typ
    if value is None and optional:
        return True
    return TYPE_CHECKS[base](value)


def validate_record(rec: Any, schema: dict) -> List[str]:
    errs: List[str] = []
    if not isinstance(rec, dict):
        return ["record is not an object"]
    required = set(schema.get("required", []))
    fields = schema.get("fields", {})
    for r in required:
        if r not in rec or rec.get(r) is None:
            errs.append(f"missing required field '{r}'")
    for name, typ in fields.items():
        if name in rec and rec[name] is not None:
            if not check_value(typ, rec[name]):
                errs.append(f"field '{name}': expected {typ}, got {type(rec[name]).__name__}")
        elif name in rec and rec[name] is None:
            # A required field present as null already produced its error above
            # ("missing required field"); do not count the same fact twice.
            optional = typ.endswith("?")
            if not optional and name not in required:
                errs.append(f"field '{name}': null not allowed")
    # unknown fields? warn-level: not an error, but collect as info? keep strict-ish: allow extras
    return errs


def main(argv: List[str]) -> int:
    ap = argparse.ArgumentParser(description="Validate a JSONL file against a schema.")
    ap.add_argument("schema")
    ap.add_argument("input")
    ap.add_argument("--max-errors", type=int, default=50)
    args = ap.parse_args(argv)

    try:
        schema = load_json(args.schema)
    except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
        print(f"schema error: {e}", file=sys.stderr)
        return 2
    ok, msg = validate_schema(schema)
    if not ok:
        print(f"schema error: {msg}", file=sys.stderr)
        return 2

    total = 0
    bad = 0
    err_count = 0
    stopped = False
    try:
        # utf-8-sig: a leading BOM is an encoding marker, not invalid data, so it
        # is stripped rather than graded as a bad record on line 1.
        with open(args.input, "r", encoding="utf-8-sig") as f:
            for lineno, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                total += 1
                try:
                    rec = json.loads(line)
                except json.JSONDecodeError as e:
                    bad += 1
                    err_count += 1
                    print(f"line {lineno}: invalid JSON: {e}")
                    if err_count >= args.max_errors:
                        stopped = True
                        break
                    continue
                errs = validate_record(rec, schema)
                if errs:
                    bad += 1
                    for e in errs:
                        err_count += 1
                        print(f"line {lineno}: {e}")
                        if err_count >= args.max_errors:
                            stopped = True
                            break
                if stopped:
                    break
    except UnicodeDecodeError as e:
        # Undecodable bytes are an IO/encoding failure, not a validation error:
        # rc 2, no traceback, and no summary line that a caller could mistake
        # for a verdict on the rows it never reached.
        print(f"input error: not valid UTF-8 text ({e.reason})", file=sys.stderr)
        return 2
    except OSError as e:
        print(f"input error: {e}", file=sys.stderr)
        return 2

    print(f"checked {total} records, {bad} invalid" + (" (truncated)" if stopped else ""))
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))