def basic_binary_operations():
    """基礎(chǔ)二進(jìn)制文件操作"""
    # 寫(xiě)入二進(jìn)制數(shù)據(jù)
    with open('binary_data.bin', 'wb') as f:
        # 寫(xiě)入字節(jié)數(shù)據(jù)
        f.write(b'\x00\x01\x02\x03\x04\x05')  # 十六進(jìn)制字節(jié)
        f.write(bytes([10, 20, 30, 40, 50]))  # 十進(jìn)制字節(jié)數(shù)組
        f.write(bytearray([100, 200, 255]))   # 字節(jié)數(shù)組
        
        # 寫(xiě)入文本數(shù)據(jù)的二進(jìn)制形式
        text_data = "Hello, 世界!"
        f.write(text_data.encode('utf-8'))
    
    # 讀取二進(jìn)制數(shù)據(jù)
    with open('binary_data.bin', 'rb') as f:
        # 讀取全部數(shù)據(jù)
        all_data = f.read()
        print(f"全部數(shù)據(jù): {all_data}")
        print(f"數(shù)據(jù)長(zhǎng)度: {len(all_data)} 字節(jié)")
        
        # 重新定位到文件開(kāi)頭
        f.seek(0)
        
        # 讀取指定字節(jié)數(shù)
        first_6_bytes = f.read(6)
        print(f"前6字節(jié): {first_6_bytes}")
        
        # 繼續(xù)讀取
        next_bytes = f.read(5)
        print(f"接下來(lái)5字節(jié): {next_bytes}")
        
        # 讀取剩余數(shù)據(jù)
        remaining = f.read()
        print(f"剩余數(shù)據(jù): {remaining}")
        
        # 嘗試解碼文本數(shù)據(jù)
        try:
            text_content = remaining.decode('utf-8')
            print(f"解碼文本: {text_content}")
        except UnicodeDecodeError as e:
            print(f"解碼錯(cuò)誤: {e}")

# 執(zhí)行示例
basic_binary_operations()

def byte_manipulation():
    """字節(jié)數(shù)據(jù)操作與轉(zhuǎn)換"""
    # 創(chuàng)建字節(jié)數(shù)據(jù)的多種方式
    data1 = bytes([0, 1, 2, 3, 4, 5])
    data2 = bytearray([10, 20, 30, 40, 50])
    data3 = b'\x00\x01\x02\x03\x04\x05'
    data4 = "Hello".encode('ascii')
    
    print("字節(jié)數(shù)據(jù)示例:")
    print(f"bytes: {data1}")
    print(f"bytearray: {data2}")
    print(f"字節(jié)字面量: {data3}")
    print(f"編碼文本: {data4}")
    
    # 字節(jié)數(shù)據(jù)操作
    combined = data1 + data2 + data3 + data4
    print(f"合并數(shù)據(jù): {combined}")
    print(f"合并長(zhǎng)度: {len(combined)} 字節(jié)")
    
    # 切片操作
    slice1 = combined[5:10]
    slice2 = combined[-5:]
    print(f"切片[5:10]: {slice1}")
    print(f"切片[-5:]: {slice2}")
    
    # 修改字節(jié)數(shù)據(jù) (bytearray可變)
    mutable_data = bytearray(combined)
    mutable_data[0] = 255  # 修改第一個(gè)字節(jié)
    mutable_data[5:10] = b'\xFF\xFE\xFD'  # 替換范圍
    print(f"修改后數(shù)據(jù): {mutable_data}")
    
    # 搜索和替換
    if b'Hello' in combined:
        position = combined.index(b'Hello')
        print(f"找到 'Hello' 在位置: {position}")
    
    # 計(jì)數(shù)和統(tǒng)計(jì)
    count_zeros = combined.count(0)
    print(f"字節(jié)0出現(xiàn)次數(shù): {count_zeros}")
    
    # 轉(zhuǎn)換為不同表示形式
    print(f"十六進(jìn)制: {combined.hex()}")
    print(f"整數(shù)列表: {list(combined)}")

byte_manipulation()

import struct

def struct_module_usage():
    """使用struct模塊處理結(jié)構(gòu)化二進(jìn)制數(shù)據(jù)"""
    # 打包數(shù)據(jù)
    packed_data = struct.pack('>I f 4s', 123456, 3.14159, b'TEST')
    print(f"打包數(shù)據(jù): {packed_data}")
    print(f"打包長(zhǎng)度: {len(packed_data)} 字節(jié)")
    
    # 解包數(shù)據(jù)
    unpacked_data = struct.unpack('>I f 4s', packed_data)
    print(f"解包數(shù)據(jù): {unpacked_data}")
    
    # 計(jì)算大小
    calc_size = struct.calcsize('>I f 4s')
    print(f"格式大小: {calc_size} 字節(jié)")
    
    # 復(fù)雜數(shù)據(jù)結(jié)構(gòu)
    complex_format = '>I I 10s d'  # 大端序，2個(gè)整數(shù)，10字節(jié)字符串，雙精度浮點(diǎn)數(shù)
    sample_data = (1001, 2002, b'HelloWorld', 2.71828)
    
    packed_complex = struct.pack(complex_format, *sample_data)
    print(f"復(fù)雜打包: {packed_complex}")
    
    # 寫(xiě)入文件
    with open('structured.bin', 'wb') as f:
        f.write(packed_complex)
    
    # 從文件讀取并解包
    with open('structured.bin', 'rb') as f:
        file_data = f.read()
        unpacked_complex = struct.unpack(complex_format, file_data)
        print(f"文件解包: {unpacked_complex}")
        
        # 處理字符串?dāng)?shù)據(jù)
        text_data = unpacked_complex[2].decode('ascii').rstrip('\x00')
        print(f"解碼文本: {text_data}")
    
    # 處理多個(gè)結(jié)構(gòu)
    multiple_records = []
    for i in range(5):
        record = struct.pack('>H d', i, i * 1.5)
        multiple_records.append(record)
    
    # 寫(xiě)入多個(gè)記錄
    with open('multiple_records.bin', 'wb') as f:
        for record in multiple_records:
            f.write(record)
    
    # 讀取多個(gè)記錄
    record_format = '>H d'
    record_size = struct.calcsize(record_format)
    
    with open('multiple_records.bin', 'rb') as f:
        while True:
            record_data = f.read(record_size)
            if not record_data:
                break
            unpacked_record = struct.unpack(record_format, record_data)
            print(f"記錄: {unpacked_record}")

