package com.ruoshui.bigdata.tool.flinkx;


import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.ruoshui.bigdata.dto.*;
import com.ruoshui.bigdata.entity.JobDatasource;
import com.ruoshui.bigdata.tool.flinkx.reader.*;
import com.ruoshui.bigdata.tool.flinkx.writer.*;
import com.ruoshui.bigdata.tool.pojo.FlinkxHbasePojo;
import com.ruoshui.bigdata.tool.pojo.FlinkxHivePojo;
import com.ruoshui.bigdata.tool.pojo.FlinkxMongoDBPojo;
import com.ruoshui.bigdata.tool.pojo.FlinkxRdbmsPojo;
import com.ruoshui.bigdata.util.JdbcConstants;
import com.ruoshui.core.util.Constants;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * JSON的构建类
 */
@Data
public class SeatunnelJsonHelper implements JsonInterface {

    /**
     * 读取的表，根据flinkx示例，支持多个表（先不考虑，后面再去实现， 这里先用list保存吧）
     * <p>
     * 目的表的表名称。支持写入一个或者多个表。当配置为多张表时，必须确保所有表结构保持一致
     */
    private List<String> readerTables;
    /**
     * 读取的字段列表
     */
    private List<String> readerColumns;
    /**
     * reader jdbc 数据源
     */
    private JobDatasource readerDatasource;
    /**
     * writer jdbc 数据源
     */
    private JobDatasource writerDatasource;
    /**
     * 写入的表
     */
    private List<String> writerTables;
    /**
     * 写入的字段列表
     */
    private List<String> writerColumns;

    private Map<String, Object> buildReader;

    private Map<String, Object> buildWriter;

    private BaseFlinkxPlugin readerPlugin;

    private BaseFlinkxPlugin writerPlugin;

    private HiveReaderDto hiveReaderDto;

    private HiveWriterDto hiveWriterDto;

    private HbaseReaderDto hbaseReaderDto;

    private HbaseWriterDto hbaseWriterDto;

    private RdbmsReaderDto rdbmsReaderDto;

    private RdbmsWriterDto rdbmsWriterDto;

    private MongoDBReaderDto mongoDBReaderDto;

    private MongoDBWriterDto mongoDBWriterDto;

    private ClickhouseReaderDto clickhouseReaderDto;

    private ClickhouseWriterDto clickhouseWriterDto;


    //用于保存额外参数
    private Map<String, Object> extraParams = Maps.newHashMap();

    public void initReader(JsonBuildDto flinkxJsonDto, JobDatasource readerDatasource) {

        this.readerDatasource = readerDatasource;
        this.readerTables = flinkxJsonDto.getReaderTables();
        this.readerColumns = flinkxJsonDto.getReaderColumns();
        this.hiveReaderDto = flinkxJsonDto.getHiveReader();
        this.rdbmsReaderDto = flinkxJsonDto.getRdbmsReader();
        this.hbaseReaderDto = flinkxJsonDto.getHbaseReader();
        this.clickhouseReaderDto = flinkxJsonDto.getClickhouseReader();
        // reader 插件
        String datasource = readerDatasource.getDatasource();

//        this.readerColumns = convertKeywordsColumns(datasource, this.readerColumns);
        if (JdbcConstants.MYSQL.equals(datasource)) {
            readerPlugin = new MysqlReader();
            buildReader = buildReader();
        } else if (JdbcConstants.ORACLE.equals(datasource)) {
            readerPlugin = new OracleReader();
            buildReader = buildReader();
        } else if (JdbcConstants.HANA.equals(datasource)) {
            readerPlugin = new HanaReader();
            buildReader = buildReader();
        } else if (JdbcConstants.SQL_SERVER.equals(datasource)) {
            readerPlugin = new SqlServerReader();
            buildReader = buildReader();
        } else if (JdbcConstants.POSTGRESQL.equals(datasource)) {
            readerPlugin = new PostgresqlReader();
            buildReader = buildReader();
        } else if (JdbcConstants.CLICKHOUSE.equals(datasource)) {
            readerPlugin = new ClickHouseReader();
            buildReader = buildReader();
        } else if (JdbcConstants.HIVE.equals(datasource)) {
            readerPlugin = new HiveReader();
            buildReader = buildHiveReader();
        } else if (JdbcConstants.HBASE.equals(datasource)) {
            readerPlugin = new HBaseReader();
            buildReader = buildHBaseReader();
        } else if (JdbcConstants.MONGODB.equals(datasource)) {
            readerPlugin = new MongoDBReader();
            buildReader = buildMongoDBReader();
        }
    }

