75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
# -*- 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,
|
||
errors='replace',
|
||
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))
|