"""
mysql_type_converter.py
将文本结果集的字符串值转换为 Python 类型
"""
from datetime import datetime, date, time, timedelta
from decimal import Decimal


def convert_text_value(value: str, column_type: int, flags: int = 0,
                       decimals: int = 0, charset: int = 45):
    """
    将文本结果集的字符串值转换为 Python 类型

    参数:
        value: 文本值 (可能为 None)
        column_type: 列类型 ID
        flags: 列标志
        decimals: 小数位数
        charset: 字符集 ID

    返回:
        转换后的 Python 值
    """
    if value is None:
        return None

    # 数值类型
    if column_type == 0x01:  # TINYINT
        return int(value)
    elif column_type == 0x02:  # SMALLINT
        return int(value)
    elif column_type == 0x03:  # INT
        return int(value)
    elif column_type == 0x04:  # FLOAT
        return float(value)
    elif column_type == 0x05:  # DOUBLE
        return float(value)
    elif column_type == 0x08:  # BIGINT
        return int(value)
    elif column_type == 0x09:  # MEDIUMINT
        return int(value)

    # 定点数
    elif column_type == 0x00:  # DECIMAL
        return Decimal(value)
    elif column_type == 0xF6:  # NEWDECIMAL
        return Decimal(value)

    # 日期时间
    elif column_type == 0x0A:  # DATE
        return datetime.strptime(value, '%Y-%m-%d').date()
    elif column_type == 0x0B:  # TIME
        return parse_mysql_time(value)
    elif column_type == 0x0C:  # DATETIME
        return datetime.strptime(value, '%Y-%m-%d %H:%M:%S')
    elif column_type == 0x07:  # TIMESTAMP
        return datetime.strptime(value, '%Y-%m-%d %H:%M:%S')
    elif column_type == 0x0D:  # YEAR
        return int(value)

    # 字符串
    elif column_type in (0x0F, 0xF5, 0xF7, 0xF8, 0xFD, 0xFE):
        return value

    # BLOB
    elif column_type in (0xF9, 0xFA, 0xFB, 0xFC):
        if charset == 63:  # binary
            return value.encode('latin1')
        return value

    # BIT
    elif column_type == 0x10:
        return int.from_bytes(value.encode('latin1'), 'big')

    else:
        return value


def parse_mysql_time(value: str) -> timedelta:
    """解析 MySQL TIME 值为 timedelta"""
    negative = value.startswith('-')
    if negative:
        value = value[1:]

    parts = value.split(':')
    hours = int(parts[0])
    minutes = int(parts[1])
    seconds = float(parts[2]) if len(parts) > 2 else 0

    total_seconds = hours * 3600 + minutes * 60 + seconds
    td = timedelta(seconds=total_seconds)
    return -td if negative else td


def demo():
    """演示类型转换"""
    print("MySQL 文本值类型转换演示")
    print()

    test_cases = [
        ("42",        0x03, 0, 0, "INT → Python int"),
        ("3.14",      0x05, 0, 0, "DOUBLE → Python float"),
        ("12345.67",  0xF6, 0, 2, "DECIMAL(10,2) → Python Decimal"),
        ("hello",     0xFD, 0, 0, "VARCHAR → Python str"),
        ("2024-01-15", 0x0A, 0, 0, "DATE → Python date"),
        ("2024-01-15 10:30:00", 0x0C, 0, 0, "DATETIME → Python datetime"),
        ("-838:59:59", 0x0B, 0, 0, "TIME → Python timedelta"),
        ("1",         0x01, 0, 0, "TINYINT → Python int"),
    ]

    for value, type_id, flags, decimals, desc in test_cases:
        result = convert_text_value(value, type_id, flags, decimals)
        print(f"  {desc}")
        print(f"    输入: {value!r}")
        print(f"    输出: {result!r} ({type(result).__name__})")
        print()


if __name__ == '__main__':
    demo()

"""
mysql_streaming.py
流式读取 MySQL 结果集
"""
import socket
import struct
from typing import Generator, List, Any


