草庐IT

python - 通过 subprocess.check_output 调用的可执行文件打印在控制台上但不返回结果

coder 2024-06-12 原文

在 Windows 机器上,我试图从 Python 调用外部可执行文件并收集其输出以进行进一步处理。因为在调用可执行文件之前必须设置本地路径变量,所以我创建了一个批处理脚本

  • 首先调用另一个脚本来设置 %PATH% 和
  • 然后使用给定的参数调用可执行文件。

*.bat 文件如下所示:

@echo off
call set_path.bat
@echo on
executable.exe %*

Python 代码如下:

print("before call");
result = subprocess.check_output([batfile, parameters], stderr=subprocess.STDOUT, shell=True);
print("after call");

print("------- ------- ------- printing result ------- ------- ------- ");
print(result);
print("------- ------- ------- /printing result ------- ------- ------- ");

现在,从技术上讲,这是可行的。可执行文件使用预期参数调用、运行、完成并产生结果。我知道这一点,因为它们被 mock 地显示在运行 Python 脚本的控制台中。

但是,结果字符串只包含批处理脚本返回的内容,而不是可执行文件的输出:

before call

hello? yes, this is executable.exe

after call

------- ------- ------- printing result ------- ------- -------

C:\Users\me\Documents\pythonscript\execute\executable.exe "para1|para2|para3"

------- ------- ------- /printing result ------- ------- -------

subprocess.check_output 命令本身以某种方式将预期输出打印到控制台,它返回的内容仅包含@echo 再次打开后批处理文件的输出。

如何访问可执行文件的输出并将其保存到字符串以供进一步工作?

或者我是否必须以某种方式修改批处理文件以捕获并打印输出,以便它最终出现在 check_output 的结果中?如果是这样,我该怎么做呢?

最佳答案

如果程序直接写入控制台(例如,通过打开 CONOUT$ 设备)而不是进程标准句柄,唯一的选择是直接读取控制台屏幕缓冲区。为了使这更简单,从一个新的空屏幕缓冲区开始。通过以下函数创建、调整大小、初始化和激活新的屏幕缓冲区:

确保请求 GENERIC_READ | GENERIC_WRITE 在调用 CreateConsoleScreenBuffer 时访问。稍后您将需要读取权限才能读取屏幕内容。

专门针对 Python,使用 ctypes在 Windows 控制台 API 中调用函数。此外,如果通过 msvcrt.open_osfhandle 用 C 文件描述符包装句柄,则可以将其作为 stdoutstderr 参数传递subprocess.Popen

屏幕缓冲区的文件描述符或句柄无法通过readReadFile 甚至ReadConsole 直接读取。如果您有文件描述符,请通过 msvcrt.get_osfhandle 获取底层句柄。给定屏幕缓冲区句柄,调用 ReadConsoleOutputCharacter从屏幕上阅读。下面示例代码中的 read_screen 函数演示了从屏幕缓冲区的开头到光标位置的读取。

进程需要附加到控制台才能使用控制台 API。为此,我包含了一个简单的 allocate_console 上下文管理器来临时打开一个控制台。这在通常不附加到控制台的 GUI 应用程序中很有用。

以下示例在 Windows 7 和 10、Python 2.7 和 3.5 中进行了测试。

ctypes 定义

import os
import contextlib
import msvcrt
import ctypes
from ctypes import wintypes

kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)

GENERIC_READ  = 0x80000000
GENERIC_WRITE = 0x40000000
FILE_SHARE_READ  = 1
FILE_SHARE_WRITE = 2
CONSOLE_TEXTMODE_BUFFER = 1
INVALID_HANDLE_VALUE = wintypes.HANDLE(-1).value
STD_OUTPUT_HANDLE = wintypes.DWORD(-11)
STD_ERROR_HANDLE = wintypes.DWORD(-12)

def _check_zero(result, func, args):
    if not result:
        raise ctypes.WinError(ctypes.get_last_error())
    return args

def _check_invalid(result, func, args):
    if result == INVALID_HANDLE_VALUE:
        raise ctypes.WinError(ctypes.get_last_error())
    return args

if not hasattr(wintypes, 'LPDWORD'): # Python 2
    wintypes.LPDWORD = ctypes.POINTER(wintypes.DWORD)
    wintypes.PSMALL_RECT = ctypes.POINTER(wintypes.SMALL_RECT)

class COORD(ctypes.Structure):
    _fields_ = (('X', wintypes.SHORT),
                ('Y', wintypes.SHORT))