struct_module_usage()

def binary_protocol_implementation():
    """二進(jìn)制協(xié)議實(shí)現(xiàn)"""
    # 定義協(xié)議格式
    class BinaryProtocol:
        """簡(jiǎn)單的二進(jìn)制協(xié)議"""
        HEADER_FORMAT = '>I I'  # 大端序，消息類(lèi)型 + 數(shù)據(jù)長(zhǎng)度
        HEADER_SIZE = struct.calcsize(HEADER_FORMAT)
        
        def __init__(self):
            self.buffer = bytearray()
        
        def encode_message(self, msg_type, data):
            """編碼消息"""
            # 編碼數(shù)據(jù)
            if isinstance(data, str):
                encoded_data = data.encode('utf-8')
            else:
                encoded_data = data
            
            # 打包頭部
            header = struct.pack(self.HEADER_FORMAT, msg_type, len(encoded_data))
            
            # 返回完整消息
            return header + encoded_data
        
        def decode_message(self, data):
            """解碼消息"""
            messages = []
            self.buffer.extend(data)
            
            while len(self.buffer) >= self.HEADER_SIZE:
                # 解析頭部
                header_data = self.buffer[:self.HEADER_SIZE]
                msg_type, data_length = struct.unpack(self.HEADER_FORMAT, header_data)
                
                # 檢查是否有足夠的數(shù)據(jù)
                total_needed = self.HEADER_SIZE + data_length
                if len(self.buffer) < total_needed:
                    break  # 等待更多數(shù)據(jù)
                
                # 提取數(shù)據(jù)部分
                data_part = self.buffer[self.HEADER_SIZE:total_needed]
                
                # 嘗試解碼文本
                try:
                    decoded_data = data_part.decode('utf-8')
                except UnicodeDecodeError:
                    decoded_data = data_part  # 保持為字節(jié)
                
                messages.append((msg_type, decoded_data))
                
                # 從緩沖區(qū)移除已處理的數(shù)據(jù)
                self.buffer = self.buffer[total_needed:]
            
            return messages
    
    # 使用協(xié)議
    protocol = BinaryProtocol()
    
    # 編碼多個(gè)消息
    messages = [
        (1, "Hello, World!"),
        (2, "Binary Protocol Test"),
        (3, "結(jié)束消息"),
        (4, b'\x00\x01\x02\x03\x04')  # 二進(jìn)制數(shù)據(jù)
    ]
    
    encoded_packets = []
    for msg_type, data in messages:
        packet = protocol.encode_message(msg_type, data)
        encoded_packets.append(packet)
        print(f"編碼包 {msg_type}: {packet[:20]}... (長(zhǎng)度: {len(packet)})")
    
    # 模擬網(wǎng)絡(luò)傳輸（連接所有包）
    transmitted_data = b''.join(encoded_packets)
    print(f"\n傳輸數(shù)據(jù)總長(zhǎng)度: {len(transmitted_data)} 字節(jié)")
    
    # 解碼消息
    print("\n解碼消息:")
    decoded_messages = protocol.decode_message(transmitted_data)
    for msg_type, data in decoded_messages:
        print(f"類(lèi)型: {msg_type}, 數(shù)據(jù): {data}")

binary_protocol_implementation()

import mmap

def memory_mapped_operations():
    """內(nèi)存映射文件處理大型二進(jìn)制數(shù)據(jù)"""
    # 創(chuàng)建大型二進(jìn)制文件
    with open('large_binary.bin', 'wb') as f:
        # 生成測(cè)試數(shù)據(jù) (1MB)
        data = bytes([i % 256 for i in range(1024 * 1024)])
        f.write(data)
    
    # 使用內(nèi)存映射讀取
    with open('large_binary.bin', 'r+b') as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            print(f"文件大小: {len(mm)} 字節(jié)")
            
            # 隨機(jī)訪問(wèn)
            print(f"位置1000的字節(jié): {mm[1000]}")
            print(f"位置5000-5010的切片: {mm[5000:5010]}")
            
            # 搜索模式
            pattern = b'\x00\x01\x02\x03'
            position = mm.find(pattern)
            if position != -1:
                print(f"找到模式在位置: {position}")
            
            # 批量處理
            chunk_size = 4096
            for offset in range(0, len(mm), chunk_size):
                chunk = mm[offset:offset + chunk_size]
                # 處理塊數(shù)據(jù)
                if b'\xff' in chunk:
                    print(f"在塊 {offset}-{offset+chunk_size} 中找到 0xFF")
    
    # 可寫(xiě)內(nèi)存映射
    with open('mutable_binary.bin', 'w+b') as f:
        # 初始化文件
        f.write(b'\x00' * 1000)  # 1000字節(jié)空文件
        
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_WRITE) as mm:
            # 修改數(shù)據(jù)
            mm[0:5] = b'HELLO'  # 修改前5字節(jié)
            mm.seek(100)
            mm.write(b'WORLD')   # 在位置100寫(xiě)入
            
            # 同步到磁盤(pán)
            mm.flush()
            
            # 驗(yàn)證修改
            mm.seek(0)
            print(f"修改后開(kāi)始: {mm.read(10)}")
            mm.seek(100)
            print(f"位置100: {mm.read(5)}")

