| | from __future__ import annotations
|
| |
|
| | import os
|
| | import folder_paths
|
| | import glob
|
| | from aiohttp import web
|
| | import json
|
| | import logging
|
| | from functools import lru_cache
|
| |
|
| | from utils.json_util import merge_json_recursive
|
| |
|
| |
|
| |
|
| | EXTRA_LOCALE_FILES = [
|
| | "nodeDefs.json",
|
| | "commands.json",
|
| | "settings.json",
|
| | ]
|
| |
|
| |
|
| | def safe_load_json_file(file_path: str) -> dict:
|
| | if not os.path.exists(file_path):
|
| | return {}
|
| |
|
| | try:
|
| | with open(file_path, "r", encoding="utf-8") as f:
|
| | return json.load(f)
|
| | except json.JSONDecodeError:
|
| | logging.error(f"Error loading {file_path}")
|
| | return {}
|
| |
|
| |
|
| | class CustomNodeManager:
|
| | @lru_cache(maxsize=1)
|
| | def build_translations(self):
|
| | """Load all custom nodes translations during initialization. Translations are
|
| | expected to be loaded from `locales/` folder.
|
| |
|
| | The folder structure is expected to be the following:
|
| | - custom_nodes/
|
| | - custom_node_1/
|
| | - locales/
|
| | - en/
|
| | - main.json
|
| | - commands.json
|
| | - settings.json
|
| |
|
| | returned translations are expected to be in the following format:
|
| | {
|
| | "en": {
|
| | "nodeDefs": {...},
|
| | "commands": {...},
|
| | "settings": {...},
|
| | ...{other main.json keys}
|
| | }
|
| | }
|
| | """
|
| |
|
| | translations = {}
|
| |
|
| | for folder in folder_paths.get_folder_paths("custom_nodes"):
|
| |
|
| | for custom_node_dir in sorted(glob.glob(os.path.join(folder, "*/"))):
|
| | locales_dir = os.path.join(custom_node_dir, "locales")
|
| | if not os.path.exists(locales_dir):
|
| | continue
|
| |
|
| | for lang_dir in glob.glob(os.path.join(locales_dir, "*/")):
|
| | lang_code = os.path.basename(os.path.dirname(lang_dir))
|
| |
|
| | if lang_code not in translations:
|
| | translations[lang_code] = {}
|
| |
|
| |
|
| | main_file = os.path.join(lang_dir, "main.json")
|
| | node_translations = safe_load_json_file(main_file)
|
| |
|
| |
|
| | for extra_file in EXTRA_LOCALE_FILES:
|
| | extra_file_path = os.path.join(lang_dir, extra_file)
|
| | key = extra_file.split(".")[0]
|
| | json_data = safe_load_json_file(extra_file_path)
|
| | if json_data:
|
| | node_translations[key] = json_data
|
| |
|
| | if node_translations:
|
| | translations[lang_code] = merge_json_recursive(
|
| | translations[lang_code], node_translations
|
| | )
|
| |
|
| | return translations
|
| |
|
| | def add_routes(self, routes, webapp, loadedModules):
|
| |
|
| | @routes.get("/workflow_templates")
|
| | async def get_workflow_templates(request):
|
| | """Returns a web response that contains the map of custom_nodes names and their associated workflow templates. The ones without templates are omitted."""
|
| | files = [
|
| | file
|
| | for folder in folder_paths.get_folder_paths("custom_nodes")
|
| | for file in glob.glob(
|
| | os.path.join(folder, "*/example_workflows/*.json")
|
| | )
|
| | ]
|
| | workflow_templates_dict = (
|
| | {}
|
| | )
|
| | for file in files:
|
| | custom_nodes_name = os.path.basename(
|
| | os.path.dirname(os.path.dirname(file))
|
| | )
|
| | workflow_name = os.path.splitext(os.path.basename(file))[0]
|
| | workflow_templates_dict.setdefault(custom_nodes_name, []).append(
|
| | workflow_name
|
| | )
|
| | return web.json_response(workflow_templates_dict)
|
| |
|
| |
|
| | for module_name, module_dir in loadedModules:
|
| | workflows_dir = os.path.join(module_dir, "example_workflows")
|
| | if os.path.exists(workflows_dir):
|
| | webapp.add_routes(
|
| | [
|
| | web.static(
|
| | "/api/workflow_templates/" + module_name, workflows_dir
|
| | )
|
| | ]
|
| | )
|
| |
|
| | @routes.get("/i18n")
|
| | async def get_i18n(request):
|
| | """Returns translations from all custom nodes' locales folders."""
|
| | return web.json_response(self.build_translations())
|
| |
|
