#!/usr/bin/env python3
"""
Скрипт для генерации модуля узлов из JSON файлов в папке N8N_Components.
"""
import json
import os
from pathlib import Path
from typing import Dict, List, Any
from collections import defaultdict
def parse_node_type(node_type: str) -> str:
"""Парсит тип узла и возвращает категорию."""
if node_type.startswith('@n8n/n8n-nodes-langchain'):
return 'ai'
elif 'trigger' in node_type.lower():
return 'trigger'
elif 'manual' in node_type.lower():
return 'trigger'
elif any(keyword in node_type.lower() for keyword in ['http', 'webhook', 'api']):
return 'trigger'
elif any(keyword in node_type.lower() for keyword in ['email', 'gmail', 'smtp']):
return 'action'
elif any(keyword in node_type.lower() for keyword in ['database', 'sql', 'mongo']):
return 'action'
elif any(keyword in node_type.lower() for keyword in ['file', 'ftp', 's3']):
return 'action'
elif any(keyword in node_type.lower() for keyword in ['set', 'merge', 'filter']):
return 'transform'
else:
return 'action'
def extract_node_info(node: Dict[str, Any]) -> Dict[str, Any]:
"""Извлекает информацию об узле из JSON определения."""
# Проверяем, что node является словарем
if not isinstance(node, dict):
print(f"⚠️ Пропускаем некорректный узел: {type(node)}")
return None
node_type = node.get('type', '')
name = node.get('name', 'Unknown')
# Определяем категорию
category = parse_node_type(node_type)
# Извлекаем параметры (упрощаем для демонстрации)
parameters = node.get('parameters', {})
# Упрощаем параметры для лучшего восприятия
simplified_params = {}
for key, value in parameters.items():
if isinstance(value, dict):
if 'values' in value and isinstance(value['values'], list):
simplified_params[key] = f"Array with {len(value['values'])} items"
elif '__rl' in value:
simplified_params[key] = "Dynamic list"
else:
simplified_params[key] = "Object"
elif isinstance(value, list):
simplified_params[key] = f"Array with {len(value)} items"
elif isinstance(value, str):
simplified_params[key] = value[:50] + "..." if len(value) > 50 else value
else:
simplified_params[key] = str(value)
return {
"name": name,
"type": node_type,
"typeVersion": node.get('typeVersion', 1),
"description": f"{name} - {node_type}",
"parameters": simplified_params,
"category": category
}
def generate_nodes_module():
"""Генерирует модуль узлов из JSON файлов."""
components_dir = Path(r"C:\Users\lebed\Desktop\N8N_Tool\N8N_Components")
all_nodes = []
# Проходим по всем JSON файлам
for json_file in components_dir.rglob("*.json"):
if json_file.name == "note.json":
continue
try:
with open(json_file, 'r', encoding='utf-8') as f:
data = json.load(f)
nodes = data.get('nodes', [])
print(f"📄 Обработка {json_file.name}: {len(nodes)} узлов")
for node in nodes:
node_info = extract_node_info(node)
if node_info: # Проверяем, что node_info не None
all_nodes.append(node_info)
except Exception as e:
print(f"❌ Ошибка обработки {json_file}: {e}")
# Группируем по категориям
nodes_by_category = defaultdict(list)
for node in all_nodes:
category = node['category']
nodes_by_category[category].append(node)
# Определяем категории
categories = {
"ai": {
"name": "AI & Language",
"description": "Искусственный интеллект и обработка языка"
},
"trigger": {
"name": "Triggers",
"description": "Триггеры для запуска workflow"
},
"action": {
"name": "Actions",
"description": "Действия для выполнения задач"
},
"transform": {
"name": "Transform",
"description": "Преобразование данных"
},
"aggregate": {
"name": "Aggregate",
"description": "Агрегация и группировка данных"
}
}
# Генерируем код модуля
module_code = '''"""
Модуль для работы с узлами (nodes) n8n.
Предоставляет информацию о категориях узлов и доступных узлах в каждой категории.
"""
from typing import Dict, List, Any
# Статические данные о категориях и узлах n8n
NODE_CATEGORIES = {
'''
for category_id, category_info in categories.items():
module_code += f''' "{category_id}": {{
"name": "{category_info['name']}",
"description": "{category_info['description']}"
}},
'''
module_code = module_code.rstrip(',\n') + '\n}\n\n'
# Добавляем узлы по категориям
for category_id in categories.keys():
nodes = nodes_by_category.get(category_id, [])
module_code += f'{category_id.upper()}_NODES = [\n'
for node in nodes:
module_code += f''' {{
"name": "{node['name']}",
"type": "{node['type']}",
"typeVersion": {node['typeVersion']},
"description": "{node['description']}",
"parameters": {repr(node['parameters'])},
"category": "{node['category']}"
}},
'''
module_code = module_code.rstrip(',\n') + '\n]\n\n'
# Объединяем все узлы по категориям
module_code += '# Объединяем все узлы по категориям\nNODES_BY_CATEGORY = {\n'
for category_id in categories.keys():
module_code += f' "{category_id}": {category_id.upper()}_NODES,\n'
module_code = module_code.rstrip(',\n') + '\n}\n\n'
# Добавляем функции
module_code += '''
def get_node_categories() -> List[Dict[str, Any]]:
"""Возвращает список категорий узлов."""
return [
{
"id": category_id,
"name": category_info["name"],
"description": category_info["description"],
"node_count": len(NODES_BY_CATEGORY.get(category_id, []))
}
for category_id, category_info in NODE_CATEGORIES.items()
]
def get_nodes_by_category(category_id: str) -> List[Dict[str, Any]]:
"""Возвращает список узлов для указанной категории."""
if category_id not in NODES_BY_CATEGORY:
return []
nodes = NODES_BY_CATEGORY[category_id]
# Добавляем информацию о категории к каждому узлу
for node in nodes:
node_copy = node.copy()
node_copy["category_info"] = NODE_CATEGORIES.get(category_id, {})
yield node_copy
def get_all_nodes() -> List[Dict[str, Any]]:
"""Возвращает все доступные узлы."""
all_nodes = []
for category_id in NODES_BY_CATEGORY:
all_nodes.extend(get_nodes_by_category(category_id))
return all_nodes
def get_node_by_type(node_type: str) -> Dict[str, Any]:
"""Возвращает информацию об узле по его типу."""
all_nodes = get_all_nodes()
for node in all_nodes:
if node["type"] == node_type:
return node
return {}
'''
# Сохраняем сгенерированный модуль
output_path = Path(r"C:\Users\lebed\Desktop\N8N_Tool\MCP\n8n_mcp_server\nodes_module.py")
with open(output_path, 'w', encoding='utf-8') as f:
f.write(module_code)
print("\n✅ Модуль узлов сгенерирован!")
print(f"📊 Всего узлов: {len(all_nodes)}")
print(f"📂 Категорий: {len(categories)}")
for category_id, nodes in nodes_by_category.items():
print(f" 📁 {categories[category_id]['name']}: {len(nodes)} узлов")
if __name__ == "__main__":
generate_nodes_module()
