當前位置：首頁 > 编程语言 > python >内容正文

python

python打印字节流_Python 调用系统命令的模块 Subprocess

發布時間：2024/8/5 python 43 豆豆

生活随笔收集整理的這篇文章主要介紹了 python打印字节流_Python 调用系统命令的模块 Subprocess 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

有些時候需要調用系統內部的一些命令，或者給某個應用命令傳不定參數時可以使用該模塊。

初識 Subprocess 模塊

Subprocess 模塊提供了多個方法來運行額外的進程。在 Python2.7 的時候使用的方法主要有 call(),check_call(), check_output(),到了 Python3.5 的時候加入了一個更高級的方法 run()，該方法可以運行一個額外的進程同時它還能收集到運行之后的結果。Popen 類最為一個低級 API，它主要用于構建其他 API，在更復雜的流程交互中非常有用。Popen 的構造函數接受參數來設置新進程，以便父進程可以通過管道與它通信。它替換了其他模塊和函數的所有功能，甚至更多。Subprocess 子進程模塊旨在替換 os.system(), os.spawnv()等函數，os 和 popen2 模塊中 popen()的變體，以及 commands()模塊
注意:在 Unix 和 Windows 系統上工作的應用編程接口大致相同，但是底層的實現是不同的，因為操作系統中的過程模型不同。這里顯示的所有示例都是在 Mac 操作系統上測試的。在非 Unix 操作系統上的行為可能會有所不同。例如 unix 系統查看文件列表使用 ls,windows 只能使用 dir.

Run 方法使用

運行外部命令

要實現和 os.system()命令相同的方式,運行外部命令而不與之交互時候，我們可以使用 run()函數。前面提到了這是一個高級函數
先看一下其語法結構。

subprocess.run(args,?*,?stdin=None,?input=None,?stdout=None,?stderr=None,?capture_output=False,?shell=False,?cwd=None,?timeout=None,?check=False,?encoding=None,?errors=None,?text=None,?env=None,?universal_newlines=None)

運行被 arg 描述的指令。等待指令完成，然后返回一個 CompletedProcess 示例。run 方法的參數和 Popen 的構造函數一樣，接受的大多數參數都被傳遞給該接口。(timeout, input, check 和 capture_output 除外)。

import?subprocess
completed?=?subprocess.run(['ls',?'-1'])
print('returncode:',?completed.returncode)

輸出內容:

subprocess_demo.py
returncode:?0

第一個參數傳入的就是我們要運行的命令,其格式推薦使用列表字符串的形式，將命令進行分割。這避免了轉義引號或 shell 可能解釋的其他特殊字符的需要。
如果將 shell 參數設置為 true 值將導致子進程生成一個中間 shell 進程，然后運行該命令。默認情況下是直接運行命令。

import?subprocess

completed?=?subprocess.run('echo?$HOME',?shell=True)
print('returncode:',?completed.returncode)

輸出

/Users/chenxiangan
returncode:?0

使用中間 shell 意味著在運行命令之前要處理命令字符串中的變量、glob 模式和其他特殊的 shell 特性。

錯誤處理

CompletedProcess 的 returncode 屬性是程序的退出代碼。調用者負責解釋它以檢測錯誤。如果 run()的 check 參數為 True，則檢查退出代碼，如果它指示發生錯誤，則引發 CalledProcessError 異常。

#公眾號:python?學習開發
#author:陳祥安
import?subprocess

try:
????subprocess.run(['false'],?check=True)
except?subprocess.CalledProcessError?as?err:
????print('ERROR:',?err)

運行結果

ERROR:?Command?'['false']'?returned?non-zero?exit?status?1.

false 命令總是以非零狀態代碼退出，run()將其解釋為錯誤。
將 run()函數的 check 屬性設置為 True,等同于使用 check_call()方法。

獲取結果

由于 run()啟動的進程的標準輸入和輸出通道綁定到父輸入和輸出。這意味著調用程序無法捕獲命令的輸出。可以通過調整 stdout 和 stderr 參數來捕獲輸出的值。

