🐍 第14课：子进程

—— Python 调用系统命令

📌 本课目标

1️⃣ subprocess.run() 基础

import subprocess

# 基本调用
result = subprocess.run(["echo", "Hello, World!"], capture_output=True, text=True)
print(result.stdout)       # Hello, World!\n
print(result.returncode)   # 0

# 检查返回码
result = subprocess.run(["ls", "/nonexistent"], capture_output=True, text=True)
if result.returncode != 0:
    print(f"错误: {result.stderr}")

# check=True 自动抛异常
try:
    subprocess.run(["false"], check=True)
except subprocess.CalledProcessError as e:
    print(f"命令失败，返回码: {e.returncode}")

2️⃣ 输出捕获

import subprocess

# capture_output=True + text=True (推荐)
result = subprocess.run(
    ["ls", "-la"],
    capture_output=True,
    text=True,
)
print(result.stdout)

# 分离 stdout 和 stderr
result = subprocess.run(
    ["python3", "-c", "import sys; print('out'); print('err', file=sys.stderr)"],
    capture_output=True,
    text=True,
)
print(f"stdout: {result.stdout.strip()}")  # stdout: out
print(f"stderr: {result.stderr.strip()}")  # stderr: err

# 合并 stdout + stderr
result = subprocess.run(
    ["python3", "-c", "import sys; print('out'); print('err', file=sys.stderr)"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
)
print(result.stdout)  # 包含 out 和 err

3️⃣ 输入传递

import subprocess

# 传递输入
result = subprocess.run(
    ["python3", "-c", "data = input(); print(f'收到: {data}')"],
    input="Hello from Python",
    capture_output=True,
    text=True,
)
print(result.stdout)  # 收到: Hello from Python

# 管道方式（更灵活）
process = subprocess.Popen(
    ["python3", "-c", "print(input().upper())"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    text=True,
)
out, err = process.communicate("hello world")
print(out.strip())  # HELLO WORLD

4️⃣ 管道操作

import subprocess

# 模拟 shell 管道: cat file | grep pattern | sort
# 方法1: 使用 shell=True（简单但不安全）
# ⚠️ 仅用于可信输入！
result = subprocess.run(
    "echo 'hello\nworld\nhello' | sort | uniq",
    shell=True,
    capture_output=True,
    text=True,
)

# 方法2: 串联 Popen（推荐，更安全）
p1 = subprocess.Popen(
    ["echo", "hello\nworld\nhello"],
    stdout=subprocess.PIPE,
    text=True,
)
p2 = subprocess.Popen(
    ["sort"],
    stdin=p1.stdout,
    stdout=subprocess.PIPE,
    text=True,
)
p3 = subprocess.Popen(
    ["uniq"],
    stdin=p2.stdout,
    stdout=subprocess.PIPE,
    text=True,
)
p1.stdout.close()  # 允许 p1 接收 SIGPIPE
p2.stdout.close()  # 允许 p2 接收 SIGPIPE
output = p3.communicate()[0]
print(output)

5️⃣ 超时与环境变量

import subprocess
import os

# 超时控制
try:
    result = subprocess.run(
        ["sleep", "10"],
        timeout=3,
        capture_output=True,
    )
except subprocess.TimeoutExpired:
    print("命令超时！")

# 自定义环境变量
env = os.environ.copy()
env["MY_VAR"] = "custom_value"
result = subprocess.run(
    ["python3", "-c", "import os; print(os.environ['MY_VAR'])"],
    env=env,
    capture_output=True,
    text=True,
)
print(result.stdout.strip())  # custom_value

# 指定工作目录
result = subprocess.run(
    ["ls"],
    cwd="/tmp",
    capture_output=True,
    text=True,
)

6️⃣ 实战：系统信息采集

import subprocess
import json

def get_disk_usage():
    """获取磁盘使用情况"""
    result = subprocess.run(
        ["df", "-h", "--output=target,size,used,avail,pcent"],
        capture_output=True, text=True,
    )
    lines = result.stdout.strip().split("\n")
    disks = []
    for line in lines[1:]:
        parts = line.split()
        if len(parts) >= 5:
            disks.append({
                "挂载点": parts[0],
                "总大小": parts[1],
                "已用": parts[2],
                "可用": parts[3],
                "使用率": parts[4],
            })
    return disks

def get_memory_info():
    """获取内存信息"""
    result = subprocess.run(
        ["free", "-h"],
        capture_output=True, text=True,
    )
    return result.stdout

def get_top_processes(n=10):
    """获取占用 CPU 最多的进程"""
    result = subprocess.run(
        ["ps", "aux", "--sort=-%cpu"],
        capture_output=True, text=True,
    )
    lines = result.stdout.strip().split("\n")
    return lines[:n + 1]  # 包含表头

7️⃣ 验证脚本

#!/usr/bin/env python3
"""第14课 子进程验证"""
import subprocess
import sys

def test_run_basic():
    """基本调用测试"""
    result = subprocess.run(["echo", "hello"], capture_output=True, text=True)
    assert result.returncode == 0
    assert "hello" in result.stdout
    print("✅ 基本调用测试通过")

def test_capture_output():
    """输出捕获测试"""
    result = subprocess.run(
        [sys.executable, "-c", "print('test output')"],
        capture_output=True, text=True,
    )
    assert "test output" in result.stdout
    print("✅ 输出捕获测试通过")

def test_input():
    """输入传递测试"""
    result = subprocess.run(
        [sys.executable, "-c", "print(input().upper())"],
        input="hello", capture_output=True, text=True,
    )
    assert "HELLO" in result.stdout
    print("✅ 输入传递测试通过")

def test_error_handling():
    """错误处理测试"""
    result = subprocess.run(
        [sys.executable, "-c", "import sys; sys.exit(1)"],
        capture_output=True,
    )
    assert result.returncode == 1
    print("✅ 错误处理测试通过")

def test_timeout():
    """超时测试"""
    try:
        subprocess.run(["sleep", "10"], timeout=0.1)
        assert False, "应该超时"
    except subprocess.TimeoutExpired:
        pass
    print("✅ 超时测试通过")

def test_env():
    """环境变量测试"""
    import os
    env = os.environ.copy()
    env["TEST_VAR_123"] = "test_value"
    result = subprocess.run(
        [sys.executable, "-c", "import os; print(os.environ['TEST_VAR_123'])"],
        env=env, capture_output=True, text=True,
    )
    assert "test_value" in result.stdout
    print("✅ 环境变量测试通过")

if __name__ == "__main__":
    test_run_basic()
    test_capture_output()
    test_input()
    test_error_handling()
    test_timeout()
    test_env()
    print("\n🎉 第14课全部验证通过！")

🔑 本课要点