    public void initWriter(JsonBuildDto flinkxJsonDto, JobDatasource readerDatasource) {
        this.writerDatasource = readerDatasource;
        this.writerTables = flinkxJsonDto.getWriterTables();
        this.writerColumns = flinkxJsonDto.getWriterColumns();
        this.hiveWriterDto = flinkxJsonDto.getHiveWriter();
        this.rdbmsWriterDto = flinkxJsonDto.getRdbmsWriter();
        this.hbaseWriterDto = flinkxJsonDto.getHbaseWriter();
        this.mongoDBWriterDto = flinkxJsonDto.getMongoDBWriter();
        // writer
        String datasource = readerDatasource.getDatasource();
//        this.writerColumns = convertKeywordsColumns(datasource, this.writerColumns);
        if (JdbcConstants.MYSQL.equals(datasource)) {
            writerPlugin = new MysqlWriter();
            buildWriter = this.buildWriter();
        } else if (JdbcConstants.ORACLE.equals(datasource)) {
            writerPlugin = new OraclelWriter();
            buildWriter = this.buildWriter();
        } else if (JdbcConstants.HANA.equals(datasource)) {
            writerPlugin = new HanaWriter();
            buildWriter = this.buildWriter();
        } else if (JdbcConstants.SQL_SERVER.equals(datasource)) {
            writerPlugin = new SqlServerlWriter();
            buildWriter = this.buildWriter();
        } else if (JdbcConstants.POSTGRESQL.equals(datasource)) {
            writerPlugin = new PostgresqllWriter();
            buildWriter = this.buildWriter();
        } else if (JdbcConstants.CLICKHOUSE.equals(datasource)) {
            writerPlugin = new ClickHouseWriter();
            buildWriter = buildWriter();
        } else if (JdbcConstants.HIVE.equals(datasource)) {
            writerPlugin = new HiveWriter();
            buildWriter = this.buildHiveWriter();
        } else if (JdbcConstants.HBASE.equals(datasource)) {
            writerPlugin = new HBaseWriter();
            buildWriter = this.buildHBaseWriter();
        } else if (JdbcConstants.MONGODB.equals(datasource)) {
            writerPlugin = new MongoDBWriter();
            buildWriter = this.buildMongoDBWriter();
        }
    }

    private List<String> convertKeywordsColumns(String datasource, List<String> columns) {
        if (columns == null) {
            return null;
        }

        List<String> toColumns = new ArrayList<>();
        columns.forEach(s -> {
            toColumns.add(doConvertKeywordsColumn(datasource, s));
        });
        return toColumns;
    }

    private String doConvertKeywordsColumn(String dbType, String column) {
        if (column == null) {
            return null;
        }

        column = column.trim();
        column = column.replace("[", "");
        column = column.replace("]", "");
        column = column.replace("`", "");
        column = column.replace("\"", "");
        column = column.replace("'", "");

        switch (dbType) {
            case JdbcConstants.MYSQL:
                return String.format("`%s`", column);
            case JdbcConstants.SQL_SERVER:
                return String.format("[%s]", column);
            case JdbcConstants.POSTGRESQL:
            case JdbcConstants.ORACLE:
                return String.format("\"%s\"", column);
            case JdbcConstants.HANA:
                return String.format("\"%s\"", column);
            default:
                return column;
        }
    }

    @Override
    public Map<String, Object> buildJob() {
        Map<String, Object> res = Maps.newLinkedHashMap();
        Map<String, Object> jobMap = Maps.newLinkedHashMap();
        jobMap.put("setting", buildSetting());
        jobMap.put("content", ImmutableList.of(buildContent()));
        res.put("job", jobMap);
        return res;
    }

    @Override
    public Map<String, Object> buildSetting() {
        Map<String, Object> res = Maps.newLinkedHashMap();
        Map<String, Object> speedMap = Maps.newLinkedHashMap();
        Map<String, Object> errorLimitMap = Maps.newLinkedHashMap();

        Map<String, Object> restoreMap = Maps.newLinkedHashMap();
        Map<String, Object> logMap = Maps.newLinkedHashMap();
        speedMap.putAll(ImmutableMap.of("channel", 1, "bytes", 0));
        errorLimitMap.putAll(ImmutableMap.of("record", 100));
        restoreMap.putAll(ImmutableMap.of("maxRowNumForCheckpoint", 0,"isRestore",false,"restoreColumnName","","restoreColumnIndex",0));
        logMap.putAll(ImmutableMap.of("isLogger", false,"level","debug","path","","pattern",""));
        res.put("speed", speedMap);
        res.put("errorLimit", errorLimitMap);
        res.put("restore",restoreMap);
        res.put("log",logMap);
        return res;
    }

    @Override
    public Map<String, Object> buildContent() {
        Map<String, Object> res = Maps.newLinkedHashMap();
        res.put("reader", this.buildReader);
        res.put("writer", this.buildWriter);
        return res;
    }