class StreamingResultSet:
    """流式结果集读取器"""

    def __init__(self, sock: socket.socket):
        self.sock = sock
        self.columns = []
        self.status_flags = 0
        self.warnings = 0
        self._read_column_count()
        self._read_column_definitions()

    def _read_packet(self) -> bytes:
        """读取一个数据包"""
        header = b''
        while len(header) < 4:
            header += self.sock.recv(4 - len(header))
        pkt_len = struct.unpack('<I', header[0:3] + b'\x00')[0]
        seq = header[3]
        payload = b''
        while len(payload) < pkt_len:
            payload += self.sock.recv(pkt_len - len(payload))
        return payload

    def _read_column_count(self):
        """读取列数"""
        payload = self._read_packet()
        self.column_count = payload[0]

    def _read_column_definitions(self):
        """读取列定义"""
        for _ in range(self.column_count):
            payload = self._read_packet()
            # 简化解析，只提取列名
            self.columns.append(self._parse_column_name(payload))

        # EOF
        eof = self._read_packet()

    def _parse_column_name(self, payload: bytes) -> str:
        """从列定义中提取列名（简化版）"""
        offset = 0
        for _ in range(5):  # 跳过 catalog, schema, table_alias, table_name
            length = payload[offset]
            if length >= 0xFB:
                offset += 2
            else:
                offset += 1 + length
        # 现在 offset 指向 column_name
        length = payload[offset]
        return payload[offset+1:offset+1+length].decode('utf-8', errors='replace')

    def rows(self) -> Generator[List[Any], None, None]:
        """生成器：逐行返回结果"""
        while True:
            payload = self._read_packet()

            # 检查 EOF
            if payload[0] == 0xFE and len(payload) < 9:
                if len(payload) >= 5:
                    self.status_flags = struct.unpack('<H', payload[3:5])[0]
                break

            # 解析行数据
            row = self._parse_row(payload)
            yield row

    def _parse_row(self, payload: bytes) -> List[Any]:
        """解析行数据（文本协议）"""
        row = []
        offset = 0
        for _ in range(self.column_count):
            if payload[offset] == 0xFB:  # NULL
                row.append(None)
                offset += 1
            else:
                # 长度编码字符串
                length, offset = self._read_le_int(payload, offset)
                value = payload[offset:offset+length].decode('utf-8', errors='replace')
                row.append(value)
                offset += length
        return row

    def _read_le_int(self, data, offset):
        """读取长度编码整数"""
        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 process_large_table(sock, sql):
    """流式处理大表"""
    # 发送 COM_QUERY
    payload = b'\x03' + sql.encode('utf-8')
    packet = struct.pack('<I', len(payload))[:3] + struct.pack('B', 0) + payload
    sock.send(packet)

    # 创建流式读取器
    reader = StreamingResultSet(sock)

    print(f"列名: {reader.columns}")
    print("逐行处理中...")

    total = 0
    for row in reader.rows():
        # 处理每一行
        total += 1
        if total % 10000 == 0:
            print(f"  已处理 {total:,} 行")

    print(f"处理完成: {total:,} 行")

# 使用游标 + FETCH 分批获取
stmt = prepare("SELECT * FROM huge_table")
execute_with_cursor(stmt, cursor_type=READ_ONLY)

while True:
    rows = fetch(stmt, num_rows=1000)
    if not rows:
        break
    process(rows)

def paginate_query(conn, sql, page_size=1000):
    """分页查询"""
    offset = 0
    while True:
        page_sql = f"{sql} LIMIT {page_size} OFFSET {offset}"
        rows = execute(conn, page_sql)
        if not rows:
            break
        yield from rows
        offset += page_size

def keyset_pagination(conn, table, last_id=0, page_size=1000):
    """基于键的游标分页"""
    while True:
        sql = f"SELECT * FROM {table} WHERE id > %s ORDER BY id LIMIT %s"
        rows = execute(conn, sql, [last_id, page_size])
        if not rows:
            break
        last_id = rows[-1]['id']
        yield from rows

def read_all_result_sets(sock):
    """读取所有结果集"""
    results = []
    while True:
        result = read_result_set(sock)
        results.append(result)

        if not (result.status_flags & 0x0008):  # SERVER_MORE_RESULTS_EXISTS
            break

    return results


# 示例：存储过程返回多个结果集
# CALL get_user_stats()
# 结果集1: 用户总数
# 结果集2: 活跃用户
# 结果集3: 最近注册

def handle_local_infile(sock, filename):
    """处理 LOCAL INFILE 请求"""
    with open(filename, 'rb') as f:
        while True:
            chunk = f.read(65535)
            if not chunk:
                break
            packet = encode_packet(chunk, 0)
            sock.send(packet)

    # 发送空包表示文件结束
    sock.send(encode_packet(b'', 0))

def export_to_csv(sock, table, output_file, batch_size=10000):
    """将表数据导出为 CSV"""
    import csv

    sql = f"SELECT * FROM {table}"
    send_query(sock, sql)

    reader = StreamingResultSet(sock)

    with open(output_file, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(reader.columns)

        count = 0
        for row in reader.rows():
            writer.writerow(row)
            count += 1

    print(f"导出 {count} 行到 {output_file}")

def replicate_table(source_sock, dest_sock, table):
    """复制表数据到目标数据库"""
    send_query(source_sock, f"SELECT * FROM {table}")
    reader = StreamingResultSet(source_sock)

    # 在目标端预处理 INSERT
    placeholders = ", ".join(["?" for _ in reader.columns])
    stmt = send_prepare(dest_sock, f"INSERT INTO {table} VALUES ({placeholders})")

    for row in reader.rows():
        send_execute(dest_sock, stmt, row)

    send_close(dest_sock, stmt.statement_id)