package com.highdatas.mdm.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.highdatas.mdm.entity.*;
import com.highdatas.mdm.mapper.TableInfoMapper;
import com.highdatas.mdm.pojo.CodeMsg;
import com.highdatas.mdm.pojo.DbAccessType;
import com.highdatas.mdm.pojo.Result;
import com.highdatas.mdm.pojo.SysAssembleLogStatus;
import com.highdatas.mdm.pojo.kettle.KettleDBTrans;
import com.highdatas.mdm.pojo.kettle.UnBigDataDataSourceInfo;
import com.highdatas.mdm.service.*;
import com.highdatas.mdm.util.Constant;
import com.highdatas.mdm.util.DbUtils;
import org.apache.commons.lang3.StringUtils;
import org.pentaho.di.core.KettleClientEnvironment;
import org.pentaho.di.core.database.Database;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.LoggingObjectInterface;
import org.pentaho.di.core.logging.LoggingObjectType;
import org.pentaho.di.core.logging.SimpleLoggingObject;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.dummytrans.DummyTransMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.awt.*;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author kimi
 * @description
 * @date 2020-02-15 14:26
 */

@Service
public class TaskServixceImpl  {
    @Autowired
    ISysAssembleService assembleService;
    @Autowired
    ISysAssembleDetailService assembleDetailService;
    @Autowired
    ISysAssembleLogService assembleLogService;
    @Autowired
    ISysDbtypeService dbtypeService;
    @Autowired
    IMenuMappingService menuMappingService;
    @Autowired
    UnBigDataDataSourceInfo unBigDataDataSourceInfo;
    @Autowired
    MasterDataService masterDataService;
    @Autowired
    TableInfoMapper infoMapper;

    @Override
    public Result run(String assembleId) {
        SysAssembleLog sysAssembleLog = new SysAssembleLog();
        sysAssembleLog.setId(DbUtils.getUUID());
        sysAssembleLog.setCreateTime(new Date());
        sysAssembleLog.setAssembleId(assembleId);
        sysAssembleLog.setStatus(SysAssembleLogStatus.working);
        sysAssembleLog.insert();


        if (StringUtils.isEmpty(assembleId)) {
            sysAssembleLog.setMessage("assembleId is null").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
            return Result.error(CodeMsg.ERROR_PARAMS_NOT_MATHED);
        }

        SysAssembleLog runtimeLog = assembleLogService.selectOne(new EntityWrapper<SysAssembleLog>().eq("assemble_id", assembleId).orderBy("create_time desc"));
        if (runtimeLog != null) {
            if (runtimeLog.getStatus().equals(SysAssembleLogStatus.working)) {
                sysAssembleLog.setMessage("当前正在有同一任务在运行").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
                return Result.error(new CodeMsg(3001, "当前正在有同一任务在运行"));
            }
        }

        KettleDBTrans kettleDBTrans = null;
        try {
            SysAssemble assemble = assembleService.selectById(assembleId);
            if (assemble == null) {
                //false
                sysAssembleLog.setMessage("assemble is not found").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
                return Result.error(CodeMsg.ERROR_PARAMS_NOT_MATHED);
            }
            String sqlStr = "", tableName = "";

            Database dsInput = null;
            Database dsOutput = null;
            TableInputMeta tableInputMeta = new TableInputMeta();
            TableOutputMeta talbeOutputMeta = new TableOutputMeta();
            TransMeta transMeta = new TransMeta();

            kettleDBTrans	= new KettleDBTrans();
            kettleDBTrans.setAssemble(assemble);
            kettleDBTrans.setDsInput(dsInput);
            kettleDBTrans.setDsOutput(dsOutput);
            kettleDBTrans.setTableInputMeta(tableInputMeta);
            kettleDBTrans.setTalbeOutputMeta(talbeOutputMeta);
            kettleDBTrans.setTransMeta(transMeta);

            boolean connected = kettleDBConn(kettleDBTrans);
            if (!connected){
                //false
                sysAssembleLog.setMessage("connection failed").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
            }
            String dbType = "";
//            assemble.getDbType();
            SysDbtype sysDbtype = dbtypeService.selectOne(new EntityWrapper<SysDbtype>().eq("name", dbType));
            DbAccessType type = sysDbtype.getType();
            switch (type) {
                case unRelational:
                    break;
                case relational:
                    extractDBData(kettleDBTrans, sysAssembleLog);

            }
        } catch (KettleException e) {
            e.printStackTrace();
            sysAssembleLog.setMessage(e.getMessage()).setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
        }finally {
            Database dsInput = kettleDBTrans.getDsInput();
            if (dsInput != null) {
                dsInput.disconnect();
            }
            Database dsOutput = kettleDBTrans.getDsOutput();
            if (dsOutput != null) {
                dsOutput.disconnect();
            }

        }
        return null;
    }

