Python 执行命令并保存输出到文件

# 方法 2：先捕获输出再写入文件
def save_output_to_file_capture(command, filename):
    """ 先捕获命令输出，再写入文件，便于处理输出内容
    :param command: 要执行的命令字符串
    :param filename: 输出文件名
    """
    try:
        # 执行命令并捕获输出
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        # 将标准输出和标准错误分别处理
        with open(filename, 'w') as f:
            f.write("=== 标准输出 === ") 
            f.write(result.stdout)
            if result.stderr:
                f.write(" === 标准错误 === ") 
                f.write(result.stderr)
        print(f"命令执行完成，返回码：{result.returncode}")
        print(f"输出已保存到：{filename}")
    except Exception as e:
        print(f"命令执行异常：{e}")

# 使用示例
save_output_to_file_capture('pip list', 'installed_packages.txt')

2. 使用 subprocess.Popen() 进行高级控制

当需要更细粒度的控制时，Popen 类提供了最大的灵活性。

import subprocess

def save_output_with_realtime(command, filename):
    """ 实时处理命令输出并保存到文件
    :param command: 要执行的命令字符串
    :param filename: 输出文件名
    """
    with open(filename, 'w') as output_file:
        # 启动进程
        process = subprocess.Popen(
            command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            text=True, bufsize=1, # 行缓冲
            universal_newlines=True
        )
        # 实时读取输出
        try:
            while True:
                # 读取标准输出
                output = process.stdout.readline()
                if output:
                    print(output.strip()) # 实时显示在终端
                    output_file.write(output) # 同时写入文件
                    output_file.flush() # 确保立即写入
                # 检查进程是否结束
                if process.poll() is not None:
                    # 读取剩余输出
                    remaining_output = process.stdout.read()
                    if remaining_output:
                        print(remaining_output.strip())
                        output_file.write(remaining_output)
                    # 处理错误输出
                    error_output = process.stderr.read()
                    if error_output:
                        print(f"错误信息：{error_output}")
                        output_file.write(f" === 错误信息 === {error_output}")
                    break
        except KeyboardInterrupt:
            process.terminate()
            print("命令执行被用户中断")

# 使用示例：监控系统日志（需要适当权限）
# save_output_with_realtime('tail -f /var/log/syslog', 'system_log.txt')

3. 使用 os.popen() 方法（传统方式）

虽然较老，但在简单场景中仍然可用。

import os

def save_output_os_popen(command, filename):
    """ 使用 os.popen() 执行命令并保存输出
    :param command: 要执行的命令字符串
    :param filename: 输出文件名
    """
    try:
        # 执行命令并获取输出
        stream = os.popen(command)
        output = stream.read()
        return_code = stream.close()
        # 写入文件
        with open(filename, 'w') as f:
            f.write(output)
        print(f"命令执行完成，输出已保存到：{filename}")
        if return_code is not None:
            print(f"命令返回码：{return_code}")
    except Exception as e:
        print(f"执行命令时发生错误：{e}")

# 使用示例
save_output_os_popen('date', 'current_date.txt')

🛠️ 高级应用场景

场景 1：批量执行命令并分别保存输出

import subprocess
from datetime import datetime

def batch_commands_with_logging(commands_config):
    """ 批量执行多个命令，每个命令输出保存到单独文件
    :param commands_config: 命令配置列表，每个元素为 (命令，输出文件名)
    """
    log_entries = []
    for command, filename in commands_config:
        try:
            start_time = datetime.now()
            # 执行命令
            result = subprocess.run(
                command, shell=True, capture_output=True, text=True, timeout=30 # 30 秒超时
            )
            end_time = datetime.now()
            duration = (end_time - start_time).total_seconds()
            # 保存输出到文件
            with open(filename, 'w', encoding='utf-8') as f:
                f.write(f"命令：{command} ") 
                f.write(f"执行时间：{start_time.strftime('%Y-%m-%d %H:%M:%S')} ") 
                f.write(f"耗时：{duration:.2f}秒 ") 
                f.write(f"返回码：{result.returncode} ") 
                f.write("=== 标准输出 === ") 
                f.write(result.stdout)
                if result.stderr:
                    f.write(" === 标准错误 === ") 
                    f.write(result.stderr)
            # 记录执行日志
            log_entry = {
                'command': command,
                'filename': filename,
                'returncode': result.returncode,
                'duration': duration,
                'timestamp': start_time,
                'status': 'SUCCESS' if result.returncode == 0 else 'FAILED'
            }
            log_entries.append(log_entry)
            print(f"✓ {command} -> {filename} (耗时：{duration:.2f}s)")
        except subprocess.TimeoutExpired:
            print(f"✗ {command} 执行超时")
        except Exception as e:
            print(f"✗ {command} 执行异常：{e}")

    # 生成执行摘要
    generate_execution_summary(log_entries)

