从基础到高级详解Python实现输出重定向的完全指南

import sys

def basic_redirection():
    """基础输出重定向示例"""
    # 保存原始标准输出
    original_stdout = sys.stdout
    
    try:
        # 重定向到文件
        with open('output.txt', 'w', encoding='utf-8') as f:
            sys.stdout = f
            print("这条消息将写入文件")
            print("当前时间:", "2024-01-15 10:30:00")
            print("状态: 正常")
    
    finally:
        # 恢复标准输出
        sys.stdout = original_stdout
    
    print("这条消息将显示在控制台")

# 执行示例
basic_redirection()

def print_file_parameter():
    """使用print函数的file参数进行重定向"""
    # 直接重定向到文件
    with open('direct_output.txt', 'w', encoding='utf-8') as f:
        print("直接使用file参数", file=f)
        print("多行输出示例", file=f)
        print("结束输出", file=f)
    
    # 同时输出到控制台和文件
    with open('dual_output.txt', 'w', encoding='utf-8') as f:
        print("同时输出到文件和屏幕:")
        for i in range(3):
            msg = f"消息 {i+1}"
            print(msg)  # 输出到控制台
            print(msg, file=f)  # 输出到文件
    
    print("所有操作完成")

print_file_parameter()

def stderr_redirection():
    """标准错误输出重定向"""
    import sys
    
    # 保存原始stderr
    original_stderr = sys.stderr
    
    try:
        # 重定向stderr到文件
        with open('error_log.txt', 'w', encoding='utf-8') as f:
            sys.stderr = f
            # 模拟错误输出
            print("这是一条错误消息", file=sys.stderr)
            print("程序遇到问题", file=sys.stderr)
            # 模拟异常
            try:
                1 / 0
            except Exception as e:
                print(f"捕获到异常: {e}", file=sys.stderr)
    
    finally:
        # 恢复stderr
        sys.stderr = original_stderr
    
    print("错误日志已保存到文件")

stderr_redirection()

class RedirectOutput:
    """输出重定向上下文管理器"""
    def __init__(self, stdout=None, stderr=None, mode='w', encoding='utf-8'):
        self.stdout_file = stdout
        self.stderr_file = stderr
        self.mode = mode
        self.encoding = encoding
        self.original_stdout = None
        self.original_stderr = None
        self.stdout_handle = None
        self.stderr_handle = None
    
    def __enter__(self):
        import sys
        
        # 保存原始输出
        self.original_stdout = sys.stdout
        self.original_stderr = sys.stderr
        
        # 设置新的输出
        if self.stdout_file:
            if isinstance(self.stdout_file, str):
                self.stdout_handle = open(self.stdout_file, self.mode, encoding=self.encoding)
                sys.stdout = self.stdout_handle
            else:
                sys.stdout = self.stdout_file
        
        if self.stderr_file:
            if isinstance(self.stderr_file, str):
                self.stderr_handle = open(self.stderr_file, self.mode, encoding=self.encoding)
                sys.stderr = self.stderr_handle
            else:
                sys.stderr = self.stderr_file
        
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        import sys
        
        # 恢复原始输出
        if self.original_stdout:
            sys.stdout = self.original_stdout
        if self.original_stderr:
            sys.stderr = self.original_stderr
        
        # 关闭文件句柄
        if self.stdout_handle:
            self.stdout_handle.close()
        if self.stderr_handle:
            self.stderr_handle.close()
        
        return False  # 不抑制异常

# 使用示例
def context_manager_example():
    """上下文管理器使用示例"""
    with RedirectOutput('output_ctx.txt', 'error_ctx.txt'):
        print("普通输出到文件")
        print("另一条输出消息", file=sys.stdout)
        print("错误信息", file=sys.stderr)
    
    print("回到控制台输出")

context_manager_example()

class MultiOutput:
    """多目标输出重定向"""
    def __init__(self, *outputs):
        self.outputs = outputs
    
    def write(self, message):
        for output in self.outputs:
            output.write(message)
    
    def flush(self):
        for output in self.outputs:
            if hasattr(output, 'flush'):
                output.flush()

