# coding: utf-8

from datetime import datetime, timedelta
import re
import logging

from airflow.models import Variable
from airflow.hooks.base import BaseHook
from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.exceptions import AirflowFailException

from utils import clickhouse
from common.feishu import FeishuMessageTool

logger = logging.getLogger(__name__)
feishu_group_code = 'ods_table_structure_alter'


def detect_source_table_structure(src_conn_id, src_table):
    """
    检测源表结构，包括列名、类型和Decimal精度，保持列顺序

    Args:
        src_conn_id: 源数据库连接ID
        src_table: 源表名

    Returns:
        tuple: (列名有序列表, {列名: {'type': 类型, 'precision': Decimal精度, 'scale': Decimal小数位数, 'nullable': 是否可空, 'comment': 列注释}})
    """
    logger.info(f"开始检测源表结构: 连接ID={src_conn_id}, 表名={src_table}")
    src_conn = BaseHook.get_connection(src_conn_id)
    src_type = src_conn.conn_type
    src_schema = src_conn.schema
    logger.info(f"源连接信息: 类型={src_type}, 数据库={src_schema}")

    src_hook = MySqlHook(mysql_conn_id=src_conn_id)

    # 获取表结构（包含列注释）
    desc_sql = f'SHOW FULL COLUMNS FROM `{src_schema}`.`{src_table}`'

    logger.info(f"执行源表结构查询: {desc_sql}")

    columns_order = []  # 保持列顺序
    columns_info = {}
    records = src_hook.get_records(sql=desc_sql)
    logger.info(f"MySQL源表结构查询完成，返回 {len(records)} 行记录")
    for record in records:
        col_name = record[0]
        col_type = record[1]
        nullable_str = record[3]  # NULL 或 NOT NULL
        comment = record[8] if len(record) > 8 else ''
        columns_order.append(col_name)  # 保持顺序
        # 提取Decimal精度
        precision = None
        scale = None
        if 'decimal' in col_type.lower() or 'numeric' in col_type.lower():
            match = re.search(r'decimal\((\d+),(\d+)\)', col_type.lower())
            if match:
                precision = int(match.group(1))
                scale = int(match.group(2))
        # 检测是否允许NULL（MySQL DESCRIBE 返回 'YES' 或 'NO'）
        nullable = nullable_str and nullable_str.upper() == 'YES'
        columns_info[col_name] = {
            'type': col_type,
            'precision': precision,
            'scale': scale,
            'nullable': nullable,
            'comment': comment or ''
        }

    logger.info(f"源表结构解析完成: 共 {len(columns_order)} 列，包含 {sum(1 for v in columns_info.values() if v.get('nullable'))} 个可空字段")
    return columns_order, columns_info


def check_table_exists(db_name, dst_conn_id, dst_table):
    """
    检查目标表是否存在

    Args:
        db_name: 数据库名
        dst_conn_id: 目标数据库连接ID
        dst_table: 目标表名

    Returns:
        bool: 表是否存在
    """
    ch_hook = clickhouse.get_clickhouse_hook(dst_conn_id)
    check_sql = f"SELECT COUNT(*) FROM system.tables WHERE database = '{db_name}' AND name = '{dst_table}'"
    records = ch_hook.get_records(sql=check_sql)
    exists = records[0][0] > 0 if records else False
    logger.info(f"检查表是否存在: 数据库={db_name}, 表名={dst_table}, 存在={exists}")
    return exists


def detect_target_table_structure(db_name, dst_conn_id, dst_table):
    """
    检测目标表结构，dag_id作为数据库名，保持列顺序

    Args:
        db_name: 数据库名
        dst_conn_id: 目标数据库连接ID
        dst_table: 目标表名

    Returns:
        tuple: (列名有序列表, {列名: {'type': 类型}})
    """
    logger.info(f"开始检测目标表结构: 数据库={db_name}, 连接ID={dst_conn_id}, 表名={dst_table}")

    ch_hook = clickhouse.get_clickhouse_hook(dst_conn_id)

    # 获取表结构
    desc_sql = f'DESCRIBE TABLE `{db_name}`.`{dst_table}`'
    logger.info(f"执行目标表结构查询: {desc_sql}")
    records = ch_hook.get_records(sql=desc_sql)
    logger.info(f"目标表结构查询完成，返回 {len(records)} 行记录")

    columns_order = []  # 保持列顺序
    columns_info = {}
    for record in records:
        col_name = record[0]
        col_type = record[1]
        columns_order.append(col_name)  # 保持顺序
        columns_info[col_name] = {
            'type': col_type,
            'comment': record[4] if len(record) > 4 else ''
        }

    logger.info(f"目标表结构解析完成: 共 {len(columns_order)} 列")
    return columns_order, columns_info


