utils.py

import importlib.util
import json
from pathlib import Path
import subprocess
import sys


def import_and_call(folder: Path, action: str, **params) -> str | None:
    """Dynamically import Python modules and call the action function."""
    folder_str = str(folder)
    if folder_str not in sys.path:
        sys.path.insert(0, folder_str)

    # Find all Python files in the folder
    py_files = list(folder.glob("*.py"))

    for py_file in py_files:
        if py_file.name.startswith("__"):
            continue

        # Reuse an already-loaded module if present in sys.modules. This is
        # necessary for unittest.mock.patch to work — patch modifies the
        # attribute on the module object in sys.modules, so we must use that
        # same object rather than creating a fresh one each call.
        module_name = py_file.stem
        if module_name in sys.modules:
            module = sys.modules[module_name]
        else:
            try:
                spec = importlib.util.spec_from_file_location(module_name, py_file)
                if spec is None or spec.loader is None:
                    continue
                module = importlib.util.module_from_spec(spec)
                sys.modules[module_name] = module
                spec.loader.exec_module(module)
            except Exception:
                continue

        # If the function is here, call it. Do NOT swallow its exceptions —
        # surface them so the caller can see what went wrong instead of the
        # misleading "No executable found" fallback.
        if hasattr(module, action):
            func = getattr(module, action)
            try:
                return str(func(**params))
            except Exception as e:
                return f"Error calling {action}: {type(e).__name__}: {e}"

    return None


def run_script(folder: Path, action: str, **params) -> str | None:
    """Try to execute scripts as subprocess (fallback method)."""
    for script in [f"{action}.py", "run.py", f"{action}.sh"]:
        path = folder / script
        if not path.exists():
            continue

        try:
            # Use python to run .py scripts instead of direct execution
            if script.endswith(".py"):
                args = [sys.executable, str(path), action]
            else:
                args = [str(path), action]

            for k, v in params.items():
                args.extend([f"--{k}", str(v)])

            result = subprocess.run(
                args, capture_output=True, text=True, timeout=30, cwd=str(folder)
            )
            try:
                return json.loads(result.stdout)
            except json.JSONDecodeError:
                return None
        except Exception as e:
            return None
    return None