/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.seatunnel.transform.regular;

import com.google.auto.service.AutoService;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
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.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
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.SeaTunnelRowAccessor;
import org.apache.seatunnel.transform.common.SingleFieldOutputTransform;
import org.apache.seatunnel.transform.regular.ReplaceValueTransformFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;


@AutoService(SeaTunnelTransform.class)
@NoArgsConstructor
public class ReplaceValueTransform extends SingleFieldOutputTransform {

    private ReadonlyConfig config;

    private int inputFieldIndex;

    public ReplaceValueTransform(
            @NonNull ReadonlyConfig config, @NonNull CatalogTable inputCatalogTable) {
        super(inputCatalogTable);
        this.config = config;
        initOutputFields(
                inputCatalogTable.getTableSchema().toPhysicalRowDataType(),
                this.config.get(ReplaceValueTransformConfig.KEY_REPLACE_FIELD));
    }

    @Override
    public String getPluginName() {
        return "ReplaceRegular";
    }

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

    @Override
    protected void setInputRowType(SeaTunnelRowType inputRowType) {
        initOutputFields(inputRowType, config.get(ReplaceValueTransformConfig.KEY_REPLACE_FIELD));
    }

    @Override
    protected String getOutputFieldName() {
        return config.get(ReplaceValueTransformConfig.KEY_REPLACE_FIELD);
    }

    @Override
    protected SeaTunnelDataType getOutputFieldDataType() {
        return BasicType.STRING_TYPE;
    }

    @Override
    protected Object getOutputFieldValue(SeaTunnelRowAccessor inputRow) {
        Object inputFieldValue = inputRow.getField(inputFieldIndex);
        // 该转换组件只支持字符串类型
        if(inputFieldValue != null && !(inputFieldValue instanceof String)) {
            return inputFieldValue;
        }
        List<ReplaceCondition> conditions = config.get(ReplaceValueTransformConfig.KEY_REPLACE_CONDITIONS_LIST);
        String defaultValue = config.get(ReplaceValueTransformConfig.KEY_COLUMN_DEFAULT_VALUE);
        Object retObj = inputFieldValue != null ? inputFieldValue.toString() : null;
        List<Boolean> restList = new ArrayList<>();
        for (ReplaceCondition condition : conditions) {
            //null字符串识别为Null值替换
            String compareValue = "null".equals(condition.getCompareValue()) ? null : condition.getCompareValue();
            boolean isContains = Optional.ofNullable(condition.getIsContains()).orElse(false);
            boolean symbol = false;
            do{
                //是否满足替换Null值列条件
                if(symbol = compareValue == null && inputFieldValue == null){
                    break;
                }
                //是否满足替换空字符串列条件
                if(symbol = "".equals(compareValue) && inputFieldValue != null && inputFieldValue.toString().trim().equals("")){
                    break;
                }
                //包含/匹配字符串
                if(symbol = isContains && StringUtils.isNotEmpty(compareValue) && inputFieldValue != null && inputFieldValue.toString().contains(compareValue)){
                    break;
                }
                //精确匹配字符串
                if(symbol = !isContains && StringUtils.isNotEmpty(compareValue) && inputFieldValue != null && inputFieldValue.toString().equals(compareValue)){
                    break;
                }
            }while(false);
            restList.add(symbol);
            //如果匹配成功，需要去处理替换该列的值
            if(symbol) {
                String replacement = Optional.ofNullable(condition.getReplacement()).orElse("");
                boolean isCover = Optional.ofNullable(condition.getIsCover()).orElse(false);
                //当满足覆盖值，或替换Null值列，或替换空字符串列。 全部以覆盖字段值为准
                if (isCover || compareValue == null || "".equals(compareValue)) {
                    retObj = replacement;
                } else if (inputFieldValue != null) { //当原值!=Null时，直接返回即可
                    retObj = inputFieldValue.toString().replaceAll(compareValue, replacement);
                }
            }
        }
        if(!restList.contains(true) && StringUtils.isNotEmpty(defaultValue)) {
            retObj = defaultValue;
        }
        return retObj;
    }

    @Override
    protected Column getOutputColumn() {
        List<Column> columns = inputCatalogTable.getTableSchema().getColumns();
        List<Column> collect =
                columns.stream()
                        .filter(
                                column ->
                                        column.getName()
                                                .equals(
                                                        config.get(ReplaceValueTransformConfig.KEY_REPLACE_FIELD)))
                        .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(collect)) {
            throw new IllegalArgumentException(
                    "Cannot find ["
                            + config.get(ReplaceValueTransformConfig.KEY_REPLACE_FIELD)
                            + "] field in input catalog table");
        }
        return collect.get(0).copy();
    }

    private void initOutputFields(SeaTunnelRowType inputRowType, String replaceColumn) {
        inputFieldIndex = inputRowType.indexOf(replaceColumn);
        if (inputFieldIndex == -1) {
            throw new IllegalArgumentException(
                    "Cannot find [" + replaceColumn + "] field in input row type");
        }
    }
}
