使用 struct 编解码二进制记录

功能说明

二进制协议通常由定长头部和变长载荷组成。这个示例定义魔数 SNSR、版本号、消息类型、载荷长度和时间戳五个字段，使用小端格式打包。解码时逐段读取头部，校验魔数与版本，再按长度截取载荷，因此可以一次解析多条连续记录。

定长头部让解码器在读到长度之前就能判断缓冲区是否足够，长度上限则防止恶意数据导致超大内存分配。该模式适合设备上报、消息队列帧、文件内嵌记录等需要紧凑序列化的场景。

关键代码

HEADER 用 struct.Struct 预先编译格式字符串，避免每次调用重复解析。编码时先检查载荷长度，再把头部和载荷拼接。解码循环用 offset 前进：先检查剩余字节是否够一个头部，再用 unpack_from 读取字段，随后校验长度并确认载荷完整。任何不满足条件的情况都抛出带偏移量的 ValueError。

使用注意

格式字符串中的 B 表示无符号单字节，H 是两字节，I 是四字节，字段顺序与 pack 参数必须一致。跨语言通信时要注意字节序，这里固定小端，对端也必须使用相同约定。载荷长度上限应按业务设定，过大容易被滥用。解码器不校验载荷内容的语义，业务层仍需二次校验。版本升级时新增字段应同时提升版本号，否则旧客户端会错位解析。

验证方式

两条记录共四十四字节：第一条载荷为 temp=21.5，第二条为 humidity=48，时间戳分别是 1700000000 和 1700000060。打印结果应逐行显示类型、时间戳和文本。把输入末尾截掉三个字节会触发 truncated payload 错误，说明长度校验生效；篡改首字节则提示 bad magic。

代码示例

import struct
from dataclasses import dataclass

MAGIC = b"SNSR"
HEADER = struct.Struct("<4sBBHI")
MAX_PAYLOAD = 4096


@dataclass(frozen=True)
class Packet:
    kind: int
    timestamp: int
    payload: bytes


def encode_packet(packet: Packet) -> bytes:
    if len(packet.payload) > MAX_PAYLOAD:
        raise ValueError("payload exceeds maximum size")
    header = HEADER.pack(MAGIC, 1, packet.kind, len(packet.payload), packet.timestamp)
    return header + packet.payload


def decode_packets(buffer: bytes) -> list[Packet]:
    packets: list[Packet] = []
    offset = 0
    while offset < len(buffer):
        if len(buffer) - offset < HEADER.size:
            raise ValueError(f"truncated header at offset {offset}")
        magic, version, kind, length, timestamp = HEADER.unpack_from(buffer, offset)
        if magic != MAGIC:
            raise ValueError(f"bad magic at offset {offset}")
        if version != 1:
            raise ValueError(f"unsupported version {version}")
        if length > MAX_PAYLOAD:
            raise ValueError(f"payload length {length} exceeds limit")
        offset += HEADER.size
        end = offset + length
        if end > len(buffer):
            raise ValueError(f"truncated payload at offset {offset}")
        packets.append(
            Packet(kind=kind, timestamp=timestamp, payload=buffer[offset:end])
        )
        offset = end
    return packets


def main() -> None:
    stream = (
        encode_packet(Packet(kind=1, timestamp=1_700_000_000, payload=b"temp=21.5"))
        + encode_packet(
            Packet(kind=2, timestamp=1_700_000_060, payload=b"humidity=48")
        )
    )
    print(f"encoded {len(stream)} bytes")
    for packet in decode_packets(stream):
        print(packet.kind, packet.timestamp, packet.payload.decode("ascii"))

    try:
        decode_packets(stream[:-3])
    except ValueError as error:
        print("rejected:", error)


if __name__ == "__main__":
    main()