memory_mapped_operations()

def binary_stream_processing():
    """二進(jìn)制數(shù)據(jù)流處理"""
    import io
    
    class BinaryStreamProcessor:
        """二進(jìn)制流處理器"""
        def __init__(self, stream):
            self.stream = stream
            self.buffer = bytearray()
            self.position = 0
        
        def read_bytes(self, size):
            """讀取指定字節(jié)數(shù)"""
            while len(self.buffer) < size:
                chunk = self.stream.read(4096)
                if not chunk:
                    break
                self.buffer.extend(chunk)
            
            if len(self.buffer) < size:
                raise EOFError("無(wú)法讀取足夠的數(shù)據(jù)")
            
            data = self.buffer[:size]
            self.buffer = self.buffer[size:]
            self.position += size
            return data
        
        def read_until(self, delimiter):
            """讀取直到遇到分隔符"""
            while delimiter not in self.buffer:
                chunk = self.stream.read(4096)
                if not chunk:
                    break
                self.buffer.extend(chunk)
            
            if delimiter not in self.buffer:
                raise EOFError("未找到分隔符")
            
            index = self.buffer.index(delimiter)
            data = self.buffer[:index]
            self.buffer = self.buffer[index + len(delimiter):]
            self.position += index + len(delimiter)
            return data
        
        def seek(self, position):
            """定位到指定位置"""
            if position < self.position:
                raise ValueError("不支持向后定位")
            
            # 跳過(guò)不需要的數(shù)據(jù)
            skip_bytes = position - self.position
            if skip_bytes > 0:
                self.read_bytes(skip_bytes)
        
        def get_position(self):
            """獲取當(dāng)前位置"""
            return self.position
    
    # 使用示例
    # 創(chuàng)建測(cè)試數(shù)據(jù)
    test_data = b'START' + b'\x00\x01\x02\x03' * 1000 + b'END'
    stream = io.BytesIO(test_data)
    
    processor = BinaryStreamProcessor(stream)
    
    # 讀取起始標(biāo)記
    start_marker = processor.read_bytes(5)
    print(f"起始標(biāo)記: {start_marker}")
    
    # 讀取直到結(jié)束標(biāo)記
    try:
        content = processor.read_until(b'END')
        print(f"內(nèi)容長(zhǎng)度: {len(content)} 字節(jié)")
        print(f"最后幾個(gè)字節(jié): {content[-10:]}")
    except EOFError as e:
        print(f"錯(cuò)誤: {e}")
    
    print(f"最終位置: {processor.get_position()}")

binary_stream_processing()

def png_file_analysis():
    """PNG文件格式解析"""
    # PNG文件簽名
    PNG_SIGNATURE = b'\x89PNG\r\n\x1a\n'
    
    def parse_png_header(file_path):
        """解析PNG文件頭"""
        with open(file_path, 'rb') as f:
            # 檢查簽名
            signature = f.read(8)
            if signature != PNG_SIGNATURE:
                raise ValueError("不是有效的PNG文件")
            
            print("有效的PNG文件")
            
            # 讀取第一個(gè)數(shù)據(jù)塊 (IHDR)
            while True:
                # 讀取數(shù)據(jù)塊長(zhǎng)度
                length_data = f.read(4)
                if not length_data:
                    break
                
                chunk_length = struct.unpack('>I', length_data)[0]
                chunk_type = f.read(4)
                chunk_data = f.read(chunk_length)
                crc = f.read(4)
                
                print(f"\n數(shù)據(jù)塊類(lèi)型: {chunk_type.decode('ascii')}")
                print(f"數(shù)據(jù)塊長(zhǎng)度: {chunk_length}")
                
                if chunk_type == b'IHDR':
                    # 解析IHDR數(shù)據(jù)塊
                    width, height, bit_depth, color_type, compression, filter_method, interlace_method = \
                        struct.unpack('>IIBBBBB', chunk_data)
                    
                    print("=== IHDR 數(shù)據(jù)塊 ===")
                    print(f"寬度: {width} 像素")
                    print(f"高度: {height} 像素")
                    print(f"位深度: {bit_depth}")
                    print(f"顏色類(lèi)型: {color_type}")
                    print(f"壓縮方法: {compression}")
                    print(f"濾波方法: {filter_method}")
                    print(f"隔行掃描方法: {interlace_method}")
                    
                    break
    
    # 使用示例 (需要實(shí)際PNG文件)
    try:
        parse_png_header('example.png')
    except FileNotFoundError:
        print("示例PNG文件不存在，創(chuàng)建測(cè)試文件...")
        # 創(chuàng)建最小PNG文件用于測(cè)試
        with open('test.png', 'wb') as f:
            # PNG簽名
            f.write(b'\x89PNG\r\n\x1a\n')
            # IHDR數(shù)據(jù)塊
            f.write(struct.pack('>I', 13))  # 數(shù)據(jù)長(zhǎng)度
            f.write(b'IHDR')  # 數(shù)據(jù)塊類(lèi)型
            # IHDR數(shù)據(jù): 寬100, 高50, 8位深度, 2顏色類(lèi)型, 無(wú)壓縮, 無(wú)濾波, 無(wú)隔行
            f.write(struct.pack('>IIBBBBB', 100, 50, 8, 2, 0, 0, 0))
            # CRC (簡(jiǎn)化處理)
            f.write(b'\x00\x00\x00\x00')
        
        parse_png_header('test.png')

