package org.springblade.datasource.transform.clickhouse;

import cn.hutool.core.collection.CollectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.datasource.support.DynamicDataSourceSupport;
import org.springblade.datasource.transform.*;
import org.springblade.datasource.transform.util.ListUtils;
import org.springblade.etl.entity.bi.ReportDatasource;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author 喜欢番茄火锅
 * @date 2023/8/24 14:55
 */
@Slf4j
public class ClickhouseCreateTableHandle {
    /**
     * 根据通用表信息创建表对象
     *
     * @param tableInfo
     * @param targetDatasource
     * @param deleteSourceTable
     * @return
     * @author 王瑞
     * @date 2023/8/21
     */
    @SneakyThrows
    public void creatTable(TableInfo tableInfo, ReportDatasource targetDatasource, boolean deleteSourceTable) {
        List<ColumnInfo> columnInfo = tableInfo.getColumnInfo();
        List<IndexInfo> indexInfo = tableInfo.getIndexInfo();
        String columSql = buildColumnSql(tableInfo.getTableName(), tableInfo.getTableComment(), columnInfo, indexInfo);
        log.info("创建表sql 语句,sql:{}", columSql);
        Connection connection = null;
        connection = DynamicDataSourceSupport.createConnection(targetDatasource);
        ResultSet tables = connection.getMetaData().getTables(null, targetDatasource.getDsDb(), tableInfo.getTableName(), new String[]{"TABLE", "VIEW"});
        Statement statement = connection.createStatement();
        if (tables.next()) {
            List<ColumnInfo> sourceColumn = new ClickhouseParseTableHandle().getSourceColumn(tableInfo.getTableName(), connection.getMetaData(), targetDatasource.getDbSchema());
            List<String> collect = tableInfo.getColumnInfo().stream().map(ColumnInfo::getColumnName).collect(Collectors.toList());
            List<String> collect2 = sourceColumn.stream().map(ColumnInfo::getColumnName).collect(Collectors.toList());
            if (!ListUtils.equalLists(collect, collect2)) {
                log.error("Clickhouse源,表名称:+" + tableInfo.getTableName() + "已存在且字段不符!");
                throw new ServiceException("Clickhouse源,表名称:" + tableInfo.getTableName() + "已存在且字段不符!");
            }
            return;
        }
        statement.executeUpdate(columSql);
        statement.close();

    }


    private String buildColumnSql(String tableName, String tableComment, List<ColumnInfo> columnInfos, List<IndexInfo> indexInfo) {
        List<IndexInfo> collect = null;
        StringBuilder primaryKey = new StringBuilder();
        if (CollectionUtil.isNotEmpty(indexInfo)) {
            collect = indexInfo.stream().filter(IndexInfo::getPrimaryKey).collect(Collectors.toList());
            if (CollectionUtil.isNotEmpty(collect)) {
                collect.get(0).getIndexColumn().stream().forEach(v -> {
                    primaryKey.append(v.getColumnName() + ",");
                });
                if (StringUtils.isNotBlank(primaryKey)) {
                    primaryKey.deleteCharAt(primaryKey.length() - 1);
                }
            }
        } else {
            primaryKey.append("tuple()");
        }

        StringBuilder fieldSql = new StringBuilder();
        for (ColumnInfo columnInfo : columnInfos) {
            if (CollectionUtil.isNotEmpty(collect) && collect.get(0).getIndexColumn().stream().anyMatch(v -> columnInfo.getColumnName().equalsIgnoreCase(v.getColumnName()))) {
                fieldSql.append(String.format(
                        "%s %s COMMENT '%s' ,",
                        columnInfo.getColumnName(),
                        getColumnType(columnInfo),
                        columnInfo.getColumnComments()
                ));
            } else {
                fieldSql.append(String.format(
                        "%s Nullable(%s) COMMENT '%s' ,",
                        columnInfo.getColumnName(),
                        getColumnType(columnInfo),
                        columnInfo.getColumnComments()
                ));
            }
        }

        fieldSql.deleteCharAt(fieldSql.length() - 1);
        return String.format("CREATE TABLE IF NOT EXISTS %s (%s) ENGINE = MergeTree() ORDER BY %s", tableName, fieldSql, primaryKey);
    }


    public String getColumnType(ColumnInfo columnInfo) {
        ColumnEnum columnEnum = columnInfo.getColumnType();
        Integer columnSize = columnInfo.getColumnSize();
        Integer decimalDigits = columnInfo.getDecimalDigits();
        DefColumnSize defColumnSize = null;
        switch (columnEnum) {

            case DECIMAL:
                defColumnSize = DefColumnSize.parameterCorrection(new DefColumnSize(38, 8), columnSize, decimalDigits);
                return String.format("Decimal(%s,%s)", TransformUtil.getColumnSize(defColumnSize.getDefColumnSize(), "10"), TransformUtil.getDecimalDigits(defColumnSize.getDefDecimalDigits(), "2"));
            case STRING:
                return "String";
            case LONG:
                return "Int64";
            case INTEGER:
                return "Int32";
            case DOUBLE:
                return "Float64";
            case FLOAT:
                return "Float32";
            case DATE:
                return "DateTime";
            case BOOL:
                return "Bool";
            default:
                throw new ServiceException("暂不支持该数据库类型!");
        }

    }


}
