如果程序直接写入控制台(例如,通过打开
CONOUT$设备)而不是写入过程标准句柄,则唯一的选择是直接读取控制台屏幕缓冲区。为了简化此过程,请从新的空白屏幕缓冲区开始。通过以下功能创建,调整大小,初始化和激活新的屏幕缓冲区:
CreateConsoleScreenBuffer
GetConsoleScreenBufferInfoEx
(支持的最低客户端:Windows Vista)SetConsoleScreenBufferInfoEx
(支持的最低客户端:Windows Vista)SetConsoleWindowInfo
FillConsoleOutputCharacter
SetConsoleActiveScreenBuffer
请确保
GENERIC_READ |GENERIC_WRITE在致电时请求访问权限
CreateConsoleScreenBuffer。您稍后需要读取权限才能读取屏幕内容。
专门针对Python,使用ctypes调用Windows控制台API中的函数。另外,如果您通过C将文件句柄包装为C文件描述符
msvcrt.open_osfhandle,则可以将其作为
stdout或
stderr参数传递
subprocess.Popen。
文件描述符或手柄屏幕缓冲区无法通过直接读取
read,
ReadFile或甚
ReadConsole。如果您有文件描述符,请通过获取基本句柄
msvcrt.get_osfhandle。给定屏幕缓冲区句柄,调用
ReadConsoleOutputCharacter以从屏幕读取。
read_screen下面的示例代码中的函数演示了从屏幕缓冲区的开始到光标位置的读取。
为了使用控制台API,需要将进程附加到控制台。为此,我提供了一个简单的
allocate_console上下文管理器来临时打开控制台。这在GUI应用程序中很有用,而GUI应用程序通常未连接到控制台。
以下示例在Windows 7和10,Python 2.7和3.5中进行了测试。
ctypes定义
import osimport contextlibimport msvcrtimport ctypesfrom ctypes import wintypeskernel32 = ctypes.WinDLL('kernel32', use_last_error=True)GENERIC_READ = 0x80000000GENERIC_WRITE = 0x40000000FILE_SHARE_READ = 1FILE_SHARE_WRITE = 2CONSOLE_TEXTMODE_BUFFER = 1INVALID_HANDLE_VALUE = wintypes.HANDLE(-1).valueSTD_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 argsdef _check_invalid(result, func, args): if result == INVALID_HANDLE_VALUE: raise ctypes.WinError(ctypes.get_last_error()) return argsif 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.LPVOIDkernel32.GetStdHandle.errcheck = _check_invalidkernel32.GetStdHandle.restype = wintypes.HANDLEkernel32.GetStdHandle.argtypes = ( wintypes.DWORD,) # _In_ nStdHandlekernel32.CreateConsoleScreenBuffer.errcheck = _check_invalidkernel32.CreateConsoleScreenBuffer.restype = wintypes.HANDLEkernel32.CreateConsoleScreenBuffer.argtypes = ( wintypes.DWORD, # _In_ dwDesiredAccess wintypes.DWORD, # _In_ dwShareMode LPSECURITY_ATTRIBUTES, # _In_opt_ lpSecurityAttributes wintypes.DWORD, # _In_ dwFlags wintypes.LPVOID) # _Reserved_ lpScreenBufferDatakernel32.GetConsoleScreenBufferInfoEx.errcheck = _check_zerokernel32.GetConsoleScreenBufferInfoEx.argtypes = ( wintypes.HANDLE, # _In_ hConsoleOutput PCONSOLE_SCREEN_BUFFER_INFOEX) # _Out_ lpConsoleScreenBufferInfokernel32.SetConsoleScreenBufferInfoEx.errcheck = _check_zerokernel32.SetConsoleScreenBufferInfoEx.argtypes = ( wintypes.HANDLE, # _In_ hConsoleOutput PCONSOLE_SCREEN_BUFFER_INFOEX) # _In_ lpConsoleScreenBufferInfokernel32.SetConsoleWindowInfo.errcheck = _check_zerokernel32.SetConsoleWindowInfo.argtypes = ( wintypes.HANDLE, # _In_ hConsoleOutput wintypes.BOOL, # _In_ bAbsolute wintypes.PSMALL_RECT) # _In_ lpConsoleWindowkernel32.FillConsoleOutputCharacterW.errcheck = _check_zerokernel32.FillConsoleOutputCharacterW.argtypes = ( wintypes.HANDLE, # _In_ hConsoleOutput wintypes.WCHAR, # _In_ cCharacter wintypes.DWORD, # _In_ nLength COORD, # _In_ dwWriteCoord wintypes.LPDWORD) # _Out_ lpNumberOfCharsWrittenkernel32.ReadConsoleOutputCharacterW.errcheck = _check_zerokernel32.ReadConsoleOutputCharacterW.argtypes = ( wintypes.HANDLE, # _In_ hConsoleOutput wintypes.LPWSTR, # _Out_ lpCharacter wintypes.DWORD, # _In_ nLength COORD, # _In_ dwReadCoord wintypes.LPDWORD) # _Out_ lpNumberOfCharsRead职能
@contextlib.contextmanagerdef allocate_console(): allocated = kernel32.AllocConsole() try: yield allocated finally: if allocated: kernel32.FreeConsole()@contextlib.contextmanagerdef 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' ', 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' ') 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).depre() 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 outputconout:Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed doeiusmod tempor incididunt ut labore et dolore magna aliqua. Utenim ad minim veniam, quis nostrud exercitation ullamco laborisnisi ut aliquip ex ea commodo consequat. Duis aute irure dolorin reprehenderit in voluptate velit esse cillum dolore eufugiat nulla pariatur. Excepteur sint occaecat cupidatat nonproident, sunt in culpa qui officia deserunt mollit anim id estlaborum.



