Python中subprocess.Popen执行脚本无法实时输出的问题及日志需求

Great question! The difference between ping and your Python script boils down to output buffering:

• ping uses line buffering by default (even when output isn’t a terminal), so it sends output line by line immediately.
• Python scripts, however, switch to block buffering when stdout is redirected to a pipe (like subprocess.PIPE). That means output gets held in a buffer until it’s full or the process exits—hence why you only see results after test.py finishes.

We’ll cover two reliable approaches, depending on whether you can modify the child script or not.

1. Force Line Buffering in the Child Script (Simplest) #

If you can adjust test.py or control how it’s executed, you can make Python flush output immediately:

Option A: Modify the Child Script #

Add flush=True to your print calls, or enable line buffering globally:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import time
import sys
sys.stdout.line_buffering = True  # Enable line buffering for all stdout output

print('hello')
time.sleep(2)
print('hello, again')

Option B: Run Python with the -u Flag #

If you can’t edit the script, run it with python -u to disable buffering entirely:

python -u test.py

Then update your execute_cmd function to forward output to your logger in real time:

Updated execute_cmd with Logger Integration #

import shlex
import subprocess
import logging

def execute_cmd(cmd, logger):
    cmd_args = shlex.split(cmd)
    # Use text=True to get string output instead of raw bytes
    proc = subprocess.Popen(
        cmd_args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )

    # Read stdout line by line as it's generated
    for line in iter(proc.stdout.readline, ''):
        if line.strip():  # Skip empty lines
            logger.info(line.rstrip('\n'))  # Remove trailing newline (logger adds its own)

    # Handle any stderr output after the process finishes
    stderr_output = proc.stderr.read().rstrip('\n')
    if stderr_output:
        logger.error(stderr_output)
        raise Exception(f"Command failed with error: {stderr_output}")

    proc.wait()

2. Simulate a Terminal (No Child Script Modifications) #

If you can’t modify the child script or use -u, use the pty module to create a pseudo-terminal. This tricks the child process into thinking it’s writing to a real terminal, which forces line buffering.

Note: This works on Unix-like systems (Linux/macOS); Windows has limited support for pty.

execute_cmd with Pseudo-Terminal #

import os
import pty
import select
import shlex
import subprocess
import logging

def execute_cmd_pty(cmd, logger):
    cmd_args = shlex.split(cmd)
    # Create a pseudo-terminal pair
    master_fd, slave_fd = pty.openpty()

    # Attach the child process to the slave terminal
    proc = subprocess.Popen(
        cmd_args,
        stdout=slave_fd,
        stderr=slave_fd,
        stdin=slave_fd,
        text=True
    )
    os.close(slave_fd)  # Close the slave end in the parent process

    # Read output in real time using select to monitor the terminal
    while True:
        # Wait for data to be available on the master terminal
        readable, _, _ = select.select([master_fd], [], [], 0.1)
        if readable:
            data = os.read(master_fd, 1024).rstrip('\n')
            if data:
                logger.info(data)
        
        # Check if the child process has exited
        if proc.poll() is not None:
            # Read any remaining output from the terminal
            remaining_data = os.read(master_fd, 1024).rstrip('\n')
            if remaining_data:
                logger.info(remaining_data)
            break

    os.close(master_fd)
    # Raise an exception if the command failed
    if proc.returncode != 0:
        raise Exception(f"Command failed with exit code {proc.returncode}")

First, configure a logger that writes to both the console and a file:

def setup_logger():
    logger = logging.getLogger('cmd_executor')
    logger.setLevel(logging.INFO)
    logger.handlers.clear()  # Avoid duplicate handlers if the function is called multiple times

    # Console handler (output to screen)
    stream_handler = logging.StreamHandler()
    # File handler (output to log file)
    file_handler = logging.FileHandler('cmd_output.log')

    # Set a consistent log format
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    stream_handler.setFormatter(formatter)
    file_handler.setFormatter(formatter)

    logger.addHandler(stream_handler)
    logger.addHandler(file_handler)
    return logger

if __name__ == '__main__':
    logger = setup_logger()
    try:
        # Option 1: Use the -u flag for unbuffered output
        execute_cmd('python -u test.py', logger)
        # Option 2: Use pseudo-terminal (Unix-only, no script modifications)
        # execute_cmd_pty('python test.py', logger)
    except Exception as e:
        logger.error(f"Execution failed: {str(e)}")

• Buffering is the root cause: Python switches to block buffering when stdout isn’t a terminal.
• Simplest fix: Use python -u or enable line buffering in the child script.
• No-modification fix: Use pty to simulate a terminal for the child process.

内容的提问来源于stack exchange，提问作者CodeUnsolved