Python实现字节数据写入文本文件的方法完全指南

def demonstrate_bytes_vs_text():
    """
    演示字节和文本数据的区别
    """
    # 文本数据（字符串）
    text_data = "Hello, 世界! 🌍"
    print(f"文本类型: {type(text_data)}")
    print(f"文本内容: {text_data}")
    print(f"文本长度: {len(text_data)} 字符")
    
    # 字节数据
    byte_data = text_data.encode('utf-8')
    print(f"\n字节类型: {type(byte_data)}")
    print(f"字节内容: {byte_data}")
    print(f"字节长度: {len(byte_data)} 字节")
    
    # 显示编码的重要性
    print("\n=== 不同编码比较 ===")
    encodings = ['utf-8', 'gbk', 'iso-8859-1']
    for encoding in encodings:
        try:
            encoded = text_data.encode(encoding)
            decoded = encoded.decode(encoding)
            print(f"{encoding}: {len(encoded)} 字节, 往返成功: {decoded == text_data}")
        except UnicodeEncodeError:
            print(f"{encoding}: 无法编码")
        except UnicodeDecodeError:
            print(f"{encoding}: 无法解码")

# 运行演示
demonstrate_bytes_vs_text()

def demonstrate_byte_sources():
    """
    演示常见的字节数据来源
    """
    sources = {
        '网络请求': b'HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello World',
        '文件读取': open(__file__, 'rb').read(100),  # 读取自身的前100字节
        '序列化数据': b'\x80\x04\x95\x0c\x00\x00\x00\x00\x00\x00\x00\x8c\x0bHello World\x94.',
        '加密数据': b'x\x9c\xf3H\xcd\xc9\xc9\xd7Q\x08\xcf/\xcaI\x01\x00\x18\xab\x04=',
        '二进制协议': b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f'
    }
    
    print("=== 字节数据来源示例 ===")
    for source_name, data in sources.items():
        print(f"{source_name:15}: {len(data):4} 字节, 示例: {data[:20]!r}...")
        
        # 尝试解码为文本（可能失败）
        try:
            decoded = data.decode('utf-8', errors='replace')
            if len(decoded) > 20:
                decoded = decoded[:20] + '...'
            print(f"              尝试解码: {decoded}")
        except Exception as e:
            print(f"              解码失败: {e}")

demonstrate_byte_sources()

def write_bytes_as_text_basic(byte_data, file_path, encoding='utf-8'):
    """
    基础方法：解码后写入文本文件
    """
    try:
        # 将字节数据解码为文本
        text_content = byte_data.decode(encoding)
        
        # 写入文本文件
        with open(file_path, 'w', encoding=encoding) as text_file:
            text_file.write(text_content)
            
        print(f"成功写入 {len(byte_data)} 字节到 {file_path}")
        return True
        
    except UnicodeDecodeError as e:
        print(f"解码错误: {e}")
        return False
    except IOError as e:
        print(f"文件写入错误: {e}")
        return False

# 使用示例
def demo_basic_write():
    """基础写入演示"""
    # 创建测试字节数据
    sample_text = "这是测试内容\n包含中文和特殊字符: ®™¶§\n以及多行文本"
    byte_data = sample_text.encode('utf-8')
    
    # 写入文件
    success = write_bytes_as_text_basic(byte_data, 'basic_output.txt')
    if success:
        # 验证写入内容
        with open('basic_output.txt', 'r', encoding='utf-8') as f:
            content = f.read()
            print(f"写入验证: {content[:50]}...")
    
    # 清理
    import os
    if os.path.exists('basic_output.txt'):
        os.remove('basic_output.txt')

demo_basic_write()

def write_bytes_with_error_handling(byte_data, file_path, encoding='utf-8'):
    """
    带错误处理的字节数据写入
    """
    error_handlers = [
        ('strict', "严格模式 - 遇到错误抛出异常"),
        ('ignore', "忽略模式 - 跳过无效字节"),
        ('replace', "替换模式 - 用替换字符(�)代替"),
        ('backslashreplace', "反斜杠替换 - 使用Python转义序列"),
        ('surrogateescape', "代理转义 - 保留字节信息")
    ]
    
    print(f"=== 处理 {len(byte_data)} 字节数据 ===")
    
    for error_handler, description in error_handlers:
        try:
            # 使用不同的错误处理策略解码
            text_content = byte_data.decode(encoding, errors=error_handler)
            
            # 写入文件
            output_file = f"{file_path}.{error_handler}"
            with open(output_file, 'w', encoding=encoding) as f:
                f.write(text_content)
            
            print(f"{error_handler:15} {description:30} → 成功")
            
        except Exception as e:
            print(f"{error_handler:15} {description:30} → 失败: {e}")
    
    return True

