def parse_column_count(payload: bytes) -> int:
    """解析列数"""
    if payload[0] < 0xFB:
        return payload[0]
    elif payload[0] == 0xFC:
        return struct.unpack('<H', payload[1:3])[0]
    # ... 其他情况

"""
mysql_column_definition.py
解析 Column Definition 数据包
"""
import struct
from dataclasses import dataclass, field
from typing import Dict


@dataclass
class ColumnDefinition:
    """列定义"""
    catalog: str = "def"
    schema: str = ""
    table_alias: str = ""
    table_name: str = ""
    column_alias: str = ""
    column_name: str = ""
    character_set: int = 0
    column_length: int = 0
    column_type: int = 0
    flags: int = 0
    decimals: int = 0


# 类型名称映射
COLUMN_TYPE_NAMES: Dict[int, str] = {
    0x00: "DECIMAL",   0x01: "TINYINT",    0x02: "SMALLINT",
    0x03: "INT",       0x04: "FLOAT",      0x05: "DOUBLE",
    0x06: "NULL",      0x07: "TIMESTAMP",  0x08: "BIGINT",
    0x09: "MEDIUMINT", 0x0A: "DATE",       0x0B: "TIME",
    0x0C: "DATETIME",  0x0D: "YEAR",       0x0F: "VARCHAR",
    0x10: "BIT",       0xF5: "JSON",       0xF6: "NEWDECIMAL",
    0xF7: "ENUM",      0xF8: "SET",        0xF9: "TINYBLOB",
    0xFA: "MEDIUMBLOB",0xFB: "LONGBLOB",   0xFC: "BLOB",
    0xFD: "VAR_STRING",0xFE: "STRING",     0xFF: "GEOMETRY",
}

# 字符集名称映射
CHARSET_NAMES: Dict[int, str] = {
    8: "latin1", 33: "utf8mb3", 45: "utf8mb4", 255: "utf8mb4",
    63: "binary",
}


def read_length_encoded_str(data: bytes, offset: int) -> tuple:
    """读取长度编码字符串"""
    length, offset = read_length_encoded_int(data, offset)
    value = data[offset:offset + length].decode('utf-8', errors='replace')
    return value, 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 parse_column_definition(payload: bytes) -> ColumnDefinition:
    """解析 Column Definition 数据包"""
    col = ColumnDefinition()
    offset = 0

    # catalog (长度编码字符串, 固定为 "def")
    col.catalog, offset = read_length_encoded_str(payload, offset)

    # schema (数据库名)
    col.schema, offset = read_length_encoded_str(payload, offset)

    # table_alias (表别名)
    col.table_alias, offset = read_length_encoded_str(payload, offset)

    # table_name (真实表名)
    col.table_name, offset = read_length_encoded_str(payload, offset)

    # column_alias (列别名)
    col.column_alias, offset = read_length_encoded_str(payload, offset)

    # column_name (真实列名)
    col.column_name, offset = read_length_encoded_str(payload, offset)

    # 固定长度字段的长度标记 (0x0C = 12)
    fixed_len = payload[offset]
    offset += 1

    # character_set (2 字节)
    col.character_set = struct.unpack('<H', payload[offset:offset+2])[0]
    offset += 2

    # column_length (4 字节)
    col.column_length = struct.unpack('<I', payload[offset:offset+4])[0]
    offset += 4

    # column_type (1 字节)
    col.column_type = payload[offset]
    offset += 1

    # flags (2 字节)
    col.flags = struct.unpack('<H', payload[offset:offset+2])[0]
    offset += 2

    # decimals (1 字节)
    col.decimals = payload[offset]

    return col


def format_column(col: ColumnDefinition) -> str:
    """格式化列定义为可读字符串"""
    type_name = COLUMN_TYPE_NAMES.get(col.column_type, f"UNKNOWN(0x{col.column_type:02X})")
    charset = CHARSET_NAMES.get(col.character_set, f"charset({col.character_set})")

    flags_str = []
    if col.flags & 0x0001: flags_str.append("NOT_NULL")
    if col.flags & 0x0002: flags_str.append("PRI_KEY")
    if col.flags & 0x0020: flags_str.append("UNSIGNED")
    if col.flags & 0x0200: flags_str.append("AUTO_INC")

    return (
        f"{col.schema}.{col.table_name}.{col.column_name} "
        f"({type_name}, {charset}, len={col.column_length}, "
        f"dec={col.decimals}, flags=[{','.join(flags_str)}])"
    )