def multi_output_example():
    """多目标输出示例"""
    # 创建多个输出目标
    with open('file1.txt', 'w', encoding='utf-8') as f1, \
         open('file2.txt', 'w', encoding='utf-8') as f2:
        
        # 创建多目标输出对象
        multi = MultiOutput(sys.stdout, f1, f2)
        
        # 临时重定向
        original_stdout = sys.stdout
        sys.stdout = multi
        
        try:
            print("这条消息将同时输出到:")
            print("1. 控制台")
            print("2. file1.txt")
            print("3. file2.txt")
            
        finally:
            sys.stdout = original_stdout
    
    print("恢复单目标输出")

multi_output_example()

def real_time_logging():
    """实时日志重定向系统"""
    import time
    from datetime import datetime
    
    class TimestampedOutput:
        """带时间戳的输出重定向"""
        def __init__(self, filename):
            self.filename = filename
            self.file = open(filename, 'a', encoding='utf-8')
        
        def write(self, message):
            timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            log_message = f"[{timestamp}] {message}"
            self.file.write(log_message)
            self.file.flush()  # 确保实时写入
        
        def flush(self):
            self.file.flush()
        
        def close(self):
            self.file.close()
    
    # 使用示例
    log_output = TimestampedOutput('app.log')
    
    # 重定向输出
    original_stdout = sys.stdout
    sys.stdout = log_output
    
    try:
        # 模拟应用输出
        for i in range(5):
            print(f"应用日志消息 {i+1}")
            time.sleep(1)  # 模拟处理时间
        
        print("任务完成")
    
    finally:
        sys.stdout = original_stdout
        log_output.close()
    
    print("日志记录完成，查看 app.log 文件")

real_time_logging()

def rotating_log_system():
    """循环日志文件系统"""
    import os
    from datetime import datetime
    
    class RotatingFileOutput:
        """循环文件输出"""
        def __init__(self, base_filename, max_size=1024 * 1024, backup_count=5):
            self.base_filename = base_filename
            self.max_size = max_size
            self.backup_count = backup_count
            self.current_file = None
            self.current_size = 0
            self.open_file()
        
        def open_file(self):
            """打开当前日志文件"""
            if self.current_file:
                self.current_file.close()
            
            self.current_file = open(self.base_filename, 'a', encoding='utf-8')
            self.current_size = os.path.getsize(self.base_filename)
        
        def write(self, message):
            """写入消息，处理文件循环"""
            # 检查文件大小
            if self.current_size + len(message) > self.max_size:
                self.rotate_file()
            
            self.current_file.write(message)
            self.current_size += len(message)
            self.current_file.flush()
        
        def rotate_file(self):
            """循环日志文件"""
            self.current_file.close()
            
            # 重命名现有备份文件
            for i in range(self.backup_count - 1, 0, -1):
                old_file = f"{self.base_filename}.{i}"
                new_file = f"{self.base_filename}.{i+1}"
                if os.path.exists(old_file):
                    os.rename(old_file, new_file)
            
            # 重命名当前文件
            os.rename(self.base_filename, f"{self.base_filename}.1")
            
            # 创建新文件
            self.open_file()
        
        def flush(self):
            if self.current_file:
                self.current_file.flush()
        
        def close(self):
            if self.current_file:
                self.current_file.close()
    
    # 使用示例
    rotating_output = RotatingFileOutput('app.log', max_size=100, backup_count=3)
    
    original_stdout = sys.stdout
    sys.stdout = rotating_output
    
    try:
        for i in range(20):
            print(f"日志消息 {i+1}: 这是一条测试日志消息")
        
        print("所有日志消息写入完成")
    
    finally:
        sys.stdout = original_stdout
        rotating_output.close()
    
    print("循环日志测试完成")
    print("生成的文件:")
    for file in os.listdir('.'):
        if file.startswith('app.log'):
            size = os.path.getsize(file)
            print(f"  {file}: {size} 字节")

rotating_log_system()