# 使用示例
def demo_error_handling():
    """错误处理演示"""
    # 创建包含无效UTF-8字节的数据
    mixed_data = "有效文本".encode('utf-8') + b'\xff\xfe' + "继续文本".encode('utf-8')
    
    write_bytes_with_error_handling(mixed_data, 'error_handling_demo')
    
    # 清理
    import os
    for handler in ['strict', 'ignore', 'replace', 'backslashreplace', 'surrogateescape']:
        filename = f"error_handling_demo.{handler}"
        if os.path.exists(filename):
            os.remove(filename)

demo_error_handling()

import io

def advanced_bytes_writing(byte_data, file_path, encoding='utf-8'):
    """
    高级字节写入：使用二进制模式和文本包装器
    """
    try:
        # 以二进制模式打开文件
        with open(file_path, 'wb') as binary_file:
            # 创建文本包装器
            text_wrapper = io.TextIOWrapper(
                binary_file,
                encoding=encoding,
                errors='replace',
                write_through=True  # 立即写入底层缓冲
            )
            
            # 写入数据
            if isinstance(byte_data, bytes):
                # 如果是字节数据，先解码
                text_content = byte_data.decode(encoding, errors='replace')
                text_wrapper.write(text_content)
            else:
                # 如果是字节数据流，逐块处理
                for chunk in byte_data:
                    if isinstance(chunk, bytes):
                        decoded_chunk = chunk.decode(encoding, errors='replace')
                        text_wrapper.write(decoded_chunk)
                    else:
                        text_wrapper.write(str(chunk))
            
            # 刷新并分离包装器
            text_wrapper.flush()
            text_wrapper.detach()
        
        print(f"高级写入完成: {file_path}")
        return True
        
    except Exception as e:
        print(f"高级写入错误: {e}")
        return False

# 使用示例
def demo_advanced_writing():
    """高级写入演示"""
    # 创建测试数据
    sample_data = [
        "第一部分文本".encode('utf-8'),
        b'\xff\xfe',  # 无效字节序列
        "第二部分文本".encode('utf-8'),
        "正常文本结尾".encode('utf-8')
    ]
    
    success = advanced_bytes_writing(sample_data, 'advanced_output.txt')
    
    if success:
        # 读取验证
        with open('advanced_output.txt', 'r', encoding='utf-8', errors='replace') as f:
            content = f.read()
            print(f"写入内容: {content}")
    
    # 清理
    import os
    if os.path.exists('advanced_output.txt'):
        os.remove('advanced_output.txt')

demo_advanced_writing()

def stream_bytes_to_text(source_bytes, target_file, encoding='utf-8', chunk_size=4096):
    """
    流式处理字节数据到文本文件
    """
    try:
        with open(target_file, 'w', encoding=encoding) as text_file:
            if isinstance(source_bytes, bytes):
                # 单个字节对象处理
                text_content = source_bytes.decode(encoding, errors='replace')
                text_file.write(text_content)
            else:
                # 字节流处理
                buffer = bytearray()
                for chunk in source_bytes:
                    if isinstance(chunk, bytes):
                        buffer.extend(chunk)
                    else:
                        # 处理非字节数据
                        text_file.write(str(chunk))
                    
                    # 处理缓冲区中的数据
                    while len(buffer) >= chunk_size:
                        # 尝试解码完整块
                        try:
                            text_chunk = buffer[:chunk_size].decode(encoding, errors='strict')
                            text_file.write(text_chunk)
                            buffer = buffer[chunk_size:]
                        except UnicodeDecodeError:
                            # 遇到解码问题，尝试找到边界
                            found = False
                            for i in range(chunk_size - 1, 0, -1):
                                try:
                                    text_chunk = buffer[:i].decode(encoding, errors='strict')
                                    text_file.write(text_chunk)
                                    buffer = buffer[i:]
                                    found = True
                                    break
                                except UnicodeDecodeError:
                                    continue
                            
                            if not found:
                                # 无法找到边界，使用替换策略
                                text_chunk = buffer[:chunk_size].decode(encoding, errors='replace')
                                text_file.write(text_chunk)
                                buffer = buffer[chunk_size:]
                
                # 处理剩余缓冲区
                if buffer:
                    try:
                        text_chunk = buffer.decode(encoding, errors='strict')
                        text_file.write(text_chunk)
                    except UnicodeDecodeError:
                        text_chunk = buffer.decode(encoding, errors='replace')
                        text_file.write(text_chunk)
        
        print(f"流式处理完成: {target_file}")
        return True
        
    except Exception as e:
        print(f"流式处理错误: {e}")
        return False

