package com.purui.prims.headless.server.utils;


import com.purui.prims.auth.api.authentication.pojo.User;
import com.purui.prims.common.jsqlparser.SqlReplaceHelper;
import com.purui.prims.common.jsqlparser.SqlSelectFunctionHelper;
import com.purui.prims.common.jsqlparser.SqlSelectHelper;
import com.purui.prims.common.pojo.Aggregator;
import com.purui.prims.common.pojo.Constants;
import com.purui.prims.common.pojo.enums.AggOperatorEnum;
import com.purui.prims.common.pojo.enums.QueryType;
import com.purui.prims.common.pojo.enums.TimeDimensionEnum;
import com.purui.prims.common.util.ContextUtils;
import com.purui.prims.headless.api.pojo.FieldParam;
import com.purui.prims.headless.api.pojo.MeasureParam;
import com.purui.prims.headless.api.pojo.MetricTable;
import com.purui.prims.headless.api.pojo.QueryParam;
import com.purui.prims.headless.api.pojo.SchemaItem;
import com.purui.prims.headless.api.pojo.enums.AggOption;
import com.purui.prims.headless.api.pojo.enums.EngineType;
import com.purui.prims.headless.api.pojo.enums.MetricDefineType;
import com.purui.prims.headless.api.pojo.enums.MetricType;
import com.purui.prims.headless.api.pojo.request.QueryMultiSqlReq;
import com.purui.prims.headless.api.pojo.request.QuerySqlReq;
import com.purui.prims.headless.api.pojo.request.QueryStructReq;
import com.purui.prims.headless.api.pojo.response.DatabaseResp;
import com.purui.prims.headless.api.pojo.response.MetricResp;
import com.purui.prims.headless.api.pojo.response.MetricSchemaResp;
import com.purui.prims.headless.api.pojo.response.SemanticSchemaResp;
import com.purui.prims.headless.core.adaptor.db.DbAdaptor;
import com.purui.prims.headless.core.adaptor.db.DbAdaptorFactory;
import com.purui.prims.headless.core.pojo.DataSetQueryParam;
import com.purui.prims.headless.core.pojo.MetricQueryParam;
import com.purui.prims.headless.core.pojo.QueryStatement;
import com.purui.prims.headless.core.utils.SqlGenerateUtils;
import com.purui.prims.headless.server.service.DataSetService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 查询请求转化
 *
 * @author charlie
 * @date 2024/6/25
 */
@Component
@Slf4j
public class QueryReqConverter {

    @Autowired
    private QueryStructUtils queryStructUtils;

    @Autowired
    private SqlGenerateUtils sqlGenerateUtils;

