package com.purui.prims.headless.core.translator.calcite.sql;


import com.purui.prims.common.pojo.Constants;
import com.purui.prims.common.pojo.enums.AggOperatorEnum;
import com.purui.prims.common.util.DatePeriodEnum;
import com.purui.prims.headless.api.pojo.MeasureParam;
import com.purui.prims.headless.api.pojo.enums.EngineType;
import com.purui.prims.headless.core.pojo.MetricQueryParam;
import com.purui.prims.headless.core.translator.calcite.s2sql.DataSource;
import com.purui.prims.headless.core.translator.calcite.s2sql.Dimension;
import com.purui.prims.headless.core.translator.calcite.s2sql.Identify;
import com.purui.prims.headless.core.translator.calcite.s2sql.Measure;
import com.purui.prims.headless.core.translator.calcite.s2sql.Metric;
import com.purui.prims.headless.core.translator.calcite.s2sql.MetricTypeParams;
import com.purui.prims.headless.core.translator.calcite.schema.SemanticSchema;
import com.purui.prims.headless.core.translator.calcite.sql.node.MeasureNode;
import com.purui.prims.headless.core.translator.calcite.sql.node.MetricNode;
import com.purui.prims.headless.core.translator.calcite.sql.node.SemanticNode;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.validate.SqlValidatorScope;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import java.util.stream.Collectors;

/**
 * process TableView
 */
@Data
@Slf4j
public abstract class Renderer {

    protected TableView tableView = new TableView();

    public static Optional<Dimension> getDimensionByName(String name, DataSource datasource) {
        return datasource.getDimensions().stream().filter(d -> d.getName().equalsIgnoreCase(name)).findFirst();
    }

    public static Optional<Measure> getMeasureByName(String name, DataSource datasource) {
        return datasource.getMeasures().stream().filter(mm -> mm.getName().equalsIgnoreCase(name)).findFirst();
    }

    public static Optional<Metric> getMetricByName(String name, SemanticSchema schema) {
        Optional<Metric> metric = schema.getMetrics().stream().filter(m -> m.getName().equalsIgnoreCase(name))
                                        .findFirst();
        return metric;
    }

    public static Optional<Identify> getIdentifyByName(String name, DataSource datasource) {
        return datasource.getIdentifiers().stream().filter(i -> i.getName().equalsIgnoreCase(name)).findFirst();
    }

    public static void buildRollOverNode(MetricNode metricNode, DataSource datasource, SqlValidatorScope scope,
                                         SemanticSchema schema, boolean nonAgg, String metricName, String metricExpr,
                                         List<String> dimensions, TableView output) throws Exception {
        EngineType engineType = EngineType.fromString(datasource.getType());
        Optional<Metric> metricOpt = getMetricByName(metricName, schema);
        // 如果环比或同比
        if (metricOpt.isPresent() && StringUtils.isNotEmpty(metricOpt.get().getPeriod()) && metricOpt.get().getMetricTypeParams()
                                                                                                     .isRollOver()) {
            String period = metricOpt.get().getPeriod();
            String finalExpression = metricOpt.get().getExpr();
            Metric metricRoll = new Metric();
            boolean isOver = false;
            String expr = "";
            int counter = 1; //环比
            if (AggOperatorEnum.RATIO_ROLL.name().equalsIgnoreCase(metricOpt.get().getMetricTypeParams().getOperations())) {
                metricRoll.setName(String.format("%s%s", metricName, Constants.ROLL_SUFFIX));
            } else {
                metricRoll.setName(String.format("%s%s", metricName, Constants.OVER_SUFFIX));
                isOver = true;
            }
            metricRoll.setMetricTypeParams(new MetricTypeParams());
            if (DatePeriodEnum.DAY.name().equalsIgnoreCase(period)) {
                // RatioOverType.DAY_ON_DAY; 与昨天(环比)
                // RatioOverType.YEAR_ON_DAY 与去年(同比)
            } else if (DatePeriodEnum.MONTH.name().equalsIgnoreCase(period)) {
                // RatioOverType.MONTH_ON_MONTH 与上个月(环比)
                // RatioOverType.YEAR_ON_MONTH 与上一年(同比)
                if (isOver) { // 同比
                    counter = 12;
                }
            } else if (DatePeriodEnum.WEEK.name().equalsIgnoreCase(period)) {
                // RatioOverType.WEEK_ON_WEEK 与上周(环比)
                // RatioOverType.MONTH_ON_WEEK 与上月(环比)
                if (isOver) { // 同比
                    counter = 4;
                }
            }
            String timeDimension = dimensions.stream().map(d -> d.toLowerCase()).filter(d -> d.toLowerCase().contains("pr_biz"))
                                             .collect(Collectors.joining(","));
            String dimension = dimensions.stream().map(d -> d.toLowerCase()).filter(d -> !d.toLowerCase().contains("pr_biz"))
                                         .collect(Collectors.joining(","));
            // 算子应由指标决定
            String function = metricNode.getAggFunction().get(metricExpr);
            if (StringUtils.isEmpty(function)) {
                function = AggOperatorEnum.SUM.getOperator();
            }
            expr =
                String.format("LAG(%s(%s),%s,0) over (partition by %s order by %s)", function, finalExpression, counter, dimension,
                    timeDimension);
            metricRoll.getMetricTypeParams().setExpr(expr);
            SqlNode metricNodeRoll = MetricNode.build(metricRoll, scope, engineType);
            // 先修改源指标
            SqlNode metricOrginalNode = metricNode.getAggNode().get(metricExpr);
            if (metricOrginalNode != null) {
                String exprOld = metricOrginalNode.toString();
                String exprNew = String.format("%s(%s)", function, exprOld);
                SqlNode metricNodeNew = MetricNode.buildAs(exprOld, MetricNode.parse(exprNew, scope, engineType));
                metricNode.getAggNode().put(metricExpr, metricNodeNew);
            }
            metricNode.getAggNode().put(metricRoll.getName(), metricNodeRoll);

            // 增加group by
            if (!CollectionUtils.isEmpty(dimensions) && output != null) {
                dimensions.stream().forEach(d -> {
                    try {
                        output.getDimension().add(SemanticNode.parse(d, scope, engineType));
                    } catch (Exception e) {
                        log.info("add group by error", e);
                    }
                });
            }
        }
    }

