You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python子进程输入输出定向至日志文件时文件提前关闭问题

问题分析

你的代码核心问题在于错误使用了subprocess.communicate():这个方法的设计是一次性完成和子进程的交互——它会向stdin写入数据(如果传入参数)、读取所有stdout/stderr输出,然后等待子进程终止,并且调用后会自动关闭子进程的stdin/stdout管道。所以第一次循环调用communicate()后,子进程的管道已经被关闭,后续的process.stdin.write()自然会失败,表现出类似“文件提前关闭”的异常。

另外,你的代码没有处理子进程的stderr输出,也没考虑子进程可能提前退出的情况。

修复后的代码

下面是修正后的版本,改用直接读取stdout的方式,同时处理超时和子进程状态:

import subprocess
import shlex
import select

def main():
    program_filename = "calculator.py" 
    input_log_filename = "user_input.log"
    output_log_filename = "program_output.log"

    with open(input_log_filename, "w") as input_log, open(output_log_filename, "w") as output_log:
        print("Calculator program is running.")
        
        command = f"python3 {program_filename}"
        process = subprocess.Popen(
            shlex.split(command),
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,  # 把stderr合并到stdout一起记录,也可单独写入日志
            text=True,
            bufsize=1,
            universal_newlines=True
        )

        while True:
            # 用select检查stdout是否有可读数据,超时1秒避免阻塞
            ready, _, _ = select.select([process.stdout], [], [], 1)
            if ready:
                program_output = process.stdout.readline()
                if program_output:
                    output_log.write(program_output)
                    print("Program Output:", program_output.strip())
            
            # 检查子进程是否已经退出
            if process.poll() is not None:
                print("子进程已退出")
                break

            user_input = input("Enter input: ")
            if user_input.lower() == "exit":
                process.stdin.write(user_input + "\n")
                process.stdin.flush()
                break

            input_log.write(user_input + "\n")  
            process.stdin.write(user_input + "\n")
            process.stdin.flush()
        
        # 读取子进程剩余的输出,确保日志完整
        remaining_output = process.stdout.read()
        if remaining_output:
            output_log.write(remaining_output)
            print("Program Output:", remaining_output.strip())
        
        process.wait()

if __name__ == "__main__":
    main()
关键修改点
  • 移除process.communicate(),改用select.select()配合readline()实现非阻塞读取,避免主循环被卡死
  • 增加子进程状态检查(process.poll()),处理子进程提前退出的情况
  • 退出时读取子进程剩余输出,保证日志记录完整
  • 可选择将stderr合并到stdout或单独写入日志,避免遗漏错误信息

内容的提问来源于stack exchange,提问作者Isabella Brooks

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.13 18:05:21