#公眾號:python?學習開發
#author:陳祥安
import?subprocess

completed?=?subprocess.run(
????['ls',?'-1'],
????stdout=subprocess.PIPE,
)
print('returncode:',?completed.returncode)
print(f"結果的字節長度?{len(completed.stdout)}:\n{?completed.stdout.decode('utf-8')}")

輸出

returncode:?0
結果的字節長度?24:
subprocess_demo.py

ls -1 命令成功運行，捕獲并返回輸出結果。
下一個示例在子 shell 中運行一系列命令。在命令退出并顯示錯誤代碼之前，消息將發送到標準輸出和標準錯誤。

#公眾號:python?學習開發

import?subprocess

try:
????completed?=?subprocess.run(
????????'echo?to?stdout;?echo?to?stderr?1>&2;?exit?1',
????????check=True,
????????shell=True,
????????stdout=subprocess.PIPE,
????)
except?subprocess.CalledProcessError?as?err:
????print('ERROR:',?err)
else:
????print('returncode:',?completed.returncode)
????print(f"stdout?中的字節長度?{len(completed.stdout)}?:?{completed.stdout.decode('utf-8')!r}")

輸出結果

to?stderr
ERROR:?Command?'echo?to?stdout;?echo?to?stderr?1>&2;?exit?1'?returned?non-zero?exit?status?1.

發送到標準錯誤的消息被打印到控制臺，但是發送到標準輸出的消息是隱藏的。
為了防止通過 run()運行的命令的錯誤消息被寫入控制臺，需要將 stderr 參數設置為 subprocess.PIPE。修改后代碼如下

#公眾號:python?學習開發

import?subprocess

try:
????completed?=?subprocess.run(
????????'echo?to?stdout;?echo?to?stderr?1>&2;?exit?1',
????????shell=True,
????????stdout=subprocess.PIPE,
????????stderr=subprocess.PIPE,
????)
except?subprocess.CalledProcessError?as?err:
????print('ERROR:',?err)
else:
????print('returncode:',?completed.returncode)
????print(f"stderr?字節長度{len(completed.stdout)}:?{completed.stdout.decode('utf-8')!r}")
????print(f"stderr?字節長度{len(completed.stderr)}:?{completed.stderr.decode('utf-8')!r}")

輸出結果

returncode:?1
stderr?字節長度?10:?'to?stdout\n'
stderr?字節長度?10:?'to?stderr\n'

本示例未設置 check=True，因此會捕獲并打印命令的輸出。若要在使用 check_output()時捕獲錯誤消息，請將 stderr 設置為 STDOUT，消息將與命令的其余輸出合并。

禁止輸出

對于不應該顯示或捕獲輸出的情況，使用 DEVNULL 來抑制輸出流，這個例子同時抑制了標準輸出和錯誤流。

#?公眾號:python?學習開發

import?subprocess

try:
????completed?=?subprocess.run(
????????'echo?to?stdout;?echo?to?stderr?1>&2;?exit?1',
????????shell=True,
????????stdout=subprocess.DEVNULL,
????????stderr=subprocess.DEVNULL,
????)
except?subprocess.CalledProcessError?as?err:
????print('ERROR:',?err)
else:
????print('returncode:',?completed.returncode)
????print(f'stdout?is?{completed.stdout!r}')
????print(f'stderr?is?{completed.stderr!r}')

輸出

returncode:?1
stdout?is?None
stderr?is?None

名稱 DEVNULL 來自于 Unix 特殊設備文件/DEVE/null，該文件在打開讀取時以文件結尾響應，并在寫入時接收但忽略任何數量的輸入。

Popen 方法的使用

函數 run()、call()、check_call()和 check_output()是 Popen 類的包裝器。直接使用 Popen 可以更好地控制命令的運行方式以及輸入和輸出流的處理方式。例如，通過傳遞 stdin、stdout 和 stderr 的不同參數，可以模擬 os.popen()。
Popen 的語法結構如下:

class?subprocess.Popen(args,?bufsize=-1,?executable=None,?stdin=None,?stdout=None,?stderr=None,?preexec_fn=None,?close_fds=True,?shell=False,?cwd=None,?env=None,?universal_newlines=None,?startupinfo=None,?creationflags=0,?restore_signals=True,?start_new_session=False,?pass_fds=(),?*,?encoding=None,?errors=None,?text=None)?

args 應當是一個程序的參數列表或者一個簡單的字符串。默認情況下，如果 args 是一個序列，將運行的程序是此序列的第一項。如果 args 是一個字符串，解釋是平臺相關的，如下所述。除非另有說明，推薦將 args 作為序列傳遞。
參數 shell (默認為 False)指定是否使用 shell 執行程序。如果 shell 為 True，更推薦將 args 作為字符串傳遞而非序列。
在 POSIX，當 shell=True， shell 默認為 /bin/sh。如果 args 是一個字符串，此字符串指定將通過 shell 執行的命令。這意味著字符串的格式必須和在命令提示符中所輸入的完全相同。這包括，例如，引號和反斜杠轉義包含空格的文件名。如果 args 是一個序列，第一項指定了命令，另外的項目將作為傳遞給 shell (而非命令) 的參數對待。也就是說， Popen 等同于:

Popen(['/bin/sh',?'-c',?args[0],?args[1],?...])

在 Windows，使用 shell=True，環境變量 COMSPEC 指定了默認 shell。在 Windows 你唯一需要指定 shell=True 的情況是你想要執行內置在 shell 中的命令(例如 dir 或者 copy)。在運行一個批處理文件或者基于控制臺的可執行文件時，不需要 shell=True。

與進程的單向通信

要運行進程并讀取其所有輸出，需要將 stdout 值設置為 PIPE 并調用。

import?subprocess

print('read:')
proc?=?subprocess.Popen(
????['echo',?'"to?stdout"'],
????stdout=subprocess.PIPE,
)
stdout_value?=?proc.communicate()[0].decode('utf-8')
print('stdout:',?repr(stdout_value))

輸出

read:
stdout:?'"to?stdout"\n'

如果要設置管道允許調用程序將數據寫入管道，需要將 stdin 設置為 pipe。

import?subprocess

print('write:')
proc?=?subprocess.Popen(
????['cat',?'-'],
????stdin=subprocess.PIPE,
)
proc.communicate('stdin:?to?stdin\n'.encode('utf-8'))

輸出

write:
stdin:?to?stdin

要一次將數據發送到進程的標準輸入通道，可以使用返回對象的 communication()方法。它與使用'w'模式的 popen()類似.

與進程的雙向通信

要同時設置 Popen 實例進行讀寫，請結合使用以前的技術。

import?subprocess

print('popen2:')

proc?=?subprocess.Popen(
????['ls',?'-l'],
????stdin=subprocess.PIPE,
????stdout=subprocess.PIPE,
)
msg?=?'through?stdin?to?stdout'.encode('utf-8')
stdout_value?=?proc.communicate(msg)[0].decode('utf-8')
print('pass?through:',?repr(stdout_value))

輸出

popen2:
pass?through:?'total?8\n-rw-r--r--??1?chenxiangan??staff??316?Jul??9?11:20?subprocess_demo.py\n'

使用 communicate() 而非 .stdin.write， .stdout.read 或者 .stderr.read 來避免由于任意其他 OS 管道緩沖區被子進程填滿阻塞而導致的死鎖。

錯誤捕獲

Popen 還可以像使用 popen3()一樣,同時監視 stdout 和 stderr 流。

import?subprocess

print('popen3:')
proc?=?subprocess.Popen(
????'ls?-l;?echo?"to?stderr"?1>&2',
????shell=True,
????stdin=subprocess.PIPE,
????stdout=subprocess.PIPE,
????stderr=subprocess.PIPE,
)
msg?=?'through?stdin?to?stdout'.encode('utf-8')
stdout_value,?stderr_value?=?proc.communicate(msg)
print('pass?through:',?repr(stdout_value.decode('utf-8')))
print('stderr??????:',?repr(stderr_value.decode('utf-8')))

