DataOps-platform/scripts/inventory_n8n_workflows.py

from __future__ import annotations

import argparse
import json
import os
import re
import sys
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Mapping

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from app.core.data_factory.n8n_client import N8nClient, N8nClientError
from app.core.data_flow.workflow_repository import workflow_hash


SCHEMA_VERSION = "1.0"
KNOWN_NODE_CATEGORIES = {
    "trigger",
    "sql",
    "python_or_script",
    "http",
    "condition",
    "notify",
    "ai",
    "subflow",
}


def _parse_timestamp(value: Any) -> datetime | None:
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def _safe_error(value: Any) -> str:
    message = re.sub(
        r"(?i)((?:api[-_ ]?key|authorization|password|token)\s*[=:]\s*)\S+",
        r"\1[redacted]",
        str(value),
    )
    message = re.sub(
        r"(?i)(authorization\s+bearer\s+)\S+",
        r"\1[redacted]",
        message,
    )
    return message[:500]


def _node_category(node_type: str) -> str:
    normalized = node_type.lower()
    if "trigger" in normalized or normalized.endswith(".webhook"):
        return "trigger"
    if any(
        name in normalized
        for name in ("postgres", "mysql", "mssql", "mariadb", "oracle", "snowflake")
    ):
        return "sql"
    if any(
        name in normalized
        for name in ("python", ".code", "executecommand", ".ssh")
    ):
        return "python_or_script"
    if "httprequest" in normalized:
        return "http"
    if normalized.endswith((".if", ".switch", ".filter", ".merge")):
        return "condition"
    if any(
        name in normalized
        for name in ("email", "slack", "teams", "telegram", "mattermost", "discord")
    ):
        return "notify"
    if "langchain" in normalized or ".ai" in normalized:
        return "ai"
    if "executeworkflow" in normalized:
        return "subflow"
    return "unknown"


def _credential_references(nodes: list[dict[str, Any]]) -> list[dict[str, str]]:
    references = []
    for node in nodes:
        credentials = node.get("credentials")
        if not isinstance(credentials, dict):
            continue
        for credential_type, credential in credentials.items():
            if not isinstance(credential, dict):
                continue
            name = str(credential.get("name") or "").strip()
            if not name:
                continue
            references.append(
                {
                    "node_name": str(node.get("name") or ""),
                    "credential_type": str(credential_type),
                    "credential_name": name,
                }
            )
    return sorted(
        references,
        key=lambda item: (
            item["node_name"],
            item["credential_type"],
            item["credential_name"],
        ),
    )


def _trigger_summary(nodes: list[dict[str, Any]]) -> list[dict[str, str]]:
    triggers = []
    for node in nodes:
        node_type = str(node.get("type") or "")
        if _node_category(node_type) != "trigger":
            continue
        trigger = {
            "node_name": str(node.get("name") or ""),
            "node_type": node_type,
        }
        parameters = node.get("parameters")
        if isinstance(parameters, dict) and node_type.lower().endswith(".webhook"):
            path = parameters.get("path")
            method = parameters.get("httpMethod")
            if path:
                trigger["webhook_path"] = str(path)
            if method:
                trigger["http_method"] = str(method)
        triggers.append(trigger)
    return triggers


def _execution_summary(
    executions: list[dict[str, Any]], *, generated_at: datetime
) -> dict[str, Any]:
    cutoff = generated_at - timedelta(days=90)
    recent = [
        execution
        for execution in executions
        if (started_at := _parse_timestamp(execution.get("startedAt")))
        and started_at >= cutoff
    ]
    recent.sort(
        key=lambda item: _parse_timestamp(item.get("startedAt"))
        or datetime.min.replace(tzinfo=timezone.utc),
        reverse=True,
    )
    statuses = Counter(str(item.get("status") or "unknown") for item in recent)
    latest = recent[0] if recent else {}
    return {
        "count_90d": len(recent),
        "status_counts": dict(sorted(statuses.items())),
        "last_status": latest.get("status"),
        "last_started_at": latest.get("startedAt"),
        "last_stopped_at": latest.get("stoppedAt"),
    }


def _collect_pages(client, method_name: str, **kwargs) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    cursor = None
    seen_cursors = set()
    for _ in range(1000):
        result = getattr(client, method_name)(
            **kwargs,
            limit=100,
            cursor=cursor,
        )
        page = result.get("data", [])
        if isinstance(page, list):
            items.extend(item for item in page if isinstance(item, dict))
        next_cursor = result.get("nextCursor")
        if not next_cursor:
            return items
        if next_cursor in seen_cursors:
            raise ValueError(f"n8n {method_name} returned a repeated cursor")
        seen_cursors.add(next_cursor)
        cursor = next_cursor
    raise ValueError(f"n8n {method_name} exceeded the page safety limit")