def filtered_output_redirection():
    """过滤式输出重定向"""
    class FilteredOutput:
        """带过滤功能的输出重定向"""
        def __init__(self, output, filter_func=None):
            self.output = output
            self.filter_func = filter_func or (lambda x: True)
        
        def write(self, message):
            if self.filter_func(message):
                self.output.write(message)
        
        def flush(self):
            if hasattr(self.output, 'flush'):
                self.output.flush()
    
    # 使用示例
    def contains_error(message):
        """过滤包含错误关键词的消息"""
        error_keywords = ['error', 'fail', 'exception', '警告']
        return any(keyword in message.lower() for keyword in error_keywords)
    
    # 创建错误日志文件
    with open('error_only.log', 'w', encoding='utf-8') as error_file:
        filtered_output = FilteredOutput(error_file, contains_error)
        
        original_stdout = sys.stdout
        sys.stdout = filtered_output
        
        try:
            print("这是一条普通信息")
            print("发现错误: 文件未找到")
            print("操作完成")
            print("警告: 内存使用率高")
            print("正常结束")
        
        finally:
            sys.stdout = original_stdout
    
    print("过滤输出完成")
    print("错误日志内容:")
    with open('error_only.log', 'r', encoding='utf-8') as f:
        print(f.read())

filtered_output_redirection()

def buffered_output_redirection():
    """缓冲输出重定向优化"""
    class BufferedOutput:
        """缓冲输出提高性能"""
        def __init__(self, output, buffer_size=8192):
            self.output = output
            self.buffer_size = buffer_size
            self.buffer = []
            self.current_size = 0
        
        def write(self, message):
            self.buffer.append(message)
            self.current_size += len(message)
            
            if self.current_size >= self.buffer_size:
                self.flush()
        
        def flush(self):
            if self.buffer:
                full_message = ''.join(self.buffer)
                self.output.write(full_message)
                if hasattr(self.output, 'flush'):
                    self.output.flush()
                self.buffer = []
                self.current_size = 0
        
        def close(self):
            self.flush()
            if hasattr(self.output, 'close'):
                self.output.close()
    
    # 性能测试
    import time
    
    # 无缓冲写入
    start_time = time.time()
    with open('unbuffered.txt', 'w', encoding='utf-8') as f:
        for i in range(10000):
            f.write(f"行 {i}: 测试数据\n")
    unbuffered_time = time.time() - start_time
    
    # 缓冲写入
    start_time = time.time()
    with open('buffered.txt', 'w', encoding='utf-8') as raw_file:
        buffered_file = BufferedOutput(raw_file, buffer_size=8192)
        for i in range(10000):
            buffered_file.write(f"行 {i}: 测试数据\n")
        buffered_file.close()
    buffered_time = time.time() - start_time
    
    print(f"无缓冲时间: {unbuffered_time:.4f}秒")
    print(f"缓冲时间: {buffered_time:.4f}秒")
    print(f"性能提升: {(unbuffered_time/buffered_time):.2f}倍")

buffered_output_redirection()

def subprocess_output_redirection():
    """子进程输出重定向"""
    import subprocess
    
    # 重定向子进程输出到文件
    with open('subprocess_output.txt', 'w', encoding='utf-8') as f:
        result = subprocess.run(
            ['python', '-c', 'print("子进程输出"); import sys; sys.stderr.write("子进程错误\\n")'],
            stdout=f,
            stderr=subprocess.STDOUT,  # 合并标准错误到标准输出
            text=True
        )
    
    print("子进程执行完成")
    print("输出文件内容:")
    with open('subprocess_output.txt', 'r', encoding='utf-8') as f:
        print(f.read())
    
    # 捕获子进程输出到变量
    result = subprocess.run(
        ['python', '-c', 'print("捕获的输出"); import sys; sys.stderr.write("捕获的错误\\n")'],
        capture_output=True,
        text=True
    )
    
    print(f"标准输出: {result.stdout}")
    print(f"标准错误: {result.stderr}")
    print(f"返回码: {result.returncode}")

subprocess_output_redirection()