def generate_execution_summary(log_entries):
    """生成批量执行摘要"""
    summary_file = 'batch_execution_summary.txt'
    with open(summary_file, 'w') as f:
        f.write("批量命令执行摘要 ") 
        f.write("=" * 50 + " ") 
        success_count = sum(1 for entry in log_entries if entry['status'] == 'SUCCESS')
        total_count = len(log_entries)
        f.write(f"总命令数：{total_count} ") 
        f.write(f"成功数：{success_count} ") 
        f.write(f"失败数：{total_count - success_count} ") 
        f.write(f"成功率：{success_count/total_count*100:.1f}% ") 
        f.write("详细执行记录： ") 
        for entry in log_entries:
            f.write(f"- {entry['command']} | {entry['status']} | {entry['duration']:.2f}s | {entry['filename']} ") 
    print(f"执行摘要已保存到：{summary_file}")

# 使用示例
commands_to_run = [
    ('ls -la', 'directory_listing.txt'),
    ('python --version', 'python_version.txt'),
    ('pip freeze', 'requirements.txt'),
    ('df -h', 'disk_usage.txt')
]
batch_commands_with_logging(commands_to_run)

场景 2：带错误处理和重试机制的命令执行

import subprocess
import time

def robust_command_execution(command, filename, max_retries=3):
    """ 带错误处理和重试机制的命令执行
    :param command: 要执行的命令
    :param filename: 输出文件名
    :param max_retries: 最大重试次数
    """
    for attempt in range(max_retries):
        try:
            print(f"第 {attempt + 1} 次尝试执行命令：{command}")
            result = subprocess.run(
                command, shell=True, capture_output=True, text=True, timeout=60
            )
            # 保存输出
            with open(filename, 'w') as f:
                f.write(f"执行尝试：{attempt + 1} ") 
                f.write(f"命令：{command} ") 
                f.write(f"返回码：{result.returncode} ") 
                f.write("标准输出： ") 
                f.write(result.stdout)
                if result.stderr:
                    f.write(" 标准错误： ") 
                    f.write(result.stderr)
            if result.returncode == 0:
                print(f"命令执行成功，输出保存到：{filename}")
                return True
            else:
                print(f"命令执行失败，返回码：{result.returncode}")
                if attempt < max_retries - 1:
                    wait_time = (attempt + 1) * 2 # 指数退避
                    print(f"{wait_time}秒后重试...")
                    time.sleep(wait_time)
        except subprocess.TimeoutExpired:
            print(f"命令执行超时")
            if attempt < max_retries - 1:
                print("稍后重试...")
                time.sleep(5)
        except Exception as e:
            print(f"执行异常：{e}")
            if attempt < max_retries - 1:
                print("稍后重试...")
                time.sleep(5)
    print(f"命令执行失败，已达到最大重试次数：{max_retries}")
    return False

# 使用示例
robust_command_execution('curl -I https://www.example.com', 'http_headers.txt')

📝 最佳实践建议

1. 安全性考虑：使用 subprocess 模块时，尽量避免 shell=True 参数，或者对用户输入进行严格的验证和转义，以防止命令注入攻击。
2. 编码处理：在处理包含非 ASCII 字符的输出时，明确指定编码格式（如 encoding='utf-8'）以避免乱码问题。
3. 资源管理：使用 with 语句确保文件正确关闭，对于长时间运行的命令，考虑使用超时机制防止进程挂起。
4. 错误处理：始终检查命令的返回码，并妥善处理标准错误输出，这对于自动化脚本的稳定性至关重要。
5. 性能优化：对于需要实时处理输出的场景，使用 Popen 配合行缓冲可以实现更好的响应性。

通过上述方法和最佳实践，您可以灵活地在 Python 脚本中执行系统命令并将输出可靠地保存到文件中，满足各种自动化任务和系统管理的需求。