python使用subprocess及delegator調用第三方程序

前言

　　python里調用第三方程序一般用subprocess模塊都可以滿足了，但是同程序的交互方面使用subprocess沒有到合適的方法，這里使用了一個第三方模塊delegator.py。它實際上是對subprocess和pexpect.popen_spawn封裝。但其subprocess封方式無法達我的要求（比如無法獲取到實時的return_code）, 因此主要使用了它里面的與程序交互的部分。其他部分直接使用subprocess

　　安裝delegator.py需要使用以下命令:

       pip install delegator.py 

　　文件頭：

#/user/bin/env python3
import delegator
import time
import subprocess
from multiprocessing import Process
import logging
import shlex

subprocess 等待調用

　　執行命令並等待返回。此種場景推薦使用subprocess.run方法，很全面。

if __name__ == '__main__':

    logging.basicConfig(level=logging.DEBUG,
                format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
                datefmt='%a, %d %b %Y %H:%M:%S',
                filename='main.log',
                filemode='w')
    console = logging.StreamHandler()
    console.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
    console.setFormatter(formatter)
    logging.getLogger('').addHandler(console)

    #shell命令執行
    ls = subprocess.run('ls -lht', stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, universal_newlines=True, cwd = None, timeout = 5)
    logging.info(ls.args)
    logging.info(ls.returncode)
    logging.info(ls.stdout)
   
    #非shell命令執行
    try:
        ffprobe = subprocess.run(shlex.split('ffprobe -show_streams -i rtmp://rtmp-push.source.test/live/i3jMugm01'), stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, shell=False, universal_newlines=True, cwd = '/usr/local/bin', timeout = 3)
    except Exception as e:
        logging.exception(e)
    else:
        logging.info(ffprobe.args)
        logging.info(ffprobe.returncode)
        logging.info(ffprobe.stdout)

參數說明

args	待執行的命令，可以試字符串或者是命令列表。
stdin,stdout,stderr	默認均為None，如果需要與程序進行交互，則需要設置stdin，如需獲取輸出和錯誤，則需要設置stdout和stderr。stdout和stderr一般可設置為subprocess.PIPE，文件對象或subprocess.DEVNULL,  另外stderr可設置為subprocess.STDOUT, 表示將錯誤重定向到stderr制定的管道。
shell	subprocess.run里默認為False。設置為True時，可以認為就和在linux控制執行命令一樣，支持grep類似操作。此時args參數可以是復雜字符串。設置為False時，如果args命令中是包含參數的，則必須以命令列表方式傳入。一般通過shlex.split(string)處理即可。
universal_newlines	設置為True時，stdin，stdout，stderr均已字符串方式讀寫，否則以字節流方式讀寫。
cwd	設置程序的執行路徑，運行非shell命令時，可以用來制定程序的所在路徑。
timeout	命令超時時間，當程序超過制定時間還沒有返回時，會拋出超時異常

 

subprocess 非等待調用

　　執行命令后不等待程序返回。這里使用的是subprocess.Popen對象。這里給出的例子加上了守護程序，適合需要長時間保證程序正常執行的情況。

　　可以設定檢查間隔時間，根據程序的returncode（None表示正常運行，0為正常退出，其他均為非正常結束）判斷是否正常執行。下面代碼中是只有kill -9或者是程序自己正常退出時才會結束，其他異常退出的情況程序會重新啟動（包括kill命令）

　　同時為了防止無限重啟，加入健康檢查機制，超過一定次數連續運行失敗則停止運行

　　這里使用多進程方式進行調用，方便按照此方式啟動多個程序

    #longtime task
    def runapp(command, logfile, check_interval=5, health_check=3):
        applog= open(logfile, 'w') #使用日志文件記錄app輸出
        app = subprocess.Popen(command, stdout=applog, stderr=subprocess.STDOUT,universal_newlines=True) #錯誤重新向到輸出一並記錄
        fail_count = 0
        while True:
            time.sleep(check_interval)  #檢查間隔
            logging.debug("args {}".format(app.args))
            logging.debug("pid {}".format(app.pid))
            app.poll() #通過poll方法設置程序的returncode，即調用poll方法后獲取到的returncode才是實時的
            if app.returncode != None:
                if app.returncode != 0:
                    fail_count +=1  #非0表示程序異常退出
                if fail_count >= health_check :
                    logging.debug("health check fail,stop app!")
                    break
                if app.returncode != 0 and app.returncode != -9: #正常結束或是被kill -9方式殺掉后退出，否則重啟程序
                    logging.debug("app restart!")
                    app.__init__(command, stdout=applog, stderr=subprocess.STDOUT,universal_newlines=True)
                else:
                    logging.debug("app eixt, code {}".format(app.returncode))
                    break
            else:
                fail_count = 0
    #命令參數處理
    command = shlex.split("ffmpeg -i rtmp://rtmp-push.source.test/live/i3jMugm0 -c:v copy -c:a copy -f flv rtmp://rtmp-push.source.test/live/pwtest12345")
    #子進程方式
    p = Process(target = runapp, args=(command, "ffmpeg.log"))
    p.start()
    #p.join()

delegator 與程序進行交互（非阻塞）

　　這里使用vi編輯器作為例子，演示了使用v編輯文件的方式。可應用在類似場景中，但在某些操作上可能需要加些延時才能保證執行的正確性。

　　需要注意block參數必須設置為False，send方法帶上參數withend=False時不發送回車(這里的send方法默認會加上一個回車，因此這里對delegator.py中的send方法進行了修改，兼容不需要發送回車的情況。修改的部分放在最后。)。

　　另外下面倒數第二條發送的其實是一個ESC字符(chr(27),顯示退格符號)，但這里顯示為空字符了。

    #send string
    command = ['vi','test.txt']
    editer = delegator.run(command, block=False) #注意block參數要設置為False
    time.sleep(1)                                #程序啟動后，首次發送信息時，加入一點延遲
    editer.send('i', withend=False)
    editer.send('12345', withend=False)
    editer.send('', withend=False)
    time.sleep(1)                                #程序退出編輯模式進入命令模式時，加入一點延遲
    editer.send(':wq!')
    editer.kill()

delegator 與程序進行交互（阻塞式）

　　調用程序的另外一個場景，就是等待程序出現某提示時，再輸入信息。下面通過一個延時設置密碼的腳本執行進程進程演示。

　　主要注意expect方法的使用，第一個參數為等待出現的信息，可以使用正則表達式，第二個參數為超時時間，超過時間但沒有出現逾期的內容，則會拋出異常

　　passwd.sh內容：

#!/bin/sh
sleep 3
passwd

　　設置密碼腳本：

    command = ['./passwd.sh']
    input =  delegator.run(command, block=False) #注意block參數要設置為False
    try:
        input.expect('New password:',timeout=5)  #腳本延遲3秒運行passwd命令，這里等待5秒 出現'New password:'
    except Exception as e:
        logging.exception(e)
    else:
        input.send("123")
    try:
        input.expect('Retype password:',timeout=2)
    except Exception as e:
        logging.exception(e)        
    else:
        input.send("456")
    input.kill()
    logging.info(input.out)

對delegator.py 的修改部分

    def send(self, s, end=os.linesep, signal=False, withend=True):
        """Sends the given string or signal to std_in."""

        if self.blocking:
            raise RuntimeError("send can only be used on non-blocking commands.")

        if not signal:
            if self._uses_subprocess:
                return self.subprocess.communicate((s + end) if withend else s)
            else:
                return self.subprocess.send((s + end) if withend else s)
        else:
            self.subprocess.send_signal(s)