新增功能:执行命令时根据命令输出返回错误提示信息

This commit is contained in:
li-xiaochen 2024-12-28 20:14:01 +08:00
parent 60f43d42be
commit 478bcd5f43
2 changed files with 48 additions and 8 deletions

View file

@ -14,3 +14,17 @@ def ensure_directory_exists(directory):
"""确保目录存在,如果不存在则创建"""
if not os.path.exists(directory):
os.makedirs(directory)
def check_command_path(command, cwd=None):
"""检查命令路径是否存在exe文件
:return 是否存在,命令exe文件全路径
"""
command_path = command.split(" ")[0]
if command_path == "start":
return True, None
exec_command_path = os.getcwd()
if cwd:
exec_command_path = os.path.join(exec_command_path, cwd)
full_command_path = os.path.abspath(os.path.join(exec_command_path, command_path)) + ".exe"
return os.path.exists(full_command_path), full_command_path

View file

@ -14,7 +14,7 @@ from launcher.constants import download_git_url, download_python_url, get_new_ve
mirror_list
from launcher.file.download import init_download, download_file
from launcher.file.extract import extract_7z_file
from launcher.file.utils import ensure_directory_exists
from launcher.file.utils import ensure_directory_exists, check_command_path
from launcher.log import logger
from launcher.sys_config import sys_config
from launcher.webview.events import custom_event, LogType
@ -35,11 +35,27 @@ command_list = {
}
def read_stream(stream, log_type):
def parse_stderr(stderr_output):
error_keywords = {
"fatal: destination path 'mower-ng' already exists and is not an empty directory.": "mower-ng文件夹已存在并且非空。",
"index.lock': File exists": "上一个git命令正在执行,请等待执行结束或在任务管理器中杀掉git进程,并确保上方提示的index.lock文件删除后再次运行。",
"Could not resolve host": "网络出现错误,请检查网络是否通畅。",
"ReadTimeoutError": "网络连接超时,请检查网络连接或尝试更换镜像源。",
"No space left on device": "磁盘空间不足。",
}
for keyword, message in error_keywords.items():
if keyword in stderr_output:
return message
return "未定义的错误"
def read_stream(stream, log_type, output_list=None):
def process_lines(text_io):
for line in iter(text_io.readline, ''):
text = line.rstrip('\n')
custom_event(log_type, f"{text.strip()}")
text = line.rstrip('\n').strip()
custom_event(log_type, text)
if output_list is not None:
output_list.append(text)
detected_encoding = 'utf-8'
text_io = io.TextIOWrapper(stream, encoding=detected_encoding, errors='replace')
@ -129,26 +145,36 @@ class Api:
if callable(command):
return "success" if command() else "failed"
if cwd is not None:
custom_event(LogType.execute_command, f"路径{cwd}")
custom_event(LogType.info, f"命令执行目录{cwd}")
custom_event(LogType.execute_command, command)
# 执行命令前先判断命令路径是否存在
exist, command_path = check_command_path(command, cwd)
if not exist:
custom_event(LogType.error, f"命令路径不存在:{command_path} 请尝试依赖修复并重新初始化。")
return "failed"
try:
stdout_stderr = []
with subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, cwd=cwd, bufsize=0,
universal_newlines=False
) as p:
stdout_thread = threading.Thread(target=read_stream, args=(p.stdout, LogType.command_out))
stderr_thread = threading.Thread(target=read_stream, args=(p.stderr, LogType.command_out))
stderr_thread = threading.Thread(target=read_stream,
args=(p.stderr, LogType.command_out, stdout_stderr))
stdout_thread.start()
stderr_thread.start()
stdout_thread.join()
stderr_thread.join()
if p.returncode == 0:
return "success"
else:
logger.error(f"{command} 运行失败")
error_message = parse_stderr("\n".join(stdout_stderr))
custom_event(LogType.error, f"命令执行失败,{error_message}")
return "failed"
except Exception as e:
logger.exception(e)
custom_event(LogType.error, repr(e))
custom_event(LogType.error, str(e))
return "failed"