    public QueryStatement convert(QuerySqlReq querySQLReq, SemanticSchemaResp semanticSchemaResp) throws Exception {

        if (semanticSchemaResp == null) {
            return new QueryStatement();
        }
        //1.convert name to bizName
        convertNameToBizName(querySQLReq, semanticSchemaResp);
        //2.functionName corrector
        functionNameCorrector(querySQLReq, semanticSchemaResp);
        //3.correct tableName
        correctTableName(querySQLReq);
        String tableName = SqlSelectHelper.getTableName(querySQLReq.getSql());
        if (StringUtils.isEmpty(tableName)) {
            return new QueryStatement();
        }
        // correct order item is same as agg alias
        String reqSql = querySQLReq.getSql();
        querySQLReq.setSql(SqlReplaceHelper.replaceAggAliasOrderItem(querySQLReq.getSql()));
        log.info("replaceOrderAggSameAlias {} -> {}", reqSql, querySQLReq.getSql());
        //4.build MetricTables
        List<String> allFields = SqlSelectHelper.getAllFields(querySQLReq.getSql());
        List<MetricSchemaResp> metricSchemas = getMetrics(semanticSchemaResp, allFields);
        List<String> metrics = metricSchemas.stream().map(m -> m.getBizName()).collect(Collectors.toList());
        QueryStructReq queryStructReq = new QueryStructReq();
        MetricTable metricTable = new MetricTable();
        metricTable.setMetrics(metrics);

        // 解析需要的维度 charlie
        Set<String> dimensions = getDimensions(semanticSchemaResp, allFields);

        metricTable.setDimensions(new ArrayList<>(dimensions));

        metricTable.setAlias(tableName.toLowerCase());
        // if metric empty , fill model default
        if (CollectionUtils.isEmpty(metricTable.getMetrics())) {
            metricTable.setMetrics(new ArrayList<>());
            metricTable.getMetrics()
                       .add(sqlGenerateUtils.generateInternalMetricName(semanticSchemaResp.getModelResps().get(0).getBizName()));
        } else {
            queryStructReq.setAggregators(
                metricTable.getMetrics().stream().map(m -> new Aggregator(m, AggOperatorEnum.UNKNOWN)).collect(Collectors.toList()));
        }
        AggOption aggOption = getAggOption(querySQLReq, metricSchemas);
        metricTable.setAggOption(aggOption);
        List<MetricTable> tables = new ArrayList<>();
        tables.add(metricTable);
        //4.build ParseSqlReq
        DataSetQueryParam result = new DataSetQueryParam();
        BeanUtils.copyProperties(querySQLReq, result);

        result.setTables(tables);
        DatabaseResp database = semanticSchemaResp.getDatabaseResp();
        if (!sqlGenerateUtils.isSupportWith(EngineType.fromString(database.getType().toUpperCase()), database.getVersion())) {
            result.setSupportWith(false);
            result.setWithAlias(false);
        }
        // 5.将派生指标转化为原始的字段
        Map<String, MeasureParam> derivedMetrics = generateDerivedMetric(semanticSchemaResp, aggOption, metricTable);

        //6.physicalSql by ParseSqlReq
        queryStructReq.setDateInfo(queryStructUtils.getDateConfBySql(querySQLReq.getSql()));
        queryStructReq.setDataSetId(querySQLReq.getDataSetId());
        queryStructReq.setQueryType(getQueryType(aggOption));
        log.info("QueryReqConverter queryStructReq[{}]", queryStructReq);
        QueryParam queryParam = new QueryParam();
        convert(queryStructReq, queryParam);
        QueryStatement queryStatement = new QueryStatement();
        // 7.derivedMetrics
        MetricQueryParam metricQueryParam = new MetricQueryParam();
        metricQueryParam.setDerivedMetrics(derivedMetrics);
        metricQueryParam.setRollOverMap(querySQLReq.getRollOverMap());
        queryStatement.setMetricQueryParam(metricQueryParam);

        queryStatement.setQueryParam(queryParam);
        queryStatement.setDataSetQueryParam(result);
        queryStatement.setIsS2SQL(true);
        queryStatement.setMinMaxTime(queryStructUtils.getBeginEndTime(queryStructReq));
        queryStatement.setDataSetId(querySQLReq.getDataSetId());
        queryStatement.setLimit(querySQLReq.getLimit());

        return queryStatement;
    }

    public void convert(QueryStructReq queryStructReq, QueryParam queryParam) {
        BeanUtils.copyProperties(queryStructReq, queryParam);
        queryParam.setOrders(queryStructReq.getOrders());
        queryParam.setMetrics(queryStructReq.getMetrics());
        queryParam.setGroups(queryStructReq.getGroups());
        queryParam.setAggregators(queryStructReq.getAggregators());
        queryParam.setModelIds(queryStructReq.getModelIdSet());
    }

    private AggOption getAggOption(QuerySqlReq databaseReq, List<MetricSchemaResp> metricSchemas) {
        // if there is no group by in S2SQL,set MetricTable's aggOption to "NATIVE"
        // if there is count() in S2SQL,set MetricTable's aggOption to "NATIVE"
        String sql = databaseReq.getSql();
        if (!SqlSelectFunctionHelper.hasAggregateFunction(sql) || SqlSelectFunctionHelper.hasFunction(sql, "count")
            || SqlSelectFunctionHelper.hasFunction(sql, "count_distinct")) {
            return AggOption.OUTER;
        }
        if (databaseReq.isInnerLayerNative()) {
            return AggOption.NATIVE;
        }
        if (SqlSelectHelper.hasSubSelect(sql) || SqlSelectHelper.hasWith(sql) || SqlSelectHelper.hasGroupBy(sql)) {
            return AggOption.OUTER;
        }
        long defaultAggNullCnt =
            metricSchemas.stream().filter(m -> Objects.isNull(m.getDefaultAgg()) || StringUtils.isBlank(m.getDefaultAgg())).count();
        if (defaultAggNullCnt > 0) {
            log.info("getAggOption find null defaultAgg metric set to NATIVE");
            return AggOption.OUTER;
        }
        return AggOption.DEFAULT;
    }

