如何在Python subprocess中实时获取ansible-playbook彩色输出？

我来帮你搞定这个问题！你当前的代码之所以只能在命令结束后拿到黑白输出，主要是两个原因：一是用了communicate()等命令执行完才一次性获取结果，没法实时输出；二是ansible-playbook默认只在TTY终端环境下才会输出彩色，而subprocess.PIPE是管道，不是TTY，所以彩色效果被自动禁用了。

下面给你几个实用的解决方案，从简单到进阶：

方案一：直接让子进程继承父进程的输出（最省心） #

这个方法最直接，让ansible-playbook的输出直接打印到当前终端，和你手动在命令行敲命令的效果完全一样，天然支持实时彩色输出。

修改你的代码如下：

import subprocess
import sys

# 记得加上--force-color，保险起见强制开启彩色输出
command = '{0} --flush-cache -f 20 -i {1}/dyn_inv.py --extra-vars @{2}/vars.json {3} --force-color'.format(ansible_playbook_loc, script_path, var_path, args.playbook[0])
print(command)

# 让子进程直接使用父进程的标准输出/错误，禁用缓冲保证实时性
process = subprocess.Popen(
    command.split(),
    stdout=sys.stdout,
    stderr=subprocess.STDOUT,
    bufsize=0
)

# 等待命令执行完成，也可以根据需要做其他操作
process.wait()

这里的--force-color是为了避免某些环境下终端检测失效的情况，bufsize=0禁用输出缓冲，确保每一行输出都能实时显示到终端。

方案二：实时读取输出流（兼顾捕获输出和实时显示） #

如果你需要在实时显示彩色输出的同时，还要捕获输出内容做后续处理（比如解析结果），可以用这个方法：

import subprocess
import sys

command = '{0} --flush-cache -f 20 -i {1}/dyn_inv.py --extra-vars @{2}/vars.json {3} --force-color'.format(ansible_playbook_loc, script_path, var_path, args.playbook[0])
print(command)

# 开启管道但实时读取每一行输出
process = subprocess.Popen(
    command.split(),
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    bufsize=1,  # 行缓冲，逐行读取
    universal_newlines=True  # 按文本模式处理，不用转二进制
)

# 循环读取每一行并打印，同时可以把line存入变量做后续处理
for line in iter(process.stdout.readline, ''):
    sys.stdout.write(line)
    sys.stdout.flush()  # 强制刷新缓冲，保证实时性

# 等待命令结束
process.wait()

这个方法既保留了实时彩色输出，又能让你拿到每一行的输出内容，方便后续的逻辑处理。

方案三：模拟TTY终端（彻底解决TTY检测问题） #

如果上面的方案还是没解决彩色问题，可能是ansible的某些输出逻辑严格依赖TTY环境，这时候可以用Python的pty模块模拟真实终端：

import pty
import os
import sys
import subprocess

command = '{0} --flush-cache -f 20 -i {1}/dyn_inv.py --extra-vars @{2}/vars.json {3}'.format(ansible_playbook_loc, script_path, var_path, args.playbook[0])
print(command)

def run_with_tty(cmd):
    pid, fd = pty.fork()
    if pid == 0:
        # 子进程执行命令
        os.execvp(cmd.split()[0], cmd.split())
    else:
        # 父进程读取终端输出并打印
        while True:
            try:
                data = os.read(fd, 1024)
                if not data:
                    break
                sys.stdout.write(data.decode())
                sys.stdout.flush()
            except OSError:
                break
        os.waitpid(pid, 0)

run_with_tty(command)

这个方法完全模拟了真实的终端环境，ansible会认为自己在命令行中执行，自然输出彩色。不过注意pty模块只支持Linux和macOS，Windows系统没法用。

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