輸出

popen3:
pass?through:?'total?8\n-rw-r--r--??1?chenxiangan??staff??447?Jul??9?11:22?subprocess_os_system.py\n'
stderr??????:?'to?stderr\n'

從 stderr 讀取的工作與 stdout 相同。通過傳入 PIPE 告訴 Popen 連接到通道，并且 communication()方法在返回結果之前可以從中讀取所有數據。

結合常規輸出和錯誤輸出

要將進程的錯誤輸出定向到其標準輸出通道，可以使用 STDOUT 代替 stderr 而不是 PIPE。

#?公眾號:python?學習開發

import?subprocess

print('popen4:')
proc?=?subprocess.Popen(
????'ls?-l;?echo?"to?stderr"?1>&2',
????shell=True,
????stdin=subprocess.PIPE,
????stdout=subprocess.PIPE,
????stderr=subprocess.STDOUT,
)
msg?=?'through?stdin?to?stdout\n'.encode('utf-8')
stdout_value,?stderr_value?=?proc.communicate(msg)
print('combined?output:',?repr(stdout_value.decode('utf-8')))
print('stderr?value???:',?repr(stderr_value))

輸出

popen4:
combined?output:?'total?8\n-rw-r--r--??1?chenxiangan??staff??441?Jul??9?11:25?subprocess_os_system.py\nto?stderr\n'
stderr?value???:?None

以這種方式組合輸出類似于 popen4()的工作方式。

管道之間的連接

通過創建單獨的 Popen 實例并將它們的輸入和輸出鏈接在一起，可以類似于 Unix shell 的工作方式將多個命令連接到管道中。

#?公眾號:python?學習開發

import?subprocess

cat?=?subprocess.Popen(
????['cat',?'subprocess_demo.py'],
????stdout=subprocess.PIPE,??#?提供輸出的方式
)

grep?=?subprocess.Popen(
????['grep',?'公眾號'],
????stdin=cat.stdout,??#?cat?的輸出最為輸入
????stdout=subprocess.PIPE,
)

cut?=?subprocess.Popen(
????['awk',?'-F',?':',?'{print?$2}'],
????stdin=grep.stdout,
????stdout=subprocess.PIPE,
)

end_of_pipe?=?cut.stdout

print(end_of_pipe.readline().decode('utf-8'))

輸出

python?學習開發

上面的內容就等價于下面的命令

cat?subprocess_demo.py?|grep?"公眾號"?|awk?-F?':'?'{print?$2}'

與另一個命令交互

前面的所有示例都假定了有限的交互量。方法讀取所有輸出并等待子進程退出后返回。在程序運行時，還可以增量地對 Popen 實例使用的各個管道句柄進行讀寫。一個簡單的 echo 程序演示了這種技術，該程序從標準輸入讀取數據并將其寫入標準輸出。
在下一個示例中，腳本 repeat.py 用作子進程。它從 stdin 讀取值并將值寫入 stdout，每次一行，直到沒有更多輸入為止。它還在啟動和停止時向 stderr 寫入一條消息，顯示子進程的生存期。

#?文件:repeater.py
#?公眾號:python?學習開發
import?sys

sys.stderr.write('repeater.py:?starting\n')
sys.stderr.flush()

while?True:
????next_line?=?sys.stdin.readline()
????sys.stderr.flush()
????if?not?next_line:
????????break
????sys.stdout.write(next_line)
????sys.stdout.flush()

sys.stderr.write('repeater.py:?exiting\n')
sys.stderr.flush()

下一個交互示例以不同的方式使用 Popen 實例擁有的 stdin 和 stdout 文件句柄.
在第一個例子中，將 0-4 依次被寫入進程的 stdin，并且在每次寫入之后讀回下一行輸出。在第二個示例中，寫入這五個數字，但是使用 communic()一次讀取所有輸出。

