无法获取从Python运行的fcoder 2PDF转换过程的进度

Question

我试图通过从 Python 子进程运行 fcoder 2 PDF 命令行工具输出来获取进度。我已成功捕获其他行，但未捕获显示进度的最后两行。这个问题与子进程标准输出捕获有关。

我的代码是

import subprocess

# Define the command to run
command = [r"C:\Program Files (x86)\2PDF\2PDF.exe","-src", r"C:\Users\Rajkumar\Desktop\a\a2.docx" ,"-dst", r"C:\Users\Rajkumar\Desktop" ,"-options", "mswildc:no" ,"-pdf", "ocr:yes", "ocr_lang:English"]

# Use subprocess to run the command and capture the output
process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

# Read and print the output while the process is running
while True:
    output = process.stdout.readline()
    if output == '' and process.poll() is not None:
        break
    if output:
        print(output.strip())

我的输出是

2PDF, Version 2.0, (c) fCoder SIA 2022

---------------------------------------------------------------

Loading files list, please wait...

1 of 1. "C:\Users\Rajkumar\Desktop\a\a2.docx" -> Pdf
...........
Ok

All files converted successfully

下面的 gif 显示了我想要捕获的进度。

编辑

经过挖掘，我发现如何捕获 Python 脚本是否用作在一行中打印进度的进程。 在下面的 gif 中，demo7.py 充当子进程来检索 demo8.py 中的数据。

代码demo7.py

import time
import sys
num=1
for i in range(11):
    text="#"*i+"."*(10-i)
    print(f"[{text}] {i*10}%",end="\r")
    time.sleep(0.5)

代码demo8.py

import subprocess,os
command=["python",r"E:\PythonDIR\PdfUtil\demo7.py"]
# Set PYTHONUNBUFFERED environment variable
env = os.environ.copy()
env['PYTHONUNBUFFERED'] = '1'
# Run demo7.py using subprocess with line buffering
process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, 
                           universal_newlines=True, env=env)
# Continuously read and print the live output
while True:
    output=process.stdout.readline()
    if output == '' and process.poll() is not None:
        break
    print(output ,end='')

两个脚本的单独输出如下所示。

我们需要使用 Windows 核心级 api 破解解决方案的类似方法可能是使用 ctypes 模块来处理外部可执行（.exe）文件，如 2PDF。

Answer 1

解决方案在于实时控制台输出重定向。但我的答案不完整，我刚刚将代码转换为python，但无法使用逻辑来读取/解析输出。我需要帮助。

1。在 python 中创建一个管道并从中连续读取

import ctypes,time
from ctypes import wintypes

# Define the required constants and structures
PIPE_ACCESS_DUPLEX = 0x00000003
PIPE_TYPE_MESSAGE = 0x00000004
PIPE_READMODE_MESSAGE = 0x00000002
PIPE_WAIT = 0x00000000
PIPE_UNLIMITED_INSTANCES = 255
BUFSIZE = 512
NMPWAIT_USE_DEFAULT_WAIT = 0x00000000
INVALID_HANDLE_VALUE = -1

class SECURITY_ATTRIBUTES(ctypes.Structure):
    _fields_ = [("nLength", wintypes.DWORD),
                ("lpSecurityDescriptor", wintypes.LPVOID),
                ("bInheritHandle", wintypes.BOOL)]

# Create a named pipe
def create_named_pipe(pipe_name):
    sa = SECURITY_ATTRIBUTES()
    sa.nLength = ctypes.sizeof(SECURITY_ATTRIBUTES)
    sa.lpSecurityDescriptor = None
    sa.bInheritHandle = True
    pipe_handle = ctypes.windll.kernel32.CreateNamedPipeW(
        pipe_name,
        PIPE_ACCESS_DUPLEX,
        PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT,
        PIPE_UNLIMITED_INSTANCES,
        BUFSIZE,
        BUFSIZE,
        NMPWAIT_USE_DEFAULT_WAIT,
        ctypes.byref(sa)
    )
    if pipe_handle == INVALID_HANDLE_VALUE:
        print("Failed to create named pipe")
        return None
    else:
        print("Named pipe created successfully")
        return pipe_handle

# Example usage