def empty_inventory(*, reason: str, generated_at: str | None = None) -> dict[str, Any]:
    timestamp = generated_at or datetime.now(timezone.utc).isoformat()
    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": timestamp,
        "source": "n8n",
        "collection_status": "pending_live_n8n_access",
        "collection_error": _safe_error(reason),
        "coverage": {
            "workflow_count": 0,
            "active_count": 0,
            "recently_executed_count": 0,
            "blocked_count": 0,
            "unbound_count": 0,
        },
        "workflows": [],
    }


def collect_inventory(
    client,
    *,
    bindings: Mapping[str, str] | None = None,
    generated_at: str | None = None,
) -> dict[str, Any]:
    timestamp = generated_at or datetime.now(timezone.utc).isoformat()
    generated = _parse_timestamp(timestamp)
    if generated is None:
        raise ValueError("generated_at must be an ISO 8601 timestamp")
    workflow_bindings = {str(key): str(value) for key, value in (bindings or {}).items()}
    workflows = _collect_pages(client, "list_workflows")
    inventory_workflows = []

    for workflow in workflows:
        workflow_id = str(workflow.get("id") or "").strip()
        if not workflow_id:
            continue
        executions = _collect_pages(
            client,
            "list_executions",
            workflow_id=workflow_id,
            include_data=False,
        )
        nodes = [
            node for node in workflow.get("nodes", []) if isinstance(node, dict)
        ]
        node_types = sorted(
            {str(node.get("type") or "").strip() for node in nodes if node.get("type")}
        )
        categories = sorted({_node_category(node_type) for node_type in node_types})
        dataflow_uid = workflow_bindings.get(workflow_id)
        blockers = [
            f"unsupported_node:{node_type}"
            for node_type in node_types
            if _node_category(node_type) == "unknown"
        ]
        if not dataflow_uid:
            blockers.append("missing_dataflow_binding")
        recent_execution = _execution_summary(executions, generated_at=generated)
        active = bool(workflow.get("active"))
        inventory_workflows.append(
            {
                "workflow_id": workflow_id,
                "name": str(workflow.get("name") or ""),
                "active": active,
                "updated_at": workflow.get("updatedAt"),
                "tags": sorted(
                    str(tag.get("name"))
                    for tag in workflow.get("tags", [])
                    if isinstance(tag, dict) and tag.get("name")
                ),
                "dataflow_uid": dataflow_uid,
                "definition_hash": workflow_hash(workflow),
                "node_types": node_types,
                "node_categories": categories,
                "triggers": _trigger_summary(nodes),
                "credential_references": _credential_references(nodes),
                "recent_execution": recent_execution,
                "migration_assessment": {
                    "status": "blocked" if blockers else "review_required",
                    "priority": "blocked"
                    if blockers
                    else ("high" if active else "normal"),
                    "production_write": "unknown",
                    "idempotent": "unknown",
                    "shadow_mode": "review_required",
                    "blockers": sorted(blockers),
                },
            }
        )

    inventory_workflows.sort(key=lambda item: item["workflow_id"])
    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": timestamp,
        "source": "n8n",
        "collection_status": "complete",
        "collection_error": None,
        "coverage": {
            "workflow_count": len(inventory_workflows),
            "active_count": sum(item["active"] for item in inventory_workflows),
            "recently_executed_count": sum(
                item["recent_execution"]["count_90d"] > 0
                for item in inventory_workflows
            ),
            "blocked_count": sum(
                item["migration_assessment"]["status"] == "blocked"
                for item in inventory_workflows
            ),
            "unbound_count": sum(
                item["dataflow_uid"] is None for item in inventory_workflows
            ),
        },
        "workflows": inventory_workflows,
    }


def _load_bindings(path: str | None) -> dict[str, str]:
    if not path:
        return {}
    payload = json.loads(Path(path).read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError("bindings file must contain a workflow-id to DataFlow-UID object")
    return {str(key): str(value) for key, value in payload.items()}


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Export a credential-safe n8n migration inventory."
    )
    parser.add_argument(
        "--output",
        default="docs/generated/n8n_migration_inventory.json",
    )
    parser.add_argument("--bindings-json")
    args = parser.parse_args()

    output = Path(args.output)
    output.parent.mkdir(parents=True, exist_ok=True)
    api_url = os.environ.get("N8N_API_URL", "")
    api_key = os.environ.get("N8N_API_KEY", "")
    if not api_url or not api_key:
        result = empty_inventory(reason="N8N_API_URL or N8N_API_KEY is not configured")
        output.write_text(
            json.dumps(result, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        return 2
    try:
        api_timeout = int(os.environ.get("N8N_API_TIMEOUT", "30"))
        if api_timeout <= 0:
            raise ValueError
    except ValueError:
        result = empty_inventory(reason="N8N_API_TIMEOUT must be a positive integer")
        output.write_text(
            json.dumps(result, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        return 2

    try:
        result = collect_inventory(
            N8nClient(api_url=api_url, api_key=api_key, timeout=api_timeout),
            bindings=_load_bindings(args.bindings_json),
        )
    except (N8nClientError, ValueError) as exc:
        result = empty_inventory(reason=str(exc))
        output.write_text(
            json.dumps(result, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        return 2

    output.write_text(
        json.dumps(result, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    return 0


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