def pipeline_output_redirection():
    """管道输出重定向"""
    import subprocess
    
    # 创建管道处理链
    process1 = subprocess.Popen(
        ['python', '-c', 'for i in range(5): print(f"数据 {i}")'],
        stdout=subprocess.PIPE,
        text=True
    )
    
    process2 = subprocess.Popen(
        ['python', '-c', '''
import sys
for line in sys.stdin:
    processed = line.strip().upper()
    print(f"处理后的: {processed}")
        '''],
        stdin=process1.stdout,
        stdout=subprocess.PIPE,
        text=True
    )
    
    # 获取最终输出
    output, errors = process2.communicate()
    
    print("管道处理结果:")
    print(output)
    
    # 保存到文件
    with open('pipeline_result.txt', 'w', encoding='utf-8') as f:
        f.write(output)
    
    print("结果已保存到 pipeline_result.txt")

pipeline_output_redirection()

def logging_system_integration():
    """日志系统与输出重定向集成"""
    import logging
    from logging.handlers import RotatingFileHandler
    
    # 配置日志系统
    logger = logging.getLogger('AppLogger')
    logger.setLevel(logging.DEBUG)
    
    # 文件处理器
    file_handler = RotatingFileHandler(
        'app.log',
        maxBytes=1024 * 1024,  # 1MB
        backupCount=5,
        encoding='utf-8'
    )
    file_handler.setLevel(logging.DEBUG)
    
    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    
    # 格式化
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)
    
    # 添加处理器
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    
    # 使用日志系统
    logger.debug("调试信息")
    logger.info("一般信息")
    logger.warning("警告信息")
    logger.error("错误信息")
    logger.critical("严重错误")
    
    print("日志记录完成")
    
    # 查看日志文件内容
    print("\n日志文件内容:")
    with open('app.log', 'r', encoding='utf-8') as f:
        for line in f:
            print(line.strip())

logging_system_integration()

def multithreaded_output_redirection():
    """多线程环境下的输出重定向"""
    import threading
    import time
    from concurrent.futures import ThreadPoolExecutor
    
    class ThreadSafeOutput:
        """线程安全的输出重定向"""
        def __init__(self, output):
            self.output = output
            self.lock = threading.Lock()
        
        def write(self, message):
            with self.lock:
                self.output.write(message)
        
        def flush(self):
            with self.lock:
                if hasattr(self.output, 'flush'):
                    self.output.flush()
    
    def worker(thread_id, output):
        """工作线程函数"""
        for i in range(5):
            message = f"线程 {thread_id} - 消息 {i+1}\n"
            output.write(message)
            time.sleep(0.1)
    
    # 创建线程安全输出
    with open('thread_output.txt', 'w', encoding='utf-8') as f:
        thread_safe_output = ThreadSafeOutput(f)
        
        # 使用线程池
        with ThreadPoolExecutor(max_workers=3) as executor:
            futures = []
            for i in range(3):
                future = executor.submit(worker, i, thread_safe_output)
                futures.append(future)
            
            # 等待所有任务完成
            for future in futures:
                future.result()
    
    print("多线程输出完成")
    print("输出文件内容:")
    with open('thread_output.txt', 'r', encoding='utf-8') as f:
        print(f.read())

multithreaded_output_redirection()

def performance_optimization():
    """输出重定向性能优化"""
    import time
    import io
    
    # 测试不同输出方法的性能
    test_data = [f"测试行 {i}\n" for i in range(10000)]
    
    # 方法1: 直接文件写入
    start_time = time.time()
    with open('direct_write.txt', 'w', encoding='utf-8') as f:
        for line in test_data:
            f.write(line)
    direct_time = time.time() - start_time
    
    # 方法2: 字符串IO缓冲
    start_time = time.time()
    buffer = io.StringIO()
    for line in test_data:
        buffer.write(line)
    with open('buffered_write.txt', 'w', encoding='utf-8') as f:
        f.write(buffer.getvalue())
    buffered_time = time.time() - start_time
    
    # 方法3: 批量写入
    start_time = time.time()
    with open('batch_write.txt', 'w', encoding='utf-8') as f:
        batch_size = 1000
        for i in range(0, len(test_data), batch_size):
            batch = ''.join(test_data[i:i+batch_size])
            f.write(batch)
    batch_time = time.time() - start_time
    
    print("性能测试结果:")
    print(f"直接写入: {direct_time:.4f}秒")
    print(f"缓冲写入: {buffered_time:.4f}秒")
    print(f"批量写入: {batch_time:.4f}秒")
    print(f"批量比直接快: {(direct_time/batch_time):.2f}倍")