class CONSOLE_SCREEN_BUFFER_INFOEX(ctypes.Structure):
    _fields_ = (('cbSize',               wintypes.ULONG),
                ('dwSize',               COORD),
                ('dwCursorPosition',     COORD),
                ('wAttributes',          wintypes.WORD),
                ('srWindow',             wintypes.SMALL_RECT),
                ('dwMaximumWindowSize',  COORD),
                ('wPopupAttributes',     wintypes.WORD),
                ('bFullscreenSupported', wintypes.BOOL),
                ('ColorTable',           wintypes.DWORD * 16))
    def __init__(self, *args, **kwds):
        super(CONSOLE_SCREEN_BUFFER_INFOEX, self).__init__(
                *args, **kwds)
        self.cbSize = ctypes.sizeof(self)

PCONSOLE_SCREEN_BUFFER_INFOEX = ctypes.POINTER(
                                    CONSOLE_SCREEN_BUFFER_INFOEX)
LPSECURITY_ATTRIBUTES = wintypes.LPVOID

kernel32.GetStdHandle.errcheck = _check_invalid
kernel32.GetStdHandle.restype = wintypes.HANDLE
kernel32.GetStdHandle.argtypes = (
    wintypes.DWORD,) # _In_ nStdHandle

kernel32.CreateConsoleScreenBuffer.errcheck = _check_invalid
kernel32.CreateConsoleScreenBuffer.restype = wintypes.HANDLE
kernel32.CreateConsoleScreenBuffer.argtypes = (
    wintypes.DWORD,        # _In_       dwDesiredAccess
    wintypes.DWORD,        # _In_       dwShareMode
    LPSECURITY_ATTRIBUTES, # _In_opt_   lpSecurityAttributes
    wintypes.DWORD,        # _In_       dwFlags
    wintypes.LPVOID)       # _Reserved_ lpScreenBufferData

kernel32.GetConsoleScreenBufferInfoEx.errcheck = _check_zero
kernel32.GetConsoleScreenBufferInfoEx.argtypes = (
    wintypes.HANDLE,               # _In_  hConsoleOutput
    PCONSOLE_SCREEN_BUFFER_INFOEX) # _Out_ lpConsoleScreenBufferInfo

kernel32.SetConsoleScreenBufferInfoEx.errcheck = _check_zero
kernel32.SetConsoleScreenBufferInfoEx.argtypes = (
    wintypes.HANDLE,               # _In_  hConsoleOutput
    PCONSOLE_SCREEN_BUFFER_INFOEX) # _In_  lpConsoleScreenBufferInfo

kernel32.SetConsoleWindowInfo.errcheck = _check_zero
kernel32.SetConsoleWindowInfo.argtypes = (
    wintypes.HANDLE,      # _In_ hConsoleOutput
    wintypes.BOOL,        # _In_ bAbsolute
    wintypes.PSMALL_RECT) # _In_ lpConsoleWindow

kernel32.FillConsoleOutputCharacterW.errcheck = _check_zero
kernel32.FillConsoleOutputCharacterW.argtypes = (
    wintypes.HANDLE,  # _In_  hConsoleOutput
    wintypes.WCHAR,   # _In_  cCharacter
    wintypes.DWORD,   # _In_  nLength
    COORD,            # _In_  dwWriteCoord
    wintypes.LPDWORD) # _Out_ lpNumberOfCharsWritten

kernel32.ReadConsoleOutputCharacterW.errcheck = _check_zero
kernel32.ReadConsoleOutputCharacterW.argtypes = (
    wintypes.HANDLE,  # _In_  hConsoleOutput
    wintypes.LPWSTR,  # _Out_ lpCharacter
    wintypes.DWORD,   # _In_  nLength
    COORD,            # _In_  dwReadCoord
    wintypes.LPDWORD) # _Out_ lpNumberOfCharsRead

函数

@contextlib.contextmanager
def allocate_console():
    allocated = kernel32.AllocConsole()
    try:
        yield allocated
    finally:
        if allocated:
            kernel32.FreeConsole()

