DataOps-platform/app/commands/migrate_n8n_workflow.py

"""Convert one exported n8n workflow into governed migration artifacts."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from app.core.orchestration.migration.dual_run import (
    MigrationBlocked,
    N8nWorkflowConverter,
)


def convert_export(
    workflow,
    *,
    dataflow_uid,
    data_source_uids,
):
    try:
        converted = N8nWorkflowConverter().convert(
            workflow,
            dataflow_uid=dataflow_uid,
            data_source_uids=data_source_uids,
        )
    except MigrationBlocked as exc:
        return exc.report
    return {
        "status": "converted",
        "source_workflow_id": converted.source_workflow_id,
        "definition_hash": converted.definition_hash,
        "workflow_spec": converted.workflow_spec,
        "schedule_plan": converted.schedule_plan,
        "deployment_performed": False,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Convert one n8n export without deploying it or copying credentials."
        )
    )
    parser.add_argument("--workflow-json", required=True)
    parser.add_argument("--dataflow-uid", required=True)
    parser.add_argument("--datasource-map-json", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()

    workflow = json.loads(
        Path(args.workflow_json).read_text(encoding="utf-8")
    )
    mapping = json.loads(
        Path(args.datasource_map_json).read_text(encoding="utf-8")
    )
    if not isinstance(mapping, dict):
        raise ValueError("datasource map must be an object")
    report = convert_export(
        workflow,
        dataflow_uid=args.dataflow_uid,
        data_source_uids=mapping,
    )
    Path(args.output).write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return 0 if report["status"] == "converted" else 2


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