DataOps-platform/app/core/data_rules/compiler.py

"""Deterministic RuleSpec-to-plan compilation for production-line release."""

from __future__ import annotations

import hashlib
import json
from typing import Any

from app.core.common.identifiers import ensure_governance_uid
from app.core.data_rules.contracts import rule_spec_hash, validate_rule_spec


COMPILER_VERSION = "dataops-rulespec-1.0"


def _canonical_hash(value: Any) -> str:
    canonical = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def compile_rule_plan(rule_version: dict[str, Any]) -> dict[str, Any]:
    """Compile a published rule into a target-neutral immutable plan.

    This compiler deliberately emits no SQL or Python. M3 runtime adapters
    translate the closed RuleSpec operators for a deployment's bound backend.
    """

    if not isinstance(rule_version, dict):
        raise ValueError("rule version must be an object")
    if rule_version.get("status") != "published":
        raise ValueError("only published rule versions may be compiled")
    try:
        version_id = ensure_governance_uid(
            {"uid": str(rule_version.get("id"))}
        )
    except ValueError as exc:
        raise ValueError("rule version id must be a valid UUIDv7") from exc
    spec = validate_rule_spec(rule_version.get("rule_spec"))
    digest = rule_spec_hash(spec)
    if rule_version.get("spec_hash") != digest:
        raise ValueError("published rule version spec hash does not match")

    quality_only = all(step["op"] == "assert" for step in spec["steps"])
    backend = "quality_check" if quality_only else "polars_batch"
    plan = {
        "schema_version": "1.0",
        "compiler_version": COMPILER_VERSION,
        "rule_version_id": version_id,
        "rule_spec_hash": digest,
        "input_schema_ref": spec["input_schema_ref"],
        "output_schema_ref": spec["output_schema_ref"],
        "null_policy": spec["null_policy"],
        "timezone": spec["timezone"],
        "steps": spec["steps"],
    }
    return {
        "backend": backend,
        "compiler_version": COMPILER_VERSION,
        "plan": plan,
        "plan_hash": _canonical_hash(plan),
    }