/**
 * 
 */
package org.springblade.etl.connector.transform;

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

import javax.validation.constraints.NotEmpty;

import org.springblade.core.tool.exception.ServiceException;
import org.springblade.datasource.enums.DsTypeEnum;
import org.springblade.datasource.transform.TransformContext;
import org.springblade.datasource.transform.seatunnel.SeatunnelColumnType;
import org.springblade.etl.connector.base.SeatunnelConnector;
import org.springblade.etl.connector.transform.config.SeatunnelRowStackTransformConfig;
import org.springblade.etl.connector.transform.config.SeatunnelRowStackTransformConfig.FieldConfig;
import org.springblade.etl.connector.transform.config.SeatunnelRowStackTransformConfig.FieldMapping;
import org.springblade.etl.engine.seatunnel.SeaTunnelConstant;
import org.springblade.etl.entity.JobTask;
import org.springblade.etl.util.EtlJsonUtil;
import org.springblade.etl.util.ValidateUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.databind.node.ObjectNode;

import cn.hutool.core.util.StrUtil;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;

/**
 * @author wookvn
 *
 */
@Component
public class PlatomixRowStackTransform implements SeatunnelConnector<SeatunnelRowStackTransformConfig> {
	
	@Autowired
	private TransformContext transformContext;

	@Data
	public static class Config{
		@ApiModelProperty("结果列集合")
		private List<DestField> columns;
		@ApiModelProperty("字段映射关系配置")
		private List<PtFieldConfig> fieldConfig;
	}
	
	@Data
	public static class PtFieldConfig{
		@ApiModelProperty("数据源对应任务code")
		@NotEmpty(message = "行堆叠组件：CODE不能为空")
		private String sourceTaskCode;
		@ApiModelProperty("字段映射关系列表")
		private List<FieldMapping> list;
	}
	
	@Data
	public static class DestField{
		@ApiModelProperty("字段名")
		private String columnName;
		@ApiModelProperty("字段类型")
		private String columnType;
		@ApiModelProperty("数据源类型")
		private Integer dsType;
	}
	
	@Override
	public SeatunnelRowStackTransformConfig buildConfig(String configJson) {
		return null;
	}

	@Override
	public Object validate(String configJson) {
		return null;
	}
	
	/**
	 * 构造Seatunnel配置
	 * @return
	 */
	public ObjectNode buildConfig(JobTask jobTask, List<JobTask> jobTaks) {
		Map<String, String> code2Tablename = new HashMap<>();
		for(JobTask item : jobTaks) {
			int nodeId = Optional.<Integer>ofNullable(item.getId()).orElse(0);
			String pluginName = item.getConnectorType().getPluginName();
			String tableName = (SeaTunnelConstant.SEATUNNEL_TABLE_NAME_PREFIX + pluginName + "_" + nodeId).replaceAll("\\-", "_").toLowerCase();
			code2Tablename.put(item.getTaskCode(), tableName);
		}
		SeatunnelRowStackTransformConfig seatunnelConfig = this.buildConfig(jobTask.getConfigJson(), code2Tablename);
		seatunnelConfig.setNodeId(Optional.<Integer>ofNullable(jobTask.getId()).orElse(0));
		seatunnelConfig.setPluginType(jobTask.getPluginType());
		seatunnelConfig.setPluginName(jobTask.getConnectorType().getPluginName());
		return EtlJsonUtil.toObjectNode(seatunnelConfig);
	}
	
	public SeatunnelRowStackTransformConfig buildConfig(String configJson, Map<String, String> code2Tablename) {
		Config config = EtlJsonUtil.parseJsonString(configJson, Config.class);
		ValidateUtil.validate(config);
		ValidateUtil.validate(config.getColumns());
		ValidateUtil.validate(config.getFieldConfig());
		SeatunnelRowStackTransformConfig seatunnelConfig = new SeatunnelRowStackTransformConfig();
		List<String> fields = new ArrayList<>();
		List<String> fieldTypes = new ArrayList<>();
		for(DestField destField : config.getColumns()) {
			fields.add(destField.getColumnName());
			SeatunnelColumnType seatunneColumnType = transformContext.transformToSeatunnelType(DsTypeEnum.resolve(Optional.ofNullable(destField.getDsType()).orElse(DsTypeEnum.common.getValue())), destField.getColumnType());
			fieldTypes.add(Optional.ofNullable(seatunneColumnType).orElse(SeatunnelColumnType.STRING).name());
		}
		seatunnelConfig.setFields(fields);
		seatunnelConfig.setFieldTypes(fieldTypes);
		List<FieldConfig> fieldConfigs = new ArrayList<>();
		for(PtFieldConfig ptFieldConfig : config.getFieldConfig()) {
			String sourceTableName = code2Tablename.get(ptFieldConfig.getSourceTaskCode());
			if(StrUtil.isBlank(sourceTableName)) {
				throw new ServiceException("行堆叠组件：传递的Code不存在");
			}
			FieldConfig fieldConfig = new FieldConfig();
			fieldConfig.setSourceTable(sourceTableName);
			fieldConfig.setList(ptFieldConfig.getList());
			fieldConfigs.add(fieldConfig);
		}
		seatunnelConfig.setFieldConfigs(fieldConfigs);
		return seatunnelConfig;
	}

}
