Refactor: consolidate utilities and fix code duplication

Co-authored-by: johndoe6345789 <224850594+johndoe6345789@users.noreply.github.com>

backend/autometabuilder/utils.py

@@ -3,6 +3,7 @@
 This module provides helper functions that are used across the codebase.
 These are pure utility functions that don't contain business logic.
 """
+import importlib
 import json
 import os
 import yaml
@@ -10,6 +11,14 @@ from pathlib import Path
 from typing import Any
 
 
+def get_package_root() -> Path:
+    """Get the AutoMetabuilder package root directory.
+    
+    Returns the absolute path to the autometabuilder package root.
+    """
+    return Path(__file__).resolve().parent
+
+
 def read_json(path: Path) -> dict[str, Any]:
     """Read JSON file."""
     if not path.exists():
@@ -18,6 +27,20 @@ def read_json(path: Path) -> dict[str, Any]:
         return json.load(f)
 
 
+def load_callable(path: str):
+    """Import and return a callable by dotted path.
+    
+    Args:
+        path: Dotted path to callable (e.g., 'module.submodule.function')
+        
+    Returns:
+        The callable object
+    """
+    module_path, attr = path.rsplit(".", 1)
+    module = importlib.import_module(module_path)
+    return getattr(module, attr)
+
+
 def load_metadata() -> dict[str, Any]:
     """Load metadata.json with optional section includes.
     
@@ -29,8 +52,8 @@ def load_metadata() -> dict[str, Any]:
         "workflow_plugins_path": "workflow_plugins",
     }
     
-    # Locate metadata.json relative to the autometabuilder package root
-    metadata_path = Path(__file__).resolve().parent / "metadata.json"
+    # Locate metadata.json in package root
+    metadata_path = get_package_root() / "metadata.json"
     metadata = read_json(metadata_path)
     base_dir = metadata_path.parent
     
@@ -60,3 +83,19 @@ def load_prompt_yaml() -> dict:
         with open(local_path, "r", encoding="utf-8") as f:
             return yaml.safe_load(f)
     raise FileNotFoundError(f"Prompt file not found at {local_path}")
+
+
+def load_tool_registry() -> list:
+    """Load tool registry entries from tool_registry.json.
+    
+    This is a utility function for loading tool registry configuration.
+    """
+    path = get_package_root() / "tool_registry.json"
+    if not os.path.exists(path):
+        return []
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+    except json.JSONDecodeError:
+        return []
+    return data if isinstance(data, list) else []

backend/autometabuilder/workflow/plugin_loader.py

@@ -1,9 +1,7 @@
 """Load workflow plugins by dotted path."""
-import importlib
+from ..utils import load_callable
 
 
 def load_plugin_callable(path: str):
     """Load a workflow plugin callable."""
-    module_path, attr = path.rsplit(".", 1)
-    module = importlib.import_module(module_path)
-    return getattr(module, attr)
+    return load_callable(path)

backend/autometabuilder/workflow/plugins/backend/backend_build_tool_map/backend_build_tool_map.py

@@ -1,29 +1,5 @@
 """Workflow plugin: build tool map."""
-import importlib
-import json
-import os
-from pathlib import Path
-
-
-def _load_callable(path: str):
-    """Import and return a callable."""
-    module_path, attr = path.rsplit(".", 1)
-    module = importlib.import_module(module_path)
-    return getattr(module, attr)
-
-
-def _load_tool_registry() -> list:
-    """Load tool registry entries."""
-    # Locate tool_registry.json relative to autometabuilder package root
-    path = Path(__file__).resolve().parents[4] / "tool_registry.json"
-    if not os.path.exists(path):
-        return []
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            data = json.load(f)
-    except json.JSONDecodeError:
-        return []
-    return data if isinstance(data, list) else []
+from .....utils import load_callable, load_tool_registry
 
 
 def _build_tool_map(gh, registry_entries: list) -> dict:
@@ -40,7 +16,7 @@ def _build_tool_map(gh, registry_entries: list) -> dict:
             continue
         if provider == "module":
             path = entry.get("callable")
-            tool_map[name] = _load_callable(path) if path else None
+            tool_map[name] = load_callable(path) if path else None
             continue
         tool_map[name] = None
     return tool_map
@@ -49,7 +25,7 @@ def _build_tool_map(gh, registry_entries: list) -> dict:
 def run(runtime, _inputs):
     """Build tool registry map."""
     gh = runtime.context.get("gh")
-    registry = _load_tool_registry()
+    registry = load_tool_registry()
     tool_map = _build_tool_map(gh, registry)
     # Store in both store (for workflow) and context (for other plugins)
     runtime.context["tool_map"] = tool_map

backend/autometabuilder/workflow/plugins/backend/backend_load_metadata/backend_load_metadata.py

@@ -1,47 +1,10 @@
 """Workflow plugin: load metadata."""