def get_after_col(col_name, src_columns_order):
    """
    确定新列应该插入的位置，返回ALTER语句中的位置表达式

    Args:
        col_name: 要添加的列名
        src_columns_order: 源表的列顺序列表

    Returns:
        str or None: ALTER语句中的位置表达式，如果返回None表示是第一列（无法添加）
    """
    # 找到新列在源表中的位置
    src_index = src_columns_order.index(col_name)
    logger.info(f"列 {col_name} 在源表中的位置: 索引={src_index}, 总列数={len(src_columns_order)}")

    # 如果新列是源表的第一列，返回None
    if src_index == 0:
        logger.warning(f"列 {col_name} 是源表的第一列，无法使用AFTER子句")
        return None

    prev_col = src_columns_order[src_index - 1]
    logger.info(f"列 {col_name} 应插入在列 {prev_col} 之后")
    return f'`{prev_col}`'


def is_type_change_compatible(old_type_info, new_type_info):
    """
    判断类型变化是否兼容（精度兼容，可以安全修改）

    Args:
        old_type_info: 旧类型信息字典 {'type': 类型, 'precision': 精度, 'scale': 小数位数}
        new_type_info: 新类型信息字典 {'type': 类型, 'precision': 精度, 'scale': 小数位数}

    Returns:
        bool: True表示兼容，可以修改；False表示不兼容，不能修改
    """
    old_type = old_type_info['type'].lower().strip()
    new_type = new_type_info['type'].lower().strip()

    old_precision = old_type_info.get('precision')
    new_precision = new_type_info.get('precision')
    old_scale = old_type_info.get('scale')
    new_scale = new_type_info.get('scale')

    # 类型完全相同，只是精度或scale变化
    if old_type == new_type:
        # Decimal/Numeric类型
        if 'decimal' in old_type or 'numeric' in old_type:
            # 精度扩大或相等，scale不变或扩大但不超过新precision
            if (new_precision is not None and old_precision is not None and
                    new_scale is not None and old_scale is not None):
                if new_precision >= old_precision and new_scale >= old_scale and new_scale <= new_precision:
                    return True

        # 字符串类型（varchar, char等）
        if 'varchar' in old_type or 'char' in old_type:
            # 长度扩大或相等视为兼容
            if new_precision is not None and old_precision is not None:
                if new_precision >= old_precision:
                    return True

        # 其他类型（int, bigint等），如果类型相同，精度变化通常不兼容，除非是整数类型扩大
        if 'int' in old_type:
            # 整数类型扩大：tinyint -> smallint -> int -> bigint
            int_type_order = {'tinyint': 1, 'smallint': 2, 'mediumint': 3, 'int': 4, 'integer': 4, 'bigint': 5}
            old_order = int_type_order.get(old_type.split('(')[0], 0)
            new_order = int_type_order.get(new_type.split('(')[0], 0)
            if new_order > old_order:
                return True

        # 其他类型相同但精度/scale变化，通常不兼容（除非是Decimal或字符串类型扩大，已在上面处理）
        return False

    # 类型不同，需要判断是否兼容转换
    # 整数类型扩大
    int_types = ['tinyint', 'smallint', 'mediumint', 'int', 'integer', 'bigint']
    if any(t in old_type for t in int_types) and any(t in new_type for t in int_types):
        int_type_order = {'tinyint': 1, 'smallint': 2, 'mediumint': 3, 'int': 4, 'integer': 4, 'bigint': 5}
        old_int_type = next((t for t in int_types if t in old_type), None)
        new_int_type = next((t for t in int_types if t in new_type), None)
        if old_int_type and new_int_type:
            old_order = int_type_order.get(old_int_type, 0)
            new_order = int_type_order.get(new_int_type, 0)
            if new_order > old_order:
                return True

    # Float32 -> Float64
    if 'float' in old_type and 'double' in new_type:
        return True

    # 其他类型变化视为不兼容
    return False


def map_mysql_type_to_clickhouse(mysql_type, precision=None, scale=None, nullable=False):
    """
    将MySQL类型映射到ClickHouse类型，正确提取Decimal精度，支持Nullable

    Args:
        mysql_type: MySQL类型字符串
        precision: Decimal精度（从源表结构获取）
        scale: Decimal小数位数（从源表结构获取）
        nullable: 是否允许NULL

    Returns:
        str: ClickHouse类型（如果nullable为True，返回Nullable(Type)格式）
    """
    mysql_type_lower = mysql_type.lower().strip()

    # Decimal类型处理
    if 'decimal' in mysql_type_lower or 'numeric' in mysql_type_lower:
        if precision is not None and scale is not None:
            ch_type = f'Decimal({precision},{scale})'
        else:
            # 尝试从类型字符串中提取
            match = re.search(r'decimal\((\d+),(\d+)\)', mysql_type_lower)
            if match:
                ch_type = f'Decimal({match.group(1)},{match.group(2)})'
            else:
                ch_type = 'Decimal(38,8)'  # 默认精度
    else:
        # 其他类型映射
        type_mapping = {
            'tinyint': 'Int8',
            'smallint': 'Int16',
            'mediumint': 'Int32',
            'int': 'Int32',
            'integer': 'Int32',
            'bigint': 'Int64',
            'float': 'Float32',
            'double': 'Float64',
            'date': 'Date',
            'datetime': 'DateTime',
            'timestamp': 'DateTime',
            'time': 'String',
            'year': 'UInt16',
            'char': 'String',
            'varchar': 'String',
            'text': 'String',
            'tinytext': 'String',
            'mediumtext': 'String',
            'longtext': 'String',
            'binary': 'String',
            'varbinary': 'String',
            'blob': 'String',
            'tinyblob': 'String',
            'mediumblob': 'String',
            'longblob': 'String',
            'enum': 'String',
            'set': 'String',
            'json': 'String',
        }

        # 处理带长度的类型
        ch_type = None
        for mysql_key, ch_type_mapped in type_mapping.items():
            if mysql_key in mysql_type_lower:
                ch_type = ch_type_mapped
                break

        # 默认返回String
        if ch_type is None:
            ch_type = 'String'

    # 如果允许NULL，返回Nullable类型
    if nullable:
        return f'Nullable({ch_type})'
    else:
        return ch_type