    public void convertNameToBizName(QuerySqlReq querySqlReq, SemanticSchemaResp semanticSchemaResp) {
        Map<String, String> fieldNameToBizNameMap = getFieldNameToBizNameMap(semanticSchemaResp);
        String sql = querySqlReq.getSql();
        log.info("dataSetId:{},convert name to bizName before:{}", querySqlReq.getDataSetId(), sql);
        String replaceFields = SqlReplaceHelper.replaceFields(sql, fieldNameToBizNameMap, true);
        log.info("dataSetId:{},convert name to bizName after:{}", querySqlReq.getDataSetId(), replaceFields);
        querySqlReq.setSql(replaceFields);
    }

    private Set<String> getDimensions(SemanticSchemaResp semanticSchemaResp, List<String> allFields) {
        Map<String, String> dimensionLowerToNameMap = semanticSchemaResp.getDimensions().stream().collect(
            Collectors.toMap(entry -> entry.getBizName().toLowerCase(), SchemaItem::getBizName, (k1, k2) -> k1));
        Map<String, String> internalLowerToNameMap =
            QueryStructUtils.internalCols.stream().collect(Collectors.toMap(String::toLowerCase, a -> a));
        dimensionLowerToNameMap.putAll(internalLowerToNameMap);
        return allFields.stream().filter(entry -> dimensionLowerToNameMap.containsKey(entry.toLowerCase()))
                        .map(entry -> dimensionLowerToNameMap.get(entry.toLowerCase())).collect(Collectors.toSet());
    }

    private List<MetricSchemaResp> getMetrics(SemanticSchemaResp semanticSchemaResp, List<String> allFields) {
        Map<String, MetricSchemaResp> metricLowerToNameMap =
            semanticSchemaResp.getMetrics().stream().collect(Collectors.toMap(entry -> entry.getBizName().toLowerCase(), entry -> entry));
        return allFields.stream().filter(entry -> metricLowerToNameMap.containsKey(entry.toLowerCase()))
                        .map(entry -> metricLowerToNameMap.get(entry.toLowerCase())).collect(Collectors.toList());
    }

    private void functionNameCorrector(QuerySqlReq databaseReq, SemanticSchemaResp semanticSchemaResp) {
        DatabaseResp database = semanticSchemaResp.getDatabaseResp();
        if (Objects.isNull(database) || Objects.isNull(database.getType())) {
            return;
        }
        String type = database.getType();
        DbAdaptor engineAdaptor = DbAdaptorFactory.getEngineAdaptor(type.toLowerCase());
        if (Objects.nonNull(engineAdaptor)) {
            String functionNameCorrector = engineAdaptor.functionNameCorrector(databaseReq.getSql());
            databaseReq.setSql(functionNameCorrector);
        }
    }

    // 将数据库中的字段转为语义中的名称
    public Map<String, String> getBizNameToNameMap(SemanticSchemaResp semanticSchemaResp) {
        // support bizName to name
        Map<String, String> dimensionResults = semanticSchemaResp.getDimensions().stream().flatMap(
                                                                     entry -> getPairStream(entry.getAlias(), entry.getBizName(), entry.getName()))
                                                                 .collect(Collectors.toMap(Pair::getLeft, Pair::getRight, (k1, k2) -> k1));

        Map<String, String> metricResults =
            semanticSchemaResp.getMetrics().stream().flatMap(entry -> getPairStream(entry.getAlias(), entry.getBizName(), entry.getName()))
                              .collect(Collectors.toMap(Pair::getLeft, Pair::getRight, (k1, k2) -> k1));

        dimensionResults.putAll(TimeDimensionEnum.getNameToChNameMap());
        dimensionResults.putAll(metricResults);
        return dimensionResults;
    }

    // 将语义名称转为数据库中的名称
    protected Map<String, String> getFieldNameToBizNameMap(SemanticSchemaResp semanticSchemaResp) {
        // support fieldName and field alias to bizName
        Map<String, String> dimensionResults = semanticSchemaResp
            .getDimensions().stream().flatMap(
                entry -> getPairStream(entry.getDescription(), entry.getName(), entry.getBizName()))
            .collect(Collectors.toMap(Pair::getLeft, Pair::getRight, (k1, k2) -> k1));

        Map<String, String> metricResults =
            semanticSchemaResp.getMetrics().stream().flatMap(entry -> getPairStream(entry.getAlias(), entry.getName(), entry.getBizName()))
                              .collect(Collectors.toMap(Pair::getLeft, Pair::getRight, (k1, k2) -> k1));

        dimensionResults.putAll(TimeDimensionEnum.getChNameToNameMap());
        dimensionResults.putAll(TimeDimensionEnum.getNameToNameMap());
        dimensionResults.putAll(metricResults);
        return dimensionResults;
    }

