"""Mini JSON Schema validator for versioned final-report data.json records. This is a deliberately narrow JSON Schema implementation that supports exactly the keywords used by ``schemas/final-report-v2.0.schema.json``: type, required, properties, additionalProperties, enum, const, pattern, minLength, minItems, maxItems, uniqueItems, items, minimum, maximum, $ref ($defs), oneOf, allOf, if/then/else, contains, minContains, maxContains, not We do NOT depend on the ``jsonschema`` PyPI package because its dependency tree (``referencing``, ``rpds-py``) includes a Rust C extension which we would have to vendor or ship pre-built per platform. The ~250 lines below cover everything the final-report schema needs; strict spec compliance is not a goal — strict validation of OUR schema is. Error messages include the JSON pointer of the failing field so report assembly can return the defect to the input owner. """ from __future__ import annotations import json import re from pathlib import Path from typing import Any class SchemaError(ValueError): """Raised when a schema itself is malformed (e.g. a $ref points nowhere). Distinct from a data validation failure. """ def _format_path(path: tuple[str | int, ...]) -> str: if not path: return "" parts: list[str] = [] for p in path: if isinstance(p, int): parts.append(f"[{p}]") else: parts.append(f".{p}" if parts else p) return "".join(parts) _TYPE_PYTHON: dict[str, tuple[type, ...]] = { "object": (dict,), "array": (list,), "string": (str,), "integer": (int,), "number": (int, float), "boolean": (bool,), "null": (type(None),), } def _check_type(value: Any, expected: str) -> bool: if expected == "integer": # JSON Schema treats booleans as not integer. return isinstance(value, int) and not isinstance(value, bool) if expected == "number": return isinstance(value, (int, float)) and not isinstance(value, bool) return isinstance(value, _TYPE_PYTHON.get(expected, ())) def _json_schema_equal(left: Any, right: Any) -> bool: if isinstance(left, bool) or isinstance(right, bool): return isinstance(left, bool) and isinstance(right, bool) and left == right if isinstance(left, (int, float)) or isinstance(right, (int, float)): return ( isinstance(left, (int, float)) and isinstance(right, (int, float)) and left == right ) if isinstance(left, list) or isinstance(right, list): return ( isinstance(left, list) and isinstance(right, list) and len(left) == len(right) and all(_json_schema_equal(a, b) for a, b in zip(left, right)) ) if isinstance(left, dict) or isinstance(right, dict): return ( isinstance(left, dict) and isinstance(right, dict) and left.keys() == right.keys() and all(_json_schema_equal(left[key], right[key]) for key in left) ) return type(left) is type(right) and left == right def _resolve_ref(ref: str, root: dict) -> dict: """Resolve a ``#/$defs/Name`` style reference against ``root``.""" if not ref.startswith("#/"): raise SchemaError(f"only local refs are supported, got: {ref}") parts = ref[2:].split("/") node: Any = root for part in parts: if not isinstance(node, dict) or part not in node: raise SchemaError(f"$ref target not found: {ref}") node = node[part] if not isinstance(node, dict): raise SchemaError(f"$ref target is not a schema: {ref}") return node class _Validator: def __init__(self, root_schema: dict): self.root = root_schema self.errors: list[str] = [] def validate(self, instance: Any, schema: dict, path: tuple[str | int, ...]) -> None: # Handle $ref first — everything else is composed under the # resolved schema. if "$ref" in schema: schema = _resolve_ref(schema["$ref"], self.root) # const / enum: strict equality / membership. if "const" in schema and instance != schema["const"]: self._err(path, f"value is not equal to const {schema['const']!r}") if "enum" in schema and instance not in schema["enum"]: self._err(path, f"value {instance!r} is not in enum {schema['enum']!r}") # type type_keyword = schema.get("type") if type_keyword is not None: types = type_keyword if isinstance(type_keyword, list) else [type_keyword] if not any(_check_type(instance, t) for t in types): self._err( path, f"value of type {type(instance).__name__} is not of expected type(s) {types}", ) return # further checks would compound the error # String constraints if isinstance(instance, str): min_length = schema.get("minLength") if min_length is not None and len(instance) < min_length: self._err(path, f"string length {len(instance)} < minLength {min_length}") pattern = schema.get("pattern") if pattern is not None and not re.search(pattern, instance): self._err(path, f"string does not match pattern {pattern!r}") # Numeric constraints if isinstance(instance, (int, float)) and not isinstance(instance, bool): minimum = schema.get("minimum") if minimum is not None and instance < minimum: self._err(path, f"value {instance} < minimum {minimum}") maximum = schema.get("maximum") if maximum is not None and instance > maximum: self._err(path, f"value {instance} > maximum {maximum}") # Object constraints if isinstance(instance, dict): self._validate_object(instance, schema, path) # Array constraints if isinstance(instance, list): self._validate_array(instance, schema, path) # Composition keywords for sub in schema.get("allOf", []): self.validate(instance, sub, path) # `if/then/else` lives inside an allOf entry in our schemas. if "if" in sub: self._validate_conditional(instance, sub, path) if "if" in schema: self._validate_conditional(instance, schema, path) if "oneOf" in schema: self._validate_one_of(instance, schema["oneOf"], path) if "not" in schema: self._validate_not(instance, schema["not"], path) def _validate_object(self, instance: dict, schema: dict, path: tuple[str | int, ...]) -> None: properties = schema.get("properties") or {} required = schema.get("required") or [] for name in required: if name not in instance: self._err(path, f"required property '{name}' is missing") for name, value in instance.items(): if name in properties: self.validate(value, properties[name], path + (name,)) elif schema.get("additionalProperties") is False: # Don't fire for keys we know are part of the conditional # branches (we still validate values when they appear). self._err(path, f"additional property '{name}' is not allowed") def _validate_array(self, instance: list, schema: dict, path: tuple[str | int, ...]) -> None: min_items = schema.get("minItems") if min_items is not None and len(instance) < min_items: self._err(path, f"array length {len(instance)} < minItems {min_items}") max_items = schema.get("maxItems") if max_items is not None and len(instance) > max_items: self._err(path, f"array length {len(instance)} > maxItems {max_items}") if schema.get("uniqueItems") is True and any( _json_schema_equal(item, previous) for index, item in enumerate(instance) for previous in instance[:index] ): self._err(path, "array items are not unique") items = schema.get("items") if items is not None: for i, value in enumerate(instance): self.validate(value, items, path + (i,)) contains = schema.get("contains") if contains is not None: matches = sum(self._matches(value, contains) for value in instance) minimum = schema.get("minContains", 1) maximum = schema.get("maxContains") if matches < minimum: self._err( path, f"array contains {matches} matching item(s), below minContains {minimum}", ) if maximum is not None and matches > maximum: self._err( path, f"array contains {matches} matching item(s), above maxContains {maximum}", ) def _validate_conditional(self, instance: Any, schema: dict, path: tuple[str | int, ...]) -> None: if_schema = schema.get("if") if if_schema is None: return if self._matches(instance, if_schema): then_schema = schema.get("then") if then_schema is not None: self.validate(instance, then_schema, path) else: else_schema = schema.get("else") if else_schema is not None: self.validate(instance, else_schema, path) def _validate_one_of(self, instance: Any, branches: list[dict], path: tuple[str | int, ...]) -> None: matches = sum(1 for b in branches if self._matches(instance, b)) if matches != 1: self._err( path, f"oneOf matched {matches} branches (expected exactly 1); " f"branches: {[self._summarise(b) for b in branches]}", ) def _validate_not(self, instance: Any, sub_schema: dict, path: tuple[str | int, ...]) -> None: if self._matches(instance, sub_schema): self._err(path, f"value must NOT match {self._summarise(sub_schema)}") def _matches(self, instance: Any, schema: dict) -> bool: """Cheap 'does this validate' probe used by if/oneOf/contains. Returns True iff the sub-schema produces zero errors. Does not mutate ``self.errors``. """ probe = _Validator(self.root) probe.validate(instance, schema, ()) return not probe.errors @staticmethod def _summarise(schema: dict) -> str: keys = sorted(k for k in schema if k not in ("description", "$comment")) return "{" + ", ".join(f"{k}={schema[k]!r}" for k in keys[:3]) + "}" def _err(self, path: tuple[str | int, ...], message: str) -> None: self.errors.append(f"{_format_path(path)}: {message}") def validate(data: Any, schema: dict) -> list[str]: """Validate ``data`` against ``schema``. Returns the list of human- readable error messages (empty when the data is valid).""" v = _Validator(schema) v.validate(data, schema, ()) return v.errors SCHEMA_FILENAMES = { "2.0": "final-report-v2.0.schema.json", "3.0": "final-report-v3.0.schema.json", } def load_named_schema(filename: str, start: Path | None = None) -> dict: here = (start or Path(__file__)).resolve() if here.is_file(): here = here.parent for parent in [here, *here.parents]: candidate = parent / "schemas" / filename if candidate.is_file(): return json.loads(candidate.read_text(encoding="utf-8")) raise SchemaError(f"could not locate schemas/{filename}") def load_schema_version(version: str, start: Path | None = None) -> dict: """Load the final-report schema identified by its data version.""" try: filename = SCHEMA_FILENAMES[version] except KeyError as exc: raise SchemaError(f"unsupported final-report schemaVersion: {version}") from exc return load_named_schema(filename, start=start) def load_schema_for_data(data: dict, start: Path | None = None) -> dict: """Select a schema from the explicit ``schemaVersion`` in *data*.""" version = data.get("schemaVersion") if not isinstance(version, str) or not version: raise SchemaError("final-report data has no schemaVersion") return load_schema_version(version, start=start) def load_schema(schema_path: Path | None = None) -> dict: """Load the compatibility schema used by historical direct callers. New write paths select contract 3.0 explicitly. Readers select from the record's ``schemaVersion`` through :func:`load_schema_for_data`. """ if schema_path is not None: return json.loads(Path(schema_path).read_text(encoding="utf-8")) return load_schema_version("2.0")