def read_from_pipe(pipe_handle):
    # time.sleep(1)
    while True:
        buffer = ctypes.create_string_buffer(BUFSIZE)
        bytes_read = wintypes.DWORD()

        success = ctypes.windll.kernel32.ReadFile(
            pipe_handle,
            buffer,
            BUFSIZE,
            ctypes.byref(bytes_read),
            None
        )
        #print(success)
        if not success :
            #print("Failed to read from the pipe or no data available.")
            
            error_code = ctypes.windll.kernel32.GetLastError()
            #print("err-",error_code)
            if error_code==536 or error_code==234:
                
                time.sleep(1)
                continue
            else:
                break

        data = buffer.raw[:bytes_read.value]

        print(f"Received data: {data.decode('utf-8')}",end="")
        print("\n\nDone\n\n")
while True:
    pipe_name = r'\\.\pipe\raj' # pipe name is "raj"
    pipe_handle = create_named_pipe(pipe_name)

    read_from_pipe(pipe_handle)

2。将 C++ 代码转换为 Python，发布于 https://www.codeproject.com/Articles/16163/Real-Time-Console-Output-Redirection 和 https://github.com/buck54321/PipeStuffer2

import ctypes
from ctypes import wintypes
import sys
import time



class COORD(ctypes.Structure):
    _fields_ = [("X", ctypes.c_short), ("Y", ctypes.c_short)]

origin = COORD(0, 0)

class SECURITY_ATTRIBUTES(ctypes.Structure):
    _fields_ = [
        ('nLength', wintypes.DWORD),
        ('lpSecurityDescriptor', ctypes.c_void_p),
        ('bInheritHandle', wintypes.BOOL)
    ]
class STARTUPINFO(ctypes.Structure):
    _fields_ = [("cb", wintypes.DWORD),
                ("lpReserved", wintypes.LPWSTR),
                ("lpDesktop", wintypes.LPWSTR),
                ("lpTitle", wintypes.LPWSTR),
                ("dwX", wintypes.DWORD),
                ("dwY", wintypes.DWORD),
                ("dwXSize", wintypes.DWORD),
                ("dwYSize", wintypes.DWORD),
                ("dwXCountChars", wintypes.DWORD),
                ("dwYCountChars", wintypes.DWORD),
                ("dwFillAttribute", wintypes.DWORD),
                ("dwFlags", wintypes.DWORD),
                ("wShowWindow", wintypes.WORD),
                ("cbReserved2", wintypes.WORD),
                ("lpReserved2", wintypes.LPBYTE),
                ("hStdInput", wintypes.HANDLE),
                ("hStdOutput", wintypes.HANDLE),
                ("hStdError", wintypes.HANDLE)]


class PROCESS_INFORMATION(ctypes.Structure):
    _fields_ = [("hProcess", wintypes.HANDLE),
                ("hThread", wintypes.HANDLE),
                ("dwProcessId", wintypes.DWORD),
                ("dwThreadId", wintypes.DWORD)]

# Define the CONSOLE_SCREEN_BUFFER_INFO structure
class CONSOLE_SCREEN_BUFFER_INFO(ctypes.Structure):
    _fields_ = [
        ("dwSize", COORD),  # Size of the console screen buffer
        ("dwCursorPosition", COORD),  # Cursor position
        ("wAttributes", wintypes.WORD),  # Screen buffer attributes
        ("srWindow", wintypes.SMALL_RECT),  # Window rectangle
        ("dwMaximumWindowSize", COORD),  # Maximum window size
    ]




STARTF_FORCEOFFFEEDBACK = 0x00000080


