Python执行系统命令并保存输出的完整指南

1. Python执行命令并保存输出到文件的核心逻辑

在自动化运维、数据处理和系统管理场景中，我们经常需要通过Python程序执行系统命令并记录执行结果。这种技术组合完美结合了系统命令的底层控制力和Python的文件处理能力，是每个Python开发者都应该掌握的基础技能。

核心实现原理其实很简单：通过Python的subprocess模块启动子进程执行命令，捕获其标准输出/错误流，然后用文件操作函数将结果持久化存储。但实际应用中会遇到各种边界情况需要处理，比如命令执行超时、输出编码问题、大文件处理等。

2. 基础实现方案与代码解析

2.1 subprocess模块基础用法

最基础的实现只需要5行代码：

import subprocess result = subprocess.run(['ls', '-l'], capture_output=True, text=True) with open('output.txt', 'w') as f: f.write(result.stdout)

这里有几个关键参数需要注意：

• capture_output=True：捕获命令输出（相当于同时设置stdout和stderr）
• text=True：将输出自动解码为字符串（Python 3.7+）
• stdout属性存储正常输出，stderr存储错误输出

注意：在Python 3.6及以下版本需要使用universal_newlines=True代替text=True

2.2 进阶参数配置

实际生产环境中，我们通常需要更精细的控制：