@contextlib.contextmanager
def console_screen(ncols=None, nrows=None):
    info = CONSOLE_SCREEN_BUFFER_INFOEX()
    new_info = CONSOLE_SCREEN_BUFFER_INFOEX()
    nwritten = (wintypes.DWORD * 1)()
    hStdOut = kernel32.GetStdHandle(STD_OUTPUT_HANDLE)
    kernel32.GetConsoleScreenBufferInfoEx(
           hStdOut, ctypes.byref(info))
    if ncols is None:
        ncols = info.dwSize.X
    if nrows is None:
        nrows = info.dwSize.Y
    elif nrows > 9999:
        raise ValueError('nrows must be 9999 or less')
    fd_screen = None
    hScreen = kernel32.CreateConsoleScreenBuffer(
                GENERIC_READ | GENERIC_WRITE,
                FILE_SHARE_READ | FILE_SHARE_WRITE,
                None, CONSOLE_TEXTMODE_BUFFER, None)
    try:
        fd_screen = msvcrt.open_osfhandle(
                        hScreen, os.O_RDWR | os.O_BINARY)
        kernel32.GetConsoleScreenBufferInfoEx(
               hScreen, ctypes.byref(new_info))
        new_info.dwSize = COORD(ncols, nrows)
        new_info.srWindow = wintypes.SMALL_RECT(
                Left=0, Top=0, Right=(ncols - 1),
                Bottom=(info.srWindow.Bottom - info.srWindow.Top))
        kernel32.SetConsoleScreenBufferInfoEx(
                hScreen, ctypes.byref(new_info))
        kernel32.SetConsoleWindowInfo(hScreen, True,
                ctypes.byref(new_info.srWindow))
        kernel32.FillConsoleOutputCharacterW(
                hScreen, u'\0', ncols * nrows, COORD(0,0), nwritten)
        kernel32.SetConsoleActiveScreenBuffer(hScreen)
        try:
            yield fd_screen
        finally:
            kernel32.SetConsoleScreenBufferInfoEx(
                hStdOut, ctypes.byref(info))
            kernel32.SetConsoleWindowInfo(hStdOut, True,
                    ctypes.byref(info.srWindow))
            kernel32.SetConsoleActiveScreenBuffer(hStdOut)
    finally:
        if fd_screen is not None:
            os.close(fd_screen)
        else:
            kernel32.CloseHandle(hScreen)

def read_screen(fd):
    hScreen = msvcrt.get_osfhandle(fd)
    csbi = CONSOLE_SCREEN_BUFFER_INFOEX()
    kernel32.GetConsoleScreenBufferInfoEx(
        hScreen, ctypes.byref(csbi))
    ncols = csbi.dwSize.X
    pos = csbi.dwCursorPosition
    length = ncols * pos.Y + pos.X + 1
    buf = (ctypes.c_wchar * length)()
    n = (wintypes.DWORD * 1)()
    kernel32.ReadConsoleOutputCharacterW(
        hScreen, buf, length, COORD(0,0), n)
    lines = [buf[i:i+ncols].rstrip(u'\0')
                for i in range(0, n[0], ncols)]
    return u'\n'.join(lines)

例子

if __name__ == '__main__':
    import io
    import textwrap
    import subprocess

    text = textwrap.dedent('''\
        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do
        eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut
        enim ad minim veniam, quis nostrud exercitation ullamco laboris
        nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor
        in reprehenderit in voluptate velit esse cillum dolore eu
        fugiat nulla pariatur. Excepteur sint occaecat cupidatat non
        proident, sunt in culpa qui officia deserunt mollit anim id est
        laborum.''')

    cmd = ("python -c \""
           "print('piped output');"
           "conout = open(r'CONOUT$', 'w');"
           "conout.write('''%s''')\"" % text)

    with allocate_console() as allocated:
        with console_screen(nrows=1000) as fd_conout:
            stdout = subprocess.check_output(cmd).decode()
            conout = read_screen(fd_conout)
            with io.open('result.txt', 'w', encoding='utf-8') as f:
                f.write(u'stdout:\n' + stdout)
                f.write(u'\nconout:\n' + conout)

输出

stdout:
piped output

conout:
Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do
eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut
enim ad minim veniam, quis nostrud exercitation ullamco laboris
nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor
in reprehenderit in voluptate velit esse cillum dolore eu
fugiat nulla pariatur. Excepteur sint occaecat cupidatat non
proident, sunt in culpa qui officia deserunt mollit anim id est
laborum.

关于python - 通过 subprocess.check_output 调用的可执行文件打印在控制台上但不返回结果,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/38698383/

