Uruchamianie polecenia powłoki i przechwytywanie wyjścia

Jest to o wiele łatwiejsze, ale działa tylko na Uniksie (w tym Cygwin) i Python2.7.

import commands
print commands.getstatusoutput('wc -l file')

Zwraca krotkę z (return_value, output).

Dla rozwiązania, które działa zarówno w Python2 jak i Python3, użyj zamiast tego modułu subprocess:

from subprocess import Popen, PIPE
output = Popen(["date"],stdout=PIPE)
response = output.communicate()
print response

Coś w tym stylu:

def runProcess(exe):    
    p = subprocess.Popen(exe, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    while(True):
        # returns None while subprocess is running
        retcode = p.poll() 
        line = p.stdout.readline()
        yield line
        if retcode is not None:
            break

Zauważ, że Przekierowuję stderr na stdout, może to nie jest dokładnie to, czego chcesz, ale chcę również komunikaty o błędach.

Ta funkcja generuje linię po linii, gdy przychodzą (normalnie musiałbyś poczekać na zakończenie podprocesu, aby uzyskać wyjście jako całość).

W Twoim przypadku użycie byłoby:

for line in runProcess('mysqladmin create test -uroot -pmysqladmin12'.split()):
    print line,

Odpowiedź Varteca nie odczytuje wszystkich linijek, więc zrobiłem wersję, która zrobiła:

def run_command(command):
    p = subprocess.Popen(command,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.STDOUT)
    return iter(p.stdout.readline, b'')

Użycie jest takie samo jak zaakceptowana odpowiedź:

command = 'mysqladmin create test -uroot -pmysqladmin12'.split()
for line in run_command(command):
    print(line)

To jest trudne ale super proste rozwiązanie, które sprawdza się w wielu sytuacjach:

import os
os.system('sample_cmd > tmp')
print open('tmp', 'r').read()

Plik tymczasowy (tutaj jest tmp) jest tworzony z wyjściem polecenia i można z niego odczytać żądane wyjście.

Extra notka z komentarzy: Plik tmp można usunąć w przypadku zadania jednorazowego. Jeśli musisz to zrobić kilka razy, nie ma potrzeby usuwania tmp.

os.remove('tmp')

Miałem ten sam problem, ale wymyśliłem bardzo prosty sposób na zrobienie tego:

import subprocess
output = subprocess.getoutput("ls -l")
print(output)

Hope it helps out

Uwaga: To rozwiązanie jest specyficzne dla Python3, ponieważ subprocess.getoutput() nie działa w Python2

Możesz użyć następujących poleceń do uruchomienia dowolnego polecenia powłoki. Używałem ich na ubuntu.

import os
os.popen('your command here').read()

Uwaga: jest to przestarzałe od wersji Pythona 2.6. Teraz musisz użyć subprocess.Popen. Poniżej znajduje się przykład

import subprocess

p = subprocess.Popen("Your command", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()[0]
print p.split("\n")

Twój przebieg może się różnić, próbowałem spin @ senderle na rozwiązaniu Vartec w Windows na Pythonie 2.6.5, ale dostawałem błędy, i żadne inne rozwiązania nie działały. Mój błąd to: WindowsError: [Error 6] The handle is invalid.

Odkryłem, że musiałem przypisać PIPE do każdego uchwytu, aby zwrócił oczekiwane wyjście - poniższe zadziałało dla mnie.

import subprocess

def run_command(cmd):
    """given shell command, returns communication tuple of stdout and stderr"""
    return subprocess.Popen(cmd, 
                            stdout=subprocess.PIPE, 
                            stderr=subprocess.PIPE, 
                            stdin=subprocess.PIPE).communicate()

I wywołanie w ten sposób, ([0] otrzymuje pierwszy element krotki, stdout):

run_command('tracert 11.1.0.1')[0]

Po dowiedzeniu się więcej, uważam, że potrzebuję tych argumentów, ponieważ Pracuję nad niestandardowym systemem, który używa różnych uchwytów, więc musiałem bezpośrednio kontrolować wszystkie choroby weneryczne.]}

Aby zatrzymać wyskakujące okienka konsoli (w systemie Windows), wykonaj następujące czynności:

def run_command(cmd):
    """given shell command, returns communication tuple of stdout and stderr"""
    # instantiate a startupinfo obj:
    startupinfo = subprocess.STARTUPINFO()
    # set the use show window flag, might make conditional on being in Windows:
    startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    # pass as the startupinfo keyword argument:
    return subprocess.Popen(cmd,
                            stdout=subprocess.PIPE, 
                            stderr=subprocess.PIPE, 
                            stdin=subprocess.PIPE, 
                            startupinfo=startupinfo).communicate()

run_command('tracert 11.1.0.1')

Miałem nieco inny smak tego samego problemu z następującymi wymaganiami:

1. przechwytywanie i zwracanie wiadomości STDOUT, gdy gromadzą się one w buforze STDOUT (np. w czasie rzeczywistym).
   • @vartec rozwiązał ten problem używając generatorów i "wydajności"
     słowo kluczowe powyżej
2. wypisuje wszystkie linie Wyjścia ( nawet jeśli proces zakończy działanie przed buforem wyjścia może być w pełni odczytany )
Nie marnuj cykli procesora badanie procesu przy wysokiej częstotliwości
3. Sprawdź kod powrotu podprocesu
4. Print STDERR (oddzielony od STDOUT) jeśli otrzymamy niezerowy kod zwracający błąd.

Połączyłem i poprawiłem poprzednie odpowiedzi, aby uzyskać następujące odpowiedzi:

import subprocess
from time import sleep

def run_command(command):
    p = subprocess.Popen(command,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         shell=True)
    # Read stdout from subprocess until the buffer is empty !
    for line in iter(p.stdout.readline, b''):
        if line: # Don't print blank lines
            yield line
    # This ensures the process has completed, AND sets the 'returncode' attr
    while p.poll() is None:                                                                                                                                        
        sleep(.1) #Don't waste CPU-cycles
    # Empty STDERR buffer
    err = p.stderr.read()
    if p.returncode != 0:
       # The run_command() function is responsible for logging STDERR 
       print("Error: " + str(err))

Ten kod będzie wykonywany tak samo jak poprzednie odpowiedzi:

for line in run_command(cmd):
    print(line)

Rozdzielenie komendy początkowej dla subprocess może być trudne i uciążliwe.

Użyj shlex.split(), aby pomóc sobie.

Przykładowe polecenie

git log -n 5 --since "5 years ago" --until "2 year ago"

Kod

from subprocess import check_output
from shlex import split

res = check_output(split('git log -n 5 --since "5 years ago" --until "2 year ago"'))
print(res)
>>> b'commit 7696ab087a163e084d6870bb4e5e4d4198bdc61a\nAuthor: Artur Barseghyan...'

BEZ shlex.split() kod wyglądałby następująco

res = check_output([
    'git', 
    'log', 
    '-n', 
    '5', 
    '--since', 
    '5 years ago', 
    '--until', 
    '2 year ago'
])
print(res)
>>> b'commit 7696ab087a163e084d6870bb4e5e4d4198bdc61a\nAuthor: Artur Barseghyan...'

Jeśli musisz uruchomić polecenie powłoki na wielu plikach, To mi się udało.

import os
import subprocess

# Define a function for running commands and capturing stdout line by line
# (Modified from Vartec's solution because it wasn't printing all lines)
def runProcess(exe):    
    p = subprocess.Popen(exe, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    return iter(p.stdout.readline, b'')

# Get all filenames in working directory
for filename in os.listdir('./'):
    # This command will be run on each file
    cmd = 'nm ' + filename

    # Run the command and capture the output line by line.
    for line in runProcess(cmd.split()):
        # Eliminate leading and trailing whitespace
        line.strip()
        # Split the output 
        output = line.split()

        # Filter the output and print relevant lines
        if len(output) > 2:
            if ((output[2] == 'set_program_name')):
                print filename
                print line

Edit: właśnie zobaczyłem rozwiązanie Maxa Perssona z sugestią J. F. Sebastiana. / Align = "left" /

Według @senderle, jeśli używasz python3. 6 Jak ja:

def sh(cmd, input=""):
    rst = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, input=input.encode("utf-8"))
    assert rst.returncode == 0, rst.stderr.decode("utf-8")
    return rst.stdout.decode("utf-8")