    private Stream<Pair<String, String>> getPairStream(String aliasStr, String name, String bizName) {
        Set<Pair<String, String>> elements = new HashSet<>();
        elements.add(Pair.of(name, bizName));
        if (StringUtils.isNotBlank(aliasStr)) {
            List<String> aliasList = SchemaItem.getAliasList(aliasStr);
            for (String alias : aliasList) {
                elements.add(Pair.of(alias, bizName));
            }
        }
        return elements.stream();
    }

    public void correctTableName(QuerySqlReq querySqlReq) {
        String sql = querySqlReq.getSql();
        String tableName = Constants.TABLE_PREFIX + querySqlReq.getDataSetId();
        if (querySqlReq.getDataSetId() == null) {
            // 无datasetId时
            tableName = Constants.TABLE_PREFIX + "tmp";
        }
        sql = SqlReplaceHelper.replaceTable(sql, tableName);
        log.info("correctTableName after:{}", sql);
        querySqlReq.setSql(sql);
    }

    private QueryType getQueryType(AggOption aggOption) {
        boolean isAgg = AggOption.isAgg(aggOption);
        QueryType queryType = QueryType.DETAIL;
        if (isAgg) {
            queryType = QueryType.METRIC;
        }
        return queryType;
    }

    private Map<String, String> generateCompositMetric(SemanticSchemaResp semanticSchemaResp, MetricTable metricTable) {
        Map<String, String> compositMetrics = new HashMap<>(2);
        List<MetricResp> allMetrics = semanticSchemaResp.getAllMetrics();
        // 处理复合指标
        List<String> metricBizNames = new ArrayList<>();
        metricBizNames.addAll(metricTable.getMetrics());

        metricBizNames.stream().forEach(bizName -> {
            MetricResp metricResp = semanticSchemaResp.getMetric(bizName);
            if (metricResp != null && metricResp.getType().equalsIgnoreCase(MetricType.COMPOSITE.name())) {
                String expression = sqlGenerateUtils.getExpr(metricResp);
                Set<String> fields = SqlSelectHelper.getColumnFromExpr(expression);
                if (fields.size() > 1) {
                    // 针对复合指标需要单独处理
                    for (String field : fields) {
                        MetricResp metric =
                            allMetrics.stream().filter(m -> m.getBizName().equalsIgnoreCase(field)).findFirst().orElse(null);
                        if (metric != null) {
                            String replace = sqlGenerateUtils.getExpr(metric);
                            // 引用的指标也加入
                            if (!metricTable.getMetrics().contains(replace)) {
                                // 自定义
                                if (metric.getMetricDefineType() == MetricDefineType.CUSTOM) {
                                    FieldParam f = metric.getMetricDefineByFieldParams().getFields().stream().findFirst().orElse(null);
                                    if (f != null) {
                                        metricTable.getMetrics().add(f.getFieldName());
                                    }
                                } else {
                                    metricTable.getMetrics().add(replace);
                                }
                            }
                            // 保持指标的聚合方式不变
                            if (StringUtils.isNotEmpty(metric.getOperations())) {
                                if (AggOperatorEnum.COUNT_DISTINCT.name().equalsIgnoreCase(metric.getOperations())) {
                                    replace = String.format("count(distinct %s)", replace);
                                } else {
                                    replace = metric.getOperations() + " ( " + replace + " ) ";
                                }
                            }
                            expression = expression.replace(field, replace);
                        }
                    }
                    compositMetrics.put(metricResp.getBizName(), expression);
                }
            }
        });

        // 清除复合指标名称
        if (compositMetrics != null) {
            metricTable.getMetrics().removeAll(compositMetrics.keySet());
        }

        return compositMetrics;
    }

    private Map<String, MeasureParam> generateDerivedMetric(SemanticSchemaResp semanticSchemaResp, AggOption aggOption,
                                                            MetricTable metricTable) {
        Map<String, MeasureParam> derivedMetrics = new HashMap<>(4);
        Set<String> measures = new HashSet<>();
        Map<String, String> replaces = new HashMap<>();
        generateDerivedMetric(semanticSchemaResp, aggOption, metricTable.getMetrics(), metricTable.getDimensions(), measures, replaces,
            derivedMetrics);
        measures.stream().forEach(measure -> {
            if (!metricTable.getMetrics().contains(measure)) {
                metricTable.getMetrics().add(measure);
            }
        });
        // 清除派生指标名称
        if (derivedMetrics != null) {
            metricTable.getMetrics().removeAll(derivedMetrics.keySet());
        }
        return derivedMetrics;
    }