png_file_analysis()

def custom_binary_format():
    """自定義二進(jìn)制格式處理"""
    # 定義自定義二進(jìn)制格式
    class CustomBinaryFormat:
        """自定義二進(jìn)制文件格式"""
        MAGIC_NUMBER = b'CBF\x01'  # 魔術(shù)數(shù)字 + 版本號(hào)
        HEADER_FORMAT = '>I I I'   # 文件大小, 記錄數(shù), 數(shù)據(jù)偏移
        
        def __init__(self):
            self.records = []
        
        def add_record(self, record_type, data):
            """添加記錄"""
            self.records.append((record_type, data))
        
        def write_file(self, filename):
            """寫(xiě)入文件"""
            with open(filename, 'wb') as f:
                # 寫(xiě)入魔術(shù)數(shù)字
                f.write(self.MAGIC_NUMBER)
                
                # 預(yù)留頭部空間
                header_pos = f.tell()
                f.write(b'\x00' * struct.calcsize(self.HEADER_FORMAT))
                
                # 寫(xiě)入記錄
                record_offsets = []
                for record_type, data in self.records:
                    record_offsets.append(f.tell())
                    
                    # 記錄頭部: 類(lèi)型 + 長(zhǎng)度
                    record_header = struct.pack('>H I', record_type, len(data))
                    f.write(record_header)
                    f.write(data)
                
                # 寫(xiě)入記錄索引
                index_offset = f.tell()
                for offset in record_offsets:
                    f.write(struct.pack('>Q', offset))  # 8字節(jié)偏移量
                
                # 回到頭部寫(xiě)入完整信息
                f.seek(header_pos)
                file_size = f.tell()
                header_data = struct.pack(self.HEADER_FORMAT, 
                                         file_size, 
                                         len(self.records),
                                         index_offset)
                f.write(header_data)
        
        def read_file(self, filename):
            """讀取文件"""
            with open(filename, 'rb') as f:
                # 檢查魔術(shù)數(shù)字
                magic = f.read(4)
                if magic != self.MAGIC_NUMBER:
                    raise ValueError("不是有效的自定義格式文件")
                
                # 讀取頭部
                header_data = f.read(struct.calcsize(self.HEADER_FORMAT))
                file_size, num_records, index_offset = \
                    struct.unpack(self.HEADER_FORMAT, header_data)
                
                print(f"文件大小: {file_size}")
                print(f"記錄數(shù)量: {num_records}")
                
                # 讀取記錄索引
                f.seek(index_offset)
                record_offsets = []
                for _ in range(num_records):
                    offset_data = f.read(8)
                    offset = struct.unpack('>Q', offset_data)[0]
                    record_offsets.append(offset)
                
                # 讀取記錄
                self.records = []
                for offset in record_offsets:
                    f.seek(offset)
                    
                    # 讀取記錄頭部
                    header_data = f.read(6)  # 2字節(jié)類(lèi)型 + 4字節(jié)長(zhǎng)度
                    record_type, data_length = struct.unpack('>H I', header_data)
                    
                    # 讀取數(shù)據(jù)
                    data = f.read(data_length)
                    
                    self.records.append((record_type, data))
                    print(f"記錄 {record_type}: {len(data)} 字節(jié)")
    
    # 使用示例
    format_handler = CustomBinaryFormat()
    
    # 添加測(cè)試記錄
    format_handler.add_record(1, b'Hello, World!')
    format_handler.add_record(2, b'Binary Format Test')
    format_handler.add_record(3, bytes([i for i in range(100)]))
    
    # 寫(xiě)入文件
    format_handler.write_file('custom_format.cbf')
    print("自定義格式文件已寫(xiě)入")
    
    # 讀取文件
    format_handler.read_file('custom_format.cbf')
    
    # 顯示記錄內(nèi)容
    for record_type, data in format_handler.records:
        try:
            text_content = data.decode('utf-8')
            print(f"記錄 {record_type} (文本): {text_content}")
        except UnicodeDecodeError:
            print(f"記錄 {record_type} (二進(jìn)制): {data[:20]}...")

custom_binary_format()

import numpy as np