    private boolean kettleDBConn(KettleDBTrans kettleDBTrans) throws KettleException {
        SysAssemble assemble = kettleDBTrans.getAssemble();

        //初始化kettle环境
        KettleClientEnvironment.init();
        //导入数据的数据库连接
        DatabaseMeta dataMetaInput = null;
//        DatabaseMeta dataMetaInput = new DatabaseMeta("Input", assemble.getDbType(), "Native",assemble.getHost(), assemble.getSchema(),assemble.getPort().toString(),
//                assemble.getUserName(), assemble.getPassword());

        Boolean bigdata = true;
//        /assemble.getBigdata();
        DatabaseMeta dataMetaOutput = null;
        if (bigdata) {
            // 导入hbase数据库
        }else {
            // 导入mysql数据库
            //导出数据的数据库连接
            dataMetaOutput = new DatabaseMeta("Output", unBigDataDataSourceInfo.getDbType(), "Native", unBigDataDataSourceInfo.getDbHostName(), unBigDataDataSourceInfo.getDbName(), unBigDataDataSourceInfo.getDbPort(),
                    unBigDataDataSourceInfo.getUsername(), unBigDataDataSourceInfo.getPassword());
        }

        if (dataMetaOutput == null) {
            return false;

        }

        //导出数据库连接
        LoggingObjectInterface loggingObject = new SimpleLoggingObject("Database factory", LoggingObjectType.GENERAL, null );
        kettleDBTrans.setDsInput(new Database(loggingObject, dataMetaInput));
        kettleDBTrans.getDsInput().connect();
        Connection connInput = kettleDBTrans.getDsInput().getConnection();
        //导入数据库连接
        kettleDBTrans.setDsOutput(new Database(loggingObject, dataMetaOutput));
        kettleDBTrans.getDsOutput().connect();
        Connection connOutput = kettleDBTrans.getDsOutput().getConnection();

        kettleDBTrans.getTransMeta().setName("数据抽取");

        kettleDBTrans.getTransMeta().addDatabase(dataMetaInput);
        kettleDBTrans.getTransMeta().addDatabase(dataMetaOutput);

        //导出数据
        kettleDBTrans.getTableInputMeta().setDatabaseMeta(kettleDBTrans.getTransMeta().findDatabase("Input"));
        //导入数据
        kettleDBTrans.getTalbeOutputMeta().setDatabaseMeta(kettleDBTrans.getTransMeta().findDatabase("Output"));
        return true;
    }

