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

import java.util.Map;

import javax.validation.constraints.NotEmpty;
import javax.validation.constraints.NotNull;

import org.springblade.etl.connector.base.SeatunnelConnector;
import org.springblade.etl.connector.source.config.SeatunnelFtpFileSourceConfig;
import org.springblade.etl.connector.source.config.SeatunnelLocalFileSourceConfig;
import org.springblade.etl.connector.source.enums.FieldType;
import org.springblade.etl.connector.source.enums.FileFormat;
import org.springblade.etl.properties.SftpProperties;
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 io.swagger.annotations.ApiModelProperty;
import lombok.Data;

/**
 * @author wookvn
 *
 */
@Component
public class PlatomixLocalFileSource implements SeatunnelConnector<SeatunnelFtpFileSourceConfig> {
	
	@Autowired
	private SftpProperties sftpConfig;
	
	@Data
	public static class Config {
		@NotEmpty(message = "Source文件路径不能为空")
		@ApiModelProperty("文件路径,比如:/tmp/sink/text")
		private String path;
		@NotNull(message = "Source文件类型不能为空")
		@ApiModelProperty("文件格式")
		private FileFormat format;
		@NotEmpty(message = "Source字段不能为空")
		@ApiModelProperty("文件格式")
		private Map<String, FieldType> fields;
		@ApiModelProperty("分隔符，主要针对csv和text")
		private String delimiter;
		@ApiModelProperty("跳过几行，主要针对excel")
		private Integer skipHeaderRowNumber;
		@ApiModelProperty("excel的sheet名称")
		private String sheetName;
	}
	
	@Override
	public SeatunnelFtpFileSourceConfig buildConfig(String configJson) {
		Config ptConfig = (Config)validate(configJson);
		SeatunnelFtpFileSourceConfig seatunnelConfig = new SeatunnelFtpFileSourceConfig();
		seatunnelConfig.setDelimiter(ptConfig.getDelimiter());
		seatunnelConfig.setSheetName(ptConfig.getSheetName());
		seatunnelConfig.setFileFormatType(ptConfig.getFormat().name());
		seatunnelConfig.setPath(ptConfig.getPath());
		seatunnelConfig.setSkipHeaderRowNumber(ptConfig.getSkipHeaderRowNumber());
		seatunnelConfig.setSchema(new SeatunnelLocalFileSourceConfig.SchemaField(ptConfig.getFields()));
		seatunnelConfig.setHost(sftpConfig.getHost());
		seatunnelConfig.setPort(sftpConfig.getPort());
		seatunnelConfig.setUser(sftpConfig.getUser());
		seatunnelConfig.setPassword(sftpConfig.getPassword());
		return seatunnelConfig;
	}

	@Override
	public Object validate(String configJson) {
		Config ptConfig = EtlJsonUtil.parseJsonString(configJson, PlatomixLocalFileSource.Config.class);
		ValidateUtil.validate(ptConfig);
		return ptConfig;
	}
}