import subprocess try: result = subprocess.run( ['ping', 'example.com'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=30, # 设置超时时间 check=True # 检查返回码非零时抛出异常 ) with open('network_test.log', 'a') as f: # 使用追加模式 f.write(f"[SUCCESS] {result.stdout}") except subprocess.TimeoutExpired: with open('network_test.log', 'a') as f: f.write("[ERROR] Command timed out\n") except subprocess.CalledProcessError as e: with open('network_test.log', 'a') as f: f.write(f"[ERROR] {e.stderr}")

3. 生产环境中的最佳实践

3.1 输出实时写入文件

对于长时间运行的命令，建议实时写入输出而不是等命令结束：

import subprocess with open('real_time.log', 'w') as f: process = subprocess.Popen( ['tail', '-f', '/var/log/syslog'], stdout=f, # 直接重定向到文件 stderr=subprocess.PIPE, text=True ) try: process.wait(timeout=60) except subprocess.TimeoutExpired: process.terminate() f.write("\nProcess terminated after timeout")

3.2 处理二进制输出

当处理二进制命令输出时（如图片处理、压缩文件等）：

import subprocess result = subprocess.run( ['file', '--brief', '--mime-type', 'image.jpg'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, # 不设置text=True以获取bytes输出 ) with open('mime_type.txt', 'wb') as f: # 注意'wb'模式 f.write(result.stdout)

3.3 多命令管道处理

实现类似Linux管道的功能：

import subprocess ps = subprocess.Popen(['ps', 'aux'], stdout=subprocess.PIPE) grep = subprocess.Popen( ['grep', 'python'], stdin=ps.stdout, stdout=subprocess.PIPE ) ps.stdout.close() # 允许ps收到SIGPIPE output = grep.communicate()[0] with open('python_processes.txt', 'w') as f: f.write(output.decode('utf-8'))

4. 常见问题与解决方案

4.1 编码问题处理

不同平台命令输出的编码可能不同：

import subprocess import locale result = subprocess.run(['cmd', '/c', 'dir'], capture_output=True) encoding = locale.getpreferredencoding() try: output = result.stdout.decode(encoding) except UnicodeDecodeError: output = result.stdout.decode('utf-8', errors='replace') with open('dir_listing.txt', 'w', encoding='utf-8') as f: f.write(output)

4.2 大文件输出处理

对于可能产生大量输出的命令：

import subprocess with open('large_output.log', 'w') as f: process = subprocess.Popen( ['dd', 'if=/dev/zero', 'bs=1M', 'count=100'], stdout=subprocess.PIPE, text=True ) for line in process.stdout: f.write(line) # 可以添加处理逻辑或进度显示 print(".", end="", flush=True)

4.3 跨平台兼容性

确保代码在Windows和Linux都能运行：

import subprocess import sys command = ['dir'] if sys.platform == 'win32' else ['ls'] shell = True if sys.platform == 'win32' else False result = subprocess.run( command, capture_output=True, text=True, shell=shell ) with open('file_list.txt', 'w') as f: f.write(result.stdout)

5. 性能优化技巧

5.1 使用缓冲区提高写入性能

import subprocess from io import StringIO result = subprocess.run(['find', '/', '-name', '*.py'], capture_output=True, text=True) # 使用内存缓冲区 buffer = StringIO() buffer.write(result.stdout) with open('python_files.txt', 'w') as f: f.write(buffer.getvalue())

5.2 异步处理命令输出

import asyncio import subprocess async def run_command(cmd, output_file): process = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) with open(output_file, 'w') as f: while True: output = await process.stdout.readline() if not output: break f.write(output.decode('utf-8')) return await process.wait() asyncio.run(run_command('ls -lR /', 'full_listing.txt'))

6. 安全注意事项

6.1 防止命令注入

绝对不要这样做：

filename = input("Enter filename: ") # 危险！可能被注入恶意命令 subprocess.run(f"rm {filename}", shell=True)

应该这样：

filename = input("Enter filename: ") subprocess.run(['rm', filename]) # 安全

6.2 敏感信息处理

当处理可能包含敏感信息的命令输出时：

import subprocess import re result = subprocess.run(['config', 'show'], capture_output=True, text=True) # 过滤密码等敏感信息 cleaned_output = re.sub(r'(password|key)=.*', r'\1=******', result.stdout) with open('config_dump.txt', 'w') as f: f.write(cleaned_output)

7. 实际应用案例

7.1 自动化服务器监控

import subprocess from datetime import datetime def log_system_stats(): commands = [ ('date', 'date.txt'), ('free -h', 'memory.txt'), ('df -h', 'disk.txt'), ('uptime', 'uptime.txt') ] timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') for cmd, filename in commands: try: result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=10 ) with open(filename, 'a') as f: f.write(f"[{timestamp}]\n{result.stdout}\n") except subprocess.TimeoutExpired: with open(filename, 'a') as f: f.write(f"[{timestamp}] Command timed out\n")

7.2 批量处理文件

import subprocess import os def convert_images(input_dir, output_dir): if not os.path.exists(output_dir): os.makedirs(output_dir) for filename in os.listdir(input_dir): if filename.endswith('.jpg'): input_path = os.path.join(input_dir, filename) output_path = os.path.join(output_dir, f"{os.path.splitext(filename)[0]}.png") result = subprocess.run( ['convert', input_path, output_path], capture_output=True, text=True ) log_entry = { 'filename': filename, 'success': result.returncode == 0, 'message': result.stdout if result.returncode == 0 else result.stderr } with open('conversion.log', 'a') as f: f.write(f"{log_entry}\n")

8. 高级技巧：上下文管理器封装

为了更好的代码复用，可以创建一个上下文管理器：

import subprocess import contextlib @contextlib.contextmanager def command_output_to_file(cmd, filename, mode='w'): try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) with open(filename, mode) as f: yield process, f # 将进程和文件对象提供给调用方 process.wait() except Exception as e: with open(filename, mode) as f: f.write(f"Error executing command: {str(e)}\n") raise # 使用示例 with command_output_to_file(['ls', '-l'], 'directory_listing.txt') as (process, output_file): for line in process.stdout: output_file.write(line) # 可以在这里添加额外的处理逻辑

9. 性能对比：不同实现方式的差异

我们比较几种常见实现方式的性能差异（测试100次'ls -l'命令）：

测试代码示例：

import time import statistics import subprocess def time_function(func, iterations=100): times = [] for _ in range(iterations): start = time.perf_counter() func() times.append((time.perf_counter() - start) * 1000) return statistics.mean(times) def method1(): result = subprocess.run(['ls', '-l'], capture_output=True, text=True) with open('temp1.txt', 'w') as f: f.write(result.stdout) print(f"subprocess.run 平均耗时: {time_function(method1):.1f}ms")

10. 错误处理与日志记录

完善的错误处理机制对于生产环境至关重要：

import subprocess import logging import sys logging.basicConfig( filename='command_runner.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) def run_command_safely(cmd, output_file): try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) with open(output_file, 'w') as f: while True: output = process.stdout.readline() if not output and process.poll() is not None: break if output: f.write(output) sys.stdout.write(output) # 同时输出到控制台 return_code = process.poll() if return_code != 0: error = process.stderr.read() logging.error( f"Command failed with code {return_code}: {cmd}\nError: {error}" ) raise subprocess.CalledProcessError(return_code, cmd, error) logging.info(f"Successfully executed: {cmd}") except FileNotFoundError as e: logging.error(f"Command not found: {cmd[0]}") raise except subprocess.TimeoutExpired: logging.error(f"Command timed out: {cmd}") raise except Exception as e: logging.error(f"Unexpected error executing {cmd}: {str(e)}") raise