def demonstrate_bytes_vs_text():
    """
    演示字節(jié)和文本數(shù)據(jù)的區(qū)別
    """
    # 文本數(shù)據(jù)（字符串）
    text_data = "Hello, 世界! ??"
    print(f"文本類型: {type(text_data)}")
    print(f"文本內(nèi)容: {text_data}")
    print(f"文本長度: {len(text_data)} 字符")
    
    # 字節(jié)數(shù)據(jù)
    byte_data = text_data.encode('utf-8')
    print(f"\n字節(jié)類型: {type(byte_data)}")
    print(f"字節(jié)內(nèi)容: {byte_data}")
    print(f"字節(jié)長度: {len(byte_data)} 字節(jié)")
    
    # 顯示編碼的重要性
    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)} 字節(jié), 往返成功: {decoded == text_data}")
        except UnicodeEncodeError:
            print(f"{encoding}: 無法編碼")
        except UnicodeDecodeError:
            print(f"{encoding}: 無法解碼")

# 運行演示
demonstrate_bytes_vs_text()

def demonstrate_byte_sources():
    """
    演示常見的字節(jié)數(shù)據(jù)來源
    """
    sources = {
        '網(wǎng)絡(luò)請求': b'HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello World',
        '文件讀取': open(__file__, 'rb').read(100),  # 讀取自身的前100字節(jié)
        '序列化數(shù)據(jù)': b'\x80\x04\x95\x0c\x00\x00\x00\x00\x00\x00\x00\x8c\x0bHello World\x94.',
        '加密數(shù)據(jù)': b'x\x9c\xf3H\xcd\xc9\xc9\xd7Q\x08\xcf/\xcaI\x01\x00\x18\xab\x04=',
        '二進制協(xié)議': b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f'
    }
    
    print("=== 字節(jié)數(shù)據(jù)來源示例 ===")
    for source_name, data in sources.items():
        print(f"{source_name:15}: {len(data):4} 字節(jié), 示例: {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'):
    """
    基礎(chǔ)方法：解碼后寫入文本文件
    """
    try:
        # 將字節(jié)數(shù)據(jù)解碼為文本
        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)} 字節(jié)到 {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():
    """基礎(chǔ)寫入演示"""
    # 創(chuàng)建測試字節(jié)數(shù)據(jù)
    sample_text = "這是測試內(nèi)容\n包含中文和特殊字符: ???§\n以及多行文本"
    byte_data = sample_text.encode('utf-8')
    
    # 寫入文件
    success = write_bytes_as_text_basic(byte_data, 'basic_output.txt')
    if success:
        # 驗證寫入內(nèi)容
        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'):
    """
    帶錯誤處理的字節(jié)數(shù)據(jù)寫入
    """
    error_handlers = [
        ('strict', "嚴格模式 - 遇到錯誤拋出異常"),
        ('ignore', "忽略模式 - 跳過無效字節(jié)"),
        ('replace', "替換模式 - 用替換字符(?)代替"),
        ('backslashreplace', "反斜杠替換 - 使用Python轉(zhuǎn)義序列"),
        ('surrogateescape', "代理轉(zhuǎn)義 - 保留字節(jié)信息")
    ]
    
    print(f"=== 處理 {len(byte_data)} 字節(jié)數(shù)據(jù) ===")
    
    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():
    """錯誤處理演示"""
    # 創(chuàng)建包含無效UTF-8字節(jié)的數(shù)據(jù)
    mixed_data = "有效文本".encode('utf-8') + b'\xff\xfe' + "繼續(xù)文本".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'):
    """
    高級字節(jié)寫入：使用二進制模式和文本包裝器
    """
    try:
        # 以二進制模式打開文件
        with open(file_path, 'wb') as binary_file:
            # 創(chuàng)建文本包裝器
            text_wrapper = io.TextIOWrapper(
                binary_file,
                encoding=encoding,
                errors='replace',
                write_through=True  # 立即寫入底層緩沖
            )
            
            # 寫入數(shù)據(jù)
            if isinstance(byte_data, bytes):
                # 如果是字節(jié)數(shù)據(jù)，先解碼
                text_content = byte_data.decode(encoding, errors='replace')
                text_wrapper.write(text_content)
            else:
                # 如果是字節(jié)數(shù)據(jù)流，逐塊處理
                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():
    """高級寫入演示"""
    # 創(chuàng)建測試數(shù)據(jù)
    sample_data = [
        "第一部分文本".encode('utf-8'),
        b'\xff\xfe',  # 無效字節(jié)序列
        "第二部分文本".encode('utf-8'),
        "正常文本結(jié)尾".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"寫入內(nèi)容: {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):
    """
    流式處理字節(jié)數(shù)據(jù)到文本文件
    """
    try:
        with open(target_file, 'w', encoding=encoding) as text_file:
            if isinstance(source_bytes, bytes):
                # 單個字節(jié)對象處理
                text_content = source_bytes.decode(encoding, errors='replace')
                text_file.write(text_content)
            else:
                # 字節(jié)流處理
                buffer = bytearray()
                for chunk in source_bytes:
                    if isinstance(chunk, bytes):
                        buffer.extend(chunk)
                    else:
                        # 處理非字節(jié)數(shù)據(jù)
                        text_file.write(str(chunk))
                    
                    # 處理緩沖區(qū)中的數(shù)據(jù)
                    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:]
                
                # 處理剩余緩沖區(qū)
                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():
    """流式處理演示"""
    # 創(chuàng)建生成器模擬字節(jié)流
    def byte_stream_generator():
        chunks = [
            "第一部分".encode('utf-8'),
            b'\xff\xfe',  # 無效序列
            "第二部分".encode('utf-8'),
            "第三部分很長的內(nèi)容".encode('utf-8') * 100  # 大塊數(shù)據(jù)
        ]
        for chunk in chunks:
            yield chunk
    
    # 處理流數(shù)據(jù)
    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} 字節(jié)")
        
        # 清理
        os.remove('stream_output.txt')

