init: docker-migrate universal migration tool

core/runner.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+"""
+runner.py — запуск shell-команд с логированием и перехватом ошибок
+"""
+
+import subprocess
+import shlex
+import os
+from core.color import log_cmd, error, info
+
+
+def run(cmd, cwd=None, env=None, check=True, capture=True, shell=True, timeout=300):
+    """
+    Выполняет команду. При ошибке (check=True) бросает RuntimeError с stdout+stderr.
+    """
+    if not shell and isinstance(cmd, str):
+        cmd = shlex.split(cmd)
+
+    if isinstance(cmd, list):
+        display = " ".join(shlex.quote(str(x)) for x in cmd)
+    else:
+        display = cmd
+
+    log_cmd(display)
+
+    merged_env = os.environ.copy()
+    if env:
+        merged_env.update(env)
+
+    try:
+        result = subprocess.run(
+            cmd,
+            cwd=cwd,
+            env=merged_env,
+            shell=shell,
+            capture_output=capture,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired:
+        raise RuntimeError(f"Команда зависла (timeout {timeout}s): {display}")
+
+    if check and result.returncode != 0:
+        raise RuntimeError(
+            f"Команда завершилась с кодом {result.returncode}: {display}\n"
+            f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
+        )
+
+    return result
+
+
+def run_quiet(cmd, **kwargs):
+    """Выполняет без вывода команды, только результат"""
+    return run(cmd, capture=True, **kwargs)
+
+
+def run_json(cmd, **kwargs):
+    """Выполняет и парсит JSON вывод"""
+    import json
+    r = run(cmd, **kwargs)
+    if not r.stdout.strip():
+        return {}
+    try:
+        return json.loads(r.stdout)
+    except json.JSONDecodeError:
+        raise RuntimeError(f"Не удалось распарсить JSON из: {cmd}\nВывод:\n{r.stdout}")
+
+
+def exists(cmd_list):
+    """Проверяет наличие команды в PATH"""
+    import shutil
+    return bool(shutil.which(cmd_list[0]) if isinstance(cmd_list, list) else shutil.which(cmd_list))