72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
|
import subprocess
|
|||
|
from typing import Union, Tuple
|
|||
|
from .system_utils import SystemUtils
|
|||
|
|
|||
|
|
|||
|
class CommandUtils:
|
|||
|
"""命令执行工具类"""
|
|||
|
|
|||
|
@staticmethod
|
|||
|
def execute(command: Union[str, list], timeout: int = None) -> Tuple[int, str, str]:
|
|||
|
"""
|
|||
|
执行系统命令并返回结果(兼容Windows/macOS/Linux)
|
|||
|
|
|||
|
参数:
|
|||
|
command: 要执行的命令,可以是字符串或列表
|
|||
|
timeout: 超时时间(秒)
|
|||
|
|
|||
|
返回:
|
|||
|
元组: (return_code, stdout, stderr)
|
|||
|
"""
|
|||
|
# 预处理命令
|
|||
|
processed_cmd = CommandUtils._prepare_command(command)
|
|||
|
|
|||
|
try:
|
|||
|
# 执行命令
|
|||
|
result = subprocess.run(
|
|||
|
processed_cmd,
|
|||
|
shell=isinstance(command, str),
|
|||
|
stdout=subprocess.PIPE,
|
|||
|
stderr=subprocess.PIPE,
|
|||
|
universal_newlines=True,
|
|||
|
timeout=timeout
|
|||
|
)
|
|||
|
return result.returncode, result.stdout, result.stderr
|
|||
|
except subprocess.TimeoutExpired:
|
|||
|
return -1, "", "Command timed out"
|
|||
|
except Exception as e:
|
|||
|
return -1, "", str(e)
|
|||
|
|
|||
|
@staticmethod
|
|||
|
def _prepare_command(command: Union[str, list]) -> Union[str, list]:
|
|||
|
"""预处理命令"""
|
|||
|
if SystemUtils.is_windows() and isinstance(command, str):
|
|||
|
return ['cmd', '/c'] + command.split()
|
|||
|
return command
|
|||
|
|
|||
|
@staticmethod
|
|||
|
def execute_with_real_time_output(command: Union[str, list], timeout: int = None) -> int:
|
|||
|
"""
|
|||
|
执行命令并实时输出(不捕获输出,直接打印到控制台)
|
|||
|
|
|||
|
参数:
|
|||
|
command: 要执行的命令
|
|||
|
timeout: 超时时间(秒)
|
|||
|
|
|||
|
返回:
|
|||
|
返回状态码
|
|||
|
"""
|
|||
|
processed_cmd = CommandUtils._prepare_command(command)
|
|||
|
|
|||
|
try:
|
|||
|
result = subprocess.run(
|
|||
|
processed_cmd,
|
|||
|
shell=isinstance(command, str),
|
|||
|
timeout=timeout
|
|||
|
)
|
|||
|
return result.returncode
|
|||
|
except subprocess.TimeoutExpired:
|
|||
|
return -1
|
|||
|
except Exception as e:
|
|||
|
return -1
|