def numpy_binary_processing():
    """使用numpy進(jìn)行高性能二進(jìn)制處理"""
    # 創(chuàng)建大型數(shù)值數(shù)據(jù)集
    large_data = np.random.rand(1000000).astype(np.float32)  # 100萬(wàn)個(gè)浮點(diǎn)數(shù)
    print(f"原始數(shù)據(jù)形狀: {large_data.shape}")
    print(f"原始數(shù)據(jù)類(lèi)型: {large_data.dtype}")
    print(f"內(nèi)存占用: {large_data.nbytes / 1024 / 1024:.2f} MB")
    
    # 保存為二進(jìn)制文件
    large_data.tofile('large_binary_data.bin')
    print("數(shù)據(jù)已保存到文件")
    
    # 從文件加載
    loaded_data = np.fromfile('large_binary_data.bin', dtype=np.float32)
    print(f"加載數(shù)據(jù)形狀: {loaded_data.shape}")
    
    # 驗(yàn)證數(shù)據(jù)完整性
    print(f"數(shù)據(jù)一致性: {np.array_equal(large_data, loaded_data)}")
    
    # 內(nèi)存映射文件處理
    mmap_data = np.memmap('large_binary_data.bin', 
                         dtype=np.float32, 
                         mode='r', 
                         shape=large_data.shape)
    
    print("內(nèi)存映射訪問(wèn):")
    print(f"前10個(gè)元素: {mmap_data[:10]}")
    print(f"平均值: {mmap_data.mean():.6f}")
    print(f"標(biāo)準(zhǔn)差: {mmap_data.std():.6f}")
    
    # 處理大型二進(jìn)制文件
    def process_large_file(filename, dtype, chunk_size=100000):
        """分塊處理大型二進(jìn)制文件"""
        total_values = 0
        total_sum = 0.0
        min_val = float('inf')
        max_val = float('-inf')
        
        with open(filename, 'rb') as f:
            while True:
                # 讀取塊數(shù)據(jù)
                chunk = np.fromfile(f, dtype=dtype, count=chunk_size)
                if len(chunk) == 0:
                    break
                
                # 處理統(tǒng)計(jì)信息
                total_values += len(chunk)
                total_sum += chunk.sum()
                min_val = min(min_val, chunk.min())
                max_val = max(max_val, chunk.max())
        
        return {
            'count': total_values,
            'mean': total_sum / total_values if total_values > 0 else 0,
            'min': min_val,
            'max': max_val
        }
    
    # 處理大型文件
    stats = process_large_file('large_binary_data.bin', np.float32)
    print("\n文件統(tǒng)計(jì)信息:")
    print(f"數(shù)據(jù)點(diǎn)數(shù): {stats['count']:,}")
    print(f"平均值: {stats['mean']:.6f}")
    print(f"最小值: {stats['min']:.6f}")
    print(f"最大值: {stats['max']:.6f}")
    
    # 結(jié)構(gòu)化數(shù)組處理
    structured_dtype = np.dtype([
        ('id', 'i4'),
        ('value', 'f8'),
        ('timestamp', 'i8'),
        ('flags', 'u1')
    ])
    
    # 創(chuàng)建結(jié)構(gòu)化數(shù)據(jù)
    structured_data = np.array([
        (1, 3.14, 1640995200, 0x01),
        (2, 2.718, 1641081600, 0x02),
        (3, 1.618, 1641168000, 0x03)
    ], dtype=structured_dtype)
    
    # 保存結(jié)構(gòu)化數(shù)據(jù)
    structured_data.tofile('structured_data.bin')
    
    # 加載結(jié)構(gòu)化數(shù)據(jù)
    loaded_structured = np.fromfile('structured_data.bin', dtype=structured_dtype)
    print("\n結(jié)構(gòu)化數(shù)據(jù):")
    for record in loaded_structured:
        print(f"ID: {record['id']}, 值: {record['value']:.3f}, "
              f"時(shí)間戳: {record['timestamp']}, 標(biāo)志: 0x{record['flags']:02x}")

numpy_binary_processing()

def parallel_binary_processing():
    """并行二進(jìn)制數(shù)據(jù)處理"""
    import concurrent.futures
    import multiprocessing
    
    # 創(chuàng)建大型測(cè)試文件
    def create_test_file(filename, size_mb=10):
        """創(chuàng)建大型測(cè)試文件"""
        chunk_size = 1024 * 1024  # 1MB
        with open(filename, 'wb') as f:
            for i in range(size_mb):
                # 生成1MB的隨機(jī)數(shù)據(jù)
                data = os.urandom(chunk_size)
                f.write(data)
                print(f"已寫(xiě)入 {(i+1)} MB")
    
    create_test_file('large_test_file.bin', 10)
    print("測(cè)試文件創(chuàng)建完成")
    
    # 并行處理函數(shù)
    def process_chunk(args):
        """處理數(shù)據(jù)塊"""
        filename, start, size = args
        with open(filename, 'rb') as f:
            f.seek(start)
            data = f.read(size)
        
        # 簡(jiǎn)單的處理：計(jì)算校驗(yàn)和
        checksum = 0
        for byte in data:
            checksum = (checksum + byte) % 256
        
        return {
            'start': start,
            'size': size,
            'checksum': checksum,
            'processed': len(data)
        }
    
    # 并行處理
    def parallel_process_file(filename, chunk_size_mb=1):
        """并行處理文件"""
        file_size = os.path.getsize(filename)
        chunk_size = chunk_size_mb * 1024 * 1024
        
        # 準(zhǔn)備任務(wù)
        tasks = []
        for start in range(0, file_size, chunk_size):
            size = min(chunk_size, file_size - start)
            tasks.append((filename, start, size))
        
        print(f"總共 {len(tasks)} 個(gè)任務(wù)")
        
        # 使用進(jìn)程池
        results = []
        with concurrent.futures.ProcessPoolExecutor() as executor:
            future_to_task = {
                executor.submit(process_chunk, task): task 
                for task in tasks
            }
            
            for future in concurrent.futures.as_completed(future_to_task):
                task = future_to_task[future]
                try:
                    result = future.result()
                    results.append(result)
                    print(f"完成塊 {task[1]}-{task[1]+task[2]}，校驗(yàn)和: {result['checksum']}")
                except Exception as e:
                    print(f"處理塊 {task[1]} 時(shí)出錯(cuò): {e}")
        
        # 匯總結(jié)果
        total_processed = sum(r['processed'] for r in results)
        overall_checksum = sum(r['checksum'] for r in results) % 256
        
        return {
            'total_processed': total_processed,
            'overall_checksum': overall_checksum,
            'chunk_count': len(results)
        }
    
    # 執(zhí)行并行處理
    result = parallel_process_file('large_test_file.bin', 2)
    print("\n并行處理結(jié)果:")
    print(f"處理總字節(jié)數(shù): {result['total_processed']:,}")
    print(f"總校驗(yàn)和: {result['overall_checksum']}")
    print(f"處理塊數(shù): {result['chunk_count']}")

