import struct

# 编码长度
length = 1024
encoded = struct.pack('<I', length)[:3]  # 3 字节小端序
print(encoded.hex())  # 000400

# 解码长度
payload_length = struct.unpack('<I', encoded + b'\x00')[0]
print(payload_length)  # 1024

def next_sequence(current):
    return (current + 1) & 0xFF  # 0xFF = 255

-- 查看当前设置
SHOW VARIABLES LIKE 'max_allowed_packet';
-- 默认值: 67108864 (64 MB)

-- 设置新的值
SET GLOBAL max_allowed_packet = 134217728;  -- 128 MB

"""
mysql_packet_splitter.py
MySQL 数据包的分包与合并实现
"""
import struct

MAX_PACKET_PAYLOAD = 0xFFFFFF  # 16 MB - 1


def create_packet(payload: bytes, sequence_id: int) -> bytes:
    """
    创建一个 MySQL 数据包（不分包）

    参数:
        payload: 有效载荷数据
        sequence_id: 序列号 (0-255)

    返回:
        完整的数据包字节序列
    """
    if len(payload) > MAX_PACKET_PAYLOAD:
        raise ValueError(f"Payload 太大 ({len(payload)} > {MAX_PACKET_PAYLOAD})，需要分包")

    # 包头: 3 字节长度 + 1 字节序列号
    header = struct.pack('<I', len(payload))[:3] + struct.pack('B', sequence_id & 0xFF)
    return header + payload


def create_packets_with_splitting(payload: bytes, start_seq: int = 0) -> list:
    """
    创建数据包，自动处理分包

    参数:
        payload: 有效载荷数据
        start_seq: 起始序列号

    返回:
        数据包列表
    """
    packets = []
    offset = 0
    seq = start_seq

    while offset < len(payload):
        # 计算本包的数据量
        chunk_size = min(MAX_PACKET_PAYLOAD, len(payload) - offset)
        chunk = payload[offset:offset + chunk_size]

        packets.append(create_packet(chunk, seq))

        offset += chunk_size
        seq = (seq + 1) & 0xFF

    # 如果 payload 长度恰好是 MAX_PACKET_PAYLOAD 的倍数，
    # 需要额外发送一个空包来标识结束
    if len(payload) > 0 and len(payload) % MAX_PACKET_PAYLOAD == 0:
        packets.append(create_packet(b'', seq))

    return packets


def parse_packets_from_stream(data: bytes, start_offset: int = 0) -> list:
    """
    从字节流中解析多个数据包，处理分包合并

    参数:
        data: 完整的字节流
        start_offset: 起始偏移量

    返回:
        (完整消息, 已消费字节数) 的列表
    """
    packets = []
    offset = start_offset

    while offset + 4 <= len(data):
        # 读取包头
        payload_length = struct.unpack('<I', data[offset:offset+3] + b'\x00')[0]
        sequence_id = data[offset + 3]

        # 检查是否有足够的数据
        if offset + 4 + payload_length > len(data):
            break  # 数据不完整

        payload = data[offset + 4:offset + 4 + payload_length]

        packets.append({
            'sequence_id': sequence_id,
            'payload_length': payload_length,
            'payload': payload,
        })

        offset += 4 + payload_length

        # 如果长度小于最大值，消息结束
        if payload_length < MAX_PACKET_PAYLOAD:
            break

    return packets, offset


def demo():
    """演示分包与合并"""
    print("=" * 60)
    print("MySQL 数据包分包演示")
    print("=" * 60)

    # 示例 1：小数据包（不需要分包）
    print("\n--- 示例 1：小数据包 ---")
    payload = b"SELECT * FROM users WHERE id = 1"
    packets = create_packets_with_splitting(payload, start_seq=0)
    for i, pkt in enumerate(packets):
        pkt_len = struct.unpack('<I', pkt[0:3] + b'\x00')[0]
        seq = pkt[3]
        print(f"  包 {i}: len={pkt_len}, seq={seq}, payload={pkt[4:30]}...")

    # 示例 2：大数据包（需要分包）
    print("\n--- 示例 2：大数据包（模拟）---")
    # 创建一个超过 16MB 的数据（这里用小数据模拟）
    big_payload = b'A' * (MAX_PACKET_PAYLOAD + 1000)
    packets = create_packets_with_splitting(big_payload, start_seq=0)
    for i, pkt in enumerate(packets):
        pkt_len = struct.unpack('<I', pkt[0:3] + b'\x00')[0]
        seq = pkt[3]
        print(f"  包 {i}: len={pkt_len}, seq={seq}")
    print(f"  总数据大小: {len(big_payload):,} 字节")
    print(f"  分包数量: {len(packets)}")

    # 示例 3：恰好 16MB 的数据（需要额外的空包）
    print("\n--- 示例 3：恰好 MAX_PACKET_PAYLOAD 的倍数 ---")
    exact_payload = b'B' * MAX_PACKET_PAYLOAD
    packets = create_packets_with_splitting(exact_payload, start_seq=0)
    for i, pkt in enumerate(packets):
        pkt_len = struct.unpack('<I', pkt[0:3] + b'\x00')[0]
        seq = pkt[3]
        print(f"  包 {i}: len={pkt_len}, seq={seq}")
    print(f"  注意: 额外的空包用于标识消息结束")


