package org.springblade.datasource.transform.mysql;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.datasource.enums.DsTypeEnum;
import org.springblade.datasource.support.DynamicDataSourceSupport;
import org.springblade.datasource.transform.ColumnEnum;
import org.springblade.datasource.transform.ColumnInfo;
import org.springblade.datasource.transform.ITransform;
import org.springblade.datasource.transform.TableInfo;
import org.springblade.datasource.transform.general.GeneralEnum;
import org.springblade.datasource.transform.general.GeneralModel;
import org.springblade.datasource.transform.seatunnel.MysqlToSeatunnelEnum;
import org.springblade.datasource.transform.seatunnel.SeatunnelColumnType;
import org.springblade.etl.entity.SqlColumn;
import org.springblade.etl.entity.bi.ReportDatasource;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
import java.util.Locale;
import java.util.stream.Collectors;

/**
 * @author 喜欢番茄火锅
 * @date 2023/8/18 17:42
 */
@Slf4j
@Component
public class MysqlTransform implements ITransform {


    /**
     * 翻译数据库字段
     *
     * @param model
     * @return
     */
    @Override
    public List<ColumnInfo> transformColumns(List<GeneralModel> model, List<SqlColumn> sqlColumns) {

        MysqlCreateTableHandle mysqlCreateTableHandle = new MysqlCreateTableHandle();
        return model.stream().map(c -> {
            ColumnInfo columnInfo = new ColumnInfo();
            columnInfo.setColumnComments(c.getColumnComments());
            columnInfo.setColumnName(c.getColumnName());
            columnInfo.setColumnSize(c.getColumnSize());
            columnInfo.setDecimalDigits(c.getDecimalDigits());
            if (c.getSourceColumn()) {
                SqlColumn sqlColumn = sqlColumns.stream().filter(v -> v.getColumnType().equalsIgnoreCase(c.getType())).findAny().orElseThrow(() -> new ServiceException("转换模型源端字段类型错误!"));
                columnInfo.setColumnType(ColumnEnum.valueOf(sqlColumn.getSqlColumnType().toUpperCase(Locale.ROOT)));
            } else {
                columnInfo.setColumnType(GeneralEnum.valueOf(c.getType().toUpperCase()).getColumnEnum());
            }
            String columnType = mysqlCreateTableHandle.getColumnType(columnInfo);
            columnInfo.setDataType(columnType);
            return columnInfo;
        }).collect(Collectors.toList());


    }

    /**
     * 根据通用表信息创建表对象
     *
     * @param tableInfo
     * @return
     * @author 王瑞
     * @date 2023/8/16
     */
    @SneakyThrows
    @Override
    public void creatTable(TableInfo tableInfo, ReportDatasource datasource, boolean deleteSourceTable) {
        new MysqlCreateTableHandle().creatTable(tableInfo, datasource, deleteSourceTable);

    }


    /**
     * 将源信息转换成通用表数据
     *
     * @param datasource
     * @return {@link TableInfo}
     * @author 王瑞
     * @date 2023/8/16
     */
    @Override
    public TableInfo parseTableInfo(ReportDatasource datasource, List<GeneralModel> model, String tableViewName, String tableComment) {
        return new MysqlParseTableHandle().parseTableInfo(datasource, model, tableViewName, tableComment);

    }

    /**
     * 获取原表字段信息
     *
     * @param tableViewName
     * @param datasource
     * @return {@link List< ColumnInfo>}
     * @author 王瑞
     * @date 2023/12/25
     */
    @Override
    @SneakyThrows
    public List<ColumnInfo> getSourceColumn(String tableViewName, ReportDatasource datasource) throws SQLException {
        Connection connection = DynamicDataSourceSupport.createConnection(datasource);

        List<ColumnInfo> sourceColumn;
        try {
            sourceColumn = new MysqlParseTableHandle().getSourceColumn(tableViewName, connection.getMetaData(), datasource.getDbSchema());
        } finally {
            connection.close();
        }
        return sourceColumn;
    }

    /**
     * 数据源类型
     *
     * @return
     */
    @Override
    public DsTypeEnum dsType() {
        return DsTypeEnum.mysql;
    }

    @Override
    public SeatunnelColumnType transformToSeatunnelType(String columnType) {
        return MysqlToSeatunnelEnum.transform(columnType);
    }
}
