Python使用subprocess处理长时运行CLI命令的Pythonic方案

原有代码核心问题 #

• 未配置stderr捕获，process.stderr始终为None，无法拿到真实错误信息
• 未跟踪单条命令的完整生命周期，命令尚未执行完成就误判为失败，重复发起新调用，既浪费资源又可能触发云接口限流
• 固定5分钟累计等待时长完全不匹配10-20分钟的实际耗时，递归实现存在栈溢出风险，还存在函数名不匹配的低级错误（定义getJsonResponse却递归调用getShellJsonResponse）
• 使用shell=True引入不必要的shell注入风险和环境依赖，无法实时获取命令运行中的输出，长耗时场景下表现为程序假死
• 重试逻辑无判断依据，属于盲目重试。

重构实现核心原则 #

• 拆分命令为参数列表，禁用不必要的shell=True，符合PEP8安全规范
• 使用subprocess.Popen替代subprocess.run，实现进程状态轮询、流式输出读取，不设置全局超时，只要命令正常运行就持续等待
• 实时解析命令输出，匹配Running标识时输出旋转状态提示，和手动终端执行体验一致
• 仅在进程确实异常退出（非0返回码、输出解析失败）时触发重试，不干扰正常运行的命令
• 命令执行完成后自动从输出中提取合法JSON结果，兼容Azure CLI先输出运行提示、末尾返回JSON的格式。

完整实现代码 #

import subprocess
import time
import json
import sys
from typing import Dict

def run_az_long_running_command(cmd: list, max_retries: int = 3, retry_interval: int = 30) -> Dict:
    """
    执行长耗时Azure CLI命令，实时输出运行状态，等待命令完成后返回解析后的JSON结果
    :param cmd: 拆分后的命令参数列表，禁止传入拼接完成的命令字符串
    :param max_retries: 进程异常退出后的最大重试次数
    :param retry_interval: 失败重试间隔，单位秒
    """
    spin_chars = ['|', '/', '-', '\\']
    spin_idx = 0

    for attempt in range(max_retries + 1):
        print(f"命令启动，当前为第{attempt + 1}次执行尝试")
        # 启动子进程，合并stderr到stdout统一捕获，开启行缓冲实现实时输出读取
        process = subprocess.Popen(
            cmd,
            shell=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
            universal_newlines=True
        )

        output_cache = []
        last_active_time = time.time()

        # 轮询进程状态直到退出
        while True:
            line = process.stdout.readline()
            if line:
                output_cache.append(line)
                last_active_time = time.time()
                # 匹配运行中提示，输出旋转状态
                if "Running" in line:
                    print(f"\r命令执行中 {spin_chars[spin_idx % len(spin_chars)]}", end="", flush=True)
                    spin_idx += 1

            # 检测到进程退出，读取剩余管道内容后跳出循环
            if process.poll() is not None:
                remaining_output = process.stdout.read()
                if remaining_output:
                    output_cache.append(remaining_output)
                break

            # 短休眠降低CPU占用
            time.sleep(0.2)

        # 清理状态行，打印执行结果概要
        print("\r" + " " * 60 + "\r", end="", flush=True)
        return_code = process.returncode
        full_output = "".join(output_cache).strip()
        print(f"进程退出，返回码：{return_code}")

        if return_code == 0:
            # 从混合输出中提取末尾的JSON段
            try:
                json_start = full_output.rfind("{")
                json_end = full_output.rfind("}") + 1
                if json_start != -1 and json_end > json_start:
                    json_content = json.loads(full_output[json_start:json_end])
                    print(f"执行完成，最终状态：{json_content.get('status', '未知')}")
                    return json_content
                print(f"未在输出中检测到合法JSON，完整输出如下：\n{full_output}")
            except json.JSONDecodeError as e:
                print(f"JSON解析失败：{str(e)}，完整输出如下：\n{full_output}")

        # 执行失败判断是否重试
        if attempt < max_retries:
            print(f"本次执行失败，{retry_interval}秒后重试")
            time.sleep(retry_interval)
        else:
            error_msg = f"累计重试{max_retries}次仍执行失败，最后返回码：{return_code}\n输出内容：\n{full_output}"
            print(error_msg)
            sys.exit(1)

调用方式 #

将命令拆分为参数列表传入，避免shell解析带来的异常：

if __name__ == "__main__":
    target_cmd = [
        "az", "resource", "invoke-action",
        "--resource-group", "someRG",
        "--resource-type", "Microsoft.VirtualMachineImages/imageTemplates",
        "-n", "somename78686786976",
        "--action", "Run"
    ]
    final_result = run_az_long_running_command(target_cmd)
    print("命令返回结果：", final_result)

关键特性说明 #

• 无强制等待时长限制：未给子进程设置固定超时，只要进程正常运行就会持续等待，完全适配任意时长的长耗时命令，不会中途打断正在执行的操作
• 平滑状态反馈：运行期间输出旋转等待标识，无多余日志刷屏，和本地终端执行体验一致
• 可靠的重试逻辑：仅在进程异常退出、结果解析失败时触发重试，不会对正常运行的命令重复发起调用
• 无资源泄漏：进程退出后读取完所有管道内容，避免管道阻塞引发的僵尸进程问题
• 结果可靠性：自动从混合输出中提取JSON段，跳过前面的运行提示内容，直接返回结构化结果。

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