Вызов внешней команды в Python - PullRequest
Новая
       559

Вызов внешней команды в Python

4261 голосов
/ 18 сентября 2008

Как я могу вызвать внешнюю команду (как если бы я набрал ее в оболочке Unix или командной строке Windows) из скрипта Python?

Ответы [ 56 ]

2 голосов
/ 14 марта 2014

После некоторых исследований у меня есть следующий код, который очень хорошо работает для меня. Он в основном печатает как stdout, так и stderr в режиме реального времени. Надеюсь, что это поможет кому-то еще, кто нуждается в этом. 

stdout_result = 1
stderr_result = 1


def stdout_thread(pipe):
    global stdout_result
    while True:
        out = pipe.stdout.read(1)
        stdout_result = pipe.poll()
        if out == '' and stdout_result is not None:
            break

        if out != '':
            sys.stdout.write(out)
            sys.stdout.flush()


def stderr_thread(pipe):
    global stderr_result
    while True:
        err = pipe.stderr.read(1)
        stderr_result = pipe.poll()
        if err == '' and stderr_result is not None:
            break

        if err != '':
            sys.stdout.write(err)
            sys.stdout.flush()


def exec_command(command, cwd=None):
    if cwd is not None:
        print '[' + ' '.join(command) + '] in ' + cwd
    else:
        print '[' + ' '.join(command) + ']'

    p = subprocess.Popen(
        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd
    )

    out_thread = threading.Thread(name='stdout_thread', target=stdout_thread, args=(p,))
    err_thread = threading.Thread(name='stderr_thread', target=stderr_thread, args=(p,))

    err_thread.start()
    out_thread.start()

    out_thread.join()
    err_thread.join()

    return stdout_result + stderr_result
2 голосов
/ 29 января 2019

Поскольку некоторые из ответов были связаны с предыдущими версиями python или использовали модуль os.system, я публикую этот ответ для таких людей, как я, которые намереваются использовать subprocess в python 3.5+. Следующее помогло мне в Linux: 

import subprocess