def main():
    
    dwDummy = ctypes.c_ulong()

    # Parse command line: skip to RTconsole's arguments
   
    command_line=r"C:\Users\Rajkumar\Downloads\Compressed\RTconsoleDemo\RTconsoleDemo\Release\DemoConsole.exe"
    #path to any console based application
    # Prepare the console window & inherited screen buffer
    
    pipe_name = r"\\.\pipe\raj"

    named_pipe = ctypes.windll.kernel32.CreateFileW(
        pipe_name, 0x40000000, 0, None, 3, 0, None
    )
    if named_pipe == -1:
        error_code = ctypes.windll.kernel32.GetLastError()
        print(f"Failed to create/open named pipe. Error code: {error_code}")
        return
    sa = SECURITY_ATTRIBUTES(
        ctypes.sizeof(SECURITY_ATTRIBUTES), None, True
    )
    hConsole = ctypes.windll.kernel32.CreateConsoleScreenBuffer(
        0x80000000 | 0x40000000, 1 | 2, ctypes.byref(sa), 1, None
    )


    ctypes.windll.kernel32.FillConsoleOutputCharacterA(
        hConsole, b'\0', 0x7fffffff, COORD(0, 0), ctypes.byref(dwDummy)

    )

    if not ctypes.windll.kernel32.SetConsoleActiveScreenBuffer(hConsole):
        ctypes.windll.kernel32.CloseHandle(hConsole)
        return 1
    
    ctypes.windll.kernel32.SetStdHandle(-11, hConsole)

    # Start the subprocess
    pi = PROCESS_INFORMATION()
    si = STARTUPINFO()
    si.cb = ctypes.sizeof(si)
    si.dwFlags = STARTF_FORCEOFFFEEDBACK
    # si.hStdOutput = hConsole
    if not ctypes.windll.kernel32.CreateProcessA(
        None,  # ApplicationName
        ctypes.create_string_buffer(command_line.encode()),  # CommandLine
        None,  # ProcessAttributes
        None,  # ThreadAttributes
        True,  # InheritHandles
        0,  # CreationFlags
        None,  # Environment
        None,  # CurrentDirectory
        ctypes.byref(si),
        ctypes.byref(pi)
    ):
        ctypes.windll.kernel32.CloseHandle(hConsole)
        return -2
    ctypes.windll.kernel32.CloseHandle(pi.hThread)

    last_pos = COORD(0, 0)
    csbi = CONSOLE_SCREEN_BUFFER_INFO()
    exit_now = False
    while not exit_now:
        if ctypes.windll.kernel32.WaitForSingleObject(pi.hProcess, 0) != 0x102:
            exit_now = True  # Exit after this last iteration

        # Get screen buffer state
        ctypes.windll.kernel32.GetConsoleScreenBufferInfo(hConsole, ctypes.byref(csbi))
        line_width = csbi.dwSize.X

        if csbi.dwCursorPosition.Y == last_pos.Y and  csbi.dwCursorPosition.X == last_pos.X:
            
            time.sleep(0.05)
            
        else:
            count = (csbi.dwCursorPosition.Y - last_pos.Y) * line_width + csbi.dwCursorPosition.X - last_pos.X
            # Read newly output characters starting from the last cursor position
            buffer = ctypes.create_string_buffer(count)
            ctypes.windll.kernel32.ReadConsoleOutputCharacterA(hConsole, buffer, count, last_pos, ctypes.byref(dwDummy))
            ctypes.windll.kernel32.FillConsoleOutputCharacterA(hConsole,b"\0", count, last_pos, ctypes.byref(dwDummy))
            last_pos = csbi.dwCursorPosition
            ctypes.windll.kernel32.GetConsoleScreenBufferInfo(hConsole, ctypes.byref(csbi))
            if (csbi.dwCursorPosition.X == last_pos.X) and (csbi.dwCursorPosition.Y == last_pos.Y):
                # Text cursor did not move since this treatment, hurry to reset it to home
                ctypes.windll.kernel32.SetConsoleCursorPosition(hConsole, origin)
                last_pos = origin
            

            '''Need help to Parse output and send to PIPE here
            For Example 
            bytes_written = wintypes.DWORD()
            success = ctypes.windll.kernel32.WriteFile(
                    named_pipe,
                    buffer.value,
                    len(buffer.value),
                    ctypes.byref(bytes_written),
                    None
                )
            '''
    ctypes.windll.kernel32.CloseHandle(hConsole)

    # Release subprocess handle
    exit_code = ctypes.c_ulong()
    if not ctypes.windll.kernel32.GetExitCodeProcess(pi.hProcess, ctypes.byref(exit_code)):
        exit_code.value = -3
    ctypes.windll.kernel32.CloseHandle(pi.hProcess)
    return exit_code.value

if __name__ == "__main__":
    sys.exit(main())