1. 项目缘起为什么我们总在命令行和程序之间“反复横跳”做开发或者运维的朋友估计都经历过这种场景你写了一个Python脚本需要调用一个系统命令比如用ffmpeg转码一个视频或者用curl获取一个API接口的数据。这时候你可能会想直接在脚本里os.system()一下不就完了确实对于一次性的、简单的、不需要交互的命令os.system()或者os.popen()勉强够用。但一旦需求变得复杂——你需要实时获取命令的输出、需要向命令输入参数、需要判断命令是否执行成功、甚至需要解析命令返回的结构化数据比如JSON——你就会发现os模块提供的那些方法就像一把生锈的瑞士军刀用起来处处掣肘。我最近就踩了这么一个大坑。一个数据处理的自动化任务需要调用一个外部的命令行工具来清洗数据这个工具不仅执行时间长还会分阶段输出进度信息JSON格式并且中途可能需要根据上一步的结果输入确认指令。一开始用os.system结果脚本“卡死”了因为命令在等待输入而我的脚本根本不知道。后来换os.popen能读到输出但又没法区分标准输出和错误输出而且命令一结束管道就关了想再交互也没门。更头疼的是那个JSON输出混在了一大堆日志文本里手动解析起来简直是一场灾难。正是这些“痛不欲生”的经历让我下定决心把Python标准库里的subprocess模块彻底吃透。subprocess才是Python与系统外壳Shell打交道的“官方指定武器库”。它不仅能替代os.system和os.popen更重要的是它提供了对子进程近乎完全的控制能力你可以重定向输入输出、等待进程结束、获取返回码甚至进行复杂的实时交互。结合Python强大的字符串处理和json模块我们就能优雅地完成“执行命令 - 交互 - 等待 - 判断状态 - 解析结果”这一整套流程。这篇文章我就结合自己趟过的坑手把手带你掌握subprocess的核心用法并重点解决如何与产生JSON输出的命令行工具进行协作。无论你是想自动化运维脚本、集成第三方命令行工具还是构建复杂的任务流水线这些内容都能让你少走弯路。2. 初识subprocess从“黑盒子”到“透明管道”在深入具体代码之前我们得先理解subprocess设计的基本哲学。它把我们要运行的命令行程序看作一个“子进程”。这个子进程有三个标准的数据流stdin (标准输入)子进程从这里读取数据。对应我们在终端里打字。stdout (标准输出)子进程把正常的输出信息写到这里。对应终端里显示的正常结果。stderr (标准错误)子进程把错误和警告信息写到这里。对应终端里红色的错误提示。subprocess的核心能力就是让我们在Python父进程中能够创建并精细地控制这些数据流的管道。我们可以选择不捕获让子进程的输出直接打印到终端和父进程共享。捕获到变量把输出内容抓取到Python的变量里方便后续处理。重定向到文件把输出写入文件。提供一个输入源我们可以预先准备好数据或者实时生成数据喂给子进程的stdin。subprocess模块提供了几个不同层次的函数来启动子进程我们从最简单常用的说起。2.1subprocess.run()一站式解决方案这是Python 3.5之后推荐的首选方法它封装了最常见的用例执行一个命令等待它完成然后返回一个包含结果的CompletedProcess对象。对于大多数“执行命令并获取结果”的场景它都是最佳选择。import subprocess import json # 示例1执行一个简单的命令并捕获其输出 result subprocess.run([ls, -l], capture_outputTrue, textTrue) print(f返回码: {result.returncode}) print(f标准输出:\n{result.stdout}) # 如果命令出错错误信息在result.stderr里 # 示例2执行一个返回JSON的命令并直接解析 # 假设我们有一个工具 data-tool 它接受一个 --format json 参数 result subprocess.run([data-tool, --format, json], capture_outputTrue, textTrue, checkTrue) if result.returncode 0: try: data json.loads(result.stdout) # 解析JSON print(f解析成功获取到数据: {data}) except json.JSONDecodeError as e: print(fJSON解析失败: {e}\n原始输出:{result.stdout}) else: print(f命令执行失败: {result.stderr})关键参数解析args: 命令列表。强烈建议以列表形式传入如[ls, -l]。这可以避免Shell注入攻击并且参数中的空格、引号会被正确处理。如果非要用字符串必须设置shellTrue但这会引入安全风险。capture_outputTrue: 这是stdoutsubprocess.PIPE, stderrsubprocess.PIPE的快捷方式。意味着同时捕获标准输出和错误输出。textTrue(或universal_newlinesTrue): 让stdout和stderr以字符串Pythonstr形式返回而不是字节序列bytes。这样我们就不用再手动.decode(utf-8)了。checkTrue: 如果子进程返回非零退出码通常表示失败它会抛出一个CalledProcessError异常。这对于需要确保命令成功的场景非常有用。timeout: 设置超时时间秒。如果命令执行超过这个时间会抛出TimeoutExpired异常。subprocess.run()是同步的它会一直阻塞直到命令执行完毕。对于长时间运行或需要交互的命令我们还需要其他工具。2.2subprocess.Popen()底层控制与实时交互当subprocess.run()无法满足需求时——比如你需要与进程进行实时对话一问一答或者需要同时管理多个长时间运行的进程——就该Popen出场了。Popen是subprocess模块的基石run()函数内部也是用它实现的。它给了你最大的灵活性但复杂度也更高。Popen创建子进程后立即返回不会等待进程结束。你需要通过它返回的Popen对象来与进程交互。import subprocess import time # 启动一个可能会等待输入的命令比如一个简单的交互式Python解释器 # 注意这里我们同时打开了stdin, stdout, stderr三个管道 proc subprocess.Popen([python, -i], # -i 进入交互模式 stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, # 同样使用文本模式 bufsize1, # 行缓冲方便实时读写 universal_newlinesTrue) # 现在proc就是一个代表子进程的对象 print(f子进程PID: {proc.pid}) # 向子进程的stdin写入命令并立即刷新缓冲区 proc.stdin.write(print(“Hello from subprocess!”)\n) proc.stdin.flush() # 确保数据被发送 # 从子进程的stdout读取一行输出 # 注意如果子进程没有输出这里会阻塞。可以使用select或线程来处理。 output_line proc.stdout.readline() print(f收到输出: {output_line}) # 再发送一个退出命令 proc.stdin.write(exit()\n) proc.stdin.flush() # 等待进程结束并获取返回码 return_code proc.wait(timeout5) # 等待最多5秒 print(f进程结束返回码: {return_code}) # 最后读取所有剩余的输出来清理管道 remaining_output, remaining_errors proc.communicate() print(f剩余输出: {remaining_output})关键方法与属性proc.stdin,proc.stdout,proc.stderr: 文件对象用于与子进程通信。proc.pid: 子进程的进程ID。proc.poll(): 检查进程是否已终止。如果已终止返回退出码否则返回None。这是判断“是否结束”的非阻塞方法。proc.wait(timeoutNone): 等待进程终止。可设置超时。proc.communicate(inputNone, timeoutNone):这是处理交互和获取输出的“瑞士军刀”。它向stdin发送数据可选然后等待进程结束同时收集所有的stdout和stderr数据。对于一次性发送所有输入并获取所有输出的场景它比手动读写stdin/stdout更安全能避免死锁。proc.terminate(): 发送SIGTERM信号请求进程终止。proc.kill(): 发送SIGKILL信号强制杀死进程。重要经验处理输出时的死锁问题这是一个经典的坑。子进程的缓冲区是有限的。如果你通过proc.stdout.read()试图读取大量数据而子进程又在等待你通过proc.stdin.write()输入数据但它的stdout缓冲区已经满了它就会卡住。双方都在等对方这就形成了死锁。黄金法则如果需要交互考虑使用proc.communicate(input_data)。它会处理好所有缓冲。如果必须手动读写请确保读写顺序正确并且考虑使用线程一个线程专门读stdout一个线程专门写stdin。对于输出量未知的命令使用iter(proc.stdout.readline, )来逐行读取而不是一次性read()。3. 实战与输出JSON的命令行工具共舞现在我们来解决一个更具体、也更常见的需求调用一个命令行工具这个工具会输出结构化的JSON数据我们需要在Python中捕获并解析它。这可能是调用一个云平台的CLI如aws cli,kubectl一个数据转换工具如jq或者任何自定义的脚本。3.1 场景一简单调用一次性获取JSON这是最简单的情况。命令执行后直接输出一个完整的JSON字符串。import subprocess import json def run_command_and_parse_json(cmd_args): 执行命令并尝试解析其标准输出为JSON。 参数: cmd_args (list): 命令及其参数的列表例如 [aws, ec2, describe-instances, --output, json] 返回: tuple: (success(bool), data(dict/list), error_message(str)) try: # 使用run设置checkTrue确保命令成功超时设为30秒 result subprocess.run(cmd_args, capture_outputTrue, textTrue, checkTrue, timeout30) # 命令执行成功尝试解析JSON try: data json.loads(result.stdout) return True, data, None except json.JSONDecodeError as e: # JSON解析失败可能是工具输出了非JSON内容如警告信息 error_msg fFailed to decode JSON from stdout. Error: {e}. Stdout preview: {result.stdout[:200]} return False, None, error_msg except subprocess.CalledProcessError as e: # 命令本身执行失败返回非零码 error_msg fCommand failed with return code {e.returncode}. Stderr: {e.stderr} return False, None, error_msg except subprocess.TimeoutExpired as e: # 命令超时 error_msg fCommand timed out after {e.timeout} seconds. return False, None, error_msg except FileNotFoundError as e: # 命令不存在 error_msg fCommand not found: {cmd_args[0]} return False, None, error_msg # 使用示例 success, data, error run_command_and_parse_json([echo, {name: Alice, age: 30}]) if success: print(fParsed data: {data}) print(fName: {data[name]}) else: print(fError: {error})这里的关键点错误处理分层我们区分了“命令执行失败”和“JSON解析失败”。有时候命令成功了但可能输出了额外的日志行污染了JSON导致解析失败。超时控制使用timeout参数避免脚本因某个命令卡死而永远挂起。文本模式textTrue确保我们拿到的是字符串可以直接给json.loads()。3.2 场景二处理混合输出JSON与日志交织很多命令行工具尤其是开发中的或者调试模式下的会把JSON数据和普通的日志信息一起输出到stdout。比如[INFO] Starting processing... {status: processing, progress: 25} [DEBUG] Fetched 100 records. {status: processing, progress: 50} {status: completed, result: [...]} [INFO] All done.我们的目标是从中提取出有效的JSON行。这里subprocess.run()仍然可用但我们需要对输出进行后处理。import subprocess import json def extract_json_from_output(cmd_args): 执行命令并从其标准输出中提取所有有效的JSON行。 result subprocess.run(cmd_args, capture_outputTrue, textTrue) json_objects [] for line in result.stdout.splitlines(): line line.strip() if not line: continue # 尝试将每一行解析为JSON try: obj json.loads(line) json_objects.append(obj) print(fFound JSON: {obj}) except json.JSONDecodeError: # 不是JSON可能是普通日志可以选择记录或忽略 # print(fIgnored log line: {line}) pass return json_objects, result.returncode # 模拟一个输出混合内容的命令 test_script import json import sys import time print(Start..., filesys.stderr) # 错误输出 print([LOG] Initializing, filesys.stdout) time.sleep(0.1) print(json.dumps({step: 1, data: sample})) print([LOG] Processing...) print(json.dumps({step: 2, data: processed})) print([LOG] Done.) # 将上面的脚本写到一个临时文件并执行 import tempfile with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(test_script) temp_file_path f.name try: objects, returncode extract_json_from_output([python, temp_file_path]) print(f\nExtracted {len(objects)} JSON objects. Return code: {returncode}) for obj in objects: print(f - Step {obj[step]}: {obj[data]}) finally: import os os.unlink(temp_file_path) # 清理临时文件策略分析这种方法简单粗暴但假设了每个JSON对象都独占一行。如果JSON被漂亮打印Pretty-print成多行这种方法就会失败。对于多行JSON我们需要更复杂的解析器或者使用专门的工具如jq来预处理输出。3.3 场景三实时流式处理JSON输出对于一些长时间运行的任务如数据导入、模型训练它们可能会持续输出进度信息或中间结果JSON格式。我们可能希望实时处理这些数据而不是等命令全部结束。这就需要用到Popen进行实时读取。import subprocess import json import select # 用于非阻塞读取Unix-like系统 import sys def stream_and_process_json(cmd_args): 启动一个命令并实时读取其标准输出尝试解析并处理每一行JSON。 注意此示例在Unix-like系统Linux, macOS上工作最佳。 Windows上select()对管道的行为不同可能需要使用线程。 # 启动进程注意stderr我们重定向到stdout一起读或者单独处理 proc subprocess.Popen(cmd_args, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, # 也可以重定向到subprocess.STDOUT合并 textTrue, bufsize1, # 行缓冲 universal_newlinesTrue) print(fStarted process {proc.pid}. Streaming output...) # 使用select来监控哪些管道有数据可读避免阻塞 # 这是一个简化的例子实际生产环境可能需要更健壮的错误处理 while True: # 检查进程是否已经结束 if proc.poll() is not None: print(f\nProcess finished with return code: {proc.poll()}) break # 监控stdout和stderr这里我们分开监控 readable, _, _ select.select([proc.stdout, proc.stderr], [], [], 0.1) # 0.1秒超时 for stream in readable: line stream.readline() if not line: # EOF continue if stream is proc.stdout: # 处理标准输出行 line line.rstrip(\n) try: data json.loads(line) # **这里是你的实时处理逻辑** print(f[STDOUT JSON] 实时处理: {data}) # 例如更新进度条、将数据存入队列、触发回调等 if data.get(status) completed: print(检测到任务完成信号) except json.JSONDecodeError: # 不是JSON当作普通日志处理 print(f[STDOUT LOG] {line}) elif stream is proc.stderr: # 处理错误输出行 print(f[STDERR] {line.rstrip()}, filesys.stderr) # 循环结束后读取可能剩余的任何输出 remaining_out, remaining_err proc.communicate() if remaining_out: print(f剩余标准输出: {remaining_out}) if remaining_err: print(f剩余错误输出: {remaining_err}, filesys.stderr) # 示例模拟一个流式输出JSON的命令 streaming_script import json import sys import time for i in range(5): data {iteration: i, value: i * 10, status: running} print(json.dumps(data)) sys.stdout.flush() # 确保立即输出而不是等缓冲区满 time.sleep(1) # 模拟耗时操作 print(json.dumps({iteration: 5, value: 50, status: completed})) sys.stdout.flush() print(Some final log message., filesys.stderr) with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(streaming_script) temp_file_path f.name try: stream_and_process_json([python, temp_file_path]) finally: import os os.unlink(temp_file_path)关键点与陷阱缓冲Buffering这是流式处理最大的敌人。子进程的输出可能被缓冲在内存中直到缓冲区满或进程结束才一次性吐出。我们通过设置bufsize1行缓冲和在子进程脚本中调用sys.stdout.flush()来缓解。对于非Python程序可能需要在调用时使用stdbuf等工具如[stdbuf, -oL, your-command]在Linux上。非阻塞读取使用select模块可以避免在readline()上无限期阻塞。注意select在Windows上对普通文件/管道的行为有限制生产环境的跨平台代码可能需要使用threading模块创建单独的读取线程。错误流处理错误输出stderr也需要被及时读取否则如果它被填满子进程也可能被阻塞。上面的例子展示了分开处理stdout和stderr。进程状态检查proc.poll()用于非阻塞地检查进程是否结束。4. 进阶技巧与避坑指南掌握了基本用法后我们来看看一些能让你代码更健壮、更高效的进阶技巧和常见陷阱。4.1 环境变量与工作目录子进程默认会继承父进程的环境变量。但有时我们需要修改。import subprocess import os # 1. 添加或修改环境变量 my_env os.environ.copy() my_env[MY_CUSTOM_VAR] some_value my_env[PATH] /my/custom/bin: my_env[PATH] # 添加自定义路径到PATH开头 result subprocess.run([some_command], capture_outputTrue, textTrue, envmy_env) # 2. 设置工作目录相当于 cd 到那个目录再执行命令 result subprocess.run([ls, -la], capture_outputTrue, textTrue, cwd/path/to/target/directory)4.2 ShellTrue便利与风险并存subprocess.run(ls -l | grep .py, shellTrue)这种写法很诱人因为它可以直接使用Shell的特性管道|、重定向、通配符*等。但这是极度危险的特别是当命令字符串来自用户输入或外部数据时极易引发Shell注入攻击。# 危险如果user_input来自不可信源比如 ; rm -rf / user_input “somefile.txt” subprocess.run(fls -l {user_input}, shellTrue) # 灾难 # 安全做法使用列表参数让subprocess处理参数转义 subprocess.run([ls, -l, user_input]) # 即使user_input是 ; rm -rf /它也会被当作一个文件名参数不会被解析为命令原则除非有非常充分的理由如必须使用复杂的Shell特性否则永远使用列表形式传递参数并避免shellTrue。4.3 超时、终止与资源清理长时间运行或失控的子进程会耗尽资源。try: # 设置超时 result subprocess.run([sleep, 10], timeout5, capture_outputTrue, textTrue) except subprocess.TimeoutExpired as e: print(f命令超时进程可能仍在运行。) # 你可以尝试终止它 # e.cmd 是命令 e.timeout 是超时时间 # 但run()已经帮我们处理了终止如果超时子进程会被kill。 # 如果使用Popen则需要手动处理 # proc.terminate() # 先友好地请求终止 # try: # proc.wait(timeout2) # 给一点时间结束 # except subprocess.TimeoutExpired: # proc.kill() # 强制杀死 # proc.wait() # 等待清理 # 使用Popen时务必确保关闭管道 proc subprocess.Popen(...) try: # ... 与进程交互 ... finally: # 确保管道被关闭避免资源泄漏 if proc.stdin: proc.stdin.close() # 对于stdout/stderrcommunicate()或wait()会自动关闭但手动关闭也无害 proc.wait()4.4 解析复杂JSON与使用jq预处理有时命令行工具输出的JSON非常庞大或复杂或者嵌套很深。直接在Python里用json.loads()解析整个字符串可能效率不高或者你只关心其中一小部分。这时可以借助强大的命令行JSON处理器jq进行预处理。import subprocess import json # 假设 big-command 输出一个巨大的JSON我们只想要其中的 data.items[].name 字段 # 方案1 全部拉取到Python再解析内存可能压力大 # result subprocess.run([big-command, --output, json], capture_outputTrue, textTrue) # all_data json.loads(result.stdout) # names [item[name] for item in all_data[data][items]] # 方案2 用jq在命令行层面过滤只把需要的数据传给Python jq_filter .data.items[].name # 一个jq过滤表达式提取所有name cmd [big-command, --output, json] # 通过管道将 big-command 的输出传给 jq result subprocess.run([sh, -c, f{ .join(cmd)} | jq -c {jq_filter}], capture_outputTrue, textTrue, shellTrue) # 这里用了shell因为用了管道 if result.returncode 0: # jq的 -c 参数输出紧凑格式每行一个JSON字符串这里是每个name names [line.strip() for line in result.stdout.splitlines()] # 去掉引号 print(fExtracted names: {names}) else: print(fjq processing failed: {result.stderr})注意方案2中我们为了使用管道|不得已使用了shellTrue。为了安全应确保cmd和jq_filter是硬编码或经过严格校验的。更好的做法是使用subprocess.Popen建立两个进程的管道但这更复杂。4.5 编码问题永恒的痛当命令输出包含非ASCII字符如中文、表情符号时编码问题就来了。result subprocess.run([command_that_outputs_utf8], capture_outputTrue) # 默认capture_output得到的是bytes output_bytes result.stdout # 尝试用utf-8解码如果失败则用系统默认编码或忽略错误 try: output_text output_bytes.decode(utf-8) except UnicodeDecodeError: # 可能不是utf-8尝试其他编码如系统locale import locale default_encoding locale.getpreferredencoding(False) output_text output_bytes.decode(default_encoding, errorsignore) # 或replace print(fWarning: Used fallback encoding {default_encoding}) # 更稳健的做法如果知道命令的编码直接指定 # 例如在Windows中文环境下许多命令输出GBK result subprocess.run([dir], capture_outputTrue, shellTrue) output_text result.stdout.decode(gbk) # Windows cmd默认编码最佳实践尽可能在UTF-8环境下工作。设置环境变量PYTHONUTF81或确保你的脚本和终端都使用UTF-8。对于不可控的命令做好解码错误的异常处理。5. 综合案例构建一个健壮的命令行工具封装器让我们把上面的所有知识点融合起来编写一个相对健壮的类用于封装任何可能输出JSON的命令行工具。import subprocess import json import threading import queue import time from typing import List, Optional, Union, Callable, Any class JsonCommandWrapper: 一个封装命令行工具的类支持同步/异步执行、JSON解析、实时输出捕获和超时控制。 def __init__(self, cmd_base: List[str]): 初始化。 参数: cmd_base: 命令的基础列表如 [aws, ec2, describe-instances] self.cmd_base cmd_base self.process None self._stdout_queue queue.Queue() self._stderr_queue queue.Queue() def run_sync(self, args: List[str] None, input_data: str None, timeout: float 60.0, env: dict None, cwd: str None) - dict: 同步执行命令等待完成并返回解析后的JSON。 参数: args: 附加参数列表。 input_data: 要发送到stdin的字符串。 timeout: 超时时间秒。 env: 环境变量字典为None则继承。 cwd: 工作目录。 返回: 解析后的JSON数据字典或列表。 异常: subprocess.CalledProcessError: 命令执行失败。 subprocess.TimeoutExpired: 命令超时。 json.JSONDecodeError: 输出无法解析为JSON。 ValueError: 命令输出为空或非JSON。 cmd self.cmd_base (args if args else []) try: result subprocess.run(cmd, inputinput_data, capture_outputTrue, textTrue, checkTrue, timeouttimeout, envenv, cwdcwd) except subprocess.CalledProcessError as e: # 可以在这里添加自定义错误日志 raise RuntimeError(fCommand { .join(cmd)} failed with code {e.returncode}. Stderr: {e.stderr}) from e except subprocess.TimeoutExpired as e: raise RuntimeError(fCommand { .join(cmd)} timed out after {timeout}s.) from e stdout result.stdout.strip() if not stdout: raise ValueError(Command produced no output on stdout.) try: return json.loads(stdout) except json.JSONDecodeError as e: # 尝试从输出中提取JSON如果被日志包围 for line in stdout.splitlines(): line line.strip() if line.startswith({) and line.endswith(}) or line.startswith([) and line.endswith(]): try: return json.loads(line) except json.JSONDecodeError: continue # 如果都没提取到抛出原始异常 raise json.JSONDecodeError(fCould not parse JSON from stdout. Error: {e}. Output preview: {stdout[:500]}, e.doc, e.pos) from e def start_async(self, args: List[str] None, env: dict None, cwd: str None): 异步启动命令开始实时捕获输出。 if self.process is not None and self.process.poll() is None: raise RuntimeError(A process is already running.) cmd self.cmd_base (args if args else []) self.process subprocess.Popen(cmd, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1, envenv, cwdcwd, universal_newlinesTrue) # 启动线程来异步读取stdout和stderr避免阻塞主线程 def enqueue_output(out, queue): for line in iter(out.readline, ): queue.put(line) out.close() self._stdout_queue queue.Queue() self._stderr_queue queue.Queue() threading.Thread(targetenqueue_output, args(self.process.stdout, self._stdout_queue), daemonTrue).start() threading.Thread(targetenqueue_output, args(self.process.stderr, self._stderr_queue), daemonTrue).start() def read_stdout_line(self, blockTrue, timeoutNone) - Optional[str]: 从stdout队列中读取一行如果可用。 try: return self._stdout_queue.get(blockblock, timeouttimeout) except queue.Empty: return None def read_stderr_line(self, blockTrue, timeoutNone) - Optional[str]: 从stderr队列中读取一行如果可用。 try: return self._stderr_queue.get(blockblock, timeouttimeout) except queue.Empty: return None def write_stdin(self, data: str): 向进程的stdin写入数据。 if self.process and self.process.stdin: self.process.stdin.write(data \n) self.process.stdin.flush() def wait_for_json_line(self, timeout: float None) - Optional[Any]: 等待并从stdout中读取并解析一行JSON。 这是一个阻塞调用直到读到有效的JSON或超时。 start_time time.time() while True: if timeout and (time.time() - start_time) timeout: return None line self.read_stdout_line(blockTrue, timeout0.1) # 短超时以便检查进程状态 if line is None: # 检查进程是否已结束 if self.process and self.process.poll() is not None: return None continue line line.strip() if not line: continue try: return json.loads(line) except json.JSONDecodeError: # 不是JSON继续读取 # 可以选择将这一行作为普通日志处理 print(f[LOG] {line}) continue def terminate(self): 终止正在运行的进程。 if self.process and self.process.poll() is None: self.process.terminate() try: self.process.wait(timeout5) except subprocess.TimeoutExpired: self.process.kill() self.process.wait() def get_return_code(self) - Optional[int]: 获取进程的返回码如果仍在运行则返回None。 if self.process: return self.process.poll() return None # 使用示例 if __name__ __main__: # 示例模拟一个长时间运行、间歇性输出JSON状态的工具 simulator_script import json import sys import time import random for i in range(1, 6): time.sleep(random.uniform(0.5, 1.5)) status {id: i, progress: i*20, message: fProcessing item {i}} print(json.dumps(status)) sys.stdout.flush() print([INFO] Task completed successfully., filesys.stderr) with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(simulator_script) sim_path f.name wrapper JsonCommandWrapper([python, sim_path]) print(Starting async process...) wrapper.start_async() try: while True: # 等待并处理JSON状态更新 status wrapper.wait_for_json_line(timeout10) if status is None: print(No more JSON output or timeout.) break print(fStatus update: {status}) # 这里可以根据status做业务逻辑 # 同时检查一下有没有错误日志 err_line wrapper.read_stderr_line(blockFalse) while err_line: print(fError stream: {err_line.strip()}) err_line wrapper.read_stderr_line(blockFalse) finally: # 清理 return_code wrapper.get_return_code() if return_code is None: print(Process still running, terminating...) wrapper.terminate() else: print(fProcess exited with code: {return_code}) import os os.unlink(sim_path)这个JsonCommandWrapper类展示了如何将subprocess的复杂功能封装成一个易于使用的接口。它处理了同步/异步执行、JSON解析、实时读取、错误流处理、超时和资源清理是一个可以在实际项目中使用的坚实基础。当然根据具体需求你还可以增加更多功能比如信号处理、更复杂的JSON提取逻辑、输出回调函数等。
1. 项目缘起为什么我们总在命令行和程序之间“反复横跳”做开发或者运维的朋友估计都经历过这种场景你写了一个Python脚本需要调用一个系统命令比如用ffmpeg转码一个视频或者用curl获取一个API接口的数据。这时候你可能会想直接在脚本里os.system()一下不就完了确实对于一次性的、简单的、不需要交互的命令os.system()或者os.popen()勉强够用。但一旦需求变得复杂——你需要实时获取命令的输出、需要向命令输入参数、需要判断命令是否执行成功、甚至需要解析命令返回的结构化数据比如JSON——你就会发现os模块提供的那些方法就像一把生锈的瑞士军刀用起来处处掣肘。我最近就踩了这么一个大坑。一个数据处理的自动化任务需要调用一个外部的命令行工具来清洗数据这个工具不仅执行时间长还会分阶段输出进度信息JSON格式并且中途可能需要根据上一步的结果输入确认指令。一开始用os.system结果脚本“卡死”了因为命令在等待输入而我的脚本根本不知道。后来换os.popen能读到输出但又没法区分标准输出和错误输出而且命令一结束管道就关了想再交互也没门。更头疼的是那个JSON输出混在了一大堆日志文本里手动解析起来简直是一场灾难。正是这些“痛不欲生”的经历让我下定决心把Python标准库里的subprocess模块彻底吃透。subprocess才是Python与系统外壳Shell打交道的“官方指定武器库”。它不仅能替代os.system和os.popen更重要的是它提供了对子进程近乎完全的控制能力你可以重定向输入输出、等待进程结束、获取返回码甚至进行复杂的实时交互。结合Python强大的字符串处理和json模块我们就能优雅地完成“执行命令 - 交互 - 等待 - 判断状态 - 解析结果”这一整套流程。这篇文章我就结合自己趟过的坑手把手带你掌握subprocess的核心用法并重点解决如何与产生JSON输出的命令行工具进行协作。无论你是想自动化运维脚本、集成第三方命令行工具还是构建复杂的任务流水线这些内容都能让你少走弯路。2. 初识subprocess从“黑盒子”到“透明管道”在深入具体代码之前我们得先理解subprocess设计的基本哲学。它把我们要运行的命令行程序看作一个“子进程”。这个子进程有三个标准的数据流stdin (标准输入)子进程从这里读取数据。对应我们在终端里打字。stdout (标准输出)子进程把正常的输出信息写到这里。对应终端里显示的正常结果。stderr (标准错误)子进程把错误和警告信息写到这里。对应终端里红色的错误提示。subprocess的核心能力就是让我们在Python父进程中能够创建并精细地控制这些数据流的管道。我们可以选择不捕获让子进程的输出直接打印到终端和父进程共享。捕获到变量把输出内容抓取到Python的变量里方便后续处理。重定向到文件把输出写入文件。提供一个输入源我们可以预先准备好数据或者实时生成数据喂给子进程的stdin。subprocess模块提供了几个不同层次的函数来启动子进程我们从最简单常用的说起。2.1subprocess.run()一站式解决方案这是Python 3.5之后推荐的首选方法它封装了最常见的用例执行一个命令等待它完成然后返回一个包含结果的CompletedProcess对象。对于大多数“执行命令并获取结果”的场景它都是最佳选择。import subprocess import json # 示例1执行一个简单的命令并捕获其输出 result subprocess.run([ls, -l], capture_outputTrue, textTrue) print(f返回码: {result.returncode}) print(f标准输出:\n{result.stdout}) # 如果命令出错错误信息在result.stderr里 # 示例2执行一个返回JSON的命令并直接解析 # 假设我们有一个工具 data-tool 它接受一个 --format json 参数 result subprocess.run([data-tool, --format, json], capture_outputTrue, textTrue, checkTrue) if result.returncode 0: try: data json.loads(result.stdout) # 解析JSON print(f解析成功获取到数据: {data}) except json.JSONDecodeError as e: print(fJSON解析失败: {e}\n原始输出:{result.stdout}) else: print(f命令执行失败: {result.stderr})关键参数解析args: 命令列表。强烈建议以列表形式传入如[ls, -l]。这可以避免Shell注入攻击并且参数中的空格、引号会被正确处理。如果非要用字符串必须设置shellTrue但这会引入安全风险。capture_outputTrue: 这是stdoutsubprocess.PIPE, stderrsubprocess.PIPE的快捷方式。意味着同时捕获标准输出和错误输出。textTrue(或universal_newlinesTrue): 让stdout和stderr以字符串Pythonstr形式返回而不是字节序列bytes。这样我们就不用再手动.decode(utf-8)了。checkTrue: 如果子进程返回非零退出码通常表示失败它会抛出一个CalledProcessError异常。这对于需要确保命令成功的场景非常有用。timeout: 设置超时时间秒。如果命令执行超过这个时间会抛出TimeoutExpired异常。subprocess.run()是同步的它会一直阻塞直到命令执行完毕。对于长时间运行或需要交互的命令我们还需要其他工具。2.2subprocess.Popen()底层控制与实时交互当subprocess.run()无法满足需求时——比如你需要与进程进行实时对话一问一答或者需要同时管理多个长时间运行的进程——就该Popen出场了。Popen是subprocess模块的基石run()函数内部也是用它实现的。它给了你最大的灵活性但复杂度也更高。Popen创建子进程后立即返回不会等待进程结束。你需要通过它返回的Popen对象来与进程交互。import subprocess import time # 启动一个可能会等待输入的命令比如一个简单的交互式Python解释器 # 注意这里我们同时打开了stdin, stdout, stderr三个管道 proc subprocess.Popen([python, -i], # -i 进入交互模式 stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, # 同样使用文本模式 bufsize1, # 行缓冲方便实时读写 universal_newlinesTrue) # 现在proc就是一个代表子进程的对象 print(f子进程PID: {proc.pid}) # 向子进程的stdin写入命令并立即刷新缓冲区 proc.stdin.write(print(“Hello from subprocess!”)\n) proc.stdin.flush() # 确保数据被发送 # 从子进程的stdout读取一行输出 # 注意如果子进程没有输出这里会阻塞。可以使用select或线程来处理。 output_line proc.stdout.readline() print(f收到输出: {output_line}) # 再发送一个退出命令 proc.stdin.write(exit()\n) proc.stdin.flush() # 等待进程结束并获取返回码 return_code proc.wait(timeout5) # 等待最多5秒 print(f进程结束返回码: {return_code}) # 最后读取所有剩余的输出来清理管道 remaining_output, remaining_errors proc.communicate() print(f剩余输出: {remaining_output})关键方法与属性proc.stdin,proc.stdout,proc.stderr: 文件对象用于与子进程通信。proc.pid: 子进程的进程ID。proc.poll(): 检查进程是否已终止。如果已终止返回退出码否则返回None。这是判断“是否结束”的非阻塞方法。proc.wait(timeoutNone): 等待进程终止。可设置超时。proc.communicate(inputNone, timeoutNone):这是处理交互和获取输出的“瑞士军刀”。它向stdin发送数据可选然后等待进程结束同时收集所有的stdout和stderr数据。对于一次性发送所有输入并获取所有输出的场景它比手动读写stdin/stdout更安全能避免死锁。proc.terminate(): 发送SIGTERM信号请求进程终止。proc.kill(): 发送SIGKILL信号强制杀死进程。重要经验处理输出时的死锁问题这是一个经典的坑。子进程的缓冲区是有限的。如果你通过proc.stdout.read()试图读取大量数据而子进程又在等待你通过proc.stdin.write()输入数据但它的stdout缓冲区已经满了它就会卡住。双方都在等对方这就形成了死锁。黄金法则如果需要交互考虑使用proc.communicate(input_data)。它会处理好所有缓冲。如果必须手动读写请确保读写顺序正确并且考虑使用线程一个线程专门读stdout一个线程专门写stdin。对于输出量未知的命令使用iter(proc.stdout.readline, )来逐行读取而不是一次性read()。3. 实战与输出JSON的命令行工具共舞现在我们来解决一个更具体、也更常见的需求调用一个命令行工具这个工具会输出结构化的JSON数据我们需要在Python中捕获并解析它。这可能是调用一个云平台的CLI如aws cli,kubectl一个数据转换工具如jq或者任何自定义的脚本。3.1 场景一简单调用一次性获取JSON这是最简单的情况。命令执行后直接输出一个完整的JSON字符串。import subprocess import json def run_command_and_parse_json(cmd_args): 执行命令并尝试解析其标准输出为JSON。 参数: cmd_args (list): 命令及其参数的列表例如 [aws, ec2, describe-instances, --output, json] 返回: tuple: (success(bool), data(dict/list), error_message(str)) try: # 使用run设置checkTrue确保命令成功超时设为30秒 result subprocess.run(cmd_args, capture_outputTrue, textTrue, checkTrue, timeout30) # 命令执行成功尝试解析JSON try: data json.loads(result.stdout) return True, data, None except json.JSONDecodeError as e: # JSON解析失败可能是工具输出了非JSON内容如警告信息 error_msg fFailed to decode JSON from stdout. Error: {e}. Stdout preview: {result.stdout[:200]} return False, None, error_msg except subprocess.CalledProcessError as e: # 命令本身执行失败返回非零码 error_msg fCommand failed with return code {e.returncode}. Stderr: {e.stderr} return False, None, error_msg except subprocess.TimeoutExpired as e: # 命令超时 error_msg fCommand timed out after {e.timeout} seconds. return False, None, error_msg except FileNotFoundError as e: # 命令不存在 error_msg fCommand not found: {cmd_args[0]} return False, None, error_msg # 使用示例 success, data, error run_command_and_parse_json([echo, {name: Alice, age: 30}]) if success: print(fParsed data: {data}) print(fName: {data[name]}) else: print(fError: {error})这里的关键点错误处理分层我们区分了“命令执行失败”和“JSON解析失败”。有时候命令成功了但可能输出了额外的日志行污染了JSON导致解析失败。超时控制使用timeout参数避免脚本因某个命令卡死而永远挂起。文本模式textTrue确保我们拿到的是字符串可以直接给json.loads()。3.2 场景二处理混合输出JSON与日志交织很多命令行工具尤其是开发中的或者调试模式下的会把JSON数据和普通的日志信息一起输出到stdout。比如[INFO] Starting processing... {status: processing, progress: 25} [DEBUG] Fetched 100 records. {status: processing, progress: 50} {status: completed, result: [...]} [INFO] All done.我们的目标是从中提取出有效的JSON行。这里subprocess.run()仍然可用但我们需要对输出进行后处理。import subprocess import json def extract_json_from_output(cmd_args): 执行命令并从其标准输出中提取所有有效的JSON行。 result subprocess.run(cmd_args, capture_outputTrue, textTrue) json_objects [] for line in result.stdout.splitlines(): line line.strip() if not line: continue # 尝试将每一行解析为JSON try: obj json.loads(line) json_objects.append(obj) print(fFound JSON: {obj}) except json.JSONDecodeError: # 不是JSON可能是普通日志可以选择记录或忽略 # print(fIgnored log line: {line}) pass return json_objects, result.returncode # 模拟一个输出混合内容的命令 test_script import json import sys import time print(Start..., filesys.stderr) # 错误输出 print([LOG] Initializing, filesys.stdout) time.sleep(0.1) print(json.dumps({step: 1, data: sample})) print([LOG] Processing...) print(json.dumps({step: 2, data: processed})) print([LOG] Done.) # 将上面的脚本写到一个临时文件并执行 import tempfile with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(test_script) temp_file_path f.name try: objects, returncode extract_json_from_output([python, temp_file_path]) print(f\nExtracted {len(objects)} JSON objects. Return code: {returncode}) for obj in objects: print(f - Step {obj[step]}: {obj[data]}) finally: import os os.unlink(temp_file_path) # 清理临时文件策略分析这种方法简单粗暴但假设了每个JSON对象都独占一行。如果JSON被漂亮打印Pretty-print成多行这种方法就会失败。对于多行JSON我们需要更复杂的解析器或者使用专门的工具如jq来预处理输出。3.3 场景三实时流式处理JSON输出对于一些长时间运行的任务如数据导入、模型训练它们可能会持续输出进度信息或中间结果JSON格式。我们可能希望实时处理这些数据而不是等命令全部结束。这就需要用到Popen进行实时读取。import subprocess import json import select # 用于非阻塞读取Unix-like系统 import sys def stream_and_process_json(cmd_args): 启动一个命令并实时读取其标准输出尝试解析并处理每一行JSON。 注意此示例在Unix-like系统Linux, macOS上工作最佳。 Windows上select()对管道的行为不同可能需要使用线程。 # 启动进程注意stderr我们重定向到stdout一起读或者单独处理 proc subprocess.Popen(cmd_args, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, # 也可以重定向到subprocess.STDOUT合并 textTrue, bufsize1, # 行缓冲 universal_newlinesTrue) print(fStarted process {proc.pid}. Streaming output...) # 使用select来监控哪些管道有数据可读避免阻塞 # 这是一个简化的例子实际生产环境可能需要更健壮的错误处理 while True: # 检查进程是否已经结束 if proc.poll() is not None: print(f\nProcess finished with return code: {proc.poll()}) break # 监控stdout和stderr这里我们分开监控 readable, _, _ select.select([proc.stdout, proc.stderr], [], [], 0.1) # 0.1秒超时 for stream in readable: line stream.readline() if not line: # EOF continue if stream is proc.stdout: # 处理标准输出行 line line.rstrip(\n) try: data json.loads(line) # **这里是你的实时处理逻辑** print(f[STDOUT JSON] 实时处理: {data}) # 例如更新进度条、将数据存入队列、触发回调等 if data.get(status) completed: print(检测到任务完成信号) except json.JSONDecodeError: # 不是JSON当作普通日志处理 print(f[STDOUT LOG] {line}) elif stream is proc.stderr: # 处理错误输出行 print(f[STDERR] {line.rstrip()}, filesys.stderr) # 循环结束后读取可能剩余的任何输出 remaining_out, remaining_err proc.communicate() if remaining_out: print(f剩余标准输出: {remaining_out}) if remaining_err: print(f剩余错误输出: {remaining_err}, filesys.stderr) # 示例模拟一个流式输出JSON的命令 streaming_script import json import sys import time for i in range(5): data {iteration: i, value: i * 10, status: running} print(json.dumps(data)) sys.stdout.flush() # 确保立即输出而不是等缓冲区满 time.sleep(1) # 模拟耗时操作 print(json.dumps({iteration: 5, value: 50, status: completed})) sys.stdout.flush() print(Some final log message., filesys.stderr) with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(streaming_script) temp_file_path f.name try: stream_and_process_json([python, temp_file_path]) finally: import os os.unlink(temp_file_path)关键点与陷阱缓冲Buffering这是流式处理最大的敌人。子进程的输出可能被缓冲在内存中直到缓冲区满或进程结束才一次性吐出。我们通过设置bufsize1行缓冲和在子进程脚本中调用sys.stdout.flush()来缓解。对于非Python程序可能需要在调用时使用stdbuf等工具如[stdbuf, -oL, your-command]在Linux上。非阻塞读取使用select模块可以避免在readline()上无限期阻塞。注意select在Windows上对普通文件/管道的行为有限制生产环境的跨平台代码可能需要使用threading模块创建单独的读取线程。错误流处理错误输出stderr也需要被及时读取否则如果它被填满子进程也可能被阻塞。上面的例子展示了分开处理stdout和stderr。进程状态检查proc.poll()用于非阻塞地检查进程是否结束。4. 进阶技巧与避坑指南掌握了基本用法后我们来看看一些能让你代码更健壮、更高效的进阶技巧和常见陷阱。4.1 环境变量与工作目录子进程默认会继承父进程的环境变量。但有时我们需要修改。import subprocess import os # 1. 添加或修改环境变量 my_env os.environ.copy() my_env[MY_CUSTOM_VAR] some_value my_env[PATH] /my/custom/bin: my_env[PATH] # 添加自定义路径到PATH开头 result subprocess.run([some_command], capture_outputTrue, textTrue, envmy_env) # 2. 设置工作目录相当于 cd 到那个目录再执行命令 result subprocess.run([ls, -la], capture_outputTrue, textTrue, cwd/path/to/target/directory)4.2 ShellTrue便利与风险并存subprocess.run(ls -l | grep .py, shellTrue)这种写法很诱人因为它可以直接使用Shell的特性管道|、重定向、通配符*等。但这是极度危险的特别是当命令字符串来自用户输入或外部数据时极易引发Shell注入攻击。# 危险如果user_input来自不可信源比如 ; rm -rf / user_input “somefile.txt” subprocess.run(fls -l {user_input}, shellTrue) # 灾难 # 安全做法使用列表参数让subprocess处理参数转义 subprocess.run([ls, -l, user_input]) # 即使user_input是 ; rm -rf /它也会被当作一个文件名参数不会被解析为命令原则除非有非常充分的理由如必须使用复杂的Shell特性否则永远使用列表形式传递参数并避免shellTrue。4.3 超时、终止与资源清理长时间运行或失控的子进程会耗尽资源。try: # 设置超时 result subprocess.run([sleep, 10], timeout5, capture_outputTrue, textTrue) except subprocess.TimeoutExpired as e: print(f命令超时进程可能仍在运行。) # 你可以尝试终止它 # e.cmd 是命令 e.timeout 是超时时间 # 但run()已经帮我们处理了终止如果超时子进程会被kill。 # 如果使用Popen则需要手动处理 # proc.terminate() # 先友好地请求终止 # try: # proc.wait(timeout2) # 给一点时间结束 # except subprocess.TimeoutExpired: # proc.kill() # 强制杀死 # proc.wait() # 等待清理 # 使用Popen时务必确保关闭管道 proc subprocess.Popen(...) try: # ... 与进程交互 ... finally: # 确保管道被关闭避免资源泄漏 if proc.stdin: proc.stdin.close() # 对于stdout/stderrcommunicate()或wait()会自动关闭但手动关闭也无害 proc.wait()4.4 解析复杂JSON与使用jq预处理有时命令行工具输出的JSON非常庞大或复杂或者嵌套很深。直接在Python里用json.loads()解析整个字符串可能效率不高或者你只关心其中一小部分。这时可以借助强大的命令行JSON处理器jq进行预处理。import subprocess import json # 假设 big-command 输出一个巨大的JSON我们只想要其中的 data.items[].name 字段 # 方案1 全部拉取到Python再解析内存可能压力大 # result subprocess.run([big-command, --output, json], capture_outputTrue, textTrue) # all_data json.loads(result.stdout) # names [item[name] for item in all_data[data][items]] # 方案2 用jq在命令行层面过滤只把需要的数据传给Python jq_filter .data.items[].name # 一个jq过滤表达式提取所有name cmd [big-command, --output, json] # 通过管道将 big-command 的输出传给 jq result subprocess.run([sh, -c, f{ .join(cmd)} | jq -c {jq_filter}], capture_outputTrue, textTrue, shellTrue) # 这里用了shell因为用了管道 if result.returncode 0: # jq的 -c 参数输出紧凑格式每行一个JSON字符串这里是每个name names [line.strip() for line in result.stdout.splitlines()] # 去掉引号 print(fExtracted names: {names}) else: print(fjq processing failed: {result.stderr})注意方案2中我们为了使用管道|不得已使用了shellTrue。为了安全应确保cmd和jq_filter是硬编码或经过严格校验的。更好的做法是使用subprocess.Popen建立两个进程的管道但这更复杂。4.5 编码问题永恒的痛当命令输出包含非ASCII字符如中文、表情符号时编码问题就来了。result subprocess.run([command_that_outputs_utf8], capture_outputTrue) # 默认capture_output得到的是bytes output_bytes result.stdout # 尝试用utf-8解码如果失败则用系统默认编码或忽略错误 try: output_text output_bytes.decode(utf-8) except UnicodeDecodeError: # 可能不是utf-8尝试其他编码如系统locale import locale default_encoding locale.getpreferredencoding(False) output_text output_bytes.decode(default_encoding, errorsignore) # 或replace print(fWarning: Used fallback encoding {default_encoding}) # 更稳健的做法如果知道命令的编码直接指定 # 例如在Windows中文环境下许多命令输出GBK result subprocess.run([dir], capture_outputTrue, shellTrue) output_text result.stdout.decode(gbk) # Windows cmd默认编码最佳实践尽可能在UTF-8环境下工作。设置环境变量PYTHONUTF81或确保你的脚本和终端都使用UTF-8。对于不可控的命令做好解码错误的异常处理。5. 综合案例构建一个健壮的命令行工具封装器让我们把上面的所有知识点融合起来编写一个相对健壮的类用于封装任何可能输出JSON的命令行工具。import subprocess import json import threading import queue import time from typing import List, Optional, Union, Callable, Any class JsonCommandWrapper: 一个封装命令行工具的类支持同步/异步执行、JSON解析、实时输出捕获和超时控制。 def __init__(self, cmd_base: List[str]): 初始化。 参数: cmd_base: 命令的基础列表如 [aws, ec2, describe-instances] self.cmd_base cmd_base self.process None self._stdout_queue queue.Queue() self._stderr_queue queue.Queue() def run_sync(self, args: List[str] None, input_data: str None, timeout: float 60.0, env: dict None, cwd: str None) - dict: 同步执行命令等待完成并返回解析后的JSON。 参数: args: 附加参数列表。 input_data: 要发送到stdin的字符串。 timeout: 超时时间秒。 env: 环境变量字典为None则继承。 cwd: 工作目录。 返回: 解析后的JSON数据字典或列表。 异常: subprocess.CalledProcessError: 命令执行失败。 subprocess.TimeoutExpired: 命令超时。 json.JSONDecodeError: 输出无法解析为JSON。 ValueError: 命令输出为空或非JSON。 cmd self.cmd_base (args if args else []) try: result subprocess.run(cmd, inputinput_data, capture_outputTrue, textTrue, checkTrue, timeouttimeout, envenv, cwdcwd) except subprocess.CalledProcessError as e: # 可以在这里添加自定义错误日志 raise RuntimeError(fCommand { .join(cmd)} failed with code {e.returncode}. Stderr: {e.stderr}) from e except subprocess.TimeoutExpired as e: raise RuntimeError(fCommand { .join(cmd)} timed out after {timeout}s.) from e stdout result.stdout.strip() if not stdout: raise ValueError(Command produced no output on stdout.) try: return json.loads(stdout) except json.JSONDecodeError as e: # 尝试从输出中提取JSON如果被日志包围 for line in stdout.splitlines(): line line.strip() if line.startswith({) and line.endswith(}) or line.startswith([) and line.endswith(]): try: return json.loads(line) except json.JSONDecodeError: continue # 如果都没提取到抛出原始异常 raise json.JSONDecodeError(fCould not parse JSON from stdout. Error: {e}. Output preview: {stdout[:500]}, e.doc, e.pos) from e def start_async(self, args: List[str] None, env: dict None, cwd: str None): 异步启动命令开始实时捕获输出。 if self.process is not None and self.process.poll() is None: raise RuntimeError(A process is already running.) cmd self.cmd_base (args if args else []) self.process subprocess.Popen(cmd, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue, bufsize1, envenv, cwdcwd, universal_newlinesTrue) # 启动线程来异步读取stdout和stderr避免阻塞主线程 def enqueue_output(out, queue): for line in iter(out.readline, ): queue.put(line) out.close() self._stdout_queue queue.Queue() self._stderr_queue queue.Queue() threading.Thread(targetenqueue_output, args(self.process.stdout, self._stdout_queue), daemonTrue).start() threading.Thread(targetenqueue_output, args(self.process.stderr, self._stderr_queue), daemonTrue).start() def read_stdout_line(self, blockTrue, timeoutNone) - Optional[str]: 从stdout队列中读取一行如果可用。 try: return self._stdout_queue.get(blockblock, timeouttimeout) except queue.Empty: return None def read_stderr_line(self, blockTrue, timeoutNone) - Optional[str]: 从stderr队列中读取一行如果可用。 try: return self._stderr_queue.get(blockblock, timeouttimeout) except queue.Empty: return None def write_stdin(self, data: str): 向进程的stdin写入数据。 if self.process and self.process.stdin: self.process.stdin.write(data \n) self.process.stdin.flush() def wait_for_json_line(self, timeout: float None) - Optional[Any]: 等待并从stdout中读取并解析一行JSON。 这是一个阻塞调用直到读到有效的JSON或超时。 start_time time.time() while True: if timeout and (time.time() - start_time) timeout: return None line self.read_stdout_line(blockTrue, timeout0.1) # 短超时以便检查进程状态 if line is None: # 检查进程是否已结束 if self.process and self.process.poll() is not None: return None continue line line.strip() if not line: continue try: return json.loads(line) except json.JSONDecodeError: # 不是JSON继续读取 # 可以选择将这一行作为普通日志处理 print(f[LOG] {line}) continue def terminate(self): 终止正在运行的进程。 if self.process and self.process.poll() is None: self.process.terminate() try: self.process.wait(timeout5) except subprocess.TimeoutExpired: self.process.kill() self.process.wait() def get_return_code(self) - Optional[int]: 获取进程的返回码如果仍在运行则返回None。 if self.process: return self.process.poll() return None # 使用示例 if __name__ __main__: # 示例模拟一个长时间运行、间歇性输出JSON状态的工具 simulator_script import json import sys import time import random for i in range(1, 6): time.sleep(random.uniform(0.5, 1.5)) status {id: i, progress: i*20, message: fProcessing item {i}} print(json.dumps(status)) sys.stdout.flush() print([INFO] Task completed successfully., filesys.stderr) with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(simulator_script) sim_path f.name wrapper JsonCommandWrapper([python, sim_path]) print(Starting async process...) wrapper.start_async() try: while True: # 等待并处理JSON状态更新 status wrapper.wait_for_json_line(timeout10) if status is None: print(No more JSON output or timeout.) break print(fStatus update: {status}) # 这里可以根据status做业务逻辑 # 同时检查一下有没有错误日志 err_line wrapper.read_stderr_line(blockFalse) while err_line: print(fError stream: {err_line.strip()}) err_line wrapper.read_stderr_line(blockFalse) finally: # 清理 return_code wrapper.get_return_code() if return_code is None: print(Process still running, terminating...) wrapper.terminate() else: print(fProcess exited with code: {return_code}) import os os.unlink(sim_path)这个JsonCommandWrapper类展示了如何将subprocess的复杂功能封装成一个易于使用的接口。它处理了同步/异步执行、JSON解析、实时读取、错误流处理、超时和资源清理是一个可以在实际项目中使用的坚实基础。当然根据具体需求你还可以增加更多功能比如信号处理、更复杂的JSON提取逻辑、输出回调函数等。
