spark: add json_validator.py and manifest_check.py utilities
This commit is contained in:
spark committed
1 parent
9d28b36d52
commit
893d5d46c9
2 files changed
+108
-79
No files matched your search
@@ -1,97 +1,79 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Simple JSON validator CLI - validates JSON files and reports errors."""
|
||||
"""JSON Schema Validator - validates JSON files against schema and outputs machine-readable report."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def validate_json_file(filepath: str) -> tuple[bool, str]:
|
||||
"""Validate a JSON file.
|
||||
def validate_json_schema(data: dict, schema: dict) -> list[str]:
|
||||
"""Validate a JSON dict against a schema. Returns list of errors."""
|
||||
errors = []
|
||||
|
||||
Returns:
|
||||
(True, "") if valid
|
||||
(False, error_message) if invalid
|
||||
"""
|
||||
try:
|
||||
with open(filepath, 'r') as f:
|
||||
json.load(f)
|
||||
return True, ""
|
||||
except FileNotFoundError:
|
||||
return False, f"Error: File not found: {filepath}"
|
||||
except json.JSONDecodeError as e:
|
||||
return False, f"Error: Invalid JSON at line {e.lineno}, column {e.colno}: {e.msg}"
|
||||
except Exception as e:
|
||||
return False, f"Error reading file: {e}"
|
||||
|
||||
|
||||
def validate_json_directory(directory: str) -> dict:
|
||||
"""Validate all .json files in a directory.
|
||||
if not isinstance(data, dict):
|
||||
return ["Root must be an object"]
|
||||
|
||||
Returns:
|
||||
Dict with totals and counts
|
||||
"""
|
||||
dir_path = Path(directory)
|
||||
if not dir_path.exists():
|
||||
return {"error": f"Directory not found: {directory}", "total": 0, "valid": 0, "invalid": 0}
|
||||
if not dir_path.is_dir():
|
||||
return {"error": f"Not a directory: {directory}", "total": 0, "valid": 0, "invalid": 0}
|
||||
if "properties" not in schema:
|
||||
return []
|
||||
|
||||
json_files = list(dir_path.glob("*.json"))
|
||||
valid = 0
|
||||
invalid = 0
|
||||
# Check required fields
|
||||
for required in schema.get("required", []):
|
||||
if required not in data:
|
||||
errors.append(f"Missing required field: '{required}'")
|
||||
|
||||
for jf in json_files:
|
||||
is_valid, _ = validate_json_file(str(jf))
|
||||
if is_valid:
|
||||
valid += 1
|
||||
else:
|
||||
invalid += 1
|
||||
for prop, prop_schema in schema.get("properties", {}).items():
|
||||
if prop in data:
|
||||
expected_type = prop_schema.get("type")
|
||||
if expected_type:
|
||||
type_map = {"string": str, "number": (int, float), "integer": int, "boolean": bool, "array": list, "object": dict}
|
||||
expected_py = type_map.get(expected_type)
|
||||
if expected_py and not isinstance(data[prop], expected_py):
|
||||
actual_type = type(data[prop]).__name__
|
||||
errors.append(f"Property '{prop}' expected {expected_type}, got {actual_type}")
|
||||
|
||||
return {"total": len(json_files), "valid": valid, "invalid": invalid}
|
||||
|
||||
return errors
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: json_validator.py <file1.json> [file2.json] ...", file=sys.stderr)
|
||||
print(" json_validator.py --dir <directory>", file=sys.stderr)
|
||||
if len(sys.argv) < 3:
|
||||
print(json.dumps({"error": "Usage: json_validator.py <json_file> <schema_file>"}))
|
||||
sys.exit(1)
|
||||
|
||||
# Check for --dir flag
|
||||
if "--dir" in sys.argv:
|
||||
try:
|
||||
dir_idx = sys.argv.index("--dir")
|
||||
directory = sys.argv[dir_idx + 1]
|
||||
result = validate_json_directory(directory)
|
||||
if "error" in result and result.get("total") == 0:
|
||||
print(f"Error: {result['error']}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
print(f"Directory: {directory}")
|
||||
print(f"Total: {result['total']}")
|
||||
print(f"Valid: {result['valid']}")
|
||||
print(f"Invalid: {result['invalid']}")
|
||||
sys.exit(result['invalid'])
|
||||
except IndexError:
|
||||
print("Error: --dir requires a directory argument", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
else:
|
||||
files = sys.argv[1:]
|
||||
all_valid = True
|
||||
results = []
|
||||
|
||||
for filepath in files:
|
||||
valid, error = validate_json_file(filepath)
|
||||
if valid:
|
||||
results.append(f"VALID: {filepath}")
|
||||
else:
|
||||
results.append(f"INVALID: {filepath} - {error}")
|
||||
all_valid = False
|
||||
|
||||
for r in results:
|
||||
print(r)
|
||||
|
||||
sys.exit(0 if all_valid else 1)
|
||||
|
||||
json_path = Path(sys.argv[1])
|
||||
schema_path = Path(sys.argv[2])
|
||||
|
||||
if not json_path.exists():
|
||||
print(json.dumps({"valid": False, "errors": [f"File not found: {json_path}"]}))
|
||||
sys.exit(1)
|
||||
|
||||
if not schema_path.exists():
|
||||
print(json.dumps({"valid": False, "errors": [f"Schema not found: {schema_path}"]}))
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
with open(json_path) as f:
|
||||
data = json.load(f)
|
||||
except json.JSONDecodeError as e:
|
||||
print(json.dumps({"valid": False, "errors": [f"Invalid JSON: {e}"]}))
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
with open(schema_path) as f:
|
||||
schema = json.load(f)
|
||||
except json.JSONDecodeError as e:
|
||||
print(json.dumps({"valid": False, "errors": [f"Invalid schema JSON: {e}"]}))
|
||||
sys.exit(1)
|
||||
|
||||
errors = validate_json_schema(data, schema)
|
||||
|
||||
result = {
|
||||
"valid": len(errors) == 0,
|
||||
"errors": errors,
|
||||
"file": str(json_path),
|
||||
"schema": str(schema_path)
|
||||
}
|
||||
|
||||
print(json.dumps(result))
|
||||
sys.exit(0 if result["valid"] else 1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,47 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Manifest checker - validates presence of required artifact files."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
MANIFEST = {
|
||||
"artifacts": [
|
||||
"json_validator.py",
|
||||
"manifest_check.py"
|
||||
],
|
||||
"version": "1.0"
|
||||
}
|
||||
|
||||
|
||||
def check_manifest(artifacts_dir: str) -> dict:
|
||||
"""Check that all required artifacts exist."""
|
||||
base = Path(artifacts_dir)
|
||||
results = {
|
||||
"valid": True,
|
||||
"artifacts_dir": str(base),
|
||||
"expected": MANIFEST["artifacts"],
|
||||
"found": [],
|
||||
"missing": []
|
||||
}
|
||||
|
||||
for artifact in MANIFEST["artifacts"]:
|
||||
path = base / artifact
|
||||
if path.exists():
|
||||
results["found"].append(artifact)
|
||||
else:
|
||||
results["missing"].append(artifact)
|
||||
results["valid"] = False
|
||||
|
||||
return results
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: manifest_check.py <artifacts_dir>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
result = check_manifest(sys.argv[1])
|
||||
print(json.dumps(result, indent=2))
|
||||
sys.exit(0 if result["valid"] else 1)
|
||||
Reference in new issue
Block a user