Schema-Constrained Integrity Scorer

The problem

It is difficult to ensure that complex, nested data structures are both correctly shaped and internally consistent. This often leads to errors when information is missing or formatted incorrectly.

What it does

It checks nested data to ensure it follows specific structural rules while simultaneously verifying that the information remains accurate.

Why it matters

It ensures that data remains reliable and structurally sound across complex systems.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 schema_constrained_integrity_scorer.py
[ OK ] scalar Zod types pass

All basic checks passed. Run  python3 schema_constrained_integrity_scorer.py demo  for the full demo.

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 383 lines, one file, standard library only.

"""
Schema-Constrained Integrity Scorer v2 — Transformation-Aware

Combines Zod-style schema validation with Hierarchical Cache-Validation
Integrity Scoring to validate nested data against both structural constraints
and checksum-based content integrity, with depth-aware penalty scaling.

NEW v2: Transformation-Aware scoring — validate output integrity after
custom mapping transforms. Define transform pipelines that compose field
mappings, then score whether the transformed data preserves schema integrity
and content fidelity using pre/post checksum comparison.

Usage:
    from schema_constrained_integrity_scorer import (
        SchemaRegistry, CompositeScorer, z_string, z_int, z_float,
        z_object, z_list, z_bool, z_nullable, z_union, z_enum,
        TransformMapper, TransformAwareScorer
    )

    schema = SchemaRegistry()
    schema.define("user", {"name": z_string(), "age": z_int()})
    result = CompositeScorer(schema).score(data, "user")

    # v2 — Transformation-Aware
    mapper = TransformMapper()
    mapper.register("rename_fields", {"full_name": "name", "years": "age"})
    result = TransformAwareScorer(schema).score_transformed(
        data, "user", "rename_fields"
    )

    python3 schema_constrained_integrity_scorer.py       # runs built-in tests
    python3 schema_constrained_integrity_scorer.py demo  # runs extended demo
"""

import hashlib
import json
import re
from copy import deepcopy
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Optional


# ---------------------------------------------------------------------------
# Schema-level primitives -- Zod-style type constructors
# ---------------------------------------------------------------------------

class ZodTypeKind(Enum):
    STRING = "string"
    INT = "int"
    FLOAT = "float"
    BOOL = "bool"
    LIST = "list"
    OBJECT = "object"
    ANY = "any"
    NULLABLE = "nullable"
    UNION = "union"
    ENUM = "enum"


@dataclass
class ZodType:
    kind: ZodTypeKind
    inner: Optional[Any] = None
    options: Optional[List[Any]] = None
    constraints: Dict[str, Any] = field(default_factory=dict)

    def describe(self) -> str:
        c = self.constraints
        parts = [self.kind.value]
        if self.kind == ZodTypeKind.LIST and self.inner:
            parts.append(f"<{self.inner.describe()}>")
        if self.kind == ZodTypeKind.NULLABLE and self.inner:
            parts.append(f"<{self.inner.describe()}>")
        if self.kind == ZodTypeKind.UNION and self.options:
            parts.append("(" + " | ".join(o.describe() for o in self.options) + ")")
        if c:
            parts.append(str(c))
        return ".".join(parts)


# Public builder functions ---------------------------------------------------

def z_string(min_len=0, max_len=None, pattern=None) -> ZodType:
    c = {"min_len": min_len}
    if max_len is not None:
        c["max_len"] = max_len
    if pattern is not None:
        c["pattern"] = pattern
    return ZodType(kind=ZodTypeKind.STRING, constraints=c)


def z_int(min_val=None, max_val=None) -> ZodType:
    c = {}
    if min_val is not None:
        c["min"] = min_val
    if max_val is not None:
        c["max"] = max_val
    return ZodType(kind=ZodTypeKind.INT, constraints=c)


def z_float(min_val=None, max_val=None) -> ZodType:
    c = {}
    if min_val is not None:
        c["min"] = min_val
    if max_val is not None:
        c["max"] = max_val
    return ZodType(kind=ZodTypeKind.FLOAT, constraints=c)


def z_bool() -> ZodType:
    return ZodType(kind=ZodTypeKind.BOOL)


def z_list(inner: ZodType) -> ZodType:
    return ZodType(kind=ZodTypeKind.LIST, inner=inner)


def z_object(shape: Dict[str, "ZodType"]) -> ZodType:
    return ZodType(kind=ZodTypeKind.OBJECT, inner=shape)


def z_any() -> ZodType:
    return ZodType(kind=ZodTypeKind.ANY)


def z_nullable(inner: ZodType) -> ZodType:
    return ZodType(kind=ZodTypeKind.NULLABLE, inner=inner)


def z_union(*options: ZodType) -> ZodType:
    return ZodType(kind=ZodTypeKind.UNION, options=list(options))