import?io
import?subprocess

print('One?line?at?a?time:')
proc?=?subprocess.Popen(
????'python3?repeater.py',
????shell=True,
????stdin=subprocess.PIPE,
????stdout=subprocess.PIPE,
)
stdin?=?io.TextIOWrapper(
????proc.stdin,
????encoding='utf-8',
????line_buffering=True,??#?send?data?on?newline
)
stdout?=?io.TextIOWrapper(
????proc.stdout,
????encoding='utf-8',
)
for?i?in?range(5):
????line?=?'{}\n'.format(i)
????stdin.write(line)
????output?=?stdout.readline()
????print(output.rstrip())
remainder?=?proc.communicate()[0].decode('utf-8')
print(remainder)

print()
print('All?output?at?once:')
proc?=?subprocess.Popen(
????'python3?repeater.py',
????shell=True,
????stdin=subprocess.PIPE,
????stdout=subprocess.PIPE,
)
stdin?=?io.TextIOWrapper(
????proc.stdin,
????encoding='utf-8',
)
for?i?in?range(5):
????line?=?f'{i}\n'
????stdin.write(line)
stdin.flush()

output?=?proc.communicate()[0].decode('utf-8')
print(output)

“repeater.py：exiting”出現在每個循環位置的不同點。

One?line?at?a?time:
repeater.py:?starting
0
1
2
3
4
repeater.py:?exiting

All?output?at?once:
repeater.py:?starting
repeater.py:?exiting
0
1
2
3
4

信號之間的進程

正在 os 模塊中的進程管理示例包括使用 os.fork()和 os.kill()在進程之間發送信號的演示。
由于每個 Popen 實例都提供了一個帶有子進程的進程 id 的 pid 屬性，所以可以對子進程執行類似的操作。下一個例子結合了兩個腳本。這個子進程為 USR 信號設置一個信號處理器。

#signal_child.py
import?os
import?signal
import?time
import?sys

pid?=?os.getpid()
received?=?False

def?signal_usr1(signum,?frame):
????"Callback?invoked?when?a?signal?is?received"
????global?received
????received?=?True
????print(f'CHILD?{pid:>6}:?Received?USR1')
????sys.stdout.flush()

print(f'CHILD?{pid:>6}:?Setting?up?signal?handler')
sys.stdout.flush()
signal.signal(signal.SIGUSR1,?signal_usr1)
print(f'CHILD?{pid:>6}:?Pausing?to?wait?for?signal')
sys.stdout.flush()
time.sleep(3)

if?not?received:
????print(f'CHILD?{pid:>6}:?Never?received?signal')

然后再寫一個文件,此腳本作為父進程運行。它啟動 signal_child.py，然后發送 USR1 信號。

import?os
import?signal
import?subprocess
import?time
import?sys

proc?=?subprocess.Popen(['python3',?'signal_child.py'])
print('PARENT??????:?Pausing?before?sending?signal...')
sys.stdout.flush()
time.sleep(1)
print('PARENT??????:?Signaling?child')
sys.stdout.flush()
os.kill(proc.pid,?signal.SIGUSR1)

運行之后

PARENT??????:?Pausing?before?sending?signal...
CHILD??46573:?Setting?up?signal?handler
CHILD??46573:?Pausing?to?wait?for?signal
PARENT??????:?Signaling?child
CHILD??46573:?Received?USR1

進程組

如果子進程是由 Popen 創建的進程產生的，那些子進程將不會收到發送給父進程的任何信號。當 Popen 使用 shell 參數時，很難通過發送 SIGINT 或 SIGTERM 來使 shell 中啟動的命令終止。
若要在不知道進程 id 的情況下向后代發送信號，請使用進程組將子進程關聯起來，以便將它們一起發送信號。使用 os.setpgrp()創建進程組,將進程組 id 設置為當前進程的進程 id。

import?os
import?signal
import?subprocess
import?tempfile
import?time
import?sys

