def build_null_bitmap(params: list, num_params: int) -> bytes:
    """构建 null bitmap"""
    bitmap_size = (num_params + 7) // 8
    bitmap = bytearray(bitmap_size)
    for i, param in enumerate(params):
        if param is None:
            bitmap[i // 8] |= (1 << (i % 8))
    return bytes(bitmap)

import struct

def encode_tinyint(value: int) -> bytes:
    """TINYINT (1 字节)"""
    return struct.pack('<b', value)

def encode_smallint(value: int) -> bytes:
    """SMALLINT (2 字节)"""
    return struct.pack('<h', value)

def encode_int(value: int) -> bytes:
    """INT (4 字节)"""
    return struct.pack('<i', value)

def encode_bigint(value: int) -> bytes:
    """BIGINT (8 字节)"""
    return struct.pack('<q', value)

def encode_float(value: float) -> bytes:
    """FLOAT (4 字节)"""
    return struct.pack('<f', value)

def encode_double(value: float) -> bytes:
    """DOUBLE (8 字节)"""
    return struct.pack('<d', value)

def encode_string(value: str) -> bytes:
    """VARCHAR/TEXT: 长度编码 + 字节数据"""
    encoded = value.encode('utf-8')
    return encode_length(len(encoded)) + encoded

def encode_length(length: int) -> bytes:
    """长度编码"""
    if length < 0xFB:
        return struct.pack('B', length)
    elif length < 0x10000:
        return b'\xFC' + struct.pack('<H', length)
    elif length < 0x1000000:
        return b'\xFD' + struct.pack('<I', length)[:3]
    else:
        return b'\xFE' + struct.pack('<Q', length)

def encode_date(year: int, month: int, day: int) -> bytes:
    """DATE: 长度 + 年(2) + 月(1) + 日(1)"""
    return struct.pack('BBBB', 4, 0, 0, 0) + struct.pack('<HBB', year, month, day)
    # 实际: length=4, year(2), month(1), day(1) = 总共 5 字节

def encode_datetime(year, month, day, hour=0, minute=0, second=0, microsecond=0) -> bytes:
    """DATETIME/TIMESTAMP"""
    if microsecond:
        length = 11  # 年月日时分秒微秒
        return (struct.pack('B', length) +
                struct.pack('<HBBBBI', year, month, day, hour, minute, second) +
                struct.pack('<I', microsecond))
    elif second:
        length = 7   # 年月日时分秒
        return (struct.pack('B', length) +
                struct.pack('<HBBBBB', year, month, day, hour, minute, second))
    else:
        length = 4   # 年月日
        return (struct.pack('B', length) +
                struct.pack('<HBB', year, month, day))

def encode_time(negative=False, hours=0, minutes=0, seconds=0, microseconds=0) -> bytes:
    """TIME"""
    if microseconds:
        length = 12
        return (struct.pack('B', length) +
                struct.pack('<BiBBBBBI', 1 if negative else 0, 0, 0, 0, hours, minutes, seconds, microseconds))
    elif seconds:
        length = 8
        return (struct.pack('B', length) +
                struct.pack('<BiBBBBB', 1 if negative else 0, 0, 0, 0, hours, minutes, seconds))
    else:
        length = 0
        return struct.pack('B', length)

"""
mysql_binary_params.py
二进制协议参数绑定完整实现
"""
import struct


# 参数类型常量
MYSQL_TYPE_DECIMAL     = 0x00
MYSQL_TYPE_TINY        = 0x01
MYSQL_TYPE_SHORT       = 0x02
MYSQL_TYPE_LONG        = 0x03
MYSQL_TYPE_FLOAT       = 0x04
MYSQL_TYPE_DOUBLE      = 0x05
MYSQL_TYPE_NULL        = 0x06
MYSQL_TYPE_TIMESTAMP   = 0x07
MYSQL_TYPE_LONGLONG    = 0x08
MYSQL_TYPE_DATE        = 0x0A
MYSQL_TYPE_TIME        = 0x0B
MYSQL_TYPE_DATETIME    = 0x0C
MYSQL_TYPE_VARCHAR     = 0x0F
MYSQL_TYPE_BIT        = 0x10
MYSQL_TYPE_JSON        = 0xF5
MYSQL_TYPE_NEWDECIMAL  = 0xF6
MYSQL_TYPE_BLOB        = 0xFC
MYSQL_TYPE_VAR_STRING  = 0xFD
MYSQL_TYPE_STRING      = 0xFE


# Python 类型到 MySQL 类型的映射
PYTHON_TYPE_MAP = {
    type(None): (MYSQL_TYPE_NULL, 0),
    bool:       (MYSQL_TYPE_TINY, 0),
    int:        (MYSQL_TYPE_LONGLONG, 0),
    float:      (MYSQL_TYPE_DOUBLE, 0),
    str:        (MYSQL_TYPE_VAR_STRING, 0),
    bytes:      (MYSQL_TYPE_BLOB, 0),
    bytearray:  (MYSQL_TYPE_BLOB, 0),
}


def infer_param_type(value) -> tuple:
    """推断参数的 MySQL 类型"""
    if value is None:
        return (MYSQL_TYPE_NULL, 0)
    for py_type, mysql_type in PYTHON_TYPE_MAP.items():
        if isinstance(value, py_type):
            return mysql_type
    return (MYSQL_TYPE_VAR_STRING, 0)


def encode_param_value(value) -> bytes:
    """编码参数值为二进制格式"""
    if value is None:
        return b''

    if isinstance(value, bool):
        return struct.pack('<b', 1 if value else 0)

    if isinstance(value, int):
        if -128 <= value <= 127:
            return struct.pack('<b', value)
        elif -32768 <= value <= 32767:
            return struct.pack('<h', value)
        elif -2147483648 <= value <= 2147483647:
            return struct.pack('<i', value)
        else:
            return struct.pack('<q', value)

    if isinstance(value, float):
        return struct.pack('<d', value)

    if isinstance(value, str):
        encoded = value.encode('utf-8')
        return encode_length_coded(len(encoded)) + encoded

    if isinstance(value, (bytes, bytearray)):
        return encode_length_coded(len(value)) + bytes(value)

    raise TypeError(f"不支持的参数类型: {type(value)}")


def encode_length_coded(length: int) -> bytes:
    """长度编码"""
    if length < 0xFB:
        return struct.pack('B', length)
    elif length < 0x10000:
        return b'\xFC' + struct.pack('<H', length)
    elif length < 0x1000000:
        return b'\xFD' + struct.pack('<I', length)[:3]
    else:
        return b'\xFE' + struct.pack('<Q', length)


def build_execute_packet(statement_id: int, params: list, cursor_type: int = 0) -> bytes:
    """
    构建 COM_STMT_EXECUTE 数据包

    参数:
        statement_id: 预处理语句 ID
        params: 参数值列表
        cursor_type: 游标类型 (0=无游标, 1=READ_ONLY, 2=FOR_UPDATE, 4=SCROLLABLE)

    返回:
        完整的数据包（含包头）
    """
    num_params = len(params)

    # === 构建有效载荷 ===
    payload = bytearray()

    # 命令类型
    payload.append(0x17)  # COM_STMT_EXECUTE

    # statement_id
    payload.extend(struct.pack('<I', statement_id))

    # flags (cursor type)
    payload.append(cursor_type & 0xFF)

    # iteration_count (固定为 1)
    payload.extend(struct.pack('<I', 1))

    # null_bitmap
    bitmap_size = (num_params + 7) // 8
    null_bitmap = bytearray(bitmap_size)
    for i, param in enumerate(params):
        if param is None:
            null_bitmap[i // 8] |= (1 << (i % 8))
    payload.extend(null_bitmap)

    # new_params_bound_flag
    if num_params > 0:
        payload.append(0x01)  # 发送新参数类型

        # 参数类型数组
        for param in params:
            type_id, unsigned_flag = infer_param_type(param)
            payload.append(type_id)
            payload.append(unsigned_flag)

        # 参数值数组
        for param in params:
            if param is not None:
                payload.extend(encode_param_value(param))
    else:
        payload.append(0x00)  # 无参数

    # === 添加包头 ===
    packet = struct.pack('<I', len(payload))[:3] + struct.pack('B', 0) + bytes(payload)
    return packet


def demo():
    """演示二进制参数绑定"""
    print("=" * 60)
    print("MySQL 二进制协议参数绑定演示")
    print("=" * 60)

    # 模拟语句: INSERT INTO users (name, age, balance, active) VALUES (?, ?, ?, ?)
    statement_id = 1
    params = ["Alice", 28, 9999.50, True]

    packet = build_execute_packet(statement_id, params)

    print(f"\n语句 ID: {statement_id}")
    print(f"参数: {params}")
    print(f"数据包大小: {len(packet)} 字节")
    print(f"数据包头: {packet[:4].hex()}")
    print(f"命令: 0x{packet[4]:02X} (COM_STMT_EXECUTE)")
    print(f"完整 payload: {packet[4:].hex()}")
    print()

    # NULL 参数
    print("--- 含 NULL 参数 ---")
    params_with_null = ["Bob", None, 100.00, None]
    packet2 = build_execute_packet(statement_id, params_with_null)
    print(f"参数: {params_with_null}")
    print(f"数据包大小: {len(packet2)} 字节")

    # 解析 null_bitmap
    bitmap = packet2[14:14 + 1]  # 4 参数 → 1 字节 bitmap
    print(f"null_bitmap: 0x{bitmap[0]:02X}")
    for i in range(4):
        is_null = (bitmap[0] >> i) & 1
        print(f"  参数 {i} ({params_with_null[i]}): {'NULL' if is_null else '非 NULL'}")


if __name__ == '__main__':
    demo()

"""
mysql_binary_resultset.py
解析二进制结果集
"""
import struct
from datetime import datetime, date, time, timedelta
from decimal import Decimal
from typing import List, Any


# Column type 定义 (同 Column Definition)
TYPE_DECIMAL    = 0x00
TYPE_TINY       = 0x01
TYPE_SHORT      = 0x02
TYPE_LONG       = 0x03
TYPE_FLOAT      = 0x04
TYPE_DOUBLE     = 0x05
TYPE_NULL       = 0x06
TYPE_TIMESTAMP  = 0x07
TYPE_LONGLONG   = 0x08
TYPE_INT24      = 0x09
TYPE_DATE       = 0x0A
TYPE_TIME       = 0x0B
TYPE_DATETIME   = 0x0C
TYPE_YEAR       = 0x0D
TYPE_VARCHAR    = 0x0F
TYPE_BIT        = 0x10
TYPE_JSON       = 0xF5
TYPE_NEWDECIMAL = 0xF6
TYPE_ENUM       = 0xF7
TYPE_SET        = 0xF8
TYPE_TINY_BLOB  = 0xF9
TYPE_MEDIUM_BLOB= 0xFA
TYPE_LONG_BLOB  = 0xFB
TYPE_BLOB       = 0xFC
TYPE_VAR_STRING = 0xFD
TYPE_STRING     = 0xFE
TYPE_GEOMETRY   = 0xFF


def read_length_encoded_bytes(data: bytes, offset: int) -> tuple:
    """读取长度编码的字节序列"""
    length, new_offset = read_length_encoded_int(data, offset)
    if length is None:
        return None, offset + 1
    return data[new_offset:new_offset + length], new_offset + length


def read_length_encoded_int(data: bytes, offset: int) -> tuple:
    first = data[offset]
    if first < 0xFB:
        return first, offset + 1
    elif first == 0xFC:
        return struct.unpack('<H', data[offset+1:offset+3])[0], offset + 3
    elif first == 0xFD:
        return struct.unpack('<I', data[offset+1:offset+4] + b'\x00')[0], offset + 4
    elif first == 0xFE:
        return struct.unpack('<Q', data[offset+1:offset+9])[0], offset + 9


def decode_binary_column(data: bytes, offset: int, column_type: int, flags: int = 0, decimals: int = 0) -> tuple:
    """
    解码二进制结果集中的一列

    返回: (值, 新偏移量)
    """
    if column_type == TYPE_TINY:
        if flags & 0x0020:  # UNSIGNED
            return data[offset], offset + 1
        return struct.unpack('<b', data[offset:offset+1])[0], offset + 1

    elif column_type == TYPE_SHORT:
        if flags & 0x0020:
            return struct.unpack('<H', data[offset:offset+2])[0], offset + 2
        return struct.unpack('<h', data[offset:offset+2])[0], offset + 2

    elif column_type in (TYPE_LONG, TYPE_INT24):
        if flags & 0x0020:
            return struct.unpack('<I', data[offset:offset+4])[0], offset + 4
        return struct.unpack('<i', data[offset:offset+4])[0], offset + 4

    elif column_type == TYPE_LONGLONG:
        if flags & 0x0020:
            return struct.unpack('<Q', data[offset:offset+8])[0], offset + 8
        return struct.unpack('<q', data[offset:offset+8])[0], offset + 8

    elif column_type == TYPE_FLOAT:
        return struct.unpack('<f', data[offset:offset+4])[0], offset + 4

    elif column_type == TYPE_DOUBLE:
        return struct.unpack('<d', data[offset:offset+8])[0], offset + 8

    elif column_type == TYPE_YEAR:
        return struct.unpack('<H', data[offset:offset+2])[0], offset + 2

    elif column_type in (TYPE_DATE, TYPE_DATETIME, TYPE_TIMESTAMP):
        return decode_binary_temporal(data, offset)

    elif column_type == TYPE_TIME:
        return decode_binary_time(data, offset)

    elif column_type == TYPE_NEWDECIMAL:
        return decode_binary_decimal(data, offset, decimals)

    elif column_type in (TYPE_VARCHAR, TYPE_STRING, TYPE_ENUM, TYPE_SET):
        raw, new_offset = read_length_encoded_bytes(data, offset)
        if raw is not None:
            return raw.decode('utf-8'), new_offset
        return None, new_offset

    elif column_type in (TYPE_BLOB, TYPE_TINY_BLOB, TYPE_MEDIUM_BLOB,
                          TYPE_LONG_BLOB, TYPE_JSON, TYPE_BIT, TYPE_GEOMETRY):
        raw, new_offset = read_length_encoded_bytes(data, offset)
        return raw, new_offset

    else:
        # 默认当作长度编码字节
        raw, new_offset = read_length_encoded_bytes(data, offset)
        return raw, new_offset


def decode_binary_temporal(data: bytes, offset: int) -> tuple:
    """解码 DATE/DATETIME/TIMESTAMP 二进制值"""
    length = data[offset]
    offset += 1

    if length == 0:
        return None, offset

    year = struct.unpack('<H', data[offset:offset+2])[0]
    month = data[offset + 2]
    day = data[offset + 3]
    offset += 4

    if length == 4:
        return date(year, month, day), offset

    hour = data[offset]
    minute = data[offset + 1]
    second = data[offset + 2]
    offset += 3

    if length == 7:
        return datetime(year, month, day, hour, minute, second), offset

    microsecond = struct.unpack('<I', data[offset:offset+4])[0]
    offset += 4
    return datetime(year, month, day, hour, minute, second, microsecond), offset


def decode_binary_time(data: bytes, offset: int) -> tuple:
    """解码 TIME 二进制值"""
    length = data[offset]
    offset += 1

    if length == 0:
        return timedelta(0), offset

    is_negative = data[offset]
    offset += 1

    days = struct.unpack('<I', data[offset:offset+4])[0]
    offset += 4
    hours = data[offset]
    minutes = data[offset + 1]
    seconds = data[offset + 2]
    offset += 3

    total_seconds = days * 86400 + hours * 3600 + minutes * 60 + seconds

    if length > 8:
        microseconds = struct.unpack('<I', data[offset:offset+4])[0]
        total_seconds += microseconds / 1000000
        offset += 4

    td = timedelta(seconds=total_seconds)
    if is_negative:
        td = -td

    return td, offset


def decode_binary_decimal(data: bytes, offset: int, decimals: int) -> tuple:
    """解码 NEWDECIMAL 二进制值"""
    raw, new_offset = read_length_encoded_bytes(data, offset)
    if raw is None:
        return None, new_offset
    # 简化处理：直接转字符串
    return Decimal(raw.decode('utf-8')), new_offset


def parse_binary_row(data: bytes, columns: list) -> List[Any]:
    """
    解析二进制结果集中的一行

    参数:
        data: 行数据 payload
        columns: 列定义列表 [(name, type_id, flags, decimals), ...]

    返回:
        值列表
    """
    offset = 0

    # 前导字节 (0x00)
    offset += 1

    # null_bitmap
    num_cols = len(columns)
    bitmap_size = (num_cols + 7 + 2) // 8  # +2 for reserved bits
    null_bitmap = data[offset:offset + bitmap_size]
    offset += bitmap_size

    # 解析每列
    row = []
    for i, (name, type_id, flags, decimals) in enumerate(columns):
        # 检查 null bit (注意: 前 2 bit 保留, 列从 bit 2 开始)
        bit_index = i + 2
        is_null = (null_bitmap[bit_index // 8] >> (bit_index % 8)) & 1

        if is_null:
            row.append(None)
        else:
            value, offset = decode_binary_column(data, offset, type_id, flags, decimals)
            row.append(value)

    return row


def demo():
    """演示二进制结果集解析"""
    print("=" * 60)
    print("MySQL 二进制结果集解析演示")
    print("=" * 60)

    # 模拟列定义
    columns = [
        ("id",      TYPE_LONGLONG,    0x0001 | 0x0002, 0),  # NOT NULL, PRI
        ("name",    TYPE_VAR_STRING,  0,               0),
        ("balance", TYPE_NEWDECIMAL,  0x0020,          2),  # UNSIGNED
        ("active",  TYPE_TINY,        0,               0),
    ]

    # 构造模拟的二进制行数据
    row_data = bytearray()

    # 前导字节
    row_data.append(0x00)

    # null_bitmap: 4 列 → (4+2+7)/8 = 1 字节, 所有非 NULL → 0x00
    row_data.append(0x00)

    # id (BIGINT, 8 字节)
    row_data.extend(struct.pack('<q', 42))

    # name (VAR_STRING, 长度编码)
    name = "Alice".encode('utf-8')
    row_data.append(len(name))
    row_data.extend(name)

    # balance (NEWDECIMAL)
    balance = "12345.67".encode('utf-8')
    row_data.append(len(balance))
    row_data.extend(balance)

    # active (TINYINT)
    row_data.append(1)

    # 解析
    row = parse_binary_row(bytes(row_data), columns)

    print("\n列定义:")
    for name, type_id, flags, dec in columns:
        print(f"  {name}: type=0x{type_id:02X}, flags=0x{flags:04X}")

    print("\n原始数据:")
    print(f"  {bytes(row_data).hex()}")

    print("\n解析结果:")
    for i, (col_def, value) in enumerate(zip(columns, row)):
        name = col_def[0]
        print(f"  {name} = {value!r} (type: {type(value).__name__})")


if __name__ == '__main__':
    demo()

# 文本协议 (慢): 每条 INSERT 都要序列化为 SQL 字符串
# INSERT INTO t VALUES (1, 'Alice', 100.00)
# INSERT INTO t VALUES (2, 'Bob', 200.00)
# ...

# 二进制协议 (快): 预处理一次, 绑定执行多次
# PREPARE: INSERT INTO t VALUES (?, ?, ?)
# EXECUTE: [1, 'Alice', 100.00]
# EXECUTE: [2, 'Bob', 200.00]
# ...