parallel_binary_processing()

def socket_binary_transfer():
    """Socket二進(jìn)制數(shù)據(jù)傳輸"""
    import socket
    import threading
    
    # 簡(jiǎn)單的二進(jìn)制協(xié)議
    class BinaryProtocol:
        def __init__(self):
            self.buffer = bytearray()
        
        def encode_message(self, data):
            """編碼消息: 長(zhǎng)度 + 數(shù)據(jù)"""
            length = len(data)
            return struct.pack('>I', length) + data
        
        def decode_messages(self, data):
            """解碼消息"""
            self.buffer.extend(data)
            messages = []
            
            while len(self.buffer) >= 4:
                # 讀取消息長(zhǎng)度
                length = struct.unpack('>I', self.buffer[:4])[0]
                
                # 檢查是否有完整消息
                if len(self.buffer) < 4 + length:
                    break
                
                # 提取消息數(shù)據(jù)
                message_data = self.buffer[4:4+length]
                messages.append(message_data)
                
                # 從緩沖區(qū)移除已處理的消息
                self.buffer = self.buffer[4+length:]
            
            return messages
    
    # 服務(wù)器端
    def server_thread():
        """服務(wù)器線程"""
        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_socket.bind(('localhost', 9999))
        server_socket.listen(1)
        
        print("服務(wù)器啟動(dòng)，等待連接...")
        conn, addr = server_socket.accept()
        print(f"連接來(lái)自: {addr}")
        
        protocol = BinaryProtocol()
        
        try:
            while True:
                data = conn.recv(4096)
                if not data:
                    break
                
                messages = protocol.decode_messages(data)
                for msg in messages:
                    try:
                        text = msg.decode('utf-8')
                        print(f"收到消息: {text}")
                    except UnicodeDecodeError:
                        print(f"收到二進(jìn)制數(shù)據(jù): {msg[:20]}...")
                    
                    # 發(fā)送響應(yīng)
                    response = f"已接收 {len(msg)} 字節(jié)".encode('utf-8')
                    conn.send(protocol.encode_message(response))
        
        finally:
            conn.close()
            server_socket.close()
    
    # 客戶(hù)端
    def client_example():
        """客戶(hù)端示例"""
        # 等待服務(wù)器啟動(dòng)
        import time
        time.sleep(0.1)
        
        client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client_socket.connect(('localhost', 9999))
        
        protocol = BinaryProtocol()
        
        # 發(fā)送文本消息
        messages = [
            "Hello, Server!",
            "這是文本消息",
            "結(jié)束通信"
        ]
        
        for msg in messages:
            encoded_msg = protocol.encode_message(msg.encode('utf-8'))
            client_socket.send(encoded_msg)
            
            # 接收響應(yīng)
            response_data = client_socket.recv(4096)
            responses = protocol.decode_messages(response_data)
            for resp in responses:
                print(f"服務(wù)器響應(yīng): {resp.decode('utf-8')}")
        
        # 發(fā)送二進(jìn)制數(shù)據(jù)
        binary_data = bytes([i for i in range(100)])
        client_socket.send(protocol.encode_message(binary_data))
        
        response_data = client_socket.recv(4096)
        responses = protocol.decode_messages(response_data)
        for resp in responses:
            print(f"二進(jìn)制響應(yīng): {resp.decode('utf-8')}")
        
        client_socket.close()
    
    # 啟動(dòng)服務(wù)器線程
    server = threading.Thread(target=server_thread)
    server.daemon = True
    server.start()
    
    # 運(yùn)行客戶(hù)端
    client_example()
    server.join()

socket_binary_transfer()