有关python - 通过 subprocess.check_output 调用的可执行文件打印在控制台上但不返回结果的更多相关文章

  1. ruby - 使用 RubyZip 生成 ZIP 文件时设置压缩级别 - 2

    我有一个Ruby程序,它使用rubyzip压缩XML文件的目录树。gem。我的问题是文件开始变得很重,我想提高压缩级别,因为压缩时间不是问题。我在rubyzipdocumentation中找不到一种为创建的ZIP文件指定压缩级别的方法。有人知道如何更改此设置吗?是否有另一个允许指定压缩级别的Ruby库? 最佳答案 这是我通过查看ruby​​zip内部创建的代码。level=Zlib::BEST_COMPRESSIONZip::ZipOutputStream.open(zip_file)do|zip|Dir.glob("**/*")d

  2. ruby - 其他文件中的 Rake 任务 - 2

    我试图在一个项目中使用rake,如果我把所有东西都放到Rakefile中,它会很大并且很难读取/找到东西,所以我试着将每个命名空间放在lib/rake中它自己的文件中,我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题,但没有任务。我现在只有一个.rake文件作为测试,名为“servers.rake”,它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时

  3. ruby-on-rails - 在 Rails 中将文件大小字符串转换为等效千字节 - 2

    我的目标是转换表单输入,例如“100兆字节”或“1GB”,并将其转换为我可以存储在数据库中的文件大小(以千字节为单位)。目前,我有这个:defquota_convert@regex=/([0-9]+)(.*)s/@sizes=%w{kilobytemegabytegigabyte}m=self.quota.match(@regex)if@sizes.include?m[2]eval("self.quota=#{m[1]}.#{m[2]}")endend这有效,但前提是输入是倍数(“gigabytes”,而不是“gigabyte”)并且由于使用了eval看起来疯狂不安全。所以,功能正常,

  4. python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声? - 2

    关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题?更新问题,以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点?我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等),但我确实想创建一个输出文件。

  5. ruby-on-rails - Rails 3 中的多个路由文件 - 2

    Rails2.3可以选择随时使用RouteSet#add_configuration_file添加更多路由。是否可以在Rails3项目中做同样的事情? 最佳答案 在config/application.rb中:config.paths.config.routes在Rails3.2(也可能是Rails3.1)中,使用:config.paths["config/routes"] 关于ruby-on-rails-Rails3中的多个路由文件,我们在StackOverflow上找到一个类似的问题

  6. ruby-openid:执行发现时未设置@socket - 2

    我在使用omniauth/openid时遇到了一些麻烦。在尝试进行身份验证时,我在日志中发现了这一点:OpenID::FetchingError:Errorfetchinghttps://www.google.com/accounts/o8/.well-known/host-meta?hd=profiles.google.com%2Fmy_username:undefinedmethod`io'fornil:NilClass重要的是undefinedmethodio'fornil:NilClass来自openid/fetchers.rb,在下面的代码片段中:moduleNetclass

  7. ruby - 将差异补丁应用于字符串/文件 - 2

    对于具有离线功能的智能手机应用程序,我正在为Xml文件创建单向文本同步。我希望我的服务器将增量/差异(例如GNU差异补丁)发送到目标设备。这是计划:Time=0Server:hasversion_1ofXmlfile(~800kiB)Client:hasversion_1ofXmlfile(~800kiB)Time=1Server:hasversion_1andversion_2ofXmlfile(each~800kiB)computesdeltaoftheseversions(=patch)(~10kiB)sendspatchtoClient(~10kiBtransferred)Cl

  8. ruby - 如何将脚本文件的末尾读取为数据文件(Perl 或任何其他语言) - 2

    我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚

  9. ruby - 通过 rvm 升级 ruby​​gems 的问题 - 2

    尝试通过RVM将RubyGems升级到版本1.8.10并出现此错误:$rvmrubygemslatestRemovingoldRubygemsfiles...Installingrubygems-1.8.10forruby-1.9.2-p180...ERROR:Errorrunning'GEM_PATH="/Users/foo/.rvm/gems/ruby-1.9.2-p180:/Users/foo/.rvm/gems/ruby-1.9.2-p180@global:/Users/foo/.rvm/gems/ruby-1.9.2-p180:/Users/foo/.rvm/gems/rub

  10. ruby - 使用 Vim Rails,您可以创建一个新的迁移文件并一次性打开它吗? - 2

    使用带有Rails插件的vim,您可以创建一个迁移文件,然后一次性打开该文件吗?textmate也可以这样吗? 最佳答案 你可以使用rails.vim然后做类似的事情::Rgeneratemigratonadd_foo_to_bar插件将打开迁移生成的文件,这正是您想要的。我不能代表textmate。 关于ruby-使用VimRails,您可以创建一个新的迁移文件并一次性打开它吗?,我们在StackOverflow上找到一个类似的问题: https://sta

随机推荐