157 lines
5.1 KiB
Python
Executable File
157 lines
5.1 KiB
Python
Executable File
#!/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.
|
|
|
|
Exit codes:
|
|
0 all records valid
|
|
1 one or more validation errors
|
|
2 usage / IO / schema error
|
|
"""
|
|
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:
|
|
with open(path, "r", encoding="utf-8") 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:
|
|
optional = typ.endswith("?")
|
|
if name in required or not optional:
|
|
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) 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:
|
|
with open(args.input, "r", encoding="utf-8") 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 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:])) |