package org.springblade.etl.util;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.file.FileVisitOption;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.springblade.core.tool.utils.StringUtil;

import com.google.common.collect.ImmutableMap;



/**
 * https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-common/src/main/java/org/apache/seatunnel/common/config/Common.java
 */
public class SeaTunnelKit {

	private static final int APP_LIB_DIR_DEPTH = 2;
	
	public static final String STARTER_JAR_NAME = "seatunnel-flink-15-starter.jar";
	public static final String TRANSFORM_JAR_NAME = "seatunnel-transforms-v2.jar";
	public static final String SEATUNNEL_MAIN_CLASS = "org.apache.seatunnel.core.starter.flink.SeaTunnelFlink";
	public static final String SEATUNNEL_PLUGIN_MAPPING = "plugin-mapping.properties";
	public static final String SEATUNNEL_HADDOP_JAR_NAME = "seatunnel-hadoop3-3.1.4-uber.jar";

	public static Path getLibDir(Path rootDir) {
		return rootDir.resolve("lib");
	}

	public static List<URL> getLibJars(Path rootDir) {
		Path libRootDir = getLibDir(rootDir);
		if (!Files.exists(libRootDir) || !Files.isDirectory(libRootDir)) {
			return Collections.emptyList();
		}
		try (Stream<Path> stream = Files.walk(libRootDir, APP_LIB_DIR_DEPTH, FileVisitOption.FOLLOW_LINKS)) {
			return stream.filter(it -> !it.toFile().isDirectory())
					.filter(it -> it.getFileName().toString().endsWith(".jar")).map(it -> {
						try {
							return it.toUri().toURL();
						} catch (MalformedURLException e) {
							throw new RuntimeException(e);
						}
					}).collect(Collectors.toList());
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}
	
	public static URL seatunnelTransform(Path rootDir) {
		List<URL> libJars = getLibJars(rootDir);
		return libJars.stream().filter(item -> item.getPath().contains(TRANSFORM_JAR_NAME)).findFirst().get();
	}
	
	public static URL seatunnelHadoop(Path rootDir) {
		List<URL> libJars = getLibJars(rootDir);
		return libJars.stream().filter(item -> item.getPath().contains(SEATUNNEL_HADDOP_JAR_NAME)).findFirst().get();
	}
	
	public static Path getStarterDir(Path rootDir) {
		return rootDir.resolve("starter");
	}

	public static Path getStarterJarPath(Path rootDir) {
		return getStarterDir(rootDir).resolve(STARTER_JAR_NAME);
	}

	public static Path connectorsRootDir(Path rootDir) {
		return Paths.get(rootDir.toString(), "connectors");
	}

	public static Path getPluginMapping(Path rootDir) {
		return connectorsRootDir(rootDir).resolve(SEATUNNEL_PLUGIN_MAPPING);
	}

	public static Path engineConnectorsRootDir(Path rootDir, String engine) {
		return connectorsRootDir(rootDir).resolve(engine.toLowerCase());
	}

	public static Path seatunnelConnectorsRootDir(Path rootDir) {
		return engineConnectorsRootDir(rootDir, "seatunnel");
	}

	public static Path seatunnelConnector(Path rootDir, String connector) {
		return seatunnelConnectorsRootDir(rootDir).resolve(connector);
	}

//	public static String seatunnelConnectorUrl(String repoUrl, String version, String connector) {
//		if (StringUtils.endsWithIgnoreCase(repoUrl, "/")) {
//			repoUrl = StringUtils.removeEndIgnoreCase(repoUrl, "/");
//		}
//		Map<String, String> variables = ImmutableMap.of("repoUrl", repoUrl, "version", version, "connector", connector,
//				"jar", convertToJar(version, connector));
//		String template = "${repoUrl}/${connector}/${version}/${jar}";
//		return StringUtil.format(template, variables);
//	}

	public static String convertToJar(String version, String connector) {
		Map<String, String> variables = ImmutableMap.of("version", version, "connector", connector);
		String template = "${connector}-${version}.jar";
		return StringUtil.format(template, variables);
	}

	public static URL seatunnelConnector(Path rootDir, String connector, String version) {
		try {
			return seatunnelConnectorsRootDir(rootDir).resolve(convertToJar(version, connector)).toFile().toURI()
					.toURL();
		} catch (MalformedURLException e) {
			e.printStackTrace();
			return null;
		}
	}

	public static URL getPluginJarPath(Path rootDir, String pluginType, String pluginJarName) {
		try {
			return Paths.get(rootDir.toString(), "plugins", pluginType, "lib", pluginJarName).toFile().toURI().toURL();
		} catch (MalformedURLException e) {
			e.printStackTrace();
			return null;
		}
	}

}
