package org.apache.seatunnel.transform.script.java;

import com.google.auto.service.AutoService;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.util.ConfigValidator;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.api.transform.SeaTunnelTransform;
import org.apache.seatunnel.shade.com.typesafe.config.Config;
import org.apache.seatunnel.transform.common.MultipleFieldOutputTransform;
import org.apache.seatunnel.transform.common.SeaTunnelRowAccessor;
import org.apache.seatunnel.transform.script.common.ScriptField;
import org.apache.seatunnel.transform.script.common.ScriptTransformConfig;
import org.apache.seatunnel.transform.script.java.support.JavaEvalUtil;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@AutoService(SeaTunnelTransform.class)
@NoArgsConstructor
public class JavaScriptTransform extends MultipleFieldOutputTransform {

    public final static String PLUGIN_NAME = "ScriptJava";

    private ScriptTransformConfig scriptTransformConfig;
    private List<String> fieldNames;
    private List<SeaTunnelDataType> fieldTypes;

    public JavaScriptTransform(
            @NonNull ReadonlyConfig config, @NonNull CatalogTable inputCatalogTable) {
        super(inputCatalogTable);
        this.scriptTransformConfig = ScriptTransformConfig.of(config);
    }

    @Override
    public String getPluginName() {
        return PLUGIN_NAME;
    }

    @Override
    protected void setConfig(Config pluginConfig) {
        ConfigValidator.of(ReadonlyConfig.fromConfig(pluginConfig))
                .validate(new JavaScriptTransformFactory().optionRule());
        this.scriptTransformConfig = ScriptTransformConfig.of(ReadonlyConfig.fromConfig(pluginConfig));
    }

    @Override
    protected void setInputRowType(SeaTunnelRowType inputRowType) { }

    @Override
    protected String[] getOutputFieldNames() {
        return fieldNames.toArray(new String[0]);
    }

    @Override
    protected SeaTunnelDataType[] getOutputFieldDataTypes() {
        return fieldTypes.toArray(new SeaTunnelDataType[0]);
    }

    @Override
    protected Column[] getOutputColumns() {
        Column[] columns = new Column[fieldNames.size()];
        for (int i = 0; i < fieldNames.size(); i++) {
            columns[i] =
                    PhysicalColumn.of(fieldNames.get(i), fieldTypes.get(i), 200, true, "", "");
        }
        return columns;
    }

    @Override
    protected Object[] getOutputFieldValues(SeaTunnelRowAccessor inputRow) {
        return null;
    }

    @Override
    protected SeaTunnelRowType transformRowType(SeaTunnelRowType inputRowType) {
        List<ScriptField> scriptFields = scriptTransformConfig.getScriptFields();
        fieldNames = Arrays.asList(inputRowType.getFieldNames());
        fieldTypes = Arrays.asList(inputRowType.getFieldTypes());
        for(ScriptField scriptField : scriptFields) {
            int index = fieldNames.indexOf(scriptField.getField());
            if(index >= 0) {
                fieldTypes.set(index, convert(scriptField.getOutputFieldType()));
            }
        }
        SeaTunnelRowType outputRowType = new SeaTunnelRowType(fieldNames.toArray(new String[0]), fieldTypes.toArray(new SeaTunnelDataType[0]));
        return outputRowType;
    }

    @Override
    protected SeaTunnelRow transformRow(SeaTunnelRow inputRow)  {
        String scriptContent = scriptTransformConfig.getScriptContent();
        List<ScriptField> scriptFields = scriptTransformConfig.getScriptFields();
        //获取脚本入参
        List<Object> argList = new ArrayList<>();
        for (ScriptField scriptField : scriptFields) {
            Object objValue = inputRow.getField(fieldNames.indexOf(scriptField.getField()));
            //如果想设置输入类型都是字符串，修改这里
            argList.add(objValue);
        }
        //获取脚本执行结果
        List<Object> resultList = new ArrayList<>();
        try {
            Object result = JavaEvalUtil.eval(scriptContent, argList.toArray());
            if(result instanceof Object[]) {
                Object[] destResult = (Object[])result;
                resultList = Stream.of(destResult).collect(Collectors.toList());
            }else{
                resultList.add(result);
            }
        } catch (Exception e) {
            throw new RuntimeException("执行java脚本失败", e);
        }
        //遍历数据赋值
        //1.如果返回结果大于设置的字段 -> 按字段顺序设置值；
        //2.如果返回结果小于设置的字段 -> 按字段顺序设置值；
        Object[] fieldValues = inputRow.getFields();
        int j = 0;
        for(ScriptField scriptField : scriptFields) {
            int i = fieldNames.indexOf(scriptField.getField());
            if(i >= 0) {
                if(resultList.size() > j){
                    Object obj = resultList.get(j++);
                    fieldValues[i] = obj != null ? obj.toString() : null;
                }else if(fieldValues[i] != null){
                    fieldValues[i] = String.valueOf(fieldValues[i]);
                }
            }
        }
        return new SeaTunnelRow(fieldValues);
    }

    private BasicType convert(String outputFieldType){
        return BasicType.STRING_TYPE;
    }
}
