Hướng dẫn python subprocess powershell multiple commands - python subprocess powershell nhiều lệnh

Đây là tôi đảm nhận nó (mà không cần phải khởi động lại shell và không có gì)

Functions:

• Chuyển hướng đường ống (trong mã ví dụ này bên dưới hoạt động dưới dạng nhật ký, nhưng có thể được sử dụng cho bất kỳ chức năng nào khác)

• Mềm và cứng gần của quá trình phụ

• Người nghe nền

Được thử nghiệm trên Windows OS, nhưng thông thường cũng nên hoạt động trên Linux.

Mã số

import glob
import os
import pathlib
import traceback
import logging
from datetime import datetime
from subprocess import Popen, PIPE, STDOUT
from threading import Thread, Event
from typing import Union
from enum import Enum

LOG_FOLDER = "{0}\{1}".format(pathlib.Path().absolute(), "log")
GDB_FOLDER = "C:\\MinGW\\bin"
GDB_EXE = "gdb.exe"
GDB_PY_EXE = "gdb-python27.exe"
CMD_EXE = "C:\\Windows\\system32\\cmd.exe"
CALC_EXE = "C:\\Windows\\system32\\win32calc.exe"


class LOG_TYPE(Enum):
    Info = 0,
    Warning = 1,
    Error = 2,
    Critical = 3


class STD_LOG(object):
    def __init__(self, name: str, log_enabled: bool = True, print_enabled: bool = True,
                 detailed_log: bool = False) -> None:
        self.log_enabled = log_enabled
        self.print_enabled = print_enabled
        self.filename = "{0}\{1}{2}.log".format(LOG_FOLDER, name,
                                                datetime.now().strftime("-%d_%m_%Y"))  # "-%d_%m_%Y-%H_%M_%S"
        self.logger = logging.getLogger('CLI_LOGGER')
        self.logger.setLevel(logging.DEBUG)
        handler = logging.FileHandler(filename=self.filename, mode="a", encoding="utf-8")
        formatter = logging.Formatter('%(message)s')
        if detailed_log:
            formatter = logging.Formatter('%(asctime)s \t %(name)s \t %(levelname)s: \n%(message)s \n')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def output(self, data: str, logtype: LOG_TYPE = LOG_TYPE.Info):
        if not data:
            return None
        if self.print_enabled:
            print(data)
        if self.log_enabled:
            if logtype == LOG_TYPE.Info:
                self.logger.info(msg=data)
            elif logtype == LOG_TYPE.Warning:
                self.logger.warning(msg=data)
            elif logtype == LOG_TYPE.Error:
                self.logger.error(msg=data)
            else:
                self.logger.critical(msg=data)

        # FOR STACKOVERFLOW Example
        # if api_call activated => run api call from a server
        # if output == "desired result":
        #   Do something
        # Etc.

    def input(self, data: str):
        pass
        # FOR STACKOVERFLOW
        # Perhaps a separate log file?
        # Or redirect to output, etc.
        # self.output(data=data)


# inspiration for killable thread -> https://stackoverflow.com/a/49877671
class CLI_THREAD(Thread):
    def __init__(self, source: str, logger: STD_LOG, sleep_interval: float = 0.25) -> None:
        super().__init__()
        self._close = Event()
        self._interval = sleep_interval
        self.base = Popen(source, stdin=PIPE, stdout=PIPE, stderr=STDOUT, text=True, shell=True, close_fds=True)
        self.logger = logger
        self.logger.output(data="CLI STARTED [Process ID: {0}]".format(self.pid()))

    def cli_alive(self) -> Union[bool, int]:
        if self.base.poll() is None:
            return True
        else:
            return False

    def pid(self) -> int:
        return self.base.pid

    def run(self) -> None:
        while True:
            try:
                if not self.cli_alive():
                    break

                if self.base.stdout.readable():
                    output = self.base.stdout.readline().strip()
                    self.logger.output(data=output)

                closing = self._close.wait(self._interval)
                if closing and self.base.stdout.closed:
                    break
            except Exception as ex:
                ex_msg = ''.join(traceback.format_exception(None, ex, ex.__traceback__))
                self.logger.output(data=ex_msg, logtype=LOG_TYPE.Error)

        self.logger.output(data="End of CLI Thread")

    def close(self) -> None:
        self._close.set()

    def terminate(self) -> None:
        if self.cli_alive():
            self.base.terminate()
            self.logger.output(data="Terminate function activated", logtype=LOG_TYPE.Warning)