sh("ls -a")

Będzie działać dokładnie tak, jak uruchomisz polecenie w bash

Tutaj rozwiązanie, działające, jeśli chcesz wydrukować wyjście, gdy proces jest uruchomiony lub nie.

Dodałem również bieżący katalog roboczy, przydał mi się nie raz.

Mam nadzieję, że rozwiązanie komuś pomoże:).

import subprocess

def run_command(cmd_and_args, print_constantly=False, cwd=None):
"""Runs a system command.

:param cmd_and_args: the command to run with or without a Pipe (|).
:param print_constantly: If True then the output is logged in continuous until the command ended.
:param cwd: the current working directory (the directory from which you will like to execute the command)
:return: - a tuple containing the return code, the stdout and the stderr of the command
"""
output = []

process = subprocess.Popen(cmd_and_args, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd)

while True:
    next_line = process.stdout.readline()
    if next_line:
        output.append(str(next_line))
        if print_constantly:
            print(next_line)
    elif not process.poll():
        break

error = process.communicate()[1]

return process.returncode, '\n'.join(output), error

Jeśli używasz modułu Pythona subprocess, możesz obsługiwać STDOUT, STDERR i Kod polecenia osobno. Możesz zobaczyć przykład pełnej implementacji wywołującego polecenia. Oczywiście możesz go rozszerzyć za pomocą try..except, jeśli chcesz.

Poniższa funkcja zwraca kod STDOUT, STDERR I Return, abyś mógł obsłużyć je w innym skrypcie.

import subprocess

def command_caller(command=None)
    sp = subprocess.Popen(command, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=False)
    out, err = sp.communicate()
    if sp.returncode:
        print(
            "Return code: %(ret_code)s Error message: %(err_msg)s"
            % {"ret_code": sp.returncode, "err_msg": err}
            )
    return sp.returncode, out, err

Eg, execute ('ls-ahl') zróżnicowane trzy / cztery możliwe zwroty i platformy OS:

1. brak wyjścia, ale Uruchom pomyślnie
2. wyjście pustej linii, Uruchom pomyślnie
3. run failed
4. wypuść coś, Uruchom pomyślnie

Funkcja poniżej

def execute(cmd, output=True, DEBUG_MODE=False):
"""Executes a bash command.
(cmd, output=True)
output: whether print shell output to screen, only affects screen display, does not affect returned values
return: ...regardless of output=True/False...
        returns shell output as a list with each elment is a line of string (whitespace stripped both sides) from output
        could be 
        [], ie, len()=0 --> no output;    
        [''] --> output empty line;     
        None --> error occured, see below

        if error ocurs, returns None (ie, is None), print out the error message to screen
"""
if not DEBUG_MODE:
    print "Command: " + cmd

    # https://stackoverflow.com/a/40139101/2292993
    def _execute_cmd(cmd):
        if os.name == 'nt' or platform.system() == 'Windows':
            # set stdin, out, err all to PIPE to get results (other than None) after run the Popen() instance
            p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
        else:
            # Use bash; the default is sh
            p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, executable="/bin/bash")

        # the Popen() instance starts running once instantiated (??)
        # additionally, communicate(), or poll() and wait process to terminate
        # communicate() accepts optional input as stdin to the pipe (requires setting stdin=subprocess.PIPE above), return out, err as tuple
        # if communicate(), the results are buffered in memory

        # Read stdout from subprocess until the buffer is empty !
        # if error occurs, the stdout is '', which means the below loop is essentially skipped
        # A prefix of 'b' or 'B' is ignored in Python 2; 
        # it indicates that the literal should become a bytes literal in Python 3 
        # (e.g. when code is automatically converted with 2to3).
        # return iter(p.stdout.readline, b'')
        for line in iter(p.stdout.readline, b''):
            # # Windows has \r\n, Unix has \n, Old mac has \r
            # if line not in ['','\n','\r','\r\n']: # Don't print blank lines
                yield line
        while p.poll() is None:                                                                                                                                        
            sleep(.1) #Don't waste CPU-cycles
        # Empty STDERR buffer
        err = p.stderr.read()
        if p.returncode != 0:
            # responsible for logging STDERR 
            print("Error: " + str(err))
            yield None

    out = []
    for line in _execute_cmd(cmd):
        # error did not occur earlier
        if line is not None:
            # trailing comma to avoid a newline (by print itself) being printed
            if output: print line,
            out.append(line.strip())
        else:
            # error occured earlier
            out = None
    return out