    private void generateDerivedMetric(SemanticSchemaResp semanticSchemaResp, AggOption aggOption, List<String> metrics,
                                       List<String> dimensions, Set<String> measures, Map<String, String> replaces,
                                       Map<String, MeasureParam> derivedMetrics) {
        List<MetricSchemaResp> metricResps =
            semanticSchemaResp.getMetrics().stream().filter(m -> metrics.contains(m.getBizName())).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(metricResps)) {
            return;
        }
        log.info("begin to generateDerivedMetric {} [{}]", aggOption, metrics);
        for (MetricResp metricResp : metricResps) {
            if (metrics.contains(metricResp.getBizName())) {
                if (MetricType.DERIVED.name().equals(metricResp.getType())) {
                    String expr = sqlGenerateUtils.getExpr(metricResp);
                    replaces.put(metricResp.getBizName(), expr);
                    List<String> measureNames = getMeasureParam(metricResp, derivedMetrics);
                    if (!CollectionUtils.isEmpty(measureNames)) {
                        measures.addAll(measureNames);
                    }
                    log.info("derived metric {}->{}", metricResp.getBizName(), expr);
                } else {
                    measures.add(metricResp.getBizName());
                }
            }
        }
    }

    private List<String> getMeasureParam(MetricResp metricResp, Map<String, MeasureParam> derivedMetrics) {
        List<String> measures = new ArrayList<>();
        final String constraint = metricResp.getBizConditionResp() != null ? metricResp.getBizConditionResp().getDefaultConfig() : null;
        if (MetricDefineType.CUSTOM.equals(metricResp.getMetricDefineType())) {
            // 自定义类型
            log.info("derivedMetrics {} is custom type", metricResp.getBizName());
            MeasureParam measureParam = new MeasureParam();
            measureParam.setBizName(sqlGenerateUtils.getExpr(metricResp));
            measureParam.setConstraint(constraint);
            derivedMetrics.put(metricResp.getBizName(), measureParam);
            metricResp.getMetricDefineByFieldParams().getFields().stream().forEach(f -> {
                measures.add(f.getFieldName());
            });
        }
        if (MetricDefineType.MEASURE.equals(metricResp.getMetricDefineType())) {
            // measures只有一个
            log.info("derivedMetrics {} is measure type", metricResp.getBizName());
            metricResp.getMetricDefineByMeasureParams().getMeasures().stream().forEach(m -> {
                if (StringUtils.isEmpty(m.getConstraint())) {
                    m.setConstraint(constraint);
                }
                derivedMetrics.put(metricResp.getBizName(), m);
                measures.add(m.getBizName());
            });
        }
        if (MetricDefineType.METRIC.equals(metricResp.getMetricDefineType())) {
            // 也必须是一个
            log.info("derivedMetrics {} is metric type", metricResp.getBizName());
            metricResp.getMetricDefineByMetricParams().getMetrics().stream().forEach(m -> {
                MeasureParam param = new MeasureParam();
                param.setConstraint(constraint);
                param.setBizName(m.getBizName());
                derivedMetrics.put(metricResp.getBizName(), param);
            });
        }
        return measures;
    }

    public QueryMultiSqlReq buildMultiStructReq(String sql, Set<Long> modelIds) throws Exception {
        QueryMultiSqlReq queryMultiSqlReq = new QueryMultiSqlReq();
        List<String> subQueries = SqlSelectHelper.splitSelectFields(sql);
        DataSetService dataSetService = ContextUtils.getBean(DataSetService.class);
        Long dataSetId = null;
        if (CollectionUtils.isEmpty(modelIds)) {
            dataSetId = dataSetService.getDataSetIdFromSql(sql, User.getFakeUser());
        }
        List<QuerySqlReq> querySqlReqs = new ArrayList<>();
        for (String subQuery : subQueries) {
            QuerySqlReq querySqlReq = new QuerySqlReq();
            querySqlReq.setDataSetId(dataSetId);
            querySqlReq.setModelIds(modelIds);
            // 仅针对指标查询时
            if (dataSetId == null || dataSetId < 1L) {
                querySqlReq.setLimit(-1);
            }
            querySqlReq.setSql(subQuery);
            querySqlReqs.add(querySqlReq);
        }
        queryMultiSqlReq.setQuerySqlReqs(querySqlReqs);
        return queryMultiSqlReq;
    }
}