def format_structure_change_alert(src_table, changes_summary, src_schema=None, dst_schema=None):
    """
    格式化表结构变化告警消息

    Args:
        src_table: 源表名
        changes_summary: 变化摘要字典，包含以下键（可选）:
            - removed_columns: 删除的列列表
            - added_columns: 新增的列字典 {列名: {'src_type': 源表类型, 'type': ClickHouse类型, 'position': 字段位置, 'comment': 注释}}
            - compatible_type_changes: 兼容的类型变化字典 {列名: {'old_type': 旧类型, 'new_type': 新类型}}
            - incompatible_type_changes: 不兼容的类型变化字典 {列名: {'old_type': 旧类型, 'new_type': 新类型}}
            - comment_changes: 列注释变化字典 {列名: {'src_comment': 源表注释, 'dst_comment': 目标表注释}}
            - column_order_changes: 列顺序变化字典 {列名: {'old_position': 旧位置, 'new_position': 新位置}}
            - first_column_columns: 第一列的新列列表
        src_schema: 源数据库名（可选）
        dst_schema: 目标数据库名（可选）

    Returns:
        str: 格式化后的Markdown消息
    """
    logger.info(f"开始格式化表结构变化告警消息: 表名={src_table}, 源库={src_schema}, 目标库={dst_schema}")
    msg_lines = [
        f"## 表结构变化告警",
    ]

    # 显示库名和表名
    if src_schema:
        msg_lines.append(f"**源库**: `{src_schema}`")
    msg_lines.append(f"**表名**: `{src_table}`")
    msg_lines.append("")

    # 列顺序变化
    if changes_summary.get('column_order_changes'):
        msg_lines.append("### 🔴 列顺序变化（需要手动处理）")
        for col, order_info in changes_summary['column_order_changes'].items():
            msg_lines.append(f"- `{col}`: 位置 {order_info.get('old_position')} → {order_info.get('new_position')}")
        msg_lines.append("")

    # 删除的列
    if changes_summary.get('removed_columns'):
        removed_count = len(changes_summary['removed_columns'])
        msg_lines.append(f"### ❌ 删除的列（共 {removed_count} 列）")
        for idx, col in enumerate(changes_summary['removed_columns'], start=1):
            msg_lines.append(f"{idx}. `{col}`")
        msg_lines.append("")

    # 新增的列
    if changes_summary.get('added_columns'):
        added_count = len(changes_summary['added_columns'])
        msg_lines.append(f"### ✅ 新增的列（共 {added_count} 列）")
        # 按照位置顺序排序新增列
        sorted_added_columns = sorted(
            changes_summary['added_columns'].items(),
            key=lambda x: x[1].get('position', 999)
        )
        for idx, (col, col_info) in enumerate(sorted_added_columns, start=1):
            # 显示源表类型，如果没有则显示ClickHouse类型
            display_type = col_info.get('src_type') or col_info.get('type', 'Unknown')
            position = col_info.get('position', '')
            position_info = f"（位置：{position}）" if position else ""
            msg_lines.append(f"{idx}. `{col}`: {display_type}{position_info}")
        msg_lines.append("")

    # 兼容的类型变化（已自动修改）
    if changes_summary.get('compatible_type_changes'):
        msg_lines.append("### ✅ 兼容的类型变化（已自动修改）")
        for idx, (col, type_info) in enumerate(changes_summary['compatible_type_changes'].items(), start=1):
            msg_lines.append(f"{idx}. `{col}`: {type_info.get('old_type')} → {type_info.get('new_type')}")
        msg_lines.append("")

    # 列注释变化（已自动同步）
    if changes_summary.get('comment_changes'):
        msg_lines.append("### 📝 列注释变化（已自动修改）")
        for idx, (col, comment_info) in enumerate(changes_summary['comment_changes'].items(), start=1):
            msg_lines.append(f"{idx}. `{col}`: `{comment_info.get('dst_comment') or '空'}` → `{comment_info.get('src_comment') or '空'}`")
        msg_lines.append("")

    # 不兼容的类型变化（仅告警，未修改）
    if changes_summary.get('incompatible_type_changes'):
        msg_lines.append("### ⚠️ 不兼容的类型变化（仅告警，未修改）")
        for idx, (col, type_info) in enumerate(changes_summary['incompatible_type_changes'].items(), start=1):
            msg_lines.append(f"{idx}. `{col}`: {type_info.get('old_type')} → {type_info.get('new_type')}")
        msg_lines.append("")

    # 第一列的新列（无法自动添加）
    if changes_summary.get('first_column_columns'):
        msg_lines.append("### 🔴 第一列的新列（无法自动添加）")
        msg_lines.append("以下列是源表的第一列，无法使用 `AFTER` 子句自动添加，需要手动处理：")
        for col in changes_summary['first_column_columns']:
            msg_lines.append(f"- `{col}`")
        msg_lines.append("")

    msg_lines.append(f"**检测时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    alert_msg = "\n".join(msg_lines)
    logger.info(f"告警消息格式化完成，消息长度={len(alert_msg)} 字符")
    return alert_msg


def detect_removed_columns(src_column_set, cached_src_structure):
    """
    检测删除的列（源表中上次有但现在没有的列）

    Args:
        src_column_set: 当前源表的列名集合
        cached_src_structure: 上次保存的源表结构缓存（必须存在）

    Returns:
        list: 删除的列名列表
    """
    cached_src_column_set = set(cached_src_structure.keys())
    removed_columns = list(cached_src_column_set - src_column_set)
    if removed_columns:
        logger.warning(f"检测到删除的列: {', '.join(removed_columns)}")

    return removed_columns


def detect_added_columns(src_structure, src_column_set, target_column_set, cached_src_structure, src_columns_order):
    """
    检测新增的列（与缓存的上游表结构比较，且在目标表中不存在）

    Args:
        src_structure: 当前源表结构字典
        src_column_set: 当前源表的列名集合
        target_column_set: 目标表的列名集合
        cached_src_structure: 上次保存的源表结构缓存（必须存在）
        src_columns_order: 源表的列顺序列表

    Returns:
        dict: 新增列信息字典 {列名: {'src_type': 源表类型, 'type': ClickHouse类型, 'position': 字段位置}}
    """
    added_columns = {}
    cached_src_column_set = set(cached_src_structure.keys())
    # 新增列：当前源表有但缓存中没有的列，且目标表中也不存在
    new_src_columns = src_column_set - cached_src_column_set

    # 按照源表的列顺序处理新增列
    for position, col_name in enumerate(src_columns_order, start=1):
        if col_name not in new_src_columns:
            continue  # 不是新增列，跳过

        if col_name not in target_column_set:
            col_info = src_structure[col_name]
            # 保存源表类型和ClickHouse类型
            # MySQL的DESCRIBE返回的类型字符串已经包含精度信息（如 decimal(18,8)）
            src_type = col_info['type']
            nullable = col_info.get('nullable', False)
            ck_type = map_mysql_type_to_clickhouse(
                col_info['type'],
                col_info.get('precision'),
                col_info.get('scale'),
                nullable
            )

            logger.info(f"检测到新增列: {col_name}, 位置={position}, 源表类型={src_type}, nullable={nullable}, ClickHouse类型={ck_type}")

            added_columns[col_name] = {
                'src_type': src_type,  # 源表类型（用于显示，如 decimal(18,8), varchar(255)）
                'type': ck_type,  # ClickHouse类型（用于ALTER TABLE，如 Nullable(Decimal(18,8))）
                'comment': col_info.get('comment', ''),
                'position': position  # 字段在源表中的位置（从1开始）
            }
        elif col_name in target_column_set:
            logger.info(f"列 {col_name} 已在目标表中存在，跳过添加")

    if added_columns:
        logger.info(f"共检测到 {len(added_columns)} 个新增列（与缓存比较）: {', '.join(added_columns.keys())}")

    return added_columns


def detect_column_order_changes(src_columns_order, cached_src_columns_order):
    """
    检测列顺序变化

    Args:
        src_columns_order: 当前源表的列顺序列表
        cached_src_columns_order: 上次保存的源表列顺序列表

    Returns:
        tuple: (是否有顺序变化, 顺序变化详情)
               顺序变化详情: {列名: {'old_position': 旧位置, 'new_position': 新位置}}
    """
    if not cached_src_columns_order:
        return False, {}

    # 只比较两个列表中都存在的列
    curr_common_columns = [col for col in src_columns_order if col in cached_src_columns_order]
    cached_common_columns = [col for col in cached_src_columns_order if col in src_columns_order]

    # 如果顺序相同，返回False
    if curr_common_columns == cached_common_columns:
        return False, {}

    # 检测顺序变化
    order_changes = {}
    for idx, col_name in enumerate(curr_common_columns):
        old_idx = cached_common_columns.index(col_name)
        if old_idx != idx:
            order_changes[col_name] = {
                'old_position': old_idx + 1,  # 从1开始计数
                'new_position': idx + 1
            }

    if order_changes:
        logger.warning(f"检测到列顺序变化，共 {len(order_changes)} 列位置发生变化")

    return len(order_changes) > 0, order_changes


def detect_type_changes(src_structure, target_column_set, cached_src_structure):
    """
    检测字段类型变化（与上次保存的上游表结构进行比较）
    如果数据类型精度兼容，则修改目标表数据类型；否则只告警，不修改

    Args:
        src_structure: 当前源表结构字典
        target_column_set: 目标表的列名集合
        cached_src_structure: 上次保存的源表结构缓存（必须存在）

    Returns:
        tuple: (不兼容的类型变化字典, 兼容的类型变化字典)
               不兼容类型变化: {列名: {'old_type': 旧类型, 'new_type': 新类型}}
               兼容类型变化: {列名: {'old_type': 旧类型, 'new_type': 新类型, 'ck_type': ClickHouse类型}}
    """
    incompatible_type_changes = {}  # 不兼容的类型变化（只告警）
    compatible_type_changes = {}  # 兼容的类型变化（需要修改目标表）

    logger.info(f"开始比较源表结构变化，缓存中共有 {len(cached_src_structure)} 列")
    # 比较当前源表结构和上次保存的结构
    for col_name in src_structure.keys():
        if col_name in cached_src_structure:
            current_col_info = src_structure[col_name]
            cached_col_info = cached_src_structure[col_name]

            # 比较源表类型（MySQL类型），包括精度、小数位数和nullable属性
            current_type = current_col_info['type']
            cached_type = cached_col_info['type']
            current_precision = current_col_info.get('precision')
            cached_precision = cached_col_info.get('precision')
            current_scale = current_col_info.get('scale')
            cached_scale = cached_col_info.get('scale')
            current_nullable = current_col_info.get('nullable', False)
            cached_nullable = cached_col_info.get('nullable', False)

            # 类型不同，或者精度/小数位数/nullable属性不同，都认为是类型变化
            if (current_type != cached_type or
                    current_precision != cached_precision or
                    current_scale != cached_scale or
                    current_nullable != cached_nullable):

                # 判断类型变化是否兼容
                is_compatible = is_type_change_compatible(cached_col_info, current_col_info)

                # 保存源表类型用于显示（MySQL类型，如 decimal(18,8)）
                old_type_display = cached_type
                new_type_display = current_type
                # 如果nullable属性变化，在类型显示中标注
                if current_nullable != cached_nullable:
                    old_type_display = f"{cached_type} ({'NULL' if cached_nullable else 'NOT NULL'})"
                    new_type_display = f"{current_type} ({'NULL' if current_nullable else 'NOT NULL'})"

                if is_compatible:
                    # 兼容的类型变化，需要修改目标表（但需要检查目标表中是否存在该列）
                    if col_name in target_column_set:
                        logger.info(f"列 {col_name} 类型变化兼容: {cached_type} -> {current_type} (nullable: {cached_nullable} -> {current_nullable})，将自动修改目标表")

                        # 计算新的ClickHouse类型
                        new_ck_type = map_mysql_type_to_clickhouse(
                            current_type,
                            current_precision,
                            current_scale,
                            current_nullable
                        )

                        compatible_type_changes[col_name] = {
                            'old_type': old_type_display,  # 源表旧类型
                            'new_type': new_type_display,  # 源表新类型
                            'ck_type': new_ck_type  # ClickHouse新类型（用于ALTER TABLE）
                        }
                    else:
                        # 兼容但目标表中不存在该列，跳过（可能是新列，会通过新增列逻辑处理）
                        logger.info(f"列 {col_name} 类型变化兼容，但目标表中不存在该列，跳过修改（可能通过新增列逻辑处理）")
                else:
                    # 不兼容的类型变化，只告警
                    logger.warning(f"列 {col_name} 类型变化不兼容: {cached_type} -> {current_type} (nullable: {cached_nullable} -> {current_nullable})，仅发送告警，不修改表结构")

                    incompatible_type_changes[col_name] = {
                        'old_type': old_type_display,  # 源表旧类型
                        'new_type': new_type_display  # 源表新类型
                    }

    if compatible_type_changes:
        logger.info(f"兼容的类型变化检测完成，共检测到 {len(compatible_type_changes)} 个列的类型变化（将自动修改）: {', '.join(compatible_type_changes.keys())}")
    if incompatible_type_changes:
        logger.info(f"不兼容的类型变化检测完成，共检测到 {len(incompatible_type_changes)} 个列的类型变化（仅告警）: {', '.join(incompatible_type_changes.keys())}")
    if not compatible_type_changes and not incompatible_type_changes:
        logger.info("类型变化检测完成，未发现类型变化")

    return incompatible_type_changes, compatible_type_changes


def detect_comment_changes(src_structure, target_structure):
    """
    检测列注释变化（与当前目标表进行比较）

    Args:
        src_structure: 当前源表结构字典
        target_structure: 当前目标表结构字典

    Returns:
        dict: 注释变化 {列名: {'src_comment': 源表注释, 'dst_comment': 目标表注释}}
    """
    comment_changes = {}
    for col_name, src_info in src_structure.items():
        if col_name not in target_structure:
            continue
        src_comment = (src_info.get('comment') or '').strip()
        dst_comment = (target_structure[col_name].get('comment') or '').strip()
        if src_comment != dst_comment:
            comment_changes[col_name] = {
                'src_comment': src_comment,
                'dst_comment': dst_comment
            }

    if comment_changes:
        logger.info(f"检测到 {len(comment_changes)} 个列的注释与ClickHouse不一致: {', '.join(comment_changes.keys())}")

    return comment_changes


def handle_removed_columns(src_table, removed_columns):
    """
    处理删除的列，如果存在删除的列则抛出错误

    Args:
        src_table: 源表名
        removed_columns: 删除的列名列表
    """
    if removed_columns:
        error_msg = f"源表 {src_table} 删除了以下列: {', '.join(removed_columns)}。请检查并更新配置。"
        logger.error(error_msg)
        raise AirflowFailException(error_msg)


def handle_column_order_changes(src_table, column_order_changes, src_schema, dst_schema):
    """
    处理列顺序变化，如果存在顺序变化则发送告警并抛出错误

    Args:
        src_table: 源表名
        column_order_changes: 列顺序变化详情字典
        src_schema: 源数据库名
        dst_schema: 目标数据库名
    """
    if not column_order_changes:
        return

    error_msg = f"源表 {src_table} 的列顺序发生变化。请检查并更新配置。"
    logger.error(error_msg)

    # 发送飞书告警
    try:
        feishu_tool = FeishuMessageTool.from_group_code(feishu_group_code)
        changes_summary = {
            'column_order_changes': column_order_changes,
            'removed_columns': [],
            'added_columns': {},
            'incompatible_type_changes': {},
            'compatible_type_changes': {}
        }
        alert_msg = format_structure_change_alert(src_table, changes_summary, src_schema, dst_schema)
        feishu_tool.send_markdown_message(alert_msg)
        logger.info("列顺序变化告警已发送")
    except Exception as e:
        logger.error(f"发送飞书告警失败: {e}")

    raise AirflowFailException(error_msg)


def handle_added_columns(conf, src_schema, ods_db_name, src_columns_order, src_structure,
                         added_columns):
    """
    处理新增的列，生成并执行ALTER TABLE ADD COLUMN语句

    Args:
        conf: 配置字典
        src_schema: 源数据库名
        ods_db_name: ODS数据库名
        src_columns_order: 源表列顺序列表
        src_structure: 源表结构字典
        added_columns: 新增列信息字典
    """
    if not added_columns:
        return

    src_table = conf['src_table']
    dst_conn_id = conf['dst_conn_id']
    dst_table = conf['dst_table']
    dst_schema = ods_db_name

    logger.info(f"发现 {len(added_columns)} 个新列，准备添加到目标表 {dst_schema}.{dst_table}")

    ch_hook = clickhouse.get_clickhouse_hook(dst_conn_id)
    db_name = ods_db_name

    logger.info(f"开始按源表列顺序添加新列，源表共有 {len(src_columns_order)} 列")

    # 按照源表的列顺序添加新列
    for col_name in src_columns_order:
        if col_name not in added_columns:
            continue  # 不是新列，跳过

        col_info = added_columns[col_name]
        src_col_info = src_structure[col_name]
        ck_type = map_mysql_type_to_clickhouse(
            src_col_info['type'],
            src_col_info.get('precision'),
            src_col_info.get('scale'),
            src_col_info.get('nullable', False)
        )
        comment = src_col_info.get('comment', '')
        escaped_comment = comment.replace("'", "''") if comment else ''
        comment_clause = f" COMMENT '{escaped_comment}'" if comment else ''

        # 确定应该插入在哪个列后面
        after_col = get_after_col(col_name, src_columns_order)

        # 如果是第一列，发送飞书告警并抛出错误
        if after_col is None:
            error_msg = f"列 `{col_name}` 是源表 `{src_table}` 的第一列，无法使用 `AFTER` 子句自动添加，需要手动处理"
            logger.error(error_msg)

            # 发送飞书告警
            try:
                feishu_tool = FeishuMessageTool.from_group_code(feishu_group_code)
                changes_summary = {
                    'removed_columns': [],
                    'added_columns': {},
                    'incompatible_type_changes': {},
                    'first_column_columns': [col_name]
                }
                alert_msg = format_structure_change_alert(src_table, changes_summary, src_schema, dst_schema)
                feishu_tool.send_markdown_message(alert_msg)
                logger.info("第一列告警已发送至飞书")
            except Exception as e:
                logger.error(f"发送飞书告警失败: {e}")

            raise AirflowFailException(error_msg)

        # 构建ALTER语句
        alter_sql = (
            f'ALTER TABLE `{db_name}`.`{dst_table}` ON CLUSTER default '
            f'ADD COLUMN `{col_name}` {ck_type}{comment_clause} AFTER {after_col}'
        )

        logger.info(f"准备执行ALTER TABLE: {alter_sql}")
        try:
            ch_hook.run(sql=alter_sql)
            logger.info(f"成功添加列: {col_name} (类型={ck_type}, 位置=AFTER {after_col})")
        except Exception as e:
            logger.error(f"添加列 {col_name} 失败: {e}")
            raise

    logger.info("所有新列已添加完成")


def handle_compatible_type_changes(conf, ods_db_name, compatible_type_changes):
    """
    处理兼容的类型变化，生成并执行ALTER TABLE MODIFY COLUMN语句

    Args:
        conf: 配置字典
        ods_db_name: ODS数据库名
        compatible_type_changes: 兼容的类型变化字典
    """
    if not compatible_type_changes:
        return

    dst_conn_id = conf['dst_conn_id']
    dst_table = conf['dst_table']
    dst_schema = ods_db_name

    logger.info(f"发现 {len(compatible_type_changes)} 个兼容的类型变化，准备修改目标表 {dst_schema}.{dst_table}")

    ch_hook = clickhouse.get_clickhouse_hook(dst_conn_id)
    db_name = ods_db_name

    for col_name, change_info in compatible_type_changes.items():
        ck_type = change_info['ck_type']

        # 构建ALTER语句
        alter_sql = f'ALTER TABLE `{db_name}`.`{dst_table}` ON CLUSTER default MODIFY COLUMN `{col_name}` {ck_type}'

        logger.info(f"准备执行ALTER TABLE MODIFY COLUMN: {alter_sql}")
        logger.info(f"列 {col_name} 类型变化: {change_info['old_type']} -> {change_info['new_type']} (ClickHouse类型: {ck_type})")

        try:
            ch_hook.run(sql=alter_sql)
            logger.info(f"成功修改列类型: {col_name} (新类型={ck_type})")
        except Exception as e:
            logger.error(f"修改列 {col_name} 类型失败: {e}")
            raise

    logger.info("所有兼容的类型变化已修改完成")


def handle_comment_changes(conf, ods_db_name, comment_changes):
    """
    处理列注释变化，生成并执行ALTER TABLE COMMENT COLUMN语句

    Args:
        conf: 配置字典
        ods_db_name: ODS数据库名
        comment_changes: 注释变化字典
    """
    if not comment_changes:
        return

    dst_conn_id = conf['dst_conn_id']
    dst_table = conf['dst_table']
    dst_schema = ods_db_name

    logger.info(f"发现 {len(comment_changes)} 个列的注释需要更新，目标表 {dst_schema}.{dst_table}")

    ch_hook = clickhouse.get_clickhouse_hook(dst_conn_id)
    db_name = dst_schema

    for col_name, comment_info in comment_changes.items():
        src_comment = comment_info['src_comment']
        escaped_comment = src_comment.replace("'", "''")
        alter_sql = (
            f"ALTER TABLE `{db_name}`.`{dst_table}` ON CLUSTER default "
            f"COMMENT COLUMN `{col_name}` '{escaped_comment}'"
        )

        logger.info(
            f"准备更新列注释: {col_name}, ClickHouse注释='{comment_info['dst_comment']}', "
            f"源表注释='{src_comment}'"
        )
        logger.info(f"执行ALTER语句: {alter_sql}")

        try:
            ch_hook.run(sql=alter_sql)
            logger.info(f"成功更新列注释: {col_name}")
        except Exception as e:
            logger.error(f"更新列 {col_name} 注释失败: {e}")
            raise

    logger.info("所有列注释已更新完成")


def _save_cache_and_return(conf, variable_name, src_structure, src_columns_order, src_schema, dst_schema, src_table, reason):
    """
    保存源表结构缓存到Variable并返回

    Args:
        conf: 配置字典
        variable_name: Variable名称
        src_structure: 源表结构字典
        src_columns_order: 源表列顺序列表
        src_schema: 源数据库名
        dst_schema: 目标数据库名
        src_table: 源表名
        reason: 跳过检查的原因
    """
    logger.info(f"保存源表结构缓存到Variable: {variable_name}，共 {len(src_structure)} 列")
    conf['src_structure_cache'] = src_structure
    conf['src_columns_order_cache'] = src_columns_order
    conf['time_cache'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    try:
        Variable.set(variable_name, conf, serialize_json=True)
        logger.info(f"✅ 源表结构缓存已成功保存到Variable: {variable_name}，缓存列数: {len(src_structure)}，列顺序: {len(src_columns_order)}，详细配置信息: {conf}")
    except Exception as e:
        logger.error(f"❌ 保存源表结构缓存到Variable失败: {e}")
        raise
    logger.info(f"表结构检查完成（{reason}，保存缓存）: 源库={src_schema}, 目标库={dst_schema}, 表名={src_table}")


def check_and_handle_structure_changes(ods_db_name, variable_name, is_only_check_comment_changes=False):
    """
    检测并处理表结构变化

    Args:
        ods_db_name: DAG ID，用于推导数据库名
        variable_name: Airflow Variable名称
    """
    logger.info(f"开始表结构检查，Variable名称={variable_name}, ODS数据库名={ods_db_name}")
    conf = Variable.get(variable_name, deserialize_json=True)
    src_conn_id = conf['src_conn_id']
    src_table = conf['src_table']
    dst_conn_id = conf['dst_conn_id']
    dst_table = conf['dst_table']
    src_columns_config = conf.get('src_columns')

    # 获取源库名和目标库名
    src_conn = BaseHook.get_connection(src_conn_id)
    src_type = src_conn.conn_type
    src_schema = src_conn.schema  # 源数据库名
    dst_schema = ods_db_name  # 目标数据库名（ODS数据库名）

    logger.info(f"开始检测表结构变化: 源库={src_schema}, 目标库={dst_schema}, 表名={src_table}")
    logger.info(f"源连接信息: 类型={src_type}, 数据库={src_schema}")

    if src_type != 'mysql':
        logger.error(f'不支持的数据源类型: {src_type}')
        return None
    # 获取源表结构（包含列顺序）
    try:
        src_columns_order, src_structure = detect_source_table_structure(src_conn_id, src_table)
        logger.info(f"源表结构检测完成，共 {len(src_structure)} 列")
    except Exception as e:
        logger.error(f"检测源表结构失败: {e}")
        raise
    # 如果src_columns不包含*，跳过结构检查
    if src_columns_config and '*' not in src_columns_config:
        logger.info(f"src_columns不包含*，跳过表结构检查: {src_columns_config}")
        _save_cache_and_return(conf, variable_name, src_structure, src_columns_order, src_schema, dst_schema, src_table, "src_columns不包含*")
        return

    # 检查目标表是否存在
    if not check_table_exists(ods_db_name, dst_conn_id, dst_table):
        logger.warning(f"目标表不存在: {ods_db_name}.{dst_table}，跳过结构变化检查")
        _save_cache_and_return(conf, variable_name, src_structure, src_columns_order, src_schema, dst_schema, src_table, "目标表不存在")
        return

    # 获取目标表结构（包含列顺序）
    target_columns_order, target_structure = detect_target_table_structure(ods_db_name, dst_conn_id, dst_table)
    logger.info(f"目标表结构检测完成，共 {len(target_structure)} 列")

    # 获取上次保存的源表结构缓存
    cached_src_structure = conf.get('src_structure_cache')
    cached_src_columns_order = conf.get('src_columns_order_cache')
    if not cached_src_structure:
        logger.info("未找到上次保存的源表结构缓存，跳过结构变化检测（可能是第一次执行）")
        _save_cache_and_return(conf, variable_name, src_structure, src_columns_order, src_schema, dst_schema, src_table, "首次执行")
        return

    # 检测列顺序变化
    has_order_changes, column_order_changes = detect_column_order_changes(src_columns_order, cached_src_columns_order)
    handle_column_order_changes(src_table, column_order_changes, src_schema, dst_schema)

    # 检测变化
    src_column_set = set(src_structure.keys())
    target_column_set = set(target_structure.keys())

    # 检测删除的列
    removed_columns = detect_removed_columns(src_column_set, cached_src_structure)

    # 检测新增的列
    added_columns = detect_added_columns(src_structure, src_column_set, target_column_set, cached_src_structure, src_columns_order)

    # 检测类型变化
    incompatible_type_changes, compatible_type_changes = detect_type_changes(
        src_structure, target_column_set, cached_src_structure
    )

    # 检测列注释变化（源表 vs 目标表）
    comment_changes = detect_comment_changes(src_structure, target_structure)

    if is_only_check_comment_changes:
        # 准备告警消息
        changes_summary = {
            'removed_columns': [],
            'added_columns': [],
            'incompatible_type_changes': [],  # 不兼容的类型变化（只告警）
            'compatible_type_changes': [],  # 兼容的类型变化（已自动修改）
            'comment_changes': comment_changes  # 注释变化（已自动修改）
        }
        has_changes = (
            comment_changes
        )
    else:
        # 准备告警消息
        changes_summary = {
            'removed_columns': list(removed_columns),
            'added_columns': added_columns,
            'incompatible_type_changes': incompatible_type_changes,  # 不兼容的类型变化（只告警）
            'compatible_type_changes': compatible_type_changes,  # 兼容的类型变化（已自动修改）
            'comment_changes': comment_changes  # 注释变化（已自动修改）
        }

        has_changes = (
                removed_columns or added_columns or incompatible_type_changes or
                compatible_type_changes or comment_changes
        )

    # 汇总检测结果
    logger.info(
        f"表结构变化检测完成: 删除列={len(removed_columns)}, 新增列={len(added_columns)}, 兼容类型变化={len(compatible_type_changes)}, 不兼容类型变化={len(incompatible_type_changes)}")

    if not has_changes:
        logger.info("未检测到表结构变化，表结构保持一致")

    # 如果有变化，发送告警
    if has_changes:
        logger.info(f"准备发送飞书告警，飞书群代码={feishu_group_code}")
        try:
            feishu_tool = FeishuMessageTool.from_group_code(feishu_group_code)
            alert_msg = format_structure_change_alert(src_table, changes_summary, src_schema, dst_schema)
            feishu_tool.send_markdown_message(alert_msg)
            logger.info("表结构变化告警已发送")
        except Exception as e:
            logger.error(f"发送飞书告警失败: {e}")

    # 处理删除的列
    handle_removed_columns(src_table, removed_columns)

    # 处理新增的列
    handle_added_columns(conf, src_schema, ods_db_name, src_columns_order, src_structure,
                         added_columns)

    # 处理兼容的类型变化
    handle_compatible_type_changes(conf, ods_db_name, compatible_type_changes)

    # 处理列注释变化
    handle_comment_changes(conf, ods_db_name, comment_changes)

    # 更新Variable中的源表结构信息
    logger.info(f"保存源表结构缓存到Variable: {variable_name}，共 {len(src_structure)} 列")
    _save_cache_and_return(conf, variable_name, src_structure, src_columns_order, src_schema, dst_schema, src_table, "保存源表结构缓存")

    logger.info(f"表结构检查完成: 源库={src_schema}, 目标库={dst_schema}, 表名={src_table}")