demo_stream_processing()

import base64
import binascii

def write_bytes_with_encoding(byte_data, file_path, output_format='text'):
    """
    以不同格式寫入字節(jié)數(shù)據(jù)
    """
    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:
        # 轉(zhuǎn)換數(shù)據(jù)
        if output_format == 'base64_lines':
            # 特殊處理：每76字符換行（Base64標(biāo)準(zhǔn)）
            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)} 字節(jié) → {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:
            # 顯示部分內(nèi)容
            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):
    """
    以結(jié)構(gòu)化格式寫入字節(jié)數(shù)據(jù)（類似hexdump）
    """
    try:
        with open(file_path, 'w', encoding='utf-8') as f:
            # 寫入文件頭
            f.write(f"字節(jié)數(shù)據(jù)轉(zhuǎn)儲 - 長度: {len(byte_data)} 字節(jié)\n")
            f.write("=" * 70 + "\n")
            f.write("偏移量   十六進制值                          ASCII\n")
            f.write("=" * 70 + "\n")
            
            # 處理每行數(shù)據(jù)
            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"結(jié)構(gòu)化轉(zhuǎn)儲完成: {file_path}")
        return True
        
    except Exception as e:
        print(f"結(jié)構(gòu)化轉(zhuǎn)儲錯誤: {e}")
        return False

# 使用示例
def demo_structured_output():
    """結(jié)構(gòu)化輸出演示"""
    # 創(chuàng)建包含各種字節(jié)的測試數(shù)據(jù)
    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:
    """
    網(wǎng)絡(luò)數(shù)據(jù)包日志記錄器
    """
    
    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'):
        """
        記錄網(wǎng)絡(luò)數(shù)據(jù)包
        """
        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)} 字節(jié)二進制數(shù)據(jù)"
                content_type = 'binary'
            
            # 寫入日志
            with open(self.log_file, 'a', encoding='utf-8') as f:
                f.write(f"\n{'='*80}\n")
                f.write(f"數(shù)據(jù)包 #{self.packet_count} - {timestamp}\n")
                f.write(f"來源: {source_ip} → 目標(biāo): {destination_ip} ({protocol})\n")
                f.write(f"長度: {len(packet_data)} 字節(jié), 類型: {content_type}\n")
                f.write(f"{'-'*80}\n")
                
                if content_type == 'text':
                    f.write(text_content)
                else:
                    # 二進制數(shù)據(jù)以十六進制格式寫入
                    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"記錄數(shù)據(jù)包 #{self.packet_count}: {len(packet_data)} 字節(jié)")
            return True
            
        except Exception as e:
            print(f"記錄數(shù)據(jù)包錯誤: {e}")
            return False
    
    def clear_log(self):
        """清空日志文件"""
        with open(self.log_file, 'w', encoding='utf-8') as f:
            f.write("網(wǎng)絡(luò)數(shù)據(jù)包日志\n")
            f.write("=" * 80 + "\n")
        self.packet_count = 0
        print("日志已清空")