if __name__ == '__main__':
    demo()

"""
length_encoded.py
MySQL 长度编码整数的编解码实现
"""
import struct


def read_length_encoded_integer(data: bytes, offset: int = 0) -> tuple:
    """
    读取长度编码整数

    参数:
        data: 字节序列
        offset: 起始偏移量

    返回:
        (值, 新偏移量) 元组
    """
    if offset >= len(data):
        raise ValueError("偏移量超出数据范围")

    first = data[offset]

    if first < 0xFB:
        # 直接值 (0-250)
        return first, offset + 1

    elif first == 0xFC:
        # 2 字节值
        value = struct.unpack('<H', data[offset+1:offset+3])[0]
        return value, offset + 3

    elif first == 0xFD:
        # 3 字节值
        value = struct.unpack('<I', data[offset+1:offset+4] + b'\x00')[0]
        return value, offset + 4

    elif first == 0xFE:
        # 8 字节值
        value = struct.unpack('<Q', data[offset+1:offset+9])[0]
        return value, offset + 9

    else:
        raise ValueError(f"无效的长度编码前缀: 0x{first:02X}")


def write_length_encoded_integer(value: int) -> bytes:
    """
    写入长度编码整数

    参数:
        value: 要编码的整数

    返回:
        编码后的字节序列
    """
    if value < 0xFB:
        return struct.pack('B', value)
    elif value < 0x10000:
        return b'\xFC' + struct.pack('<H', value)
    elif value < 0x1000000:
        return b'\xFD' + struct.pack('<I', value)[:3]
    elif value < 0x10000000000000000:
        return b'\xFE' + struct.pack('<Q', value)
    else:
        raise ValueError(f"值太大: {value}")


def read_length_encoded_string(data: bytes, offset: int = 0) -> tuple:
    """
    读取长度编码字符串

    格式: [长度编码的长度] + [字符串字节]

    返回:
        (字符串, 新偏移量) 元组
    """
    length, offset = read_length_encoded_integer(data, offset)
    string = data[offset:offset + length].decode('utf-8')
    return string, offset + length


def write_length_encoded_string(s: str) -> bytes:
    """
    写入长度编码字符串
    """
    encoded = s.encode('utf-8')
    return write_length_encoded_integer(len(encoded)) + encoded


def demo():
    """演示长度编码"""
    print("=" * 60)
    print("MySQL 长度编码整数演示")
    print("=" * 60)

    test_values = [0, 1, 42, 250, 251, 1000, 65535, 65536, 1000000, 16777215, 16777216]

    for value in test_values:
        encoded = write_length_encoded_integer(value)
        decoded, _ = read_length_encoded_integer(encoded)
        print(f"  {value:>10,} → {encoded.hex():<24s} ({len(encoded)} 字节) → {decoded:>10,}")

    # 字符串编码
    print("\n字符串编码:")
    test_strings = ["", "hello", "你好世界", "A" * 1000]
    for s in test_strings:
        encoded = write_length_encoded_string(s)
        decoded, _ = read_length_encoded_string(encoded)
        print(f"  '{s[:20]}...' → {len(encoded)} 字节 → '{decoded[:20]}...'")


if __name__ == '__main__':
    demo()

import struct

def read_uint1(data, offset=0):
    return data[offset], offset + 1

def read_uint2(data, offset=0):
    return struct.unpack('<H', data[offset:offset+2])[0], offset + 2

def read_uint3(data, offset=0):
    return struct.unpack('<I', data[offset:offset+3] + b'\x00')[0], offset + 3

def read_uint4(data, offset=0):
    return struct.unpack('<I', data[offset:offset+4])[0], offset + 4

def read_uint8(data, offset=0):
    return struct.unpack('<Q', data[offset:offset+8])[0], offset + 8

def read_int1(data, offset=0):
    return struct.unpack('<b', data[offset:offset+1])[0], offset + 1

def read_int2(data, offset=0):
    return struct.unpack('<h', data[offset:offset+2])[0], offset + 2

def read_int4(data, offset=0):
    return struct.unpack('<i', data[offset:offset+4])[0], offset + 4

def read_int8(data, offset=0):
    return struct.unpack('<q', data[offset:offset+8])[0], offset + 8

def read_float(data, offset=0):
    return struct.unpack('<f', data[offset:offset+4])[0], offset + 4

def read_double(data, offset=0):
    return struct.unpack('<d', data[offset:offset+8])[0], offset + 8

"""
mysql_packet_parser.py
完整的 MySQL 数据包解析器
"""
import struct
from enum import IntEnum
from dataclasses import dataclass
from typing import List, Optional


