import cn.seaboot.commons.core.Converter;

import cn.seaboot.commons.exception.SystemError;

import cn.seaboot.commons.file.FileUtils;

import cn.seaboot.commons.file.IOUtils;

import cn.seaboot.commons.file.PropertiesUtils;

import org.pentaho.di.core.KettleEnvironment;

import org.pentaho.di.core.database.DatabaseMeta;

import org.pentaho.di.core.exception.KettleException;

import org.pentaho.di.core.logging.ChannelLogTable;

import org.pentaho.di.core.logging.JobEntryLogTable;

import org.pentaho.di.core.logging.JobLogTable;

import org.pentaho.di.job.Job;

import org.pentaho.di.job.JobMeta;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

import javax.annotation.PreDestroy;

import java.io.File;

import java.io.IOException;

import java.io.InputStream;

import java.util.HashMap;

import java.util.Map;

import java.util.Properties;

/**

 * Kettle桥接器，通过这个类，调用使用Kettle提供的jar包运行脚本

 * <p>

 * KettleBridge kettleBridge = new KettleBridge();

 * kettleBridge.init();

 * kettleBridge.execute(filepath, null);

 *

 * @author Mr.css

 * @version 2022-03-28 14:44

 */

@Service

public class KettleBridge {

    private Logger logger = LoggerFactory.getLogger(KettleBridge.class);

    /**

     * Kettle 环境初始化

     */

    @PostConstruct

    public void init() {

        try {

            KettleEnvironment.init();

            logger.debug("【Configuration】Kettle environment init succeed！");

        } catch (KettleException e) {

            throw new SystemError("【Configuration】Kettle Environment build failed!", e);

        }

    }

    /**

     * 设置输出日志，将日志输出到数据库表里

     *

     * @param jobMeta -

     */

    private void initDatabase(JobMeta jobMeta) {

        DatabaseMeta databaseMeta = new DatabaseMeta();

        databaseMeta.setName("med");

        databaseMeta.setDatabaseType("MySQL");

        databaseMeta.setAccessType(DatabaseMeta.TYPE_ACCESS_NATIVE);

        databaseMeta.setHostname("localhost");

        databaseMeta.setDBName("med");

        databaseMeta.setDBPort("3306");

        databaseMeta.setUsername("root");

        databaseMeta.setPassword("root");

        jobMeta.addDatabase(databaseMeta);

        // databaseMeta.name 对应于下面的 connectionName

        // 任务日志

        JobLogTable jobLogTable = JobLogTable.getDefault(jobMeta, jobMeta);

        jobLogTable.setConnectionName("med");

        jobLogTable.setSchemaName("med");

        jobLogTable.setTableName("t_kettle_job_log");

        jobMeta.setJobLogTable(jobLogTable);

        // 任务节点日志

        JobEntryLogTable jobEntryLogTable = JobEntryLogTable.getDefault(jobMeta, jobMeta);

        jobEntryLogTable.setConnectionName("med");

        jobEntryLogTable.setSchemaName("med");

        jobEntryLogTable.setTableName("t_kettle_item_log");

        jobMeta.setJobEntryLogTable(jobEntryLogTable);

        // 任通道日志

        ChannelLogTable channelLogTable = ChannelLogTable.getDefault(jobMeta, jobMeta);

        channelLogTable.setConnectionName("med");

        channelLogTable.setSchemaName("med");

        channelLogTable.setTableName("t_kettle_channel_log");

        jobMeta.setChannelLogTable(channelLogTable);

    }

    /**

     * Kettle 环境销毁

     */

    @PreDestroy

    public void shutdown() {

        KettleEnvironment.shutdown();

    }

    /**

     * 执行kettle脚本，Kettle以文件作为脚本的最小单位，提供脚本所在的绝对路即可

     *

     * @param path   脚本路径

     * @param params 脚本运行所需的参数变量

     * @return 执行结果

     * @throws KettleException run kettle cause any exception

     * @throws IOException     can not read kettle.properties

     */

    public JobResult execute(String path, Map<String, Object> params) throws KettleException, IOException {

        // 初始化job路径

        JobMeta jobMeta = new JobMeta(path, null);

        Job job = new Job(null, jobMeta);

        // 设置环境变量

        Map<String, String> en = this.loadVariable(params);

        for (Map.Entry<String, String> entry : en.entrySet()) {

            job.setVariable(entry.getKey(), entry.getValue());

        }

        // 日志设置

        this.initDatabase(jobMeta);

        // 启动等待直到结束

        job.start();

        job.waitUntilFinished();

        // 为了避免出现意外的编程，通过对象打包执行结果，不返回Job对象

        JobResult result = new JobResult();

        result.setBatchId(job.getBatchId());

        result.setErrors(job.getErrors());

        result.setPassedBatchId(job.getPassedBatchId());

        result.setStatus(job.getStatus());

        result.setParams(en);

        return result;

    }

    /**

     * 载入环境变量

     * 这种写法不是最优的，会损耗一部分性能，主要为了方便整理代码

     *

     * @param params 用户指定的变量

     * @return 最终使用的变量

     * @throws IOException 读取配置异常

     */

    private Map<String, String> loadVariable(Map<String, Object> params) throws IOException {

        Map<String, String> result = new HashMap<>();

        // 用户文件夹下的kettle.properties

        Properties properties = this.readKettleProperties();

        if (properties != null) {

            for (Map.Entry<Object, Object> entry : properties.entrySet()) {

                result.put(Converter.toString(entry.getKey()), Converter.toString(entry.getValue()));

            }

        }

        // 用户指定的环境变量

        if (params != null) {

            for (Map.Entry<String, Object> entry : params.entrySet()) {

                result.put(entry.getKey(), Converter.toString(entry.getValue()));

            }

        }

        return result;

    }

    /**

     * 获取kettle配置参数，读取用户文件夹下的kettle.properties

     *

     * @return 参数

     * @throws IOException can not read kettle.properties

     */

    private Properties readKettleProperties() throws IOException {

        String home = FileUtils.getUserDirectoryPath();

        if (home != null) {

            File file = new File(home, ".kettle/kettle.properties");

            if (file.exists()) {

                try (InputStream is = IOUtils.openFileInputStream(file)) {

                    return PropertiesUtils.load(is);

                }

            }

        } else {

            logger.debug("Can not found user.home, had batter find out why!");

        }

        return null;

    }

}