# 使用示例
def demo_stream_processing():
    """流式处理演示"""
    # 创建生成器模拟字节流
    def byte_stream_generator():
        chunks = [
            "第一部分".encode('utf-8'),
            b'\xff\xfe',  # 无效序列
            "第二部分".encode('utf-8'),
            "第三部分很长的内容".encode('utf-8') * 100  # 大块数据
        ]
        for chunk in chunks:
            yield chunk
    
    # 处理流数据
    success = stream_bytes_to_text(byte_stream_generator(), 'stream_output.txt')
    
    if success:
        # 检查文件大小
        import os
        file_size = os.path.getsize('stream_output.txt')
        print(f"输出文件大小: {file_size} 字节")
        
        # 清理
        os.remove('stream_output.txt')

demo_stream_processing()

import base64
import binascii

def write_bytes_with_encoding(byte_data, file_path, output_format='text'):
    """
    以不同格式写入字节数据
    """
    formats = {
        'text': lambda d: d.decode('utf-8', errors='replace'),
        'hex': lambda d: binascii.hexlify(d).decode('ascii'),
        'base64': lambda d: base64.b64encode(d).decode('ascii'),
        'base64_lines': lambda d: base64.b64encode(d).decode('ascii') + '\n',
        'c_style': lambda d: ''.join(f'\\x{b:02x}' for b in d)
    }
    
    if output_format not in formats:
        print(f"不支持的格式: {output_format}")
        return False
    
    try:
        # 转换数据
        if output_format == 'base64_lines':
            # 特殊处理：每76字符换行（Base64标准）
            encoded = base64.b64encode(byte_data).decode('ascii')
            formatted = '\n'.join([encoded[i:i+76] for i in range(0, len(encoded), 76)])
        else:
            formatted = formats[output_format](byte_data)
        
        # 写入文件
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(formatted)
        
        print(f"{output_format:12} 格式写入完成: {len(byte_data)} 字节 → {len(formatted)} 字符")
        return True
        
    except Exception as e:
        print(f"{output_format} 格式写入错误: {e}")
        return False

# 使用示例
def demo_formatted_output():
    """格式化输出演示"""
    sample_data = b'\x00\x01\x02\x03\x04\x05Hello World!\xff\xfe\xfd\xfc\xfb\xfa'
    
    formats = ['text', 'hex', 'base64', 'base64_lines', 'c_style']
    
    for fmt in formats:
        filename = f'formatted_{fmt}.txt'
        success = write_bytes_with_encoding(sample_data, filename, fmt)
        
        if success:
            # 显示部分内容
            with open(filename, 'r', encoding='utf-8') as f:
                content = f.read(50)
                print(f"{fmt:12}: {content}...")
        
        # 清理
        import os
        if os.path.exists(filename):
            os.remove(filename)

demo_formatted_output()

def write_structured_byte_data(byte_data, file_path, bytes_per_line=16):
    """
    以结构化格式写入字节数据（类似hexdump）
    """
    try:
        with open(file_path, 'w', encoding='utf-8') as f:
            # 写入文件头
            f.write(f"字节数据转储 - 长度: {len(byte_data)} 字节\n")
            f.write("=" * 70 + "\n")
            f.write("偏移量   十六进制值                          ASCII\n")
            f.write("=" * 70 + "\n")
            
            # 处理每行数据
            for i in range(0, len(byte_data), bytes_per_line):
                chunk = byte_data[i:i + bytes_per_line]
                
                # 十六进制部分
                hex_part = ' '.join(f'{b:02x}' for b in chunk)
                hex_part = hex_part.ljust(bytes_per_line * 3 - 1)  # 保持对齐
                
                # ASCII部分（可打印字符）
                ascii_part = ''.join(chr(b) if 32 <= b <= 126 else '.' for b in chunk)
                
                # 写入行
                f.write(f"{i:08x}  {hex_part}  |{ascii_part}|\n")
        
        print(f"结构化转储完成: {file_path}")
        return True
        
    except Exception as e:
        print(f"结构化转储错误: {e}")
        return False