class CLI(object):
    def __init__(self, name: str, source: str, close_arg: str = None, echo: bool = True) -> None:
        self.logger = STD_LOG(name)
        self._cli_thread = CLI_THREAD(source=source, logger=self.logger)
        self._close_arg = close_arg
        self._cli_thread.start()  # start listening to console
        if not echo:
            self.execute("@echo off")

    def close(self):
        if self._close_arg:
            self.execute(command=self._close_arg)
        self._cli_thread.close()

    def cleanup(self):
        self._cli_thread.base.terminate()
        del self._cli_thread

    def __exit__(self, exc_type, exc_value, traceback):
        self.cleanup()

    def execute(self, command: str):
        if self._cli_thread.is_alive():
            self._cli_thread.base.stdin.write(command + "\n")
            self._cli_thread.base.stdin.flush()
        else:
            self.logger.output(data="Sending command to CLOSED THREAD", logtype=LOG_TYPE.Error)


def empty_log():
    files = glob.glob("{0}/*".format(LOG_FOLDER))
    for f in files:
        os.remove(f)


def main():
    cli = CLI(name="cli_1", source=CMD_EXE, close_arg="exit")
    try:
        cli.execute(command="cd {0}".format(GDB_FOLDER))
        cli.execute(command=GDB_EXE)
        cli.execute(command="file C:/Windows/system32/win32calc.exe")
        cli.execute(command="quit")
        cli.close()
    except Exception as e:
        msg = ''.join(traceback.format_exception(None, e, e.__traceback__))
        cli.logger.output(data=msg, logtype=LOG_TYPE.Critical)
        cli.cleanup()


if __name__ == '__main__':
    empty_log()
    main()

Đầu ra (cli-somedate.log) (cli-somedate.log)

CLI bắt đầu [ID quy trình: 9720] Microsoft Windows [Phiên bản 10.0.17763.1728] (c) 2018 Microsoft Corporation. Đã đăng ký Bản quyền. C: \ Users \ Quản trị viên \ PyCharmProjects \ ProgramWatcher \ SRC \ GDB> CD C: \ Mingw \ bin C: \ Mingw \ bin> Gdb.exe GNU GDB ( Giấy phép GPLV3+: GPL GPL phiên bản 3 trở lên http://gnu.org/licenses/gpl.html Đây là phần mềm miễn phí: Bạn có thể tự do thay đổi và phân phối lại. Không có bảo hành, trong phạm vi được pháp luật cho phép. Loại "Hiển thị sao chép" và "Hiển thị bảo hành" để biết chi tiết. GDB này được cấu hình là "Mingw32". Để biết hướng dẫn báo cáo lỗi, vui lòng xem: http://www.gnu.org/software/gdb/bugs/. . .
Microsoft Windows [Version 10.0.17763.1728] (c) 2018 Microsoft Corporation. All rights reserved.
C:\Users\Administrator\PycharmProjects\ProgramWatcher\src\GDB>cd C:\MinGW\bin
C:\MinGW\bin>gdb.exe
GNU gdb (GDB) 7.6.1 Copyright (C) 2013 Free Software Foundation, Inc. License GPLv3+: GNU GPL version 3 or later http://gnu.org/licenses/gpl.html
This is free software: you are free to change and redistribute it. There is NO WARRANTY, to the extent permitted by law. Type "show copying" and "show warranty" for details.
This GDB was configured as "mingw32". For bug reporting instructions, please see: http://www.gnu.org/software/gdb/bugs/.
(gdb) Reading symbols from C:\Windows\system32\win32calc.exe...(no debugging symbols found)...done.
(gdb) C:\MinGW\bin>exit
End of CLI Thread

Làm cách nào để chạy nhiều lệnh PowerShell trong Python?

Làm cách nào để chạy nhiều lệnh trong Python?

Sự khác biệt giữa cuộc gọi và chạy quá trình phụ là gì?

Là quy trình phụ chạy đồng bộ?