借助subprocess实现pexpect stdout与stderr分别输出到独立文件

• 核心报错是用法逻辑错误：pexpect.spawn()本身就负责启动子进程、通过伪终端(PTY)和进程交互，不需要提前用subprocess.Popen/subprocess.run启动进程再传入返回对象。你传入的Popen、CompletedProcess是Python对象，不是可执行的命令字符串，自然会抛出「对象不可迭代」的类型错误。
• 日志文件为空、密码无法自动输入的原因：你给subprocess设置了stdout/stderr直接重定向到文件，子进程的输出会直接写入文件、不会发送到PTY通道，pexpect读不到任何输出，既没法匹配密码提示，也拿不到正常运行的日志内容。
• pexpect本身确实无法原生分离stdout和stderr：PTY是单通道设备，子进程连接PTY时会默认把stdout、stderr都输出到这个通道，要实现分流需要额外做输出标记处理。

通过bash进程替换给stderr输出增加特殊识别前缀，再由pexpect读取所有输出后按标记分流，既不影响PTY交互、自动输入密码，又能100%准确分离两类日志，不需要额外安装第三方依赖。

注意事项 #

• 日志文件使用追加模式打开，避免循环处理多个数据集时覆盖之前的日志记录
• 密码提示使用正则匹配password关键词，比单纯匹配"Enter"更准确，避免误匹配提前输入密码
• 增加超时、退出状态检查逻辑，异常信息统一写入stderr日志

实现代码 #

import pexpect

# 提前打开日志文件，追加模式避免覆盖历史记录
with open("stdout.txt", "ab") as stdout_file, open("stderr.txt", "ab") as stderr_file:
    for source_ds, dest_ds in zip(SourceLines, DestLines):
        # 构造原始syncoid执行命令
        syncoid_cmd = SyncoidCommand.replace("SourceDataSet", source_ds).replace("DestDataSet", dest_ds)
        # 包装命令：通过bash进程替换，给所有stderr输出行加特殊识别前缀
        # 前缀选择不会出现在正常业务输出里的字符串即可
        marker = "__STDERR_TAG__"
        wrapped_cmd = f'bash -c \'{syncoid_cmd} 2> >(while IFS= read -r line; do echo "{marker}$line"; done)\''

        # 直接通过pexpect启动包装后的命令，不要混用subprocess
        # timeout=None 取消默认超时，适配大数据集同步的长耗时场景
        child = pexpect.spawn(wrapped_cmd, encoding="utf-8", timeout=None)

        process_end = False
        # 循环匹配输出，处理密码输入
        while not process_end:
            match_idx = child.expect([r"[Pp]assword:", pexpect.EOF, pexpect.TIMEOUT])
            if match_idx == 0:
                # 匹配到密码提示，该提示属于stderr输出
                prompt_content = f"{child.before}{child.after}"
                stderr_file.write(prompt_content.encode("utf-8") + b"\n")
                # 输入密码
                child.sendline(PassWord)
            elif match_idx == 1:
                # 进程执行结束，拿到剩余所有输出
                rest_output = child.before
                process_end = True
            else:
                # 执行超时，记录错误后强制终止进程
                stderr_file.write(f"Sync task {source_ds} -> {dest_ds} timeout\n".encode("utf-8"))
                child.close(force=True)
                rest_output = ""
                process_end = True

        # 按标记分流剩余输出
        for line in rest_output.splitlines():
            if line.startswith(marker):
                # 去掉标记后写入stderr
                clean_line = line.removeprefix(marker)
                stderr_file.write(clean_line.encode("utf-8") + b"\n")
            else:
                stdout_file.write(line.encode("utf-8") + b"\n")

        child.close()
        # 检查进程退出码，非0则记录错误
        if child.exitstatus != 0:
            stderr_file.write(
                f"Sync task {source_ds} -> {dest_ds} failed, exit code: {child.exitstatus}\n".encode("utf-8")
            )

如果你的syncoid密码提示文本和默认的password:不一致，只要修改expect里的正则规则即可，比如提示文本是"Enter passphrase for key"，可以把正则改成r"(Enter.*[Pp]assword|Enter passphrase)"覆盖所有场景。

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