#subprocess.run() returns a completed process object that can be inspected
c = subprocess.run(["ls", "-ltrh"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print(c.stdout.decode('utf-8'))

Как упомянуто в документации , значения PIPE являются байтовыми последовательностями, и для правильного их отображения следует рассмотреть декодирование. Для более поздних версий python text=True и encoding='utf-8' добавляются к kwargs subprocess.run().

Вывод вышеупомянутого кода: 

total 113M
-rwxr-xr-x  1 farzad farzad  307 Jan 15  2018 vpnscript
-rwxrwxr-x  1 farzad farzad  204 Jan 15  2018 ex
drwxrwxr-x  4 farzad farzad 4.0K Jan 22  2018 scripts
.... # some other lines
2 голосов
/ 28 апреля 2016

Я бы порекомендовал следующий метод 'run', и он поможет нам получить STDOUT, STDERR и статус выхода в качестве словаря; Вызывающий объект может прочитать словарь, возвращаемый методом run, чтобы узнать фактическое состояние процесса. 

  def run (cmd):
       print "+ DEBUG exec({0})".format(cmd)
       p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, shell=True)
       (out, err) = p.communicate()
       ret        = p.wait()
       out        = filter(None, out.split('\n'))
       err        = filter(None, err.split('\n'))
       ret        = True if ret == 0 else False
       return dict({'output': out, 'error': err, 'status': ret})
  #end
2 голосов
/ 25 октября 2017

Я написал оболочку для обработки ошибок и перенаправления вывода и прочего. 

import shlex
import psutil
import subprocess

def call_cmd(cmd, stdout=sys.stdout, quiet=False, shell=False, raise_exceptions=True, use_shlex=True, timeout=None):
    """Exec command by command line like 'ln -ls "/var/log"'
    """
    if not quiet:
        print("Run %s", str(cmd))
    if use_shlex and isinstance(cmd, (str, unicode)):
        cmd = shlex.split(cmd)
    if timeout is None:
        process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
        retcode = process.wait()
    else:
        process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
        p = psutil.Process(process.pid)
        finish, alive = psutil.wait_procs([p], timeout)
        if len(alive) > 0:
            ps = p.children()
            ps.insert(0, p)
            print('waiting for timeout again due to child process check')
            finish, alive = psutil.wait_procs(ps, 0)
        if len(alive) > 0:
            print('process {} will be killed'.format([p.pid for p in alive]))
            for p in alive:
                p.kill()
            if raise_exceptions:
                print('External program timeout at {} {}'.format(timeout, cmd))
                raise CalledProcessTimeout(1, cmd)
        retcode = process.wait()
    if retcode and raise_exceptions:
        print("External program failed %s", str(cmd))
        raise subprocess.CalledProcessError(retcode, cmd)

Вы можете назвать это так: 

cmd = 'ln -ls "/var/log"'
stdout = 'out.txt'
call_cmd(cmd, stdout)
1 голос
/ 18 апреля 2013

Модуль подпроцесса , описанный выше Эли, является очень мощным, но синтаксис, позволяющий выполнить стандартный системный вызов и проверить его вывод, излишне распространен.

Самый простой способ сделать системный вызов - воспользоваться командным модулем (только для Linux). 

> import commands
> commands.getstatusoutput("grep matter alice-in-wonderland.txt")
(0, "'Then it doesn't matter which way you go,' said the Cat.")

Первый элемент в кортеже - это код возврата процесса. Второй элемент - это его стандартный вывод (и стандартная ошибка, объединенная).

Разработчики Python "устарели" как модуль команд, но это не значит, что вы не должны его использовать. Только то, что они его больше не развивают, и это нормально, потому что он уже совершенен (при своей небольшой, но важной функции).

0 голосов
/ 31 марта 2019

Если вы НЕ , используя пользовательский ввод в командах, вы можете использовать это 

from os import getcwd
from subprocess import check_output
from shlex import quote

def sh(command):
    return check_output(quote(command), shell=True, cwd=getcwd(), universal_newlines=True).strip()

И используйте его как 

branch = sh('git rev-parse --abbrev-ref HEAD')

shell=True породит оболочку, так что вы можете использовать трубу и такие вещи оболочки sh('ps aux | grep python'). Это очень удобно для запуска жестко запрограммированных команд и обработки их вывода. universal_lines=True гарантирует, что выходные данные возвращаются в виде строки, а не в двоичном формате.

cwd=getcwd() убедится, что команда запущена в том же рабочем каталоге, что и интерпретатор. Это удобно для работы команд git, как в приведенном выше примере с именем ветви git.

Некоторые рецепты

  • свободной памяти в мегабайтах: sh('free -m').split('\n')[1].split()[1]
  • свободного места в / в процентах sh('df -m /').split('\n')[1].split()[4][0:-1]
  • загрузка процессора sum(map(float, sh('ps -ef -o pcpu').split('\n')[1:])

Но это небезопасно для ввода пользователем из документов:

Вопросы безопасности¶

В отличие от некоторых других попсовых функций, эта реализация никогда не будет неявно вызвать системную оболочку. Это означает, что все персонажи, включая метасимволы оболочки, можно безопасно передать ребенку процессы. Если оболочка вызывается явно, через shell = True, это ответственность приложения за обеспечение того, чтобы все пробелы и метасимволы указаны в кавычках, чтобы избежать внедрения в оболочку уязвимости.

Когда используется shell = True, функцию shlex.quote () можно использовать для правильно экранировать пробелы и метасимволы оболочки в строках, которые будут использоваться для создания команд оболочки.

Даже при использовании shlex.quote() полезно быть немного параноиком при использовании пользовательских вводов для команд оболочки. Одним из вариантов является использование команды с жестким кодом для получения общего вывода и фильтрации по пользовательскому вводу. В любом случае использование shell=False обеспечит выполнение только того процесса, который вы хотите выполнить, или вы получите ошибку No such file or directory.

Также есть некоторое влияние на производительность на shell=True, по моим тестам оно примерно на 20% медленнее, чем shell=False (по умолчанию). 

In [50]: timeit("check_output('ls -l'.split(), universal_newlines=True)", number=1000, globals=globals())
Out[50]: 2.6801227919995654

In [51]: timeit("check_output('ls -l', universal_newlines=True, shell=True)", number=1000, globals=globals())
Out[51]: 3.243950183999914
Страница:
...