77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Merge instance-specific n8n credential IDs into versioned workflow definitions.
|
|
|
|
n8n's CLI does not reliably resolve two credentials of the same type by name during an
|
|
update. This utility copies only credential reference IDs/names from a pre-update export;
|
|
it never reads or writes credential values.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SERVICE_NODE_CANDIDATES = {
|
|
"mobilityops-return-processing": ("Record follow-up",),
|
|
"mobilityops-scheduled-quality-scan": ("Run quality scan",),
|
|
"6wbkc4d1AouGpmWT": ("Report sync result to Fleet Ops",),
|
|
"Xppn2rAEqUuyiCJF": ("Report failure to Fleet Ops", "failure to Fleet Ops"),
|
|
}
|
|
|
|
|
|
def _workflow_list(path: Path) -> list[dict[str, Any]]:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
return payload if isinstance(payload, list) else [payload]
|
|
|
|
|
|
def merge(backup_path: Path, source_dir: Path, output_dir: Path) -> list[Path]:
|
|
backups = {workflow["id"]: workflow for workflow in _workflow_list(backup_path)}
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
written: list[Path] = []
|
|
for source_path in sorted(source_dir.glob("fleet-ops-*.json")):
|
|
updated = _workflow_list(source_path)[0]
|
|
workflow_id = updated["id"]
|
|
if workflow_id not in SERVICE_NODE_CANDIDATES or workflow_id not in backups:
|
|
continue
|
|
original_nodes = {node["name"]: node for node in backups[workflow_id]["nodes"]}
|
|
service_node = next(
|
|
(
|
|
original_nodes[name]
|
|
for name in SERVICE_NODE_CANDIDATES[workflow_id]
|
|
if name in original_nodes
|
|
),
|
|
None,
|
|
)
|
|
if service_node is None or not service_node.get("credentials"):
|
|
raise ValueError(f"No service credential reference found for {workflow_id}")
|
|
service_credentials = service_node["credentials"]
|
|
for node in updated["nodes"]:
|
|
original = original_nodes.get(node["name"])
|
|
if original and original.get("credentials"):
|
|
node["credentials"] = copy.deepcopy(original["credentials"])
|
|
elif node.get("credentials") or node["name"] == "Report workflow heartbeat":
|
|
node["credentials"] = copy.deepcopy(service_credentials)
|
|
target = output_dir / source_path.name
|
|
target.write_text(
|
|
json.dumps(updated, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
|
)
|
|
written.append(target)
|
|
return written
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("backup", type=Path, help="n8n export:workflow --all JSON file")
|
|
parser.add_argument("source_dir", type=Path, help="versioned workflow directory")
|
|
parser.add_argument("output_dir", type=Path, help="safe import output directory")
|
|
args = parser.parse_args()
|
|
for path in merge(args.backup, args.source_dir, args.output_dir):
|
|
print(path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|