def http_binary_transfer():
    """HTTP二進(jìn)制數(shù)據(jù)傳輸"""
    import requests
    from http.server import HTTPServer, BaseHTTPRequestHandler
    import threading
    
    # HTTP請(qǐng)求處理器
    class BinaryHandler(BaseHTTPRequestHandler):
        def do_POST(self):
            """處理二進(jìn)制POST請(qǐng)求"""
            content_length = int(self.headers.get('Content-Length', 0))
            binary_data = self.rfile.read(content_length)
            
            print(f"收到 {len(binary_data)} 字節(jié)數(shù)據(jù)")
            
            # 處理數(shù)據(jù) (示例: 計(jì)算SHA256)
            import hashlib
            sha256_hash = hashlib.sha256(binary_data).hexdigest()
            
            # 發(fā)送響應(yīng)
            self.send_response(200)
            self.send_header('Content-Type', 'application/octet-stream')
            self.end_headers()
            
            response_data = f"SHA256: {sha256_hash}".encode('utf-8')
            self.wfile.write(response_data)
        
        def do_GET(self):
            """提供二進(jìn)制文件下載"""
            if self.path == '/download':
                # 生成測(cè)試二進(jìn)制數(shù)據(jù)
                test_data = bytes([i % 256 for i in range(1024)])
                
                self.send_response(200)
                self.send_header('Content-Type', 'application/octet-stream')
                self.send_header('Content-Disposition', 'attachment; filename="test.bin"')
                self.send_header('Content-Length', str(len(test_data)))
                self.end_headers()
                
                self.wfile.write(test_data)
            else:
                self.send_error(404)
    
    def start_server():
        """啟動(dòng)HTTP服務(wù)器"""
        server = HTTPServer(('localhost', 8000), BinaryHandler)
        print("HTTP服務(wù)器啟動(dòng)在端口 8000")
        server.serve_forever()
    
    # 啟動(dòng)服務(wù)器線程
    server_thread = threading.Thread(target=start_server)
    server_thread.daemon = True
    server_thread.start()
    
    # 等待服務(wù)器啟動(dòng)
    import time
    time.sleep(0.1)
    
    # 客戶(hù)端測(cè)試
    def test_client():
        """測(cè)試HTTP客戶(hù)端"""
        # 上傳二進(jìn)制數(shù)據(jù)
        test_data = b'\x00\x01\x02\x03\x04\x05' * 100  # 600字節(jié)測(cè)試數(shù)據(jù)
        
        response = requests.post('http://localhost:8000/', data=test_data)
        print(f"上傳響應(yīng): {response.text}")
        
        # 下載二進(jìn)制文件
        response = requests.get('http://localhost:8000/download')
        if response.status_code == 200:
            with open('downloaded.bin', 'wb') as f:
                f.write(response.content)
            print(f"下載完成: {len(response.content)} 字節(jié)")
            
            # 驗(yàn)證下載文件
            with open('downloaded.bin', 'rb') as f:
                downloaded_data = f.read()
                print(f"驗(yàn)證下載: {downloaded_data[:10]}...")
        else:
            print(f"下載失敗: {response.status_code}")
    
    # 運(yùn)行測(cè)試
    test_client()

http_binary_transfer()

def binary_error_handling():
    """二進(jìn)制數(shù)據(jù)錯(cuò)誤處理"""
    class SafeBinaryHandler:
        """安全的二進(jìn)制數(shù)據(jù)處理"""
        def __init__(self):
            self.buffer = bytearray()
        
        def read_binary_file(self, filename):
            """安全讀取二進(jìn)制文件"""
            try:
                with open(filename, 'rb') as f:
                    return f.read()
            except FileNotFoundError:
                print(f"錯(cuò)誤: 文件 {filename} 不存在")
                return None
            except PermissionError:
                print(f"錯(cuò)誤: 沒(méi)有權(quán)限讀取 {filename}")
                return None
            except Exception as e:
                print(f"讀取文件時(shí)發(fā)生未知錯(cuò)誤: {e}")
                return None
        
        def parse_binary_data(self, data, format_string):
            """安全解析二進(jìn)制數(shù)據(jù)"""
            try:
                return struct.unpack(format_string, data)
            except struct.error as e:
                print(f"解析錯(cuò)誤: {e}")
                print(f"數(shù)據(jù)長(zhǎng)度: {len(data)}，需要: {struct.calcsize(format_string)}")
                return None
        
        def validate_data_size(self, data, expected_size):
            """驗(yàn)證數(shù)據(jù)大小"""
            if len(data) != expected_size:
                raise ValueError(f"數(shù)據(jù)大小不匹配: 期望 {expected_size}, 實(shí)際 {len(data)}")
            return True
        
        def safe_file_operations(self, operations):
            """安全文件操作"""
            results = []
            for op in operations:
                try:
                    result = op()
                    results.append(result)
                except Exception as e:
                    print(f"操作失敗: {e}")
                    results.append(None)
            return results
    
    # 使用示例
    handler = SafeBinaryHandler()
    
    # 安全讀取文件
    data = handler.read_binary_file('example.bin')
    if data is None:
        print("創(chuàng)建測(cè)試文件...")
        data = b'\x00\x01\x02\x03\x04\x05'
        with open('example.bin', 'wb') as f:
            f.write(data)
        data = handler.read_binary_file('example.bin')
    
    # 安全解析
    if data:
        # 正確的格式
        parsed = handler.parse_binary_data(data, '6B')  # 6個(gè)無(wú)符號(hào)字節(jié)
        print(f"正確解析: {parsed}")
        
        # 錯(cuò)誤的格式
        parsed = handler.parse_binary_data(data, 'I')  # 需要4字節(jié)
        if parsed is None:
            print("解析失敗已正確處理")
        
        # 驗(yàn)證數(shù)據(jù)大小
        try:
            handler.validate_data_size(data, 6)
            print("數(shù)據(jù)大小驗(yàn)證通過(guò)")
        except ValueError as e:
            print(f"驗(yàn)證錯(cuò)誤: {e}")
    
    # 批量安全操作
    operations = [
        lambda: handler.read_binary_file('nonexistent.bin'),
        lambda: handler.parse_binary_data(b'\x00', 'I'),
        lambda: handler.validate_data_size(b'test', 10)
    ]
    
    results = handler.safe_file_operations(operations)
    print(f"批量操作結(jié)果: {results}")

binary_error_handling()