    @Override
    public Map<String, Object> buildReader() {
        FlinkxRdbmsPojo flinkxPluginPojo = new FlinkxRdbmsPojo();
        flinkxPluginPojo.setJobDatasource(readerDatasource);
        flinkxPluginPojo.setTables(readerTables);
		List<Map<String, Object>> columns = Lists.newArrayList();
		readerColumns.forEach(c -> {
			Map<String, Object> column = Maps.newLinkedHashMap();
			column.put("name", c.split(Constants.SPLIT_SCOLON)[0]);
//			column.put("index", Integer.parseInt(c.split(Constants.SPLIT_SCOLON)[0]));
//			column.put("type", c.split(Constants.SPLIT_SCOLON)[2]);
			columns.add(column);
		});
        flinkxPluginPojo.setRdbmsColumns(columns);
        flinkxPluginPojo.setSplitPk(rdbmsReaderDto.getReaderSplitPk());
        if (StringUtils.isNotBlank(rdbmsReaderDto.getQuerySql())) {
            flinkxPluginPojo.setQuerySql(rdbmsReaderDto.getQuerySql());
        }
        //where
        if (StringUtils.isNotBlank(rdbmsReaderDto.getWhereParams())) {
            flinkxPluginPojo.setWhereParam(rdbmsReaderDto.getWhereParams());
        }
        return readerPlugin.build(flinkxPluginPojo);
    }

    @Override
    public Map<String, Object> buildHiveReader() {
        FlinkxHivePojo flinkxHivePojo = new FlinkxHivePojo();
        flinkxHivePojo.setJdbcDatasource(readerDatasource);
        List<Map<String, Object>> columns = Lists.newArrayList();
        readerColumns.forEach(c -> {
            Map<String, Object> column = Maps.newLinkedHashMap();
            column.put("name", c.split(Constants.SPLIT_SCOLON)[1]);
            column.put("index", Integer.parseInt(c.split(Constants.SPLIT_SCOLON)[0]));
            column.put("type", c.split(Constants.SPLIT_SCOLON)[2]);
            columns.add(column);
        });
        flinkxHivePojo.setColumns(columns);
        flinkxHivePojo.setReaderDefaultFS(hiveReaderDto.getReaderDefaultFS());
        flinkxHivePojo.setReaderFieldDelimiter(hiveReaderDto.getReaderFieldDelimiter());
        flinkxHivePojo.setReaderFileType(hiveReaderDto.getReaderFileType());
        flinkxHivePojo.setReaderPath(hiveReaderDto.getReaderPath());
        flinkxHivePojo.setSkipHeader(hiveReaderDto.getReaderSkipHeader());
        return readerPlugin.buildHive(flinkxHivePojo);
    }

    @Override
    public Map<String, Object> buildHBaseReader() {
        FlinkxHbasePojo flinkxHbasePojo = new FlinkxHbasePojo();
        flinkxHbasePojo.setJdbcDatasource(readerDatasource);
        List<Map<String, Object>> columns = Lists.newArrayList();
        for (String readerColumn : readerColumns) {
            Map<String, Object> column = Maps.newLinkedHashMap();
            column.put("name", readerColumn);
            column.put("type", "string");
            columns.add(column);
        }
        flinkxHbasePojo.setColumns(columns);
        flinkxHbasePojo.setReaderHbaseConfig(readerDatasource.getZkAdress());
        String readerTable=!CollectionUtils.isEmpty(readerTables)?readerTables.get(0):Constants.STRING_BLANK;
        flinkxHbasePojo.setReaderTable(readerTable);
        flinkxHbasePojo.setReaderMode(hbaseReaderDto.getReaderMode());
        flinkxHbasePojo.setReaderRange(hbaseReaderDto.getReaderRange());
        return readerPlugin.buildHbase(flinkxHbasePojo);
    }

    @Override
    public Map<String, Object> buildMongoDBReader() {
        FlinkxMongoDBPojo flinkxMongoDBPojo = new FlinkxMongoDBPojo();
        flinkxMongoDBPojo.setJdbcDatasource(readerDatasource);
        List<Map<String, Object>> columns = Lists.newArrayList();
        buildColumns(readerColumns, columns);
        flinkxMongoDBPojo.setColumns(columns);
        flinkxMongoDBPojo.setAddress(readerDatasource.getJdbcUrl());
        flinkxMongoDBPojo.setDbName(readerDatasource.getDatabaseName());
        flinkxMongoDBPojo.setReaderTable(readerTables.get(0));
        return readerPlugin.buildMongoDB(flinkxMongoDBPojo);
    }


