Python：调用 Linux Shell 命令

目录

• 1. os.system —— 跑命令，不取输出
• 2. subprocess.run —— 首选方案
  • 拿不到返回码才用 try/except
• 3. subprocess.Popen —— 流式输出 / 长时间进程
  • 喂入 stdin
• 4. 异步执行：asyncio
• 5. 安全：永远不要拼接用户输入
• 6. 常见需求速查
• 7. 常见错误
• 8. 一键模板

---

Python 提供了多种方式来执行 Shell 命令。选择哪种方式取决于你是否需要拿到输出、
是否需要实时流式输出、是否要并发。

方式	适用场景	推荐度
os.system	只关心退出码，不取输出	★★
subprocess.run	绝大多数日常场景（首选）	★★★
subprocess.Popen	需要流式读取、长期进程、双向通信	★★★
asyncio.subprocess	异步 / 并发	★★
os.popen	仅取输出，已经基本被 subprocess 取代	★

⚠️ commands 模块是 Python 2 时代的产物，Python 3 已删除，请使用 subprocess 替代。

---

1. os.system —— 跑命令，不取输出

import os

exit_code = os.system('echo "Hello World"')
print("exit code =", exit_code)   # 0 表示成功

• 优点：最简单。
• 缺点：拿不到 stdout / stderr；要拿输出必须用 subprocess 或重定向到文件。

---

2. subprocess.run —— 首选方案

import subprocess

# 列表形式：参数分隔交给系统，避免 shell 解析
result = subprocess.run(
    ["ls", "-l", "/tmp"],
    capture_output=True,   # 捕获 stdout / stderr
    text=True,             # 拿到 str，而不是 bytes
    timeout=5,             # 超时保护
    check=False,           # 失败是否抛异常，下面用更可控的方式
)

print("returncode:", result.returncode)
print("stdout:", result.stdout)
print("stderr:", result.stderr)

if result.returncode != 0:
    raise RuntimeError(f"命令失败: {result.stderr}")

要点：

参数	作用
args	命令列表，不要 用字符串拼接用户输入
capture_output/stdout=PIPE	捕获输出，否则直接继承父进程终端
text=True	拿到 str（默认是 bytes）
timeout=...	必加，避免命令挂起
check=True	返回非 0 时抛 CalledProcessError
cwd=...	指定工作目录
env=...	自定义环境变量
input=...	通过 stdin 喂数据

拿不到返回码才用 try/except

try:
    subprocess.run(["ls", "/no-such-dir"], check=True, capture_output=True, text=True)
except subprocess.CalledProcessError as e:
    print("命令失败:", e.returncode, e.stderr)

---

3. subprocess.Popen —— 流式输出 / 长时间进程

当你需要：

• 实时打印日志（tail -f、ffmpeg 转码进度等）
• 同时写 stdin、读 stdout
• 与进程交互式通信

import subprocess

proc = subprocess.Popen(
    ["bash", "-lc", "for i in 1 2 3; do echo line $i; sleep 1; done"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    bufsize=1,            # 行缓冲
)

# 实时读取 stdout
assert proc.stdout is not None
for line in proc.stdout:
    print(">>", line.rstrip())

# 等子进程结束
stdout, stderr = proc.communicate()
print("exit code:", proc.returncode)

text=True + bufsize=1 才能逐行读，否则会按系统块缓冲。

喂入 stdin

proc = subprocess.Popen(
    ["grep", "hello"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    text=True,
)
out, _ = proc.communicate(input="hello world\nbye\n")
print(out.rstrip())   # hello world

---

4. 异步执行：asyncio

当你要并发跑一堆命令或 IO 密集任务时：

import asyncio


async def run(cmd: str) -> tuple[int, str, str]:
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    return proc.returncode or 0, stdout.decode(), stderr.decode()


async def main():
    codes = await asyncio.gather(
        run("ls -l /tmp"),
        run("ls -l /var"),
    )
    for code, out, err in codes:
        print(code, out[:30], err[:30])


asyncio.run(main())

需要更高并发时，配合 asyncio.Semaphore 限流。

---

5. 安全：永远不要拼接用户输入

# ❌ 危险：用户输入直接拼到 shell 命令里
user = input("file: ")
os.system(f"cat {user}")           # 可以 ; rm -rf ~

# ✅ 安全 1：用列表 + subprocess.run，让系统处理转义
subprocess.run(["cat", user], check=True)

# ✅ 安全 2：必须用 shell 时，至少用 shlex.quote
import shlex
subprocess.run(f"cat {shlex.quote(user)}", shell=True, check=True)

---

6. 常见需求速查

需求	写法
拿 stdout	subprocess.run(..., capture_output=True, text=True)
抛错当失败	check=True
设超时	timeout=5
改工作目录	cwd="/path/to/work"
注入环境变量	env={"PATH": "/usr/bin", "FOO": "bar"}
走 stdin	subprocess.run(..., input="data", text=True)
实时输出	Popen + 逐行 readline 或 for line in proc.stdout
异步并发	asyncio.create_subprocess_shell / _exec
调试路径	把 args 列表打印出来确认

---

7. 常见错误

• 忘了 text=True → 拿到 bytes，打印时会出现 b'...'。
• shell=True + 拼接字符串 → 命令注入。
• 没设 timeout → 命令挂起，整个程序跟着卡住。
• 捕获了输出又没消费 → 管道缓冲写满，子进程阻塞。用 communicate() 解决。
• 把 subprocess.run 当 os.system 用 → 想取输出却没传 capture_output=True，
  结果空字符串。

---

8. 一键模板

import subprocess


def sh(cmd: list[str], timeout: float = 10) -> str:
    """执行命令并返回 stdout；失败抛异常。"""
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=True,
    )
    return result.stdout.rstrip("\n")


# 用法
print(sh(["ls", "-l"]))