def binary_best_practices():
    """二進(jìn)制數(shù)據(jù)處理最佳實(shí)踐"""
    # 1. 使用上下文管理器
    def safe_binary_io(filename, mode='rb'):
        """安全的二進(jìn)制文件操作"""
        try:
            with open(filename, mode) as f:
                yield f
        except Exception as e:
            print(f"文件操作錯(cuò)誤: {e}")
            raise
    
    # 2. 數(shù)據(jù)驗(yàn)證函數(shù)
    def validate_binary_data(data, expected_properties):
        """驗(yàn)證二進(jìn)制數(shù)據(jù)屬性"""
        if 'size' in expected_properties:
            if len(data) != expected_properties['size']:
                raise ValueError(f"數(shù)據(jù)大小不匹配")
        
        if 'signature' in expected_properties:
            signature = expected_properties['signature']
            if data[:len(signature)] != signature:
                raise ValueError("數(shù)據(jù)簽名不匹配")
        
        return True
    
    # 3. 內(nèi)存高效處理
    def process_large_binary(filename, chunk_size=8192, processor=None):
        """處理大型二進(jìn)制文件的內(nèi)存高效方法"""
        if processor is None:
            processor = lambda chunk: chunk  # 默認(rèn)處理器
        
        results = []
        with open(filename, 'rb') as f:
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                results.append(processor(chunk))
        return results
    
    # 4. 類(lèi)型安全的二進(jìn)制操作
    class TypedBinaryData:
        """類(lèi)型安全的二進(jìn)制數(shù)據(jù)包裝器"""
        def __init__(self, data, dtype='B'):
            if not isinstance(data, (bytes, bytearray)):
                raise TypeError("必須是字節(jié)數(shù)據(jù)")
            self.data = data
            self.dtype = dtype
        
        def as_int(self):
            """轉(zhuǎn)換為整數(shù)"""
            try:
                return struct.unpack(self.dtype, self.data)[0]
            except struct.error:
                raise ValueError("無(wú)法轉(zhuǎn)換為指定類(lèi)型")
        
        def as_bytes(self):
            """獲取字節(jié)數(shù)據(jù)"""
            return self.data
        
        def validate_size(self):
            """驗(yàn)證數(shù)據(jù)大小"""
            expected_size = struct.calcsize(self.dtype)
            if len(self.data) != expected_size:
                raise ValueError(f"數(shù)據(jù)大小應(yīng)為 {expected_size} 字節(jié)")
            return True
    
    # 使用示例
    print("最佳實(shí)踐示例:")
    
    # 使用上下文管理器
    try:
        with safe_binary_io('example.bin', 'wb') as f:
            f.write(b'test data')
        print("1. 上下文管理器使用成功")
    except Exception as e:
        print(f"1. 錯(cuò)誤: {e}")
    
    # 數(shù)據(jù)驗(yàn)證
    try:
        validate_binary_data(b'test', {'size': 4, 'signature': b'test'})
        print("2. 數(shù)據(jù)驗(yàn)證成功")
    except ValueError as e:
        print(f"2. 驗(yàn)證錯(cuò)誤: {e}")
    
    # 內(nèi)存高效處理
    def simple_processor(chunk):
        return len(chunk)  # 返回塊長(zhǎng)度
    
    chunk_sizes = process_large_binary('example.bin', processor=simple_processor)
    print(f"3. 塊處理結(jié)果: {chunk_sizes}")
    
    # 類(lèi)型安全操作
    try:
        typed_data = TypedBinaryData(b'\x01\x00\x00\x00', 'I')  # 小端序整數(shù)1
        typed_data.validate_size()
        value = typed_data.as_int()
        print(f"4. 類(lèi)型安全值: {value}")
    except Exception as e:
        print(f"4. 類(lèi)型安全錯(cuò)誤: {e}")

binary_best_practices()

def professional_binary_processing():
    """
    專(zhuān)業(yè)二進(jìn)制處理模板
    
    遵循最佳實(shí)踐：
    1. 使用上下文管理器
    2. 驗(yàn)證輸入數(shù)據(jù)
    3. 錯(cuò)誤處理
    4. 資源清理
    """
    def process_binary_file(input_path, output_path=None, processor=None):
        """處理二進(jìn)制文件的完整流程"""
        # 輸入驗(yàn)證
        if not os.path.exists(input_path):
            raise FileNotFoundError(f"輸入文件不存在: {input_path}")
        
        if processor is None:
            processor = lambda x: x  # 默認(rèn)處理器
        
        try:
            # 讀取數(shù)據(jù)
            with open(input_path, 'rb') as f_in:
                input_data = f_in.read()
            
            # 驗(yàn)證數(shù)據(jù)
            if not input_data:
                raise ValueError("輸入文件為空")
            
            # 處理數(shù)據(jù)
            processed_data = processor(input_data)
            
            # 可選輸出
            if output_path:
                with open(output_path, 'wb') as f_out:
                    f_out.write(processed_data)
                print(f"結(jié)果已保存到: {output_path}")
            
            return processed_data
            
        except Exception as e:
            print(f"處理過(guò)程中發(fā)生錯(cuò)誤: {e}")
            # 可以考慮記錄日志或重試
            raise
    
    # 示例處理器
    def example_processor(data):
        """示例處理函數(shù): 簡(jiǎn)單的XOR加密"""
        key = 0x55
        return bytes(b ^ key for b in data)
    
    # 使用示例
    try:
        # 創(chuàng)建測(cè)試文件
        test_data = b'\x00\x01\x02\x03\x04\x05'
        with open('test_input.bin', 'wb') as f:
            f.write(test_data)
        
        # 處理文件
        result = process_binary_file(
            'test_input.bin',
            'test_output.bin',
            example_processor
        )
        
        print(f"處理完成，結(jié)果長(zhǎng)度: {len(result)} 字節(jié)")
        
    except Exception as e:
        print(f"操作失敗: {e}")

professional_binary_processing()