python模块之 - subprocess执行unix/linux命令-FinClip官网

python模块之 - subprocess执行unix/linux命令

网友投稿 1298 2022-09-04

python模块之 - subprocess执行unix/linux命令

subprocess模块提供了一种一致的方法来创建和处理附加进程,与标准库中的其它模块相比，提供了一个更高级的接口,subprocess模块用来生成子进程，并可以通过管道连接它们的输入/输出/错误，以及获得它们的返回值.它用来代替多个旧模块和函数:os.systemos.spawn*os.popen*popen2.*commands.*

1.subprocess.call( commands ) 方法 :

subprocess的call方法可以用于执行一个外部命令，但该方法不能返回执行的结果，只能返回执行的状态码:成功（0)或错误(非0)call()方法中的commands可以是一个列表，也可以是一个字符串，作为字符串时需要用原生的shell=True来执行：

示例：

>>> import subprocess>>> subprocess.call(["ls","-lh"])total 8.0K-rw-r--r--. 1 root root 941 Oct 22 2013 socket_2.py-rw-r--r--. 1 root root 401 Oct 22 2013 socket_client.py0>>> subprocess.call("ls -lh",shell=True)total 8.0K-rw-r--r--. 1 root root 941 Oct 22 2013 socket_2.py-rw-r--r--. 1 root root 401 Oct 22 2013 socket_client.py0>>>如上实例所示，虽然我们能看到执行的结果和返回的一个状态码0，但保存为变量后实际获取的值只是状态码 >>> cmd_result = subprocess.call("ls -lh",shell=True)total 8.0K-rw-r--r--. 1 root root 941 Oct 22 2013 socket_2.py-rw-r--r--. 1 root root 401 Oct 22 2013 socket_client.py>>> print(cmd_result)0>>>

2.subprocess.check_call() 方法:

我们说过call执行返回一个状态码,我们可以通过check_call()函数来检测命令的执行结果，如果不成功将返回 subprocess.CalledProcessError 异常#如果命令执行正确则返回结果，和subprocess.call效果一样.

示例：

>>> try:... subprocess.check_call("ls -lh",shell=True)... except subprocess.CalledProcessError as err:... print("Commands error")...total 8.0K-rw-r--r--. 1 root root 941 Oct 22 2013 socket_2.py-rw-r--r--. 1 root root 401 Oct 22 2013 socket_client.py0>>>#如果命令执行错误，则抛出异常.>>> try:... subprocess.check_call("ls -jj",shell=True)... except subprocess.CalledProcessError as err:... print("Commands error")...ls: invalid option -- 'j'Try `ls --help' for more information.Commands error>>>

subprocess.call 和subprocess.check_call的区别:

subprocess.call:执行命令，返回结果check_all :执行命令，如果执行状态码是0，则返回0，否则抛出异常.

subprocess.call如果命令执行错误不会抛出异常(经测试在linux系统上执行错误命令也会有异常错误，在pycharm上面则不抛出异常)，而subprocess.check_call如果命令执行错误则抛出异常，如果命令执行正确则两者效果一样.

3.subprocess.Popen()方法：

函数call(), check_call() 和 check_output() 都是Popen类的包装器。直接使用Popen会对如何运行命令以及如何处理其输入输出有更多控制。如通过为stdin, stdout和stderr传递不同的参数。

1.与进程的单向通信