    public static MetricNode buildMetricNode(String metric, DataSource datasource, SqlValidatorScope scope,
                                             SemanticSchema schema, boolean nonAgg, String alias,
                                             Map<String, MeasureParam> derivedMetrics, List<String> dimensions, TableView output)
        throws Exception {
        Optional<Metric> metricOpt = getMetricByName(metric, schema);
        MetricNode metricNode = new MetricNode();
        EngineType engineType = EngineType.fromString(datasource.getType());
        String metricExpr = null;
        // 如果是指标
        if (metricOpt.isPresent() && metricOpt.get().getMetricTypeParams().getMeasures() != null) {
            metricNode.setMetric(metricOpt.get());
            for (Measure m : metricOpt.get().getMetricTypeParams().getMeasures()) {
                // 获取一次最新的数据
                Optional<Measure> measure = getMeasureByName(m.getName(), datasource);
                Measure newMeasure = m;
                if (measure.isPresent()) {
                    newMeasure = measure.get();
                }
                metricNode.getNonAggNode()
                          .put(newMeasure.getName(),
                              MeasureNode.buildNonAgg(alias, newMeasure, scope, engineType));
                if (newMeasure.getName().contains(newMeasure.getExpr())) {
                    // 兼容字段和表达式
                    newMeasure.setExpr(newMeasure.getName());
                }
                metricNode.getAggNode()
                          .put(newMeasure.getName(),
                              MeasureNode.buildAgg(newMeasure, nonAgg, scope, engineType));
                metricNode.getAggFunction().put(newMeasure.getName(), newMeasure.getAgg());
                if (newMeasure.getConstraint() != null && !newMeasure.getConstraint().isEmpty()) {
                    metricNode.getMeasureFilter()
                              .put(newMeasure.getName(), SemanticNode.parse(newMeasure.getConstraint(), scope, engineType));
                }
                metricExpr = newMeasure.getName();
            }

            // 业务限定的条件处理
            if (StringUtils.isNotEmpty(metricOpt.get().getBizCondition())) {
                metricNode.getMeasureFilter()
                          .put(metricOpt.get().getName(),
                              SemanticNode.parse(metricOpt.get().getBizCondition(), scope, engineType));
            }
        }
        // 如果是度量
        Optional<Measure> measure = getMeasureByName(metric, datasource);
        if (measure.isPresent()) {
            String measureName = measure.get().getName();
            metricNode.getNonAggNode()
                      .put(measureName, MeasureNode.buildNonAgg(alias, measure.get(), scope, engineType));
            metricNode.getAggNode()
                      .put(measureName, MeasureNode.buildAgg(measure.get(), nonAgg, scope, engineType));
            metricNode.getAggFunction()
                      .put(measureName, measure.get().getAgg());

            metricExpr = measureName;
            // 转换限定where条件
            if (measure.get().getConstraint() != null && !measure.get().getConstraint().isEmpty()) {
                metricNode.getMeasureFilter().put(measureName,
                    SemanticNode.parse(measure.get().getConstraint(), scope, engineType));
            }
        }
        // 自定义表达式 CUSTOM
        if (metricOpt.isPresent() && metricOpt.get().getMetricTypeParams().getFields() != null) {

            List<Measure> fields = metricOpt.get().getMetricTypeParams().getFields();
            // 多个表达式
            for (Measure f : fields) {
                metricNode.getNonAggNode()
                          .put(f.getName(), MeasureNode.buildNonAgg(alias, f, scope, engineType));
                metricNode.getAggNode()
                          .put(f.getName(), MeasureNode.buildAgg(f, nonAgg, scope, engineType));
            }
            metricExpr = metricOpt.get().getName();
        }

        // 派生指标中使用的衍生指标转换为where条件
        if (derivedMetrics != null) {
            for (String k : derivedMetrics.keySet()) {
                MeasureParam v = derivedMetrics.get(k);
                metricNode.getMeasureFilter().put(v.getBizName(),
                    SemanticNode.parse(v.getConstraint(), scope, engineType));
            }
        }
        // 同环比
        buildRollOverNode(metricNode, datasource, scope, schema, nonAgg, metric, metricExpr, dimensions, output);

        return metricNode;
    }

    public static List<String> uniqList(List<String> list) {
        // 去重,不区分大小写 2024-08-09
        Set<String> tmp = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
        list.stream().forEach(s -> tmp.add(s));
        return tmp.stream().collect(Collectors.toList());
    }

    public void setTable(SqlNode table) {
        tableView.setTable(table);
    }

    public SqlNode builder() {
        return tableView.build();
    }

    public SqlNode builderAs(String alias) throws Exception {
        return SemanticNode.buildAs(alias, tableView.build());
    }

    public abstract void render(MetricQueryParam metricCommand, List<DataSource> dataSources, SqlValidatorScope scope,
                                SemanticSchema schema, boolean nonAgg) throws Exception;
}