# 使用示例
def demo_structured_output():
    """结构化输出演示"""
    # 创建包含各种字节的测试数据
    test_data = bytes(range(256))  # 0x00 到 0xFF
    
    success = write_structured_byte_data(test_data, 'structured_dump.txt')
    
    if success:
        # 显示前几行
        with open('structured_dump.txt', 'r', encoding='utf-8') as f:
            for i in range(5):
                line = f.readline().strip()
                print(f"行 {i+1}: {line}")
        
        # 清理
        import os
        os.remove('structured_dump.txt')

demo_structured_output()

import socket
import datetime

class NetworkPacketLogger:
    """
    网络数据包日志记录器
    """
    
    def __init__(self, log_file='network_packets.log'):
        self.log_file = log_file
        self.packet_count = 0
    
    def log_packet(self, packet_data, source_ip, destination_ip, protocol='TCP'):
        """
        记录网络数据包
        """
        timestamp = datetime.datetime.now().isoformat()
        self.packet_count += 1
        
        try:
            # 尝试解码为文本（可能失败）
            try:
                text_content = packet_data.decode('utf-8', errors='replace')
                content_preview = text_content[:100] + ('...' if len(text_content) > 100 else '')
                content_type = 'text'
            except:
                content_preview = f"{len(packet_data)} 字节二进制数据"
                content_type = 'binary'
            
            # 写入日志
            with open(self.log_file, 'a', encoding='utf-8') as f:
                f.write(f"\n{'='*80}\n")
                f.write(f"数据包 #{self.packet_count} - {timestamp}\n")
                f.write(f"来源: {source_ip} → 目标: {destination_ip} ({protocol})\n")
                f.write(f"长度: {len(packet_data)} 字节, 类型: {content_type}\n")
                f.write(f"{'-'*80}\n")
                
                if content_type == 'text':
                    f.write(text_content)
                else:
                    # 二进制数据以十六进制格式写入
                    hex_data = packet_data.hex()
                    for i in range(0, len(hex_data), 80):
                        f.write(hex_data[i:i+80] + '\n')
                
                f.write(f"\n{'='*80}\n")
            
            print(f"记录数据包 #{self.packet_count}: {len(packet_data)} 字节")
            return True
            
        except Exception as e:
            print(f"记录数据包错误: {e}")
            return False
    
    def clear_log(self):
        """清空日志文件"""
        with open(self.log_file, 'w', encoding='utf-8') as f:
            f.write("网络数据包日志\n")
            f.write("=" * 80 + "\n")
        self.packet_count = 0
        print("日志已清空")

# 使用示例
def demo_network_logging():
    """网络日志记录演示"""
    logger = NetworkPacketLogger('demo_network.log')
    logger.clear_log()
    
    # 模拟网络数据包
    test_packets = [
        (b'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n<html>Hello</html>', 
         '192.168.1.1', '192.168.1.100'),
        (b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f', 
         '10.0.0.1', '10.0.0.2'),
        ('GET /api/data HTTP/1.1\r\nHost: example.com\r\n\r\n'.encode('utf-8'),
         '172.16.0.1', '93.184.216.34')
    ]
    
    for packet_data, src_ip, dst_ip in test_packets:
        logger.log_packet(packet_data, src_ip, dst_ip)
    
    print(f"记录了 {logger.packet_count} 个数据包")
    
    # 显示日志内容
    with open('demo_network.log', 'r', encoding='utf-8') as f:
        content = f.read()
        print(f"日志文件大小: {len(content)} 字符")
        print("前200字符:", content[:200] + '...')
    
    # 清理
    import os
    os.remove('demo_network.log')

demo_network_logging()