通过Popen()方法调用命令后执行的结果,可以设置stdout值为PIPE，再调用communicate()获取结果返回结果为tuple. 在python3中结果为byte类型，要得到str类型需要decode转换一下输出结果(读）

# 直接执行命令输出到屏幕 >>> subprocess.Popen("ls -lh",shell=True)>>> total 8.0K-rw-r--r--. 1 root root 941 Oct 22 2013 socket_2.py-rw-r--r--. 1 root root 401 Oct 22 2013 socket_client.py >>> # 不输出到屏幕,输出到变量#读取字符串>>> cmd_result=subprocess.Popen(["echo","welcome to china"],stdout=subprocess.PIPE)>>> print(cmd_result)>>> stdout_var=cmd_result.communicate()>>> stdout_var('welcome to china\n', None)>>> print((stdout_var[0]).decode('utf-8'))welcome to china >>> #读取命令输出结果>>> cmd_result=subprocess.Popen(["ls","-lh"],stdout=subprocess.PIPE)>>> print(cmd_result)>>> stdout_var=cmd_result.communicate()>>> print((stdout_var[0]).decode("utf-8"))total 8.0K-rw-r--r--. 1 root root 941 Oct 22 2013 socket_2.py-rw-r--r--. 1 root root 401 Oct 22 2013 socket_client.py >>> #将结果输出到文件 >>> file_handle = open("/tmp/t.logs","w+")>>> subprocess.Popen("ls -lh",shell=True,stdout=file_handle)>>> #在指定路径创建目录 >>> import subprocess>>> obj = subprocess.Popen("mkdir t3",shell=True,cwd="/tmp",)>>> obj = subprocess.Popen("mkdir t4",shell=True,cwd="/tmp")>>>

2 与进程的双向通信:

>>> cmd_result=subprocess.Popen("cat",shell=True,stdin=subprocess.PIPE,stdout=subprocess.PIPE)>>> msg="Hello World".encode("utf-8")# 写入到输入管道>>> cmd_result.stdin.write(msg)>>> stdout_var=cmd_result.communicate()>>> stdout_var('Hello World', None)>>> (stdout_var[0]).decode("utf-8")u'Hello World'>>> (stdout_var[0]).decode()u'Hello World'>>>>>> stdout_var[0]'Hello World'>>> #打开一个python终端，执行python print命令.>>> cmd_result=subprocess.Popen(["python"],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE)>>> cmd_result.stdin.write('print("Good luck!")'.encode("utf-8"))>>> stdin_var,err_var=cmd_result.communicate()>>> print(err_var) >>> print(stdin_var)Good luck!>>> stdin_var'Good luck!\n'>>>

Popen其它方法:

poll()	检查是否结束，设置返回值
wait()	等待结束，设置返回值
communicate()	参数是标准输入，返回标准输出和标准出错
send_signal()	发送信号 (主要在unix下有用)
terminate()	终止进程，unix对应的SIGTERM信号，windows下调用api函数TerminateProcess()
kill()	杀死进程(unix对应SIGKILL信号)，windows下同上
stdin stdout stderr	参数中指定PIPE时，有用
pid	进程id
returncode	进程返回值

脚本实例：

#!/usr/bin/env python#coding:utf-8import osimport sysimport subprocessimport timeimport signalTOMCAT_HOME="/usr/local/tomcat-fraud/"TOMCAT_PORT=8083remote_ip="123.56.219.121"remote_port="'-e ssh -p 6168'"try: backup_dir='/data/backup/' + sys.argv[1] + os.sep + time.strftime('%Y%m%d')except IndexError: print("example : python fraud-dev.sh < projectname >")if len(sys.argv) < 2: print("you must use like this : python fraud-dev.sh ") sys.exit()if not os.path.isdir(backup_dir): os.makedirs(backup_dir) print("Sucessfully created directory",backup_dir)else: print("directory alreaty exist")#备份的模块(将项目的war包从jenkins目录取下来放到本地目录.)def backup(): print("download fraud.war...") #Jenkins服务器存放war包的目录 packge_dir="/root/.jenkins/workspace/fraud-dev/fraud-web/target/" remote_ssh = remote_ip + ":" + packge_dir + sys.argv[1] + ".war" print(remote_ssh) #remote_ssh = remote_ip + ":" + packge_dir rsyn="/usr/bin/sudo rsync -avH --progress %s %s %s" %(remote_port,remote_ssh,backup_dir) if subprocess.call(rsyn,shell=True) == 0: print("download fraud.war sucess...") else: print("同步代码失败，check...")#代码更新函数:def rsync_java(): target_dir=TOMCAT_HOME + "webapps/" #os.chdir(target_dir) subprocess.Popen("rm -rf *", shell=True, cwd=target_dir) time.sleep(3) print("开始更新代码中...") rsync_ssh="/usr/bin/sudo rsync -avH --progress %s %s" %(backup_dir,target_dir) if subprocess.call(rsync_ssh,shell=True) == 0: print("代码更新完成...") else: print("代码更新失败...")#启停Java服务.def kill_pid(pid): try: a = os.kill(pid, signal.SIGKILL) print("已杀死pid为%s的进程,返回值是:%s" % (pid, a)) except OSError: print("没有这个进程号")def start_java(): java_service= TOMCAT_HOME + "bin/startup.sh" os.system(java_service) time.sleep(5) print("tomcat 启动完成...")if __name__ == "__main__": backup() li = [] #cmd="/usr/sbin/lsof -i:80 |egrep -v PID | awk {'print $2'} |sort -nr|uniq" cmd="/usr/sbin/lsof -i:%s | grep java |awk {'print $2'}" %(TOMCAT_PORT) pid = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) # print(pid.stdout.readlines()) for m in pid.stdout.readlines(): li.append(int(m)) for i in li: kill_pid(i) time.sleep(3) print("开始上传代码中...") rsync_java() start_java()java备份和上传代码

Java备份和上传代码

#!/usr/bin/env python#coding:utf-8import osimport timeimport subprocesspackge_dir = "/root/.jenkins/workspace/shop-H5/"remote_ip = "59.110.164.99"remote_port = "'-e ssh -p 6168'"backup_dir = '/data/backup/' + time.strftime('%Y%m%d')source_dir = "shop-H5"def backup_php(): tgz = backup_dir + os.sep + "shop-H5_" + time.strftime('%Y%m%d%H') + '.tgz' backup_tgz = "tar -zcvf %s %s" % (tgz,source_dir) #print(time.strftime('%Y-%m-%d-%H:%M')) #print(backup_dir) if not os.path.exists(backup_dir): os.mkdir(backup_dir) print("Sucessfully created directory",backup_dir) print("开始备份代码中...") os.chdir('/opt') if subprocess.call(backup_tgz,shell=True) == 0: print("Successful backup to ", tgz) else: print("Backup FAILED!!") time.sleep(3)def rsync_php(): print("开始代码同步中...") remote_ssh = remote_ip + ":" + packge_dir rsyn = "/usr/bin/sudo rsync -avH --progress %s %s %s" %(remote_port,remote_ssh,source_dir) if subprocess.call(rsyn,shell=True) == 0: print("同步代码成功...") else: print("同步代码失败，check...") #递归改变目录的权限 #os.chown(source_dir,509,509) mattr="chown -R phpci:phpci %s" %(source_dir) #subprocess.call(mattr,shell=True) subprocess.call(mattr,shell=True)if __name__ == "__main__": backup_php() rsync_php()PHP代码备份及上传PHP代码备份及上传

PHP代码备份及上传

#!/usr/bin/env python#coding:utf-8import osimport timeimport subprocesspackge_dir = "/root/.jenkins/workspace/pre-php-p/"remote_ip = "59.110.164.99"remote_port = "'-e ssh -p 6168'"backup_dir= '/data/backup/' + time.strftime('%Y%m%d')source_dir="shop-H5"source_URL="/data/web/xd"def backup_php(): tagz = backup_dir + os.sep + source_dir + "_" + time.strftime('%Y%m%d%H') + '.tgz' print "tagz=",tagz backup_tgz = "tar -zcvf %s %s" %(tagz,source_dir) if not os.path.exists(backup_dir): os.mkdir(backup_dir) print("Sucessfully created directory",backup_dir) print("开始备份代码中...") os.chdir(source_URL) #os.system(backup_tgz) #subprocess.call(backup_tgz,shell=True) if subprocess.call(backup_tgz,shell=True) == 0: print "Successful backup to ", tagz else: print "Backup FAILED!!" time.sleep(3)def rsync_php(): print("开始代码同步中...") remote_ssh = remote_ip + ":" + packge_dir time.sleep(3) rsyn="/usr/bin/sudo rsync -avH --progress --exclude=.git %s %s %s" %(remote_port,remote_ssh,source_dir) #print rsyn if subprocess.call(rsyn,shell=True) == 0: print("同步代码成功...") else: print("同步代码失败，check...") #os.chown(source_dir,504,504) mattr="chown -R phpci:phpci %s" %(source_dir) subprocess.call(mattr,shell=True)if __name__ == "__main__": backup_php() rsync_php()PHP代码备份及目录过滤

PHP代码备份及目录过滤

1.subprocess.call( commands ) 方法 :

示例：

2.subprocess.check_call() 方法:

示例：

subprocess.call 和subprocess.check_call的区别:

subprocess.call:执行命令，返回结果check_all :执行命令，如果执行状态码是0，则返回0，否则抛出异常.

3.subprocess.Popen()方法：

1.与进程的单向通信

2 与进程的双向通信:

Popen其它方法:

poll()	检查是否结束，设置返回值
wait()	等待结束，设置返回值
communicate()	参数是标准输入，返回标准输出和标准出错
send_signal()	发送信号 (主要在unix下有用)
terminate()	终止进程，unix对应的SIGTERM信号，windows下调用api函数TerminateProcess()
kill()	杀死进程(unix对应SIGKILL信号)，windows下同上
stdin stdout stderr	参数中指定PIPE时，有用
pid	进程id
returncode	进程返回值

脚本实例：

Java备份和上传代码

PHP代码备份及上传

PHP代码备份及目录过滤

微信小程序开发实现 tabs 选项卡效果实例代码解析

1298 2022-09-04

python模块之 - subprocess执行unix/linux命令

微信开发中 ACCESS TOKEN 过期失效的解决方案详解

微信小程序开发实现 tabs 选项卡效果实例代码解析

微信小程序开发中选项卡页面切换实现方法详解

最近发表

更多内容

小程序SDK

Finclip技术文档

小程序开发

小程序容器

小程序框架

Finclip小程序平台

Finclip用户投稿

车联网

推荐文章

小程序SDK是什么意思？小程序sdk和插件有什么区别？

小程序支付功能怎么实现？

企业app开发流程是什么？

app运营模式有哪些？

小程序多端引流怎么做？

小程序生态分析的机会和威胁

Flutter入门这一篇效率文章就够了

原生与跨平台解决方案分析,跨平台软件开发技术方案

热更新技术：让软件更新变得更加轻松快速

解决方案

银行解决方案

证券解决方案

互联网解决方案

政企OA解决方案

科技解决方案

loT解决方案

信任解决方案

热评文章

AppCan:基于混合模式的移动应用开发,移动混合模

Hybrid App混合模式开发的了解

小程序容器技术助力券商数字营销突围，小程序容器化的意

用mpvue开发微信小程序基础知识（vue.js开发

小程序多端框架全面测评对比，强烈推荐！

开放银行银行案例，迎接金融创新的未来