else:
    print "Simulation! The command is " + cmd
    print ""

Wyjście może zostać przekierowane do pliku tekstowego, a następnie odczytane z powrotem.

import subprocess
import os
import tempfile

def execute_to_file(command):
    """
    This function execute the command
    and pass its output to a tempfile then read it back
    It is usefull for process that deploy child process
    """
    temp_file = tempfile.NamedTemporaryFile(delete=False)
    temp_file.close()
    path = temp_file.name
    command = command + " > " + path
    proc = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
    if proc.stderr:
        # if command failed return
        os.unlink(path)
        return
    with open(path, 'r') as f:
        data = f.read()
    os.unlink(path)
    return data

if __name__ == "__main__":
    path = "Somepath"
    command = 'ecls.exe /files ' + path
    print(execute(command))

Właśnie napisałem mały skrypt bash, aby to zrobić za pomocą curl

Https://gist.github.com/harish2704/bfb8abece94893c53ce344548ead8ba5

#!/usr/bin/env bash

# Usage: gdrive_dl.sh <url>

urlBase='https://drive.google.com'
fCookie=tmpcookies

curl="curl -L -b $fCookie -c $fCookie"
confirm(){
    $curl "$1" | grep jfk-button-action | sed -e 's/.*jfk-button-action" href="\(\S*\)".*/\1/' -e 's/\&amp;/\&/g'
}

$curl -O -J "${urlBase}$(confirm $1)"

Chciałbym zaproponować simppl jako opcję do rozważenia. Jest to moduł, który jest dostępny przez pypi: pip install simppl i został uruchomiony na python3.

simppl pozwala użytkownikowi uruchamiać polecenia powłoki i odczytywać dane wyjściowe z ekranu.

Twórcy sugerują trzy rodzaje przypadków użycia:

1. najprostsze użycie będzie wyglądało tak:

    from simppl.simple_pipeline import SimplePipeline
    sp = SimplePipeline(start=0, end=100):
    sp.print_and_run('<YOUR_FIRST_OS_COMMAND>')
    sp.print_and_run('<YOUR_SECOND_OS_COMMAND>') ```

1. aby uruchomić wiele poleceń jednocześnie użycie:

    commands = ['<YOUR_FIRST_OS_COMMAND>', '<YOUR_SECOND_OS_COMMAND>']
    max_number_of_processes = 4
    sp.run_parallel(commands, max_number_of_processes) ```

1. wreszcie, jeśli twój projekt używa modułu cli, możesz uruchomić bezpośrednio inny command_line_tool jako część potoku. Inne narzędzie będzie być uruchamiany z tego samego procesu, ale będzie on wyświetlany z dzienników jako kolejne polecenie w rurociągu. Umożliwia to płynniejsze debugowanie i refaktoryzacja narzędzi wywołanie innych narzędzi.

    from example_module import example_tool
    sp.print_and_run_clt(example_tool.run, ['first_number', 'second_nmber'], 
                                 {'-key1': 'val1', '-key2': 'val2'},
                                 {'--flag'}) ```

Zauważ, że drukowanie na STDOUT/STDERR odbywa się za pomocą modułu logging Pythona.

Oto Pełny kod pokazujący jak działa simppl:

import logging
from logging.config import dictConfig

logging_config = dict(
    version = 1,
    formatters = {
        'f': {'format':
              '%(asctime)s %(name)-12s %(levelname)-8s %(message)s'}
        },
    handlers = {
        'h': {'class': 'logging.StreamHandler',
              'formatter': 'f',
              'level': logging.DEBUG}
        },
    root = {
        'handlers': ['h'],
        'level': logging.DEBUG,
        },
)
dictConfig(logging_config)

from simppl.simple_pipeline import SimplePipeline
sp = SimplePipeline(0, 100)
sp.print_and_run('ls')