    private void extractDBData(KettleDBTrans dbTrans, SysAssembleLog sysAssembleLog) {
        try {
            SysAssemble assemble = dbTrans.getAssemble();
            boolean checked = checkField(dbTrans);
            if (!checked){
                //false log
                sysAssembleLog.setMessage("source target field is not matched").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
                return;
            }

            List<SysAssembleDetail> detailList = assembleDetailService.selectList(new EntityWrapper<SysAssembleDetail>().eq("assemble_id", assemble.getId()));
            List<String> fieldList = detailList.stream().map(sysAssembleDetail -> sysAssembleDetail.getField()).collect(Collectors.toList());
            if (fieldList.contains(Constant.ID)) {
                 fieldList.remove(Constant.ID);
                 fieldList.add(MessageFormat.format(Constant.FieldAsAlias, Constant.ID, Constant.STD_ID));
                 if (!assemble.getBigdata()){
                     //MYSQL
                     fieldList.add(MessageFormat.format(Constant.MYSQL_UUID, Constant.ID));
                 }
                //fieldList = fieldList.stream().map(s -> ( s.equalsIgnoreCase(Constant.ID) ? MessageFormat.format(Constant.FieldAsAlias,Constant.ID, Constant.STD_ID) : s))
            }else {
                if (!assemble.getBigdata()) {
                    fieldList.add(MessageFormat.format(Constant.MYSQL_UUID, Constant.STD_ID));
                    fieldList.add(MessageFormat.format(Constant.MYSQL_UUID, Constant.ID));
                }
            }

            String fieldStr = fieldList.stream().collect(Collectors.joining(Constant.COMMA));
            String countSqlStr = MessageFormat.format(Constant.selectFieldSqlTemplate, fieldStr, assemble.getTableName());
            String selectSqlStr = MessageFormat.format(Constant.selectFieldSqlTemplate, fieldStr, assemble.getTableName());
            dbTrans.getTableInputMeta().setSQL(selectSqlStr);
            Database dsInput = dbTrans.getDsInput();
            Connection inputConnection = dsInput.getConnection();
            PreparedStatement cntStmt = inputConnection.prepareStatement(countSqlStr);
            cntStmt.executeQuery();
            ResultSet cntResultSet = cntStmt.executeQuery();
            Integer sorceCnt = cntResultSet.getInt(Constant.CNT);

            // StepMeta
            StepMeta inputStep = new StepMeta(Constant.STEP_READ_FROM_TABLE, dbTrans.getTableInputMeta());
            inputStep.setLocation(50, 50);
            inputStep.setDraw(true);
            dbTrans.getTransMeta().addStep(inputStep);

            String trmpTableName = getTempTableNameFromMenu(assemble.getMenuId());

            dbTrans.getTalbeOutputMeta().setTableName(trmpTableName);
            StepMeta insertUpdateStep = new StepMeta(Constant.STEP_INSERT_UPDATE, dbTrans.getTalbeOutputMeta());
            insertUpdateStep.setLocation(150, 50);
            insertUpdateStep.setDraw(true);
            dbTrans.getTransMeta().addStep(insertUpdateStep);

            DummyTransMeta dummyMeta = new DummyTransMeta();

            StepMeta dummyStep = new StepMeta(Constant.STEP_DUMMY, dummyMeta);
            dummyStep.setLocation(200, 50);
            dummyStep.setDraw(true);

            dbTrans.getTransMeta().addStep(dummyStep);

            //设置步骤直接的关系
            TransHopMeta hop = new TransHopMeta(inputStep, insertUpdateStep);
            dbTrans.getTransMeta().addTransHop(hop);
            TransHopMeta hop2 = new TransHopMeta(insertUpdateStep, dummyStep);
            dbTrans.getTransMeta().addTransHop(hop2);

            //开始执行
            Trans trans = new Trans(dbTrans.getTransMeta());
            trans.prepareExecution(null);
            trans.startThreads();
            trans.waitUntilFinished();

            int errors = trans.getErrors();
            if (errors == 0) {
                //搬运相关数据
                if (!assemble.getBigdata()){
                    masterDataService.uploadedData(assemble.getTableName(), "All", assemble.getUserId());
                }
               else {
                    // TODO: 2020/2/16 hbase 搬运数据
                }
                sysAssembleLog.setCnt(sorceCnt).setMessage("汇集成功").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.success).updateById();
            }else {
                sysAssembleLog.setMessage("汇集搬运过程中发生问题").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
            }
            boolean hasTable = dbTrans.getDsOutput().checkTableExists(assemble.getTableName());
            if (hasTable) {
                //TODO 全量还是增量
            }
            else {
                //创建表相关 目前禁用   log
                sysAssembleLog.setMessage("当前系统中不存在相关表数据,请先初始化表结构等基础主数据结构信息").setUpdateTime(new Date()).setStatus(SysAssembleLogStatus.defeat).updateById();
            }
        }
        catch (Exception e) {
            e.printStackTrace();
        }
        finally {
            Database dsOutput = dbTrans.getDsOutput();
            if (dsOutput!= null) {
                dsOutput.disconnect();
            }
            Database dsInput = dbTrans.getDsInput();
            if (dsInput != null) {
                dsInput.disconnect();
            }
        }
    }

    private String getTempTableNameFromMenu(String menuId) {
        MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
        String tableName = menuMapping.getTableName();
        return tableName + Constant.RECORD;
    }

    private boolean checkField(KettleDBTrans dbTrans) {
        SysAssemble assemble = dbTrans.getAssemble();
        boolean checked = true;
        Connection conn = null;
        Database ds = null;
        try {
            //1 check 源库中是否有相关字段
            List<SysAssembleDetail> detailList = assembleDetailService.selectList(new EntityWrapper<SysAssembleDetail>().eq("assemble_id", assemble.getId()));
            String fieldStr = detailList.stream().map(sysAssembleDetail -> sysAssembleDetail.getField()).collect(Collectors.joining(Constant.COMMA));
            List<String> fieldList = detailList.stream().map(sysAssembleDetail -> sysAssembleDetail.getField()).collect(Collectors.toList());
            String checkSqlStr = MessageFormat.format(Constant.checkFieldSqlTemplate, fieldStr, assemble.getTableName());
            //连接数据库
            KettleClientEnvironment.init();
            DatabaseMeta dataMeta = new DatabaseMeta("KettleDBRep",assemble.getDbType(), "Native", assemble.getHost(),assemble.getSchema(),assemble.getPort().toString(),assemble.getUserName(),assemble.getPassword());
            LoggingObjectInterface loggingObject = new SimpleLoggingObject("Database factory", LoggingObjectType.GENERAL, null );
            ds = new Database(loggingObject, dataMeta);
            ds.connect();
            conn = ds.getConnection();
            PreparedStatement stmt = conn.prepareStatement(checkSqlStr);
            stmt.executeQuery();
            ds.disconnect();
            ds = null;
            //2 验证 目标库中是否有相关字段
            Boolean bigdata = assemble.getBigdata();
            if (bigdata) {
                //TODO hbase
            }
            else {
                boolean exists = dbTrans.getDsOutput().checkTableExists(assemble.getTableName());
                if (exists) {
                    List<TableSchemaResult> tableFieldFromDb = infoMapper.getTableFieldFromDb(assemble.getTableName(), assemble.getSchema());
                    List<String> targetFieldList = tableFieldFromDb.stream().map(tableSchemaResult -> tableSchemaResult.getFieldName()).collect(Collectors.toList());
                    long count = fieldList.stream().filter(sourceField -> !targetFieldList.contains(sourceField)).count();
                    if (count > 0) {
                        //TODO 有部分字段未能匹配上 且是mysql数据库 暂处理为不导入   log
                        checked = false;
                    }
                }

            }
            checked = true;
        } catch(Exception e) {
            e.printStackTrace();
            checked = false;
        }finally {
            if (ds != null) {
                ds.disconnect();
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                    checked = false;
                }
            }
        }
        return checked;
    }


    @Override
    public Result updateTask(String assembleId) {
        return null;
    }

    private Database getDataBaseByKettle(SysAssemble assemble) throws KettleException {
        Database ds = null;
        KettleClientEnvironment.init();
        DatabaseMeta dataMeta = new DatabaseMeta("KettleDBRep", assemble.getDbType(), "Native",assemble.getHost(), assemble.getSchema(), assemble.getPort().toString(),
                assemble.getUserName(), assemble.getPassword());
        LoggingObjectInterface loggingObject = new SimpleLoggingObject("Database factory", LoggingObjectType.GENERAL, null );
        ds = new Database(loggingObject, dataMeta);

        return ds;
    }
    @Override
    public boolean connectionDb(SysAssemble assemble) {
        Connection conn = null;
        Database ds = null;
        boolean connected = false;
        try {
            ds = getDataBaseByKettle(assemble);
            ds.normalConnect(null);
            conn = ds.getConnection();

        if (conn == null) {
            connected = false;
        }else {
            connected = true;
        }

        ds.disconnect();
        ds = null;
        } catch (KettleException e) {
            e.printStackTrace();
            connected = false;
        }
        finally {
            if (ds != null) {
                ds.disconnect();
            }

            try {
                if (conn != null) {
                    conn.close();
                }

            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return connected;
    }

    @Override
    public boolean sqlTest(SysAssemble assemble, String sqlStr) {
        Connection conn = null;
        Database ds = null;
        boolean connected = false;
        try {
            ds = getDataBaseByKettle(assemble);
            ds.normalConnect(null);
            conn = ds.getConnection();
            PreparedStatement stmt = conn.prepareStatement(sqlStr);
            stmt.executeQuery();

            ds.disconnect();
            ds = null;
            connected = true;
        } catch (Exception e) {
            e.printStackTrace();
            connected = false;
        }
        finally {
            if (ds != null) {
                ds.disconnect();
            }

            try {
                if (conn != null) {
                    conn.close();
                }

            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return connected;
    }
}