subprocess

subprocess 模块

用来启动新进程，执行系统命令，拿到输出结果。

简单说，就是在 Python 里运行 shell 命令的工具。

subprocess:subprocess Module Code Documentation

run() 方法

最推荐的用法，简单直接。

import subprocess

# 执行命令
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)

print(result.stdout)  # 标准输出
print(result.stderr)  # 错误输出
print(result.returncode)  # 返回码，0表示成功

常用参数：

• args: 要执行的命令，列表或字符串
• shell: 是否通过 shell 执行，默认 False
• capture_output: 捕获输出，等同于 stdout=PIPE, stderr=PIPE
• text: 以文本模式返回，不用再 decode
• timeout: 超时时间（秒），超时会抛异常
• check: True 时，如果命令返回非 0 会抛异常

shell=True 的使用

有时候需要执行管道、重定向等 shell 特性：

import subprocess

# 这种情况需要 shell=True
result = subprocess.run('ls -l | grep py', shell=True, capture_output=True, text=True)
print(result.stdout)

# 或者传字符串
result = subprocess.run('echo "Hello World"', shell=True, capture_output=True, text=True)
print(result.stdout)

warning

shell=True 有安全风险，不要传入用户输入的内容，避免命令注入。

处理错误

import subprocess

try:
    result = subprocess.run(
        ['python', 'nonexist.py'],
        capture_output=True,
        text=True,
        check=True,  # 返回码非0时抛异常
        timeout=5
    )
except subprocess.CalledProcessError as e:
    print(f"命令执行失败，返回码: {e.returncode}")
    print(f"错误输出: {e.stderr}")
except subprocess.TimeoutExpired:
    print("命令执行超时")

Popen() 方法

更底层的接口，用于需要精细控制的场景。

import subprocess

# 启动进程
process = subprocess.Popen(
    ['python', '-u', 'script.py'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 实时读取输出
for line in process.stdout:
    print(line, end='')

# 等待进程结束
return_code = process.wait()
print(f"进程结束，返回码: {return_code}")

与进程交互

import subprocess

process = subprocess.Popen(
    ['python', '-u'],  # -u 让输出不缓冲
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 发送输入并获取输出
stdout, stderr = process.communicate(input='print("Hello from stdin")\n')
print(stdout)

实际例子

执行 git 命令

import subprocess

def git_status():
    result = subprocess.run(
        ['git', 'status', '--short'],
        capture_output=True,
        text=True
    )
    return result.stdout

def git_commit(message):
    subprocess.run(['git', 'add', '.'], check=True)
    subprocess.run(['git', 'commit', '-m', message], check=True)

print(git_status())

调用其他 Python 脚本

import subprocess

result = subprocess.run(
    ['python', 'data_process.py', '--input', 'data.csv'],
    capture_output=True,
    text=True,
    timeout=300  # 5分钟超时
)

if result.returncode == 0:
    print("处理成功")
    print(result.stdout)
else:
    print("处理失败")
    print(result.stderr)

Windows 特定

import subprocess
import sys

if sys.platform == 'win32':
    # Windows 下执行命令
    result = subprocess.run('dir', shell=True, capture_output=True, text=True)
    print(result.stdout)
    
    # 或者使用 cmd
    result = subprocess.run(
        ['cmd', '/c', 'dir'],
        capture_output=True,
        text=True
    )
    print(result.stdout)

常见问题

Q: 什么时候用 run()，什么时候用 Popen()？

A: 大部分情况用 run()。只有需要实时读取输出、手动控制进程生命周期时才用 Popen()。

Q: 命令参数要不要加引号？

A: 用列表形式时不需要引号，Python 会自动处理：

# 正确
subprocess.run(['echo', 'hello world'])

# 也可以，但不推荐
subprocess.run('echo "hello world"', shell=True)

Q: 为什么输出是空的？

A: 可能忘了 capture_output=True 或 stdout=subprocess.PIPE。

Q: 中文乱码怎么办？

A: 指定编码：

result = subprocess.run(
    ['python', 'script.py'],
    capture_output=True,
    encoding='utf-8'  # 或 'gbk' (Windows中文)
)

最佳实践

1. 优先用列表传参数，避免 shell 注入风险
2. 设置合理的 timeout，避免进程卡死
3. 使用 check=True 自动处理错误
4. 指定 text=True 或 encoding，避免处理字节
5. 不信任的输入不要用 shell=True

# 推荐写法
import subprocess

def run_command(cmd_list, timeout=30):
    """运行命令的封装函数"""
    try:
        result = subprocess.run(
            cmd_list,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=True
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"命令失败: {' '.join(cmd_list)}")
        print(f"错误: {e.stderr}")
        return None
    except subprocess.TimeoutExpired:
        print(f"命令超时: {' '.join(cmd_list)}")
        return None

# 使用
output = run_command(['ls', '-la'])
if output:
    print(output)