class BinaryFileAnalyzer:
    """
    二进制文件分析报告生成器
    """
    
    def __init__(self):
        self.analysis_results = []
    
    def analyze_file(self, file_path, output_report_path):
        """
        分析二进制文件并生成文本报告
        """
        try:
            with open(file_path, 'rb') as binary_file:
                file_data = binary_file.read()
            
            # 执行各种分析
            analyses = [
                self._analyze_basic_info,
                self._analyze_byte_distribution,
                self._analyze_text_content,
                self._analyze_file_signature,
                self._analyze_entropy
            ]
            
            # 执行所有分析
            for analysis_func in analyses:
                try:
                    result = analysis_func(file_data, file_path)
                    self.analysis_results.append(result)
                except Exception as e:
                    self.analysis_results.append({
                        '分析类型': analysis_func.__name__,
                        '错误': str(e)
                    })
            
            # 生成报告
            self._generate_report(output_report_path, file_path)
            
            print(f"分析完成: {file_path} → {output_report_path}")
            return True
            
        except Exception as e:
            print(f"文件分析错误: {e}")
            return False
    
    def _analyze_basic_info(self, data, file_path):
        """分析基本信息"""
        import os
        file_stats = os.stat(file_path)
        
        return {
            '分析类型': '基本信息',
            '文件大小': f"{len(data)} 字节",
            '文件修改时间': datetime.datetime.fromtimestamp(file_stats.st_mtime),
            'MD5哈希': self._calculate_md5(data)
        }
    
    def _analyze_byte_distribution(self, data, file_path):
        """分析字节分布"""
        from collections import Counter
        byte_count = Counter(data)
        common_bytes = byte_count.most_common(10)
        
        return {
            '分析类型': '字节分布',
            '最常见字节': [f"0x{b:02x} ({count}次)" for b, count in common_bytes],
            '零字节数量': byte_count.get(0, 0),
            'FF字节数量': byte_count.get(255, 0)
        }
    
    def _analyze_text_content(self, data, file_path):
        """分析文本内容"""
        try:
            # 尝试UTF-8解码
            text_content = data.decode('utf-8', errors='replace')
            text_lines = text_content.split('\n')
            
            return {
                '分析类型': '文本内容',
                '可读文本行数': len([line for line in text_lines if len(line.strip()) > 0]),
                '最长文本行': max([len(line) for line in text_lines], default=0),
                '文本预览': text_content[:200] + ('...' if len(text_content) > 200 else '')
            }
        except:
            return {
                '分析类型': '文本内容',
                '结果': '无法解码为文本'
            }
    
    def _analyze_file_signature(self, data, file_path):
        """分析文件签名（魔数）"""
        signatures = {
            b'\xff\xd8\xff': 'JPEG图像',
            b'\x89PNG': 'PNG图像',
            b'PK\x03\x04': 'ZIP压缩文件',
            b'%PDF': 'PDF文档',
            b'\x7fELF': 'ELF可执行文件',
            b'MZ': 'Windows可执行文件'
        }
        
        file_type = '未知'
        for sig, file_type_name in signatures.items():
            if data.startswith(sig):
                file_type = file_type_name
                break
        
        return {
            '分析类型': '文件签名',
            '检测到的类型': file_type,
            '文件头': data[:8].hex(' ', 1)
        }
    
    def _analyze_entropy(self, data, file_path):
        """分析文件熵（随机性）"""
        import math
        from collections import Counter
        
        if len(data) == 0:
            return {'分析类型': '熵分析', '熵值': 0}
        
        byte_count = Counter(data)
        entropy = 0.0
        
        for count in byte_count.values():
            p = count / len(data)
            entropy -= p * math.log2(p)
        
        return {
            '分析类型': '熵分析',
            '熵值': f"{entropy:.4f}",
            '解释': '高熵值可能表示加密或压缩数据' if entropy > 7.0 else '低熵值可能表示文本或结构化数据'
        }
    
    def _calculate_md5(self, data):
        """计算MD5哈希"""
        import hashlib
        return hashlib.md5(data).hexdigest()
    
    def _generate_report(self, output_path, original_file):
        """生成文本报告"""
        with open(output_path, 'w', encoding='utf-8') as report_file:
            report_file.write(f"二进制文件分析报告\n")
            report_file.write(f"文件: {original_file}\n")
            report_file.write(f"生成时间: {datetime.datetime.now().isoformat()}\n")
            report_file.write("=" * 80 + "\n\n")
            
            for result in self.analysis_results:
                report_file.write(f"{result['分析类型']}:\n")
                report_file.write("-" * 40 + "\n")
                
                for key, value in result.items():
                    if key != '分析类型':
                        if isinstance(value, list):
                            report_file.write(f"  {key}: {', '.join(value)}\n")
                        else:
                            report_file.write(f"  {key}: {value}\n")
                
                report_file.write("\n")

