package eu.dnetlib.data.mdstore.manager.utils;

import java.io.IOException;
import java.util.LinkedHashSet;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import eu.dnetlib.data.mdstore.manager.exceptions.MDStoreManagerException;

@Component
public class HdfsClient {

	@Value("${dhp.mdstore-manager.hadoop.cluster}")
	private String hadoopCluster;

	@Value("${dhp.mdstore-manager.hadoop.user}")
	private String hadoopUser;

	@Value("${dhp.mdstore-manager.hdfs.base-path}")
	private String hdfsBasePath;

	private static final Log log = LogFactory.getLog(HdfsClient.class);

	public void deletePath(final String path) throws MDStoreManagerException {

		try (final FileSystem fs = FileSystem.get(conf())) {
			fs.delete(new Path(path), true);
			log.info("HDFS Path deleted: " + path);
		} catch (IllegalArgumentException | IOException e) {
			log.error("Eror deleting path: " + path, e);
			throw new MDStoreManagerException("Eror deleting path: " + path, e);
		}
	}

	public Set<String> listHadoopDirs() {
		final Set<String> res = new LinkedHashSet<>();

		try (final FileSystem fs = FileSystem.get(conf())) {
			for (final FileStatus mdDir : fs.listStatus(new Path(hdfsBasePath))) {
				if (isValidDir(mdDir)) {
					res.add(String.format("%s/%s", hdfsBasePath, mdDir.getPath().getName()));
					for (final FileStatus verDir : fs.listStatus(mdDir.getPath())) {
						if (isValidDir(verDir)) {
							res.add(String.format("%s/%s/%s", hdfsBasePath, mdDir.getPath().getName(), verDir.getPath().getName()));
						}
					}
				}
			}
		} catch (final Exception e) {
			log.error("Error Listing path: " + hdfsBasePath, e);
		}

		return res;
	}

	private boolean isValidDir(final FileStatus fileStatus) {
		return fileStatus.isDirectory() && fileStatus.getPath().getName().startsWith("md-");
	}

	private Configuration conf() throws MDStoreManagerException {
		final Configuration conf = new Configuration();

		System.setProperty("HADOOP_USER_NAME", hadoopUser);

		if (hadoopCluster.equalsIgnoreCase("OCEAN")) {
			conf.addResource(getClass().getResourceAsStream("/hadoop/OCEAN/core-site.xml"));
			conf.addResource(getClass().getResourceAsStream("/hadoop/OCEAN/ocean-hadoop-conf.xml"));
		} else if (hadoopCluster.equalsIgnoreCase("GARR")) {
			conf.addResource(getClass().getResourceAsStream("/hadoop/GARR/core-site.xml"));
			conf.addResource(getClass().getResourceAsStream("/hadoop/GARR/garr-hadoop-conf.xml"));
		} else {
			log.error("Invalid Haddop Cluster: " + hadoopCluster);
			throw new MDStoreManagerException("Invalid Haddop Cluster: " + hadoopCluster);
		}
		return conf;
	}

	public String getHadoopCluster() {
		return hadoopCluster;
	}

	public void setHadoopCluster(final String hadoopCluster) {
		this.hadoopCluster = hadoopCluster;
	}

	public String getHadoopUser() {
		return hadoopUser;
	}

	public void setHadoopUser(final String hadoopUser) {
		this.hadoopUser = hadoopUser;
	}
}