# Python中执行Shell命令的完整指南 Python提供了多种执行Shell命令的方法，每种方法都有其特定的使用场景和优缺点。本文将详细介绍这些方法，并通过实际代码示例展示如何在不同场景下使用它们。 ## 执行Shell命令的主要方法对比 | 方法 | 功能特点 | 适用场景 | 返回值处理 | |------|----------|----------|------------| | `os.system()` | 执行命令，返回退出状态码 | 简单命令执行 | 只能获取退出码 | | `os.popen()` | 执行命令，返回文件对象 | 需要获取命令输出 | 可读取输出内容 | | `subprocess.run()` | 推荐的标准方法 | 复杂命令执行 | 可获取输出、错误、退出码 | | `subprocess.Popen()` | 底层进程控制 | 需要精细控制进程 | 完全控制进程行为 | | `commands`模块 | 获取命令输出 | Python 2.x兼容 | 返回命令输出字符串 | ## 1. os.system()方法 `os.system()`是最简单的执行Shell命令的方法，它会执行命令并返回退出状态码[ref_2]。 ```python import os # 执行简单的Shell命令 return_code = os.system('ls -l') print(f"命令退出状态码: {return_code}") # 执行带参数的命令 return_code = os.system('echo "Hello, World!"') print(f"echo命令退出状态码: {return_code}") # 执行复杂的Shell脚本 return_code = os.system(''' echo "开始执行脚本" ls -la | head -5 echo "脚本执行完毕" ''') ``` **特点分析：** - 返回值为命令的退出状态码，0表示成功执行[ref_2] - 命令输出直接显示在控制台，无法在程序中捕获 - 适合执行不需要获取输出的简单命令 ## 2. os.popen()方法 `os.popen()`方法可以执行命令并返回一个文件对象，通过该对象可以读取命令的输出[ref_4]。 ```python import os # 执行命令并获取输出 output = os.popen('ls -l').read() print("命令输出:") print(output) # 逐行处理命令输出 with os.popen('ps aux | head -10') as process: for line in process: print(f"进程信息: {line.strip()}") # 执行带管道的复杂命令 disk_usage = os.popen('df -h | grep "/dev/"').read() print("磁盘使用情况:") print(disk_usage) ``` **实际应用场景：** ```python import os def get_system_info(): """获取系统信息""" info = {} # 获取主机名 info['hostname'] = os.popen('hostname').read().strip() # 获取内核版本 info['kernel'] = os.popen('uname -r').read().strip() # 获取内存信息 memory_info = os.popen('free -h').read() info['memory'] = memory_info return info system_info = get_system_info() for key, value in system_info.items(): print(f"{key}: {value}") ``` ## 3. subprocess模块（推荐） `subprocess`模块是当前推荐使用的标准方法，提供了更强大和灵活的功能[ref_4]。 ### 3.1 subprocess.run()方法 ```python import subprocess # 基本用法 - 执行命令并等待完成 result = subprocess.run(['ls', '-l'], capture_output=True, text=True) print(f"返回码: {result.returncode}") print(f"标准输出: {result.stdout}") print(f"错误输出: {result.stderr}") # 执行Shell管道命令 result = subprocess.run( 'ps aux | grep python | head -5', shell=True, capture_output=True, text=True ) print("Python进程:") print(result.stdout) # 带超时控制的命令执行 try: result = subprocess.run( ['sleep', '10'], timeout=5, # 5秒超时 capture_output=True ) except subprocess.TimeoutExpired: print("命令执行超时！") ``` ### 3.2 subprocess.Popen()方法 `Popen`类提供了更底层的进程控制[ref_2]。 ```python import subprocess # 创建进程并实时获取输出 process = subprocess.Popen( ['tail', '-f', '/var/log/syslog'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) # 实时读取输出（示例中只读取几行） for i in range(5): line = process.stdout.readline() if line: print(f"日志: {line.strip()}") process.terminate() # 终止进程 # 使用communicate()方法 process = subprocess.Popen( ['python3', '-c', 'print("Hello"); import time; time.sleep(2); print("World")'], stdout=subprocess.PIPE, text=True ) stdout, stderr = process.communicate() print(f"命令输出: {stdout}") ``` ## 4. 实用工具函数封装 下面提供一些实用的工具函数，方便在日常开发中使用： ```python import subprocess import shlex def run_command(command, timeout=30, shell=False): """ 执行Shell命令的通用函数 参数: command: 要执行的命令字符串 timeout: 超时时间（秒） shell: 是否使用shell执行 返回: 包含执行结果的字典 """ try: if shell: # 使用shell执行 result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=timeout ) else: # 不使用shell，更安全 command_list = shlex.split(command) result = subprocess.run( command_list, capture_output=True, text=True, timeout=timeout ) return { 'success': result.returncode == 0, 'returncode': result.returncode, 'stdout': result.stdout, 'stderr': result.stderr, 'command': command } except subprocess.TimeoutExpired: return { 'success': False, 'error': f'命令执行超时 ({timeout}秒)', 'command': command } except Exception as e: return { 'success': False, 'error': str(e), 'command': command } def execute_commands(commands, parallel=False): """ 批量执行命令 参数: commands: 命令列表 parallel: 是否并行执行 返回: 执行结果列表 """ results = [] for cmd in commands: print(f"执行命令: {cmd}") result = run_command(cmd) results.append(result) if result['success']: print("✓ 命令执行成功") if result['stdout']: print(f"输出: {result['stdout'][:100]}...") # 只显示前100字符 else: print("✗ 命令执行失败") if 'error' in result: print(f"错误: {result['error']}") elif result['stderr']: print(f"错误输出: {result['stderr']}") return results # 使用示例 if __name__ == "__main__": # 测试单个命令 result = run_command('ls -la | head -10', shell=True) print("单个命令执行结果:") print(result) # 批量执行系统检查命令 system_checks = [ 'uname -a', 'df -h', 'free -h', 'uptime', 'whoami' ] print("\n系统检查结果:") execute_commands(system_checks) ``` ## 5. 安全注意事项 在执行Shell命令时，安全性是非常重要的考虑因素： ```python import subprocess import shlex def safe_command_execution(): """演示安全的命令执行方法""" # 不安全的做法 - 直接拼接用户输入 user_input = "malicious_file; rm -rf /" # ❌ 危险：os.system(f"ls {user_input}") # 安全的做法1 - 使用参数列表 safe_result = subprocess.run(['ls', user_input], capture_output=True) print("安全执行结果:", safe_result.returncode) # 安全的做法2 - 使用shlex分割 command = f"ls {shlex.quote(user_input)}" safe_result2 = subprocess.run(command, shell=True, capture_output=True) print("安全执行结果2:", safe_result2.returncode) # 输入验证函数 def validate_command(command, allowed_commands): """ 验证命令是否在允许列表中 """ base_command = command.split()[0] if command else '' return base_command in allowed_commands # 允许的命令白名单 ALLOWED_COMMANDS = {'ls', 'cat', 'echo', 'pwd', 'whoami'} user_command = "ls -la" if validate_command(user_command, ALLOWED_COMMANDS): result = run_command(user_command) print(result) else: print("命令不在允许列表中") ``` ## 6. 高级应用场景 ### 6.1 实时日志监控 ```python import subprocess import threading def monitor_logs(log_file, callback): """ 实时监控日志文件 """ def tail_process(): process = subprocess.Popen( ['tail', '-F', log_file], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) while True: line = process.stdout.readline() if line: callback(line.strip()) thread = threading.Thread(target=tail_process, daemon=True) thread.start() return thread # 使用示例 def log_callback(line): if 'ERROR' in line: print(f"🔴 发现错误: {line}") elif 'WARNING' in line: print(f"🟡 发现警告: {line}") # 开始监控（需要实际日志文件路径） # monitor_logs('/var/log/app.log', log_callback) ``` ### 6.2 命令执行器类 ```python class CommandExecutor: """命令执行器类""" def __init__(self, timeout=30, working_dir=None): self.timeout = timeout self.working_dir = working_dir self.history = [] def execute(self, command, shell=False): """执行命令并记录历史""" result = run_command(command, self.timeout, shell) self.history.append({ 'command': command, 'result': result, 'timestamp': subprocess.run(['date'], capture_output=True, text=True).stdout.strip() }) return result def get_history(self): """获取执行历史""" return self.history def clear_history(self): """清空执行历史""" self.history.clear() # 使用命令执行器 executor = CommandExecutor(timeout=60) # 执行多个命令 commands = [ 'pwd', 'ls -la', 'echo "当前用户: $(whoami)"' ] for cmd in commands: result = executor.execute(cmd, shell=True) print(f"命令: {cmd}") print(f"结果: {'成功' if result['success'] else '失败'}") print("\n执行历史:") for entry in executor.get_history(): print(f"{entry['timestamp']}: {entry['command']}") ``` ## 总结 Python执行Shell命令的方法多样，选择合适的方法取决于具体需求： 1. **简单命令**：使用`os.system()`或`os.popen()` 2. **需要完整控制**：使用`subprocess.run()` 3. **实时交互**：使用`subprocess.Popen()` 4. **生产环境**：推荐使用`subprocess`模块，配合适当的错误处理和超时控制 在实际应用中，建议封装通用的命令执行函数，加入错误处理、日志记录和安全验证，以提高代码的健壮性和安全性[ref_4]。