class PacketType(IntEnum):
    """数据包类型"""
    OK = 0x00
    EOF = 0xFE
    ERR = 0xFF
    UNKNOWN = -1


@dataclass
class MySQLPacket:
    """MySQL 数据包"""
    sequence_id: int
    payload: bytes
    packet_type: PacketType = PacketType.UNKNOWN

    def __post_init__(self):
        if len(self.payload) > 0:
            first_byte = self.payload[0]
            if first_byte == 0x00 and len(self.payload) >= 7:
                self.packet_type = PacketType.OK
            elif first_byte == 0xFF:
                self.packet_type = PacketType.ERR
            elif first_byte == 0xFE and len(self.payload) < 9:
                self.packet_type = PacketType.EOF


class PacketReader:
    """从字节流中读取 MySQL 数据包"""

    def __init__(self):
        self._buffer = b''
        self._offset = 0

    def feed(self, data: bytes):
        """向缓冲区添加数据"""
        self._buffer += data

    def read_packet(self) -> Optional[MySQLPacket]:
        """读取一个完整的数据包"""
        if self._offset + 4 > len(self._buffer):
            return None  # 数据不足

        # 读取包头
        payload_length = struct.unpack(
            '<I', self._buffer[self._offset:self._offset+3] + b'\x00'
        )[0]
        sequence_id = self._buffer[self._offset + 3]

        # 检查是否有足够的数据
        if self._offset + 4 + payload_length > len(self._buffer):
            return None  # 数据不完整

        # 提取有效载荷
        payload = self._buffer[self._offset + 4:self._offset + 4 + payload_length]
        self._offset += 4 + payload_length

        return MySQLPacket(sequence_id=sequence_id, payload=payload)

    def read_all_packets(self) -> List[MySQLPacket]:
        """读取所有可用的完整数据包"""
        packets = []
        while True:
            pkt = self.read_packet()
            if pkt is None:
                break
            packets.append(pkt)
        return packets

    def remaining(self) -> bytes:
        """返回未消费的数据"""
        remaining = self._buffer[self._offset:]
        self._buffer = remaining
        self._offset = 0
        return remaining


class PacketWriter:
    """构建 MySQL 数据包"""

    @staticmethod
    def write_packet(payload: bytes, sequence_id: int) -> bytes:
        """构建一个数据包"""
        header = struct.pack('<I', len(payload))[:3] + struct.pack('B', sequence_id & 0xFF)
        return header + payload

    @staticmethod
    def write_payload(*parts) -> bytes:
        """将多个部分组合为有效载荷"""
        return b''.join(parts)


def demo():
    """演示数据包解析器"""
    print("=" * 60)
    print("MySQL 数据包解析器演示")
    print("=" * 60)

    # 构造一些数据包
    writer = PacketWriter()

    # OK 包
    ok_payload = b'\x00\x00\x00\x02\x00\x00\x00'  # 简化的 OK 包
    ok_packet = writer.write_packet(ok_payload, 1)

    # ERR 包
    err_payload = b'\xFF\x15\x04#28000Access denied'
    err_packet = writer.write_packet(err_payload, 1)

    # 普通数据包
    data_payload = b'\x03SELECT 1'
    data_packet = writer.write_packet(data_payload, 0)

    # 拼接所有数据包
    raw_data = data_packet + ok_packet + err_packet

    # 解析
    reader = PacketReader()
    reader.feed(raw_data)
    packets = reader.read_all_packets()

    for pkt in packets:
        print(f"\n数据包:")
        print(f"  序列号: {pkt.sequence_id}")
        print(f"  载荷长度: {len(pkt.payload)}")
        print(f"  类型: {pkt.packet_type.name}")
        print(f"  原始数据: {pkt.payload[:20].hex()}")


if __name__ == '__main__':
    demo()

import zlib

def compress_data(data: bytes) -> bytes:
    """使用 zlib 压缩数据"""
    return zlib.compress(data, 6)  # 压缩级别 6

def decompress_data(data: bytes) -> bytes:
    """解压缩数据"""
    return zlib.decompress(data)

def create_compressed_packet(payload: bytes, seq: int) -> bytes:
    """创建压缩数据包"""
    compressed = compress_data(payload)

    # 如果压缩没有节省空间，发送原始数据
    if len(compressed) >= len(payload):
        header = struct.pack('<I', len(payload))[:3]  # compressed_length
        header += struct.pack('B', seq)               # compressed_sequence_id
        header += b'\x00\x00\x00'                     # uncompressed_length = 0 (未压缩)
        return header + payload

    header = struct.pack('<I', len(compressed))[:3]    # compressed_length
    header += struct.pack('B', seq)                     # compressed_sequence_id
    header += struct.pack('<I', len(payload))[:3]      # uncompressed_length
    return header + compressed

-- 调整包大小
SET GLOBAL max_allowed_packet = 256 * 1024 * 1024;  -- 256 MB