def z_enum(*values: str) -> ZodType:
    return ZodType(kind=ZodTypeKind.ENUM, options=list(values))


# Convenience: coerce raw Python types and dicts to ZodType on first use
def coerce_schema(raw: Any) -> ZodType:
    if isinstance(raw, ZodType):
        return raw
    if raw is str:
        return z_string()
    if raw is int:
        return z_int()
    if raw is float:
        return z_float()
    if raw is bool:
        return z_bool()
    if raw is list:
        return z_list(z_any())
    if raw is dict:
        return z_object({})
    if isinstance(raw, dict):
        return z_object({k: coerce_schema(v) for k, v in raw.items()})
    raise TypeError(f"Cannot coerce {raw!r} to ZodType")


# ---------------------------------------------------------------------------
# Schema repository
# ---------------------------------------------------------------------------

class SchemaDefinitionError(Exception):
    pass


class SchemaRegistry:
    """Registers named Zod schemas. Accepts dict values that get auto-coerced."""

    def __init__(self):
        self._schemas: Dict[str, ZodType] = {}

    def define(self, name: str, shape: Dict[str, Any]) -> "SchemaRegistry":
        coerced = z_object({k: coerce_schema(v) for k, v in shape.items()})
        self._schemas[name] = coerced
        return self

    def get(self, name: str) -> ZodType:
        if name not in self._schemas:
            raise KeyError(f"Schema '{name}' not registered")
        return self._schemas[name]

    def list_names(self) -> List[str]:
        return list(self._schemas.keys())


# ---------------------------------------------------------------------------
# Zod-style validator (structural constraints)
# ---------------------------------------------------------------------------

class ZodValidationError:
    def __init__(self, path: str, reason: str):
        self.path = path
        self.reason = reason

    def __str__(self) -> str:
        return f"[{self.path}] {self.reason}"


class ZodResult:
    def __init__(self, success: bool, errors: List[ZodValidationError]):
        self.success = success
        self.errors = errors


def _validate_zod(data: Any, schema: ZodType, path: str = "") -> ZodResult:
    errors: List[ZodValidationError] = []

    if schema.kind == ZodTypeKind.NULLABLE:
        if data is None:
            return ZodResult(True, [])
        if schema.inner is not None:
            return _validate_zod(data, schema.inner, path)
        return ZodResult(True, [])

    if data is None:
        errors.append(ZodValidationError(path, f"expected {schema.kind.value}, got null"))
        return ZodResult(success=False, errors=errors)

    if schema.kind == ZodTypeKind.ANY:
        return ZodResult(success=True, errors=[])

    if schema.kind == ZodTypeKind.STRING:
        if not isinstance(data, str):
            errors.append(ZodValidationError(path, f"expected string, got {type(data).__name__}"))
        else:
            c = schema.constraints
            if len(data) < c.get("min_len", 0):
                errors.append(ZodValidationError(path, f"string too short (min {c['min_len']})"))
            if c.get("max_len") is not None and len(data) > c["max_len"]:
                errors.append(ZodValidationError(path, f"string too long (max {c['max_len']})"))
            if c.get("pattern") and not re.match(c["pattern"], data):
                errors.append(ZodValidationError(path, f"string does not match pattern {c['pattern']}"))
        return ZodResult(success=len(errors) == 0, errors=errors)

    if schema.kind == ZodTypeKind.INT:
        if not isinstance(data, int) or isinstance(data, bool):
            errors.append(ZodValidationError(path, f"expected int, got {type(data).__name__}"))
        else:
            c = schema.constraints
            if c.get("min") is not None and data < c["min"]:
                errors.append(ZodValidationError(path, f"int below min {c['min']}"))
            if c.get("max") is not None and data > c["max"]:
                errors.append(ZodValidationError(path, f"int above max {c['max']}"))
        return ZodResult(success=len(errors) == 0, errors=errors)

    if schema.kind == ZodTypeKind.FLOAT:
        if not isinstance(data, (int, float)) or isinstance(data, bool):
            errors.append(ZodValidationError(path, f"expected float, got {type(data).__name__}"))
        else:
            c = schema.constraints
            if c.get("min") is not None and data < c["min"]:
                errors.append(ZodValidationError(path, f"float below min {c['min']}"))
            if c.get("max") is not None and data > c["max"]:
                errors.append(ZodValidationError(path, f"float above max {c['max']}"))
        return ZodResult(success=len(errors) == 0, errors=errors)

    if schema.kind == ZodTypeKind.BOOL:
        if not isinstance(data, bool):
            errors.append(ZodValidationError(path, f"expected bool, got {type(data).__name__}"))
        return ZodResult(success=len(errors) == 0, errors=errors)

    if schema.kind == ZodTypeKind.ENUM:
        if data not in (schema.options or []):
            errors.append(ZodValidationError(path, f"value {data!r} not in enum {schema.options}"))
        return ZodResult(success=len(errors) == 0, errors=errors)

    if schema.kind == ZodTypeKind.LIST:
        if not isinstance(data, list):
            errors.append(ZodValidationError(path, f"expected list, got {type(data).__name__}"))
        elif schema.inner:
            for i, item in enumerate(data):
                sub = _validate_zod(item, schema.inner, f"{path}[{i}]")
                errors.extend(sub.errors)
        return ZodResult(success=len(errors) == 0, errors=errors)

    if schema.kind == ZodTypeKind.OBJECT:
        if not isinstance(data, dict):
            errors.append(ZodValidationError(path, f"expected object, got {type(data).__name__}"))
        elif schema.inner:
            shape = schema.inner
            for key, field_schema in shape.items():
                field_path = f"{path}.{key}" if path else key
                if key not in data:
                    errors.append(ZodValidationError(field_path, "required field missing"))
                else:
                    sub = _validate_zod(data[key], field_schema, field_path)
                    errors.extend(sub.errors)
        return ZodResult(success=len(errors) == 0, errors=errors)

    if schema.kind == ZodTypeKind.UNION:
        for option in (schema.options or []):
            res = _validate_zod(data, option, path)
            if res.success:
                return res
        errors.append(ZodValidationError(path, "value matches no union variant"))
        return ZodResult(success=False, errors=errors)

    return ZodResult(success=True, errors=[])


