DataOps-platform/app/commands/reconcile_dual_run.py

"""Compare credential-safe n8n and Kestra execution snapshots."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from app.core.orchestration.migration.reconciliation import (
    ExecutionSnapshot,
    ReconciliationPolicy,
    WorkflowReconciler,
)


def reconcile_exports(baseline, candidate, policy=None):
    policy = policy or {}
    if not isinstance(policy, dict):
        raise ValueError("reconciliation policy must be an object")
    return WorkflowReconciler().compare(
        ExecutionSnapshot.from_dict(baseline),
        ExecutionSnapshot.from_dict(candidate),
        ReconciliationPolicy(
            ignored_paths=policy.get("ignored_paths"),
            numeric_tolerances=policy.get("numeric_tolerances"),
        ),
    )


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Reconcile bounded execution metrics without raw rows."
    )
    parser.add_argument("--n8n-snapshot", required=True)
    parser.add_argument("--kestra-snapshot", required=True)
    parser.add_argument("--policy-json")
    parser.add_argument("--output", required=True)
    args = parser.parse_args()

    baseline = json.loads(
        Path(args.n8n_snapshot).read_text(encoding="utf-8")
    )
    candidate = json.loads(
        Path(args.kestra_snapshot).read_text(encoding="utf-8")
    )
    policy = (
        json.loads(Path(args.policy_json).read_text(encoding="utf-8"))
        if args.policy_json
        else {}
    )
    report = reconcile_exports(baseline, candidate, policy)
    Path(args.output).write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return 0 if report["status"] == "passed" else 2


if __name__ == "__main__":
    raise SystemExit(main())