performance_optimization()

def error_handling_recovery():
    """输出重定向错误处理与恢复"""
    class SafeOutputRedirect:
        """安全的输出重定向"""
        def __init__(self, filename):
            self.filename = filename
            self.original_stdout = None
            self.file = None
        
        def __enter__(self):
            try:
                self.original_stdout = sys.stdout
                self.file = open(self.filename, 'w', encoding='utf-8')
                sys.stdout = self.file
                return self
            except Exception as e:
                print(f"重定向失败: {e}")
                if self.original_stdout:
                    sys.stdout = self.original_stdout
                raise
        
        def __exit__(self, exc_type, exc_val, exc_tb):
            if self.original_stdout:
                sys.stdout = self.original_stdout
            if self.file:
                try:
                    self.file.close()
                except Exception as e:
                    print(f"文件关闭错误: {e}")
            
            if exc_type:
                print(f"操作过程中发生错误: {exc_val}")
            
            return False  # 不抑制异常
    
    # 使用示例
    try:
        with SafeOutputRedirect('safe_output.txt'):
            print("安全重定向测试")
            print("多条消息")
            # 模拟错误
            # raise ValueError("测试错误")
            print("正常结束")
    
    except Exception as e:
        print(f"捕获到异常: {e}")
    
    finally:
        print("回到控制台输出")
    
    # 验证文件内容
    try:
        with open('safe_output.txt', 'r', encoding='utf-8') as f:
            content = f.read()
            print("文件内容:")
            print(content)
    except FileNotFoundError:
        print("文件不存在")

error_handling_recovery()

def professional_output_redirect(output_file, error_file=None, buffer_size=8192):
    """
    专业输出重定向模板
    
    参数:
        output_file: 输出文件路径
        error_file: 错误文件路径（可选）
        buffer_size: 缓冲区大小
    """
    import sys
    from threading import Lock
    
    class ProfessionalRedirect:
        def __init__(self):
            self.output_handle = open(output_file, 'w', encoding='utf-8')
            self.error_handle = open(error_file, 'w', encoding='utf-8') if error_file else None
            self.buffer_size = buffer_size
            self.stdout_buffer = []
            self.stderr_buffer = []
            self.stdout_lock = Lock()
            self.stderr_lock = Lock()
            self.original_stdout = sys.stdout
            self.original_stderr = sys.stderr
        
        def write_stdout(self, message):
            with self.stdout_lock:
                self.stdout_buffer.append(message)
                if sum(len(m) for m in self.stdout_buffer) >= self.buffer_size:
                    self.flush_stdout()
        
        def write_stderr(self, message):
            with self.stderr_lock:
                self.stderr_buffer.append(message)
                if sum(len(m) for m in self.stderr_buffer) >= self.buffer_size:
                    self.flush_stderr()
        
        def flush_stdout(self):
            if self.stdout_buffer:
                content = ''.join(self.stdout_buffer)
                self.output_handle.write(content)
                self.output_handle.flush()
                self.stdout_buffer = []
        
        def flush_stderr(self):
            if self.stderr_buffer and self.error_handle:
                content = ''.join(self.stderr_buffer)
                self.error_handle.write(content)
                self.error_handle.flush()
                self.stderr_buffer = []
        
        def close(self):
            self.flush_stdout()
            self.flush_stderr()
            self.output_handle.close()
            if self.error_handle:
                self.error_handle.close()
            sys.stdout = self.original_stdout
            sys.stderr = self.original_stderr
    
    return ProfessionalRedirect()

# 使用示例
redirector = professional_output_redirect('output.log', 'error.log', 4096)
sys.stdout = redirector.write_stdout
sys.stderr = redirector.write_stderr

try:
    # 业务逻辑
    print("业务输出")
    print("更多内容")
    
finally:
    redirector.close()