# 演示
def demo():
    print("Column Definition 解析演示")
    print()

    # 模拟解析结果
    columns = [
        ColumnDefinition(
            schema="test_db", table_name="users", column_name="id",
            column_alias="id", character_set=63, column_length=11,
            column_type=0x03, flags=0x0001 | 0x0002 | 0x0200, decimals=0
        ),
        ColumnDefinition(
            schema="test_db", table_name="users", column_name="name",
            column_alias="name", character_set=45, column_length=255,
            column_type=0xFD, flags=0x0000, decimals=0
        ),
        ColumnDefinition(
            schema="test_db", table_name="users", column_name="balance",
            column_alias="balance", character_set=63, column_length=14,
            column_type=0xF6, flags=0x0020, decimals=2
        ),
    ]

    for i, col in enumerate(columns):
        print(f"  列 {i}: {format_column(col)}")


if __name__ == '__main__':
    demo()

-- 查询结果
SELECT id, name, email FROM users LIMIT 2;
-- 结果:
-- | id | name  | email             |
-- |  1 | Alice | [email protected] |
-- |  2 | NULL  | [email protected]   |

"""
mysql_text_resultset.py
解析文本结果集（完整的 COM_QUERY 响应）
"""
import struct
from dataclasses import dataclass
from typing import List, Any


def read_length_encoded_int(data: bytes, offset: int) -> tuple:
    first = data[offset]
    if first == 0xFB:
        return None, offset + 1  # NULL
    elif 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 read_length_encoded_str(data: bytes, offset: int) -> tuple:
    """读取长度编码字符串，返回 (str|None, new_offset)"""
    length, offset = read_length_encoded_int(data, offset)
    if length is None:
        return None, offset  # NULL 值
    value = data[offset:offset + length].decode('utf-8', errors='replace')
    return value, offset + length


@dataclass
class ColumnDef:
    name: str
    type_id: int
    flags: int = 0
    decimals: int = 0


@dataclass
class ResultSet:
    columns: List[ColumnDef]
    rows: List[List[Any]]
    status_flags: int = 0
    warnings: int = 0


def parse_text_resultset(packets: List[bytes]) -> ResultSet:
    """
    解析一整个文本结果集

    参数:
        packets: 按顺序的数据包 payload 列表

    返回:
        ResultSet 对象
    """
    idx = 0
    result = ResultSet(columns=[], rows=[])

    # 1. 解析列数 (长度编码整数)
    column_count, _ = read_length_encoded_int(packets[idx], 0)
    idx += 1

    # 2. 解析列定义
    for i in range(column_count):
        payload = packets[idx]
        col = parse_column_def(payload)
        result.columns.append(col)
        idx += 1

    # 3. EOF 或 OK（列定义结束）
    eof1 = packets[idx]
    idx += 1
    if eof1[0] == 0xFE and len(eof1) >= 5:
        result.status_flags = struct.unpack('<H', eof1[3:5])[0]
        result.warnings = struct.unpack('<H', eof1[1:3])[0]

    # 4. 解析行数据（直到遇到 EOF/OK）
    while idx < len(packets):
        payload = packets[idx]
        idx += 1

        # 检查是否是 EOF/OK
        if payload[0] == 0xFE and len(payload) < 9:
            if len(payload) >= 5:
                result.status_flags = struct.unpack('<H', payload[3:5])[0]
            break
        if payload[0] == 0x00 and len(payload) >= 7:
            # OK with CLIENT_DEPRECATE_EOF
            break

        # 解析行数据
        row = parse_row_data(payload, len(result.columns))
        result.rows.append(row)

    return result


def parse_column_def(payload: bytes) -> ColumnDef:
    """解析单个列定义"""
    offset = 0

    # catalog
    _, offset = read_length_encoded_str(payload, offset)
    # schema
    _, offset = read_length_encoded_str(payload, offset)
    # table alias
    _, offset = read_length_encoded_str(payload, offset)
    # table name
    _, offset = read_length_encoded_str(payload, offset)
    # column alias
    name, offset = read_length_encoded_str(payload, offset)
    # column name
    _, offset = read_length_encoded_str(payload, offset)

    # fixed-length fields
    offset += 1  # 0x0C
    _ = struct.unpack('<H', payload[offset:offset+2])[0]  # charset
    offset += 2
    _ = struct.unpack('<I', payload[offset:offset+4])[0]  # column_length
    offset += 4
    type_id = payload[offset]
    offset += 1
    flags = struct.unpack('<H', payload[offset:offset+2])[0]
    offset += 2
    decimals = payload[offset]

    return ColumnDef(name=name or "", type_id=type_id, flags=flags, decimals=decimals)


def parse_row_data(payload: bytes, num_columns: int) -> List[Any]:
    """解析行数据"""
    row = []
    offset = 0

    for _ in range(num_columns):
        value, offset = read_length_encoded_str(payload, offset)
        row.append(value)

    return row


