在Python自动化脚本中，经常需要与操作系统底层交互、执行系统命令、处理管道输入输出等。subprocess模块是Python标准库中最强大的进程管理工具，但很多开发者只用了它10%的能力。本文将深入讲解如何优雅地执行Shell命令、处理复杂输出、实现交互式脚本。

一、基础命令执行：从run到Popen

Python提供了多种执行命令的方式，从简单到复杂，你需要根据场景选择合适的方法。

import subprocess

# 方法1：简单命令执行（最常用）
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print(result.stdout)  # 标准输出
print(result.stderr)  # 标准错误

# 方法2：Shell字符串执行（方便但有安全风险）
result = subprocess.run('ls -la | grep python', shell=True, capture_output=True, text=True)

# 方法3：异步执行（长时间任务）
process = subprocess.Popen(['tail', '-f', '/var/log/syslog'], 
                           stdout=subprocess.PIPE, 
                           stderr=subprocess.PIPE,
                           text=True)

# 读取实时输出
for line in iter(process.stdout.readline, ''):
    if line:
        print(line.rstrip())

最佳实践：尽量使用列表传参而非shell=True，避免命令注入风险。

二、复杂场景处理

2.1 处理需要sudo权限的命令

import subprocess
import getpass

def run_with_sudo(cmd):
    """执行需要sudo权限的命令"""
    password = getpass.getpass("Enter sudo password: ")
    
    full_cmd = f'echo {password} | sudo -S ' + ' '.join(cmd)
    
    result = subprocess.run(
        full_cmd, 
        shell=True, 
        capture_output=True, 
        text=True,
        input=password + '\n'
    )
    return result

# 使用示例
result = run_with_sudo(['apt', 'update'])

2.2 超时控制和进程管理

import subprocess
from functools import wraps
import signal

def timeout_handler(signum, frame):
    raise TimeoutError("命令执行超时")

def run_with_timeout(cmd, timeout=30):
    """带超时控制的命令执行"""
    try:
        result = subprocess.run(
            cmd, 
            timeout=timeout,
            capture_output=True, 
            text=True,
            start_new_session=True  # 创建新进程组
        )
        return result
    except subprocess.TimeoutExpired:
        print(f"命令执行超过{timeout}秒，已终止")
        return None

2.3 管道处理和数据流控制

# 同时捕获标准输出和标准错误
proc = subprocess.Popen(
    ['command1'],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,  # 合并错误到输出
    text=True
)

# 将输出传递给下一个命令
proc2 = subprocess.Popen(
    ['command2'],
    stdin=proc.stdout,
    stdout=subprocess.PIPE,
    text=True
)

output = proc2.communicate()[0]

三、交互式脚本实现

3.1 expect风格的自动化交互

import pexpect

def automate_ssh(ssh_cmd, password):
    """SSH自动登录脚本"""
    child = pexpect.spawn(ssh_cmd, timeout=30)
    
    responses = [
        (r'password:', password),
        (r'yes/no', 'yes'),
        (r'# |\$ ', 'exit\n')
    ]
    
    for pattern, response in responses:
        try:
            child.expect(pattern)
            child.sendline(response)
        except pexpect.TIMEOUT:
            print("超时")
            break
    
    print(child.before.decode())
    child.close()

3.2 读取命令输出并进行条件判断

def check_service_status(service_name):
    """检查服务状态并返回结构化结果"""
    result = subprocess.run(
        ['systemctl', 'is-active', service_name],
        capture_output=True,
        text=True
    )
    
    status_map = {
        'active': ('running', '✅ 服务正在运行', 'green'),
        'inactive': ('stopped', '⚠️ 服务已停止', 'yellow'),
        'failed': ('error', '❌ 服务启动失败', 'red')
    }
    
    status = result.stdout.strip()
    return status_map.get(status, ('unknown', f'未知状态: {status}', 'gray'))

四、生产环境最佳实践

4.1 日志记录和错误处理

import logging
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/automation.log'),
        logging.StreamHandler()
    ]
)

def execute_command(cmd, retry=3):
    """带重试机制的命令执行"""
    logger = logging.getLogger(__name__)
    
    for attempt in range(retry):
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=300
            )
            
            if result.returncode == 0:
                logger.info(f"命令执行成功: {' '.join(cmd)}")
                return result.stdout
            else:
                logger.warning(f"命令执行失败 (尝试 {attempt+1}/{retry}): {result.stderr}")
                
        except subprocess.TimeoutExpired:
            logger.error(f"命令执行超时")
        except Exception as e:
            logger.error(f"执行异常: {e}")
            
    return None

4.2 环境隔离和安全性

# 在隔离环境中执行不受信的命令
def run_in_sandbox(cmd, allowed_paths=['/tmp', '/home/user/safe']):
    """在受限环境中执行命令"""
    env = os.environ.copy()
    env['PATH'] = '/usr/bin:/bin'  # 限制可执行文件路径
    
    # 限制文件访问
    # 注意：这只提供基本保护，不能替代容器隔离
    
    result = subprocess.run(
        cmd,
        env=env,
        capture_output=True,
        text=True,
        cwd='/tmp'  # 限制工作目录
    )
    return result

五、实战案例：自动化部署脚本

class Deployer:
    """自动化部署工具类"""
    
    def __init__(self, host, user, remote_path):
        self.host = host
        self.user = user
        self.remote_path = remote_path
        
    def execute_remote(self, cmd):
        """远程执行命令"""
        ssh_cmd = f"ssh {self.user}@{self.host} '{cmd}'"
        result = subprocess.run(
            ssh_cmd, 
            shell=True, 
            capture_output=True, 
            text=True
        )
        
        if result.returncode != 0:
            raise RuntimeError(f"远程命令失败: {result.stderr}")
            
        return result.stdout
    
    def deploy(self, local_package):
        """执行完整部署流程"""
        steps = [
            ("备份旧版本", f"cp -r {self.remote_path} {self.remote_path}.bak"),
            ("上传新版本", f"scp {local_package} {self.user}@{self.host}:{self.remote_path}"),
            ("解压包", f"tar -xzf {self.remote_path}/package.tar.gz"),
            ("重启服务", f"systemctl restart myapp")
        ]
        
        for step_name, cmd in steps:
            print(f"执行: {step_name}")
            self.execute_remote(cmd)
            print(f"✓ {step_name} 完成")

总结

掌握subprocess的高级用法，能让你的Python自动化脚本如虎添翼。关键点：

1. 优先使用列表传参：避免shell=True的安全风险
2. 合理使用异步：长时间任务用Popen，非阻塞执行
3. 完善错误处理：超时控制、重试机制、日志记录
4. 注意安全性：最小权限原则、环境隔离
5. 测试边界情况：空输出、特殊字符、超长输出

灵活运用这些技巧，你就能编写出专业级的系统交互自动化脚本。