1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
"""
------------------------------------------------
describe:
CMD系统命令执行工具
利用subprocess内置包开发，使用了底层的Popen方法，便于参数交互，其subprocess的run、call等方法都基于Popen
非类的方式，对外方法：
- run_command：命令执行方法，有命令执行output返回
- run_command_no_output：命令执行方法，无命令执行output返回
- check_command_by_which：which xxxx
base_info:
__author__ = "PyGo"
__time__ = "2022/9/14"
__version__ = "v.1.0.0"
__mail__ = "gaoming971366@163.com"
__project__ = "quality-inspect"
usage:
return_code, output = run_command(_cmd, shell=True, check_exit_code=False)
- return_code 命令执行状态码，0为执行正常，其他码为执行错误
- output 命令输出信息
return_code = run_command_no_output(_cmd, shell=True, check_exit_code=False)
- return_code 命令执行状态码，0为执行正常，其他码为执行错误
return_code = check_command_by_which(_cmd)
- return_code bool类型，True or False
design:
使用内置包subprocess的Popen方法进行开发，调用底层方法，便于参数交互
目前常用的3个方法：
- run_command：命令执行方法，有命令执行output返回
- run_command_no_output：命令执行方法，无命令执行output返回
- check_command_by_which：which xxxx
对外使用的方法都基于_run_cmd，可以进行二次封装，实现其他的命令执行方法
也可以直接对Popen进行二次开发
reference urls:
python version:
python3
Enjoy the good time everyday！！!
Life is short, I use python.
------------------------------------------------
"""
import sys
import os
import inspect
from subprocess import PIPE, Popen
from deploy.logger import logger as LOG
from config import DEBUG
def get_cur_folder():
if getattr(sys, "frozen", False):
return os.path.dirname(os.path.abspath(__file__))
else:
cur_folder = os.path.dirname(inspect.getfile(inspect.currentframe()))
return os.path.abspath(cur_folder)
def _print_die(message):
"""
exit process
:param message: print error message
:return:
"""
if not message:
return
print('%s%s%s' % ('>' * 10, message, '<' * 10))
sys.exit(1)
def _run_cmd(cmd, shell=True, check_exit_code=True, cwd=None):
"""
execute command at local shell, build in method,
not allow other file to use
:param cmd: command
:param shell: is or not run shell mode
if shell is True, command type is str
else command type is list or tuple
:param check_exit_code: check is or not return code
:param cwd: current work dir
:return: command return code, command content
cmd:
- "ls -a"
- ["ls", "-a"]
"""
if type(cmd) not in [type('string'), type([1]), type((1,))]:
return -1, 'Command type is error'
if not cwd:
cwd = get_cur_folder()
if not shell:
cmd = ' '.join(cmd)
if not isinstance(cmd, str):
cmd = ' '.join(cmd)
if DEBUG:
LOG.debug(cmd)
p = Popen(cmd,
shell=False,
bufsize=0,
cwd=cwd,
stdout=PIPE)
output = p.communicate()[0]
return_code = p.returncode
if check_exit_code and return_code != 0:
_print_die('Command "%s" failed.\n%s' % (' '.join(cmd), output))
if return_code in [0, '0']:
return_code = 0
return return_code, output
def run_command(cmd, check_exit_code=True, shell=True, cwd=None):
"""
run command return command code, output
:param cmd: command
:param check_exit_code: check command output exit code
:param shell: shell mode
:param cwd: current work dir
:return: command code
"""
return _run_cmd(cmd, shell=shell, check_exit_code=check_exit_code, cwd=cwd)
def run_command_no_output(cmd, check_exit_code=True, shell=True, cwd=None):
"""
run command no return command output
:param cmd: command
:param check_exit_code: check command output exit code
:param shell: shell mode
:param cwd: current work dir
:return: command code
"""
return _run_cmd(cmd, shell=shell, check_exit_code=check_exit_code, cwd=cwd)[0]
def which_command(cmd):
"""
which the server is or not have command
:param cmd: command
:return: bool
"""
return True if run_command_no_output("which %s" % cmd, check_exit_code=False) == 0 \
else False
|