script?=?'''#!/bin/sh
echo?"Shell?script?in?process?$$"
set?-x
python3?signal_child.py
'''
script_file?=?tempfile.NamedTemporaryFile('wt')
script_file.write(script)
script_file.flush()

proc?=?subprocess.Popen(['sh',?script_file.name])
print('PARENT??????:?Pausing?before?signaling?{}...'.format(
????proc.pid))
sys.stdout.flush()
time.sleep(1)
print('PARENT??????:?Signaling?child?{}'.format(proc.pid))
sys.stdout.flush()
os.kill(proc.pid,?signal.SIGUSR1)
time.sleep(3)

輸出

PARENT??????:?Pausing?before?signaling?46600...
Shell?script?in?process?46600
+?python3?signal_child.py
CHILD??46601:?Setting?up?signal?handler
CHILD??46601:?Pausing?to?wait?for?signal
PARENT??????:?Signaling?child?46600
CHILD??46601:?Never?received?signal

用于發送信號的 pid 與等待信號的 shell 腳本的子腳本的 pid 不匹配，因為在本例中有三個獨立的進程在交互
1.程序子進程向父 shell.py 發送信號
2.shell 進程運行主 python 程序創建的腳本
3.signal_child.py 進程。
如果想在不知道其進程 ID 的情況下向后代發送信號，可以使用進程組來關聯子進程，以便它們可以一起發出信號。可以使用 os.setpgrp()創建進程組，然后將進程組 id 設置為當前進程的進程 id。這樣所有子進程都從父進程繼承它們的進程組,因為它只能在 Popen 及其后代創建的 shell 中設置，所以不應該在創建 Popen 的同一進程中調用 os.setpgrp()。更改之后的代碼如下:

import?os
import?signal
import?subprocess
import?tempfile
import?time
import?sys

def?show_setting_prgrp():
????print('Calling?os.setpgrp()?from?{}'.format(os.getpid()))
????os.setpgrp()
????print('Process?group?is?now?{}'.format(os.getpgrp()))
????sys.stdout.flush()

script?=?'''#!/bin/sh
echo?"Shell?script?in?process?$$"
set?-x
python3?signal_child.py
'''
script_file?=?tempfile.NamedTemporaryFile('wt')
script_file.write(script)
script_file.flush()

proc?=?subprocess.Popen(
????['sh',?script_file.name],
????preexec_fn=show_setting_prgrp,
)
print('PARENT??????:?Pausing?before?signaling?{}...'.format(
????proc.pid))
sys.stdout.flush()
time.sleep(1)
print('PARENT??????:?Signaling?process?group?{}'.format(
????proc.pid))
sys.stdout.flush()
os.killpg(proc.pid,?signal.SIGUSR1)
time.sleep(3)

輸出結果

+?python3?signal_child.py
Calling?os.setpgrp()?from?46618
Process?group?is?now?46618
PARENT??????:?Pausing?before?signaling?46618...
Shell?script?in?process?46618
CHILD??46619:?Setting?up?signal?handler
CHILD??46619:?Pausing?to?wait?for?signal
PARENT??????:?Signaling?process?group?46618
CHILD??46619:?Received?USR1

事件發生的順序如下:
1.父程序實例化 Popen。
2.Popen 實例派生了一個新的進程。
3.新進程運行 os.setpgrp()。
4.新進程會運行 exec()啟動 shell。
5.shell 會運行 shell 腳本
6.shell 腳本再次 fork，該進程執行 Python。
7.python 運行腳本 signal_child.py
8.父程序使用 shell 的 pid 向進程組發出信號。
9.shell 和 Python 進程接收信號
10.運行 signal child.py 的 Python 進程調用信號處理程序。

參考資料

https://docs.python.org/3.7/library/subprocess.html

總結

以上是生活随笔為你收集整理的python打印字节流_Python 调用系统命令的模块 Subprocess的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：【解决方案】校园明厨亮灶监控系统实施方案
下一篇： php数组添加数组_PHP数组