    @Override
    public Map<String, Object> buildWriter() {
        FlinkxRdbmsPojo flinkxPluginPojo = new FlinkxRdbmsPojo();
        flinkxPluginPojo.setJobDatasource(writerDatasource);
        flinkxPluginPojo.setTables(writerTables);
		List<Map<String, Object>> columns = Lists.newArrayList();
		writerColumns.forEach(c -> {
			Map<String, Object> column = Maps.newLinkedHashMap();
			column.put("name", c.split(Constants.SPLIT_SCOLON)[0]);
//			column.put("type", c.split(Constants.SPLIT_SCOLON)[2]);
			columns.add(column);
		});
        flinkxPluginPojo.setRdbmsColumns(columns);
        flinkxPluginPojo.setPreSql(rdbmsWriterDto.getPreSql());
        flinkxPluginPojo.setPostSql(rdbmsWriterDto.getPostSql());
        return writerPlugin.build(flinkxPluginPojo);
    }

    @Override
    public Map<String, Object> buildHiveWriter() {
        FlinkxHivePojo flinkxHivePojo = new FlinkxHivePojo();
        flinkxHivePojo.setJdbcDatasource(writerDatasource);
        List<Map<String, Object>> columns = Lists.newArrayList();
        writerColumns.forEach(c -> {
            Map<String, Object> column = Maps.newLinkedHashMap();
            column.put("name", c.split(Constants.SPLIT_SCOLON)[1]);
            column.put("type", c.split(Constants.SPLIT_SCOLON)[2]);
            columns.add(column);
        });
        flinkxHivePojo.setColumns(columns);
        flinkxHivePojo.setWriterDefaultFS(hiveWriterDto.getWriterDefaultFS());
        flinkxHivePojo.setWriteFieldDelimiter(hiveWriterDto.getWriteFieldDelimiter());
        flinkxHivePojo.setWriterFileType(hiveWriterDto.getWriterFileType());
        flinkxHivePojo.setWriterPath(hiveWriterDto.getWriterPath());
        flinkxHivePojo.setWriteMode(hiveWriterDto.getWriteMode());
        flinkxHivePojo.setWriterFileName(hiveWriterDto.getWriterFileName());
        return writerPlugin.buildHive(flinkxHivePojo);
    }

    @Override
    public Map<String, Object> buildHBaseWriter() {
        FlinkxHbasePojo flinkxHbasePojo = new FlinkxHbasePojo();
        flinkxHbasePojo.setJdbcDatasource(writerDatasource);
        List<Map<String, Object>> columns = Lists.newArrayList();
        for (int i = 0; i < writerColumns.size(); i++) {
            Map<String, Object> column = Maps.newLinkedHashMap();
            column.put("index", i + 1);
            column.put("name", writerColumns.get(i));
            column.put("type", "string");
            columns.add(column);
        }
        flinkxHbasePojo.setColumns(columns);
        flinkxHbasePojo.setWriterHbaseConfig(writerDatasource.getZkAdress());
        String writerTable=!CollectionUtils.isEmpty(writerTables)?writerTables.get(0):Constants.STRING_BLANK;
        flinkxHbasePojo.setWriterTable(writerTable);
        flinkxHbasePojo.setWriterVersionColumn(hbaseWriterDto.getWriterVersionColumn());
        flinkxHbasePojo.setWriterRowkeyColumn(hbaseWriterDto.getWriterRowkeyColumn());
        flinkxHbasePojo.setWriterMode(hbaseWriterDto.getWriterMode());
        return writerPlugin.buildHbase(flinkxHbasePojo);
    }


    @Override
    public Map<String, Object> buildMongoDBWriter() {
        FlinkxMongoDBPojo flinkxMongoDBPojo = new FlinkxMongoDBPojo();
        flinkxMongoDBPojo.setJdbcDatasource(writerDatasource);
        List<Map<String, Object>> columns = Lists.newArrayList();
        buildColumns(writerColumns, columns);
        flinkxMongoDBPojo.setColumns(columns);
        flinkxMongoDBPojo.setAddress(writerDatasource.getJdbcUrl());
        flinkxMongoDBPojo.setDbName(writerDatasource.getDatabaseName());
        flinkxMongoDBPojo.setWriterTable(writerTables.get(0));
        flinkxMongoDBPojo.setUpsertInfo(mongoDBWriterDto.getUpsertInfo());
        return writerPlugin.buildMongoDB(flinkxMongoDBPojo);
    }

    private void buildColumns(List<String> columns, List<Map<String, Object>> returnColumns) {
        columns.forEach(c -> {
            Map<String, Object> column = Maps.newLinkedHashMap();
            column.put("name", c.split(Constants.SPLIT_SCOLON)[0]);
            column.put("type", c.split(Constants.SPLIT_SCOLON)[1]);
            returnColumns.add(column);
        });
    }
}