def format_table(result: ResultSet) -> str:
    """将结果集格式化为 ASCII 表格"""
    if not result.columns:
        return "(空结果集)"

    # 计算列宽
    col_widths = []
    for col in result.columns:
        width = len(col.name)
        col_widths.append(width)

    for row in result.rows:
        for i, val in enumerate(row):
            val_str = str(val) if val is not None else "NULL"
            col_widths[i] = max(col_widths[i], len(val_str))

    # 格式化
    lines = []

    # 表头
    header = " | ".join(col.name.ljust(col_widths[i]) for i, col in enumerate(result.columns))
    lines.append(header)
    lines.append("-+-".join("-" * w for w in col_widths))

    # 数据行
    for row in result.rows:
        cells = []
        for i, val in enumerate(row):
            val_str = str(val) if val is not None else "NULL"
            cells.append(val_str.ljust(col_widths[i]))
        lines.append(" | ".join(cells))

    # 底部信息
    lines.append("")
    lines.append(f"({len(result.rows)} rows, status=0x{result.status_flags:04X})")

    return "\n".join(lines)


def demo():
    """演示文本结果集解析"""
    print("=" * 70)
    print("MySQL 文本结果集解析演示")
    print("=" * 70)

    # 构造模拟数据包
    packets = []

    # 列数 = 3
    packets.append(b'\x03')

    # 列定义 (简化编码)
    def make_col_def(name, type_id, charset=0x2D, col_len=255, flags=0, dec=0):
        payload = b''
        # catalog
        payload += b'\x03def'
        # schema
        payload += b'\x04test'
        # table alias
        payload += b'\x05users'
        # table name
        payload += b'\x05users'
        # column alias
        payload += bytes([len(name)]) + name.encode()
        # column name
        payload += bytes([len(name)]) + name.encode()
        # fixed fields length
        payload += b'\x0c'
        # charset
        payload += struct.pack('<H', charset)
        # column_length
        payload += struct.pack('<I', col_len)
        # type
        payload += struct.pack('B', type_id)
        # flags
        payload += struct.pack('<H', flags)
        # decimals
        payload += struct.pack('B', dec)
        # filler
        payload += b'\x00\x00'
        return payload

    packets.append(make_col_def('id', 0x03, charset=0x3F, col_len=11, flags=0x0003))
    packets.append(make_col_def('name', 0xFD, charset=0x2D, col_len=255))
    packets.append(make_col_def('age', 0x03, charset=0x3F, col_len=11, flags=0x0020))

    # EOF (列定义结束)
    packets.append(b'\xfe\x00\x00\x02\x00')

    # 行数据
    def make_row(*values):
        payload = b''
        for v in values:
            if v is None:
                payload += b'\xFB'
            else:
                s = str(v).encode('utf-8')
                payload += bytes([len(s)]) + s
        return payload

    packets.append(make_row(1, "Alice", 28))
    packets.append(make_row(2, "Bob", 35))
    packets.append(make_row(3, None, 22))

    # EOF (行数据结束)
    packets.append(b'\xfe\x00\x00\x02\x00')

    # 解析
    result = parse_text_resultset(packets)

    print(f"\n列数: {len(result.columns)}")
    for i, col in enumerate(result.columns):
        print(f"  列 {i}: {col.name} (type=0x{col.type_id:02X})")

    print(f"\n{format_table(result)}")


if __name__ == '__main__':
    demo()

def parse_ok_packet(payload: bytes) -> dict:
    """解析 OK 包"""
    offset = 1  # 跳过 0x00

    affected_rows, offset = read_length_encoded_int(payload, offset)
    last_insert_id, offset = read_length_encoded_int(payload, offset)

    status_flags = struct.unpack('<H', payload[offset:offset+2])[0]
    offset += 2

    warnings = struct.unpack('<H', payload[offset:offset+2])[0]
    offset += 2

    info = ""
    if offset < len(payload):
        info, _ = read_length_encoded_str(payload, offset)

    return {
        'affected_rows': affected_rows,
        'last_insert_id': last_insert_id,
        'status_flags': status_flags,
        'warnings': warnings,
        'info': info,
    }

def parse_err_packet(payload: bytes) -> dict:
    """解析 ERR 包"""
    error_code = struct.unpack('<H', payload[1:3])[0]
    sql_state = ""
    message = ""

    if len(payload) > 3 and payload[3:4] == b'#':
        sql_state = payload[4:9].decode('ascii', errors='replace')
        message = payload[9:].decode('utf-8', errors='replace')
    elif len(payload) > 3:
        message = payload[3:].decode('utf-8', errors='replace')

    return {
        'error_code': error_code,
        'sql_state': sql_state,
        'message': message,
    }

def read_all_resultsets(sock, reader):
    """读取所有结果集"""
    results = []
    while True:
        result = parse_text_resultset(reader.read_until_eof())
        results.append(result)
        if not (result.status_flags & 0x0008):  # SERVER_MORE_RESULTS_EXISTS
            break
    return results