# ---------------------------------------------------------------------------
# Hierarchical Cache-Validation Integrity Hasher
# ---------------------------------------------------------------------------

def _hash_value(value: Any) -> str:
    """Deterministic hash of any JSON-serializable structure."""
    canonical = json.dumps(value, sort_keys=True, default=str)
    return hashlib.sha256(canonical.encode()).hexdigest()[:16]


def _compute_integrity_tree(data: Any, depth: int = 0) -> Dict[str, Any]:
    """Recursively build a tree of hashes keyed by path, with depth annotation."""
    tree = {}
    root_hash = _hash_value(data)
    tree["__root__"] = {"hash": root_hash, "depth": depth}
    if isinstance(data, dict):
        for k, v in data.items():
            subtree = _compute_integrity_tree(v, depth + 1)
            tree[k] = subtree
            if isinstance(v, (dict, list)):
                tree[k]["__root__"] = subtree["__root__"]
    elif isinstance(data, list):
        for i, item in enumerate(data):
            subtree = _compute_integrity_tree(item, depth + 1)
            tree[str(i)] = subtree
            if isinstance(item, (dict, list)):
                tree[str(i)]["__root__"] = subtree["__root__"]
    return tree


def _compare_integrity_trees(a_tree: Dict, b_tree: Dict, path: str = "", max_penalty: float = 1.0) -> float:
    """Compare two integrity trees and return a depth-weighted penalty score.
    0.0 = identical, 1.0 = fully broken. Penalty is scaled by depth: deeper mismatches hurt less."""
    base_depth = a_tree.get("__root__", {}).get("depth", 0)
    depth_factor = 1.0 / (1.0 + base_depth * 0.3)

    if a_tree.get("__root__", {}).get("hash") == b_tree.get("__root__", {}).get("hash"):
        return 0.0

    all_keys = set(a_tree.keys()) | set(b_tree.keys())
    all_keys.discard("__root__")
    penalty = 0.0
    for k in all_keys:
        sub_path = f"{path}.{k}" if path else k
        a_sub = a_tree.get(k, {})
        b_sub = b_tree.get(k, {})
        if not a_sub:
            penalty += 1.0 * depth_factor / len(all_keys)
        elif not b_sub:
            penalty += 1.0 * depth_factor / len(all_keys)
        else:
            penalty += _compare_integrity_trees(a_sub, b_sub, sub_path, max_penalty) / len(all_keys)
    return min(penalty * depth_factor, max_penalty)


# ---------------------------------------------------------------------------
# High-level composite scorer (structure + integrity)
# ---------------------------------------------------------------------------

@dataclass
class ScoreReport:
    schema_name: str
    schema_valid: bool
    schema_errors: List[str]
    integrity_score: float  # 0.0 = perfect
    baseline_hash: str
    current_hash: str
    total_penalty: float

    def verdict(self) -> str:
        if not self.schema_valid:
            return "REJECTED — schema violation"
        if self.integrity_score >= 0.5:
            return "DEGRADED — integrity failure"
        if self.integrity_score > 0.0:
            return "PASS — minor drift"
        return "PASS — pristine"


class CompositeScorer:
    """Scores data