-import json
-from pathlib import Path
-from typing import Any
-
-
-INCLUDED_SECTIONS = {
-    "settings_descriptions_path": "settings_descriptions",
-    "suggestions_path": "suggestions",
-    "workflow_plugins_path": "workflow_plugins",
-}
-
-
-def _read_json(path: Path) -> dict[str, Any]:
-    """Read JSON file."""
-    if not path.exists():
-        return {}
-    with path.open("r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def _load_metadata() -> dict[str, Any]:
-    """Load metadata.json with optional section includes."""
-    # Locate metadata.json relative to the autometabuilder package root
-    metadata_path = Path(__file__).resolve().parents[4] / "metadata.json"
-    metadata = _read_json(metadata_path)
-    base_dir = metadata_path.parent
-    for path_key, dest_key in INCLUDED_SECTIONS.items():
-        include_path = metadata.get(path_key)
-        if include_path:
-            resolved_path = base_dir / include_path
-            if resolved_path.is_dir():
-                merged: dict[str, Any] = {}
-                for file_path in sorted(resolved_path.glob("*.json")):
-                    merged.update(_read_json(file_path))
-                metadata[dest_key] = merged
-            else:
-                metadata[dest_key] = _read_json(resolved_path)
-    return metadata
+from .....utils import load_metadata as util_load_metadata
 
 
 def run(runtime, _inputs):
     """Load metadata.json."""
-    metadata = _load_metadata()
+    metadata = util_load_metadata()
     # Store in both store (for workflow) and context (for other plugins)
     runtime.context["metadata"] = metadata
     return {"result": metadata}

backend/autometabuilder/workflow/plugins/backend/backend_load_plugins/backend_load_plugins.py

@@ -3,15 +3,15 @@ import importlib
 import inspect
 import logging
 import os
-from pathlib import Path
+from .....utils import get_package_root
 
 logger = logging.getLogger("autometabuilder")
 
 
 def _load_plugins(tool_map: dict, tools: list) -> None:
     """Load plugin tools and append metadata."""
-    # Locate plugins directory relative to autometabuilder package root
-    plugins_dir = Path(__file__).resolve().parents[4] / "plugins"
+    # Locate plugins directory in package root
+    plugins_dir = get_package_root() / "plugins"
     if not os.path.exists(plugins_dir):
         return
 

backend/autometabuilder/workflow/plugins/backend/backend_load_tool_policies/backend_load_tool_policies.py

@@ -1,13 +1,13 @@
 """Workflow plugin: load tool policies."""
 import json
 import os
-from pathlib import Path
+from .....utils import get_package_root
 
 
 def _load_tool_policies() -> dict:
     """Load tool policies JSON."""
-    # Locate tool_policies.json relative to autometabuilder package root
-    path = Path(__file__).resolve().parents[4] / "tool_policies.json"
+    # Locate tool_policies.json in package root
+    path = get_package_root() / "tool_policies.json"
     if not os.path.exists(path):
         return {"modifying_tools": []}
     try:

backend/autometabuilder/workflow/plugins/backend/backend_load_tool_registry/backend_load_tool_registry.py

@@ -1,26 +1,10 @@
 """Workflow plugin: load tool registry."""
-import json
-import os
-from pathlib import Path
-
-
-def _load_tool_registry() -> list:
-    """Load tool registry entries."""
-    # Locate tool_registry.json relative to autometabuilder package root
-    path = Path(__file__).resolve().parents[4] / "tool_registry.json"
-    if not os.path.exists(path):
-        return []
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            data = json.load(f)
-    except json.JSONDecodeError:
-        return []
-    return data if isinstance(data, list) else []
+from .....utils import load_tool_registry as util_load_tool_registry
 
 
 def run(runtime, _inputs):
     """Load tool registry entries."""
-    tool_registry = _load_tool_registry()
+    tool_registry = util_load_tool_registry()
     # Store in context for other plugins
     runtime.context["tool_registry"] = tool_registry
     return {"result": tool_registry}

backend/autometabuilder/workflow/plugins/backend/backend_load_tools/backend_load_tools.py

@@ -1,13 +1,13 @@
 """Workflow plugin: load tools."""
 import json
 import os
-from pathlib import Path
+from .....utils import get_package_root
 
 
 def _load_tools(metadata: dict) -> list:
     """Load tool specs from metadata reference."""
     # Locate tools relative to autometabuilder package root
-    base_dir = Path(__file__).resolve().parents[4]
+    base_dir = get_package_root()
     tools_path = base_dir / metadata.get("tools_path", "tools.json")
     
     if tools_path.is_dir():

backend/autometabuilder/workflow/plugins/web/web_register_blueprint/web_register_blueprint.py

@@ -1,12 +1,5 @@
 """Workflow plugin: register Flask blueprint."""
-import importlib
-
-
-def _load_callable(path: str):
-    """Import and return a callable."""
-    module_path, attr = path.rsplit(".", 1)
-    module = importlib.import_module(module_path)
-    return getattr(module, attr)
+from .....utils import load_callable
 
 
 def run(runtime, inputs):
@@ -34,7 +27,7 @@ def run(runtime, inputs):
             return {"error": "blueprint or blueprint_path is required"}
         
         try:
-            blueprint = _load_callable(blueprint_path)
+            blueprint = load_callable(blueprint_path)
         except Exception as e:
             return {"error": f"Failed to load blueprint: {str(e)}"}