# 使用示例
def demo_binary_analysis():
    """二进制文件分析演示"""
    analyzer = BinaryFileAnalyzer()
    
    # 创建一个测试二进制文件
    test_data = b'\x89PNG\r\n\x1a\n' + b'\x00' * 100 + b'TEST CONTENT' + bytes(range(256))
    with open('test_binary_file.bin', 'wb') as f:
        f.write(test_data)
    
    # 分析文件
    success = analyzer.analyze_file('test_binary_file.bin', 'analysis_report.txt')
    
    if success:
        # 显示报告内容
        with open('analysis_report.txt', 'r', encoding='utf-8') as f:
            content = f.read()
            print("分析报告生成成功:")
            print(content[:300] + "..." if len(content) > 300 else content)
    
    # 清理
    import os
    for filename in ['test_binary_file.bin', 'analysis_report.txt']:
        if os.path.exists(filename):
            os.remove(filename)

demo_binary_analysis()

class HighPerformanceByteWriter:
    """
    高性能字节数据写入器
    """
    
    def __init__(self, buffer_size=8192, encoding='utf-8'):
        self.buffer_size = buffer_size
        self.encoding = encoding
        self.byte_cache = {}
    
    def write_large_bytes(self, byte_data, output_file):
        """
        高性能写入大量字节数据
        """
        try:
            # 使用内存视图避免复制
            if isinstance(byte_data, (bytes, bytearray)):
                data_view = memoryview(byte_data)
            else:
                data_view = memoryview(byte_data.encode(self.encoding))
            
            with open(output_file, 'w', encoding=self.encoding, buffering=self.buffer_size) as f:
                # 分块处理
                total_bytes = len(data_view)
                processed = 0
                
                while processed < total_bytes:
                    chunk_end = min(processed + self.buffer_size, total_bytes)
                    chunk = data_view[processed:chunk_end]
                    
                    # 解码并写入
                    try:
                        text_chunk = chunk.tobytes().decode(self.encoding, errors='replace')
                        f.write(text_chunk)
                    except UnicodeDecodeError:
                        # 处理解码错误
                        text_chunk = chunk.tobytes().decode(self.encoding, errors='ignore')
                        f.write(text_chunk)
                    
                    processed = chunk_end
                    
                    # 进度显示（可选）
                    if processed % (self.buffer_size * 10) == 0:
                        progress = (processed / total_bytes) * 100
                        print(f"处理进度: {progress:.1f}%")
            
            print(f"高性能写入完成: {total_bytes} 字节 → {output_file}")
            return True
            
        except Exception as e:
            print(f"高性能写入错误: {e}")
            return False
    
    def batch_process_files(self, file_list, output_dir):
        """
        批量处理多个文件
        """
        import concurrent.futures
        import os
        
        os.makedirs(output_dir, exist_ok=True)
        results = []
        
        # 使用线程池并行处理
        with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
            future_to_file = {}
            
            for input_file in file_list:
                if not os.path.exists(input_file):
                    continue
                
                output_file = os.path.join(output_dir, os.path.basename(input_file) + '.txt')
                future = executor.submit(self.process_single_file, input_file, output_file)
                future_to_file[future] = (input_file, output_file)
            
            # 收集结果
            for future in concurrent.futures.as_completed(future_to_file):
                input_file, output_file = future_to_file[future]
                try:
                    result = future.result()
                    results.append({
                        'input': input_file,
                        'output': output_file,
                        'success': result,
                        'error': None
                    })
                except Exception as e:
                    results.append({
                        'input': input_file,
                        'output': output_file,
                        'success': False,
                        'error': str(e)
                    })
        
        return results
    
    def process_single_file(self, input_file, output_file):
        """
        处理单个文件
        """
        try:
            with open(input_file, 'rb') as f:
                file_data = f.read()
            
            return self.write_large_bytes(file_data, output_file)
        except Exception as e:
            print(f"处理文件 {input_file} 错误: {e}")
            return False

# 使用示例
def demo_performance_optimization():
    """性能优化演示"""
    # 创建大测试文件
    large_content = "测试数据" * 1000000  # 约8MB文本
    large_bytes = large_content.encode('utf-8')
    
    with open('large_test_file.bin', 'wb') as f:
        f.write(large_bytes)
    
    # 高性能处理
    writer = HighPerformanceByteWriter()
    success = writer.write_large_bytes(large_bytes, 'high_perf_output.txt')
    
    if success:
        # 验证文件大小
        import os
        input_size = os.path.getsize('large_test_file.bin')
        output_size = os.path.getsize('high_perf_output.txt')
        print(f"输入: {input_size} 字节, 输出: {output_size} 字符")
        print(f"压缩比: {output_size/input_size:.2f}")
    
    # 清理
    for filename in ['large_test_file.bin', 'high_perf_output.txt']:
        if os.path.exists(filename):
            os.remove(filename)

demo_performance_optimization()