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