# 使用示例
def demo_network_logging():
    """網(wǎng)絡(luò)日志記錄演示"""
    logger = NetworkPacketLogger('demo_network.log')
    logger.clear_log()
    
    # 模擬網(wǎng)絡(luò)數(shù)據(jù)包
    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} 個數(shù)據(jù)包")
    
    # 顯示日志內(nèi)容
    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()
            
            # 執(zhí)行各種分析
            analyses = [
                self._analyze_basic_info,
                self._analyze_byte_distribution,
                self._analyze_text_content,
                self._analyze_file_signature,
                self._analyze_entropy
            ]
            
            # 執(zhí)行所有分析
            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)} 字節(jié)",
            '文件修改時間': datetime.datetime.fromtimestamp(file_stats.st_mtime),
            'MD5哈希': self._calculate_md5(data)
        }
    
    def _analyze_byte_distribution(self, data, file_path):
        """分析字節(jié)分布"""
        from collections import Counter
        byte_count = Counter(data)
        common_bytes = byte_count.most_common(10)
        
        return {
            '分析類型': '字節(jié)分布',
            '最常見字節(jié)': [f"0x{b:02x} ({count}次)" for b, count in common_bytes],
            '零字節(jié)數(shù)量': byte_count.get(0, 0),
            'FF字節(jié)數(shù)量': byte_count.get(255, 0)
        }
    
    def _analyze_text_content(self, data, file_path):
        """分析文本內(nèi)容"""
        try:
            # 嘗試UTF-8解碼
            text_content = data.decode('utf-8', errors='replace')
            text_lines = text_content.split('\n')
            
            return {
                '分析類型': '文本內(nèi)容',
                '可讀文本行數(shù)': len([line for line in text_lines if len(line.strip()) > 0]),
                '最長文本行': max([len(line) for line in text_lines], default=0),
                '文本預(yù)覽': text_content[:200] + ('...' if len(text_content) > 200 else '')
            }
        except:
            return {
                '分析類型': '文本內(nèi)容',
                '結(jié)果': '無法解碼為文本'
            }
    
    def _analyze_file_signature(self, data, file_path):
        """分析文件簽名（魔數(shù)）"""
        signatures = {
            b'\xff\xd8\xff': 'JPEG圖像',
            b'\x89PNG': 'PNG圖像',
            b'PK\x03\x04': 'ZIP壓縮文件',
            b'%PDF': 'PDF文檔',
            b'\x7fELF': 'ELF可執(zhí)行文件',
            b'MZ': 'Windows可執(zhí)行文件'
        }
        
        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}",
            '解釋': '高熵值可能表示加密或壓縮數(shù)據(jù)' if entropy > 7.0 else '低熵值可能表示文本或結(jié)構(gòu)化數(shù)據(jù)'
        }
    
    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()
    
    # 創(chuàng)建一個測試二進制文件
    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:
        # 顯示報告內(nèi)容
        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:
    """
    高性能字節(jié)數(shù)據(jù)寫入器
    """
    
    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):
        """
        高性能寫入大量字節(jié)數(shù)據(jù)
        """
        try:
            # 使用內(nèi)存視圖避免復(fù)制
            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} 字節(jié) → {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)
            
            # 收集結(jié)果
            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():
    """性能優(yōu)化演示"""
    # 創(chuàng)建大測試文件
    large_content = "測試數(shù)據(jù)" * 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} 字節(jié), 輸出: {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()