
package eu.dnetlib.dhp.ircdl_extention;

import java.io.Serializable;
import java.text.Normalizer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.text.similarity.CosineDistance;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wcohen.ss.JaroWinkler;

import eu.dnetlib.dhp.common.HdfsSupport;
import eu.dnetlib.dhp.ircdl_extention.model.Orcid;
import eu.dnetlib.dhp.ircdl_extention.model.Result;
import scala.Tuple2;

public class Utils implements Serializable {

	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

	private static final Logger log = LoggerFactory
		.getLogger(eu.dnetlib.dhp.ircdl_extention.Utils.class);

	public static String normalizeString(String input) {
		if (input == null || input.equals("void"))
			return new String();
		String tmp = Normalizer
			.normalize(input, Normalizer.Form.NFKD)
			.replaceAll("[^\\p{ASCII}]", "");
		tmp = tmp
			.replaceAll("[^\\p{Alpha}]+", " ")
			.replaceAll("\\s+", " ")
			.trim();
		return tmp;

	}

	public static void removeOutputDir(SparkSession spark, String path) {
		HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
	}

	public static <R> Dataset<R> readPath(
		SparkSession spark, String inputPath, Class<R> clazz) {
		return spark
			.read()
			.textFile(inputPath)
			.map((MapFunction<String, R>) value -> OBJECT_MAPPER.readValue(value, clazz), Encoders.bean(clazz));
	}

	private static List<String> getList(List<String> input) {
		return input
			.stream()
			.map(st -> st.trim())
			.filter(st -> st.length() > 0)
			.sorted()
			.collect(Collectors.toList());

	}

	private static List<String> getListInitials(List<String> input) {
		List<String> ret = new ArrayList<>();
		List<Character> tmp = input
			.stream()
			.map(st -> st.trim())
			.filter(st -> st.length() > 0)
			.map(st -> st.charAt(0))
			.sorted()
			.collect(Collectors.toList());
		if (tmp.size() == 1)
			ret.add(String.valueOf(tmp.get(0)));
		for (int i = 0; i < tmp.size(); i++) {
			for (int j = i + 1; j < tmp.size(); j++) {
				ret.add(String.valueOf(tmp.get(i)) + String.valueOf(tmp.get(j)));
			}
		}

		return ret;

	}

	// selezione delle coppie di primi caratteri per ogni parola che compone il nome
	// se ci sono match il nome e' giusto
	// aggiungere verifica che la lunghezza delle liste non sia troppo sbilanciata: se una lista e' lunga
	public static boolean conservativeFilterFunction(Tuple2<Result, Orcid> input) {

		List<String> res = getListInitials(Arrays.asList(input._1().getFullname().split(" ")));
		Orcid or = input._2();
		List<String> tmp = new ArrayList<>();
		Collections.addAll(tmp, or.getName().split(" "));
		Collections.addAll(tmp, or.getSurname().split(" "));
		return checkContains(
			res, getListInitials(tmp), false)
			||
			checkContains(
				res, getListInitials(Arrays.asList(or.getCreditname().split(" "))), false)
			||
			or
				.getOtherNames()
				.stream()
				.anyMatch(
					on -> checkContains(
						res, getListInitials(Arrays.asList(on.split(" "))), false));

	}

	public static boolean filterFunction(Tuple2<Result, Orcid> input) throws JsonProcessingException {

		try {
			List<String> res = getList(Arrays.asList(input._1().getFullname().split(" ")));
			Orcid or = input._2();
			List<String> tmp = new ArrayList<>();
			Collections.addAll(tmp, or.getName().split(" "));
			Collections.addAll(tmp, or.getSurname().split(" "));
			return checkContains(
				res, getList(tmp)
					.stream()
					.sorted()
					.collect(Collectors.toList()))
				||
				checkContains(
					res, getList(Arrays.asList(or.getCreditname().split(" ")))
						.stream()
						.sorted()
						.collect(Collectors.toList()))
				||
				or
					.getOtherNames()
					.stream()
					.anyMatch(
						on -> checkContains(
							res, getList(Arrays.asList(on.split(" ")))
								.stream()
								.sorted()
								.collect(Collectors.toList())));
		} catch (Exception e) {

			log.info("EXCEPTIONNNN: " + new ObjectMapper().writeValueAsString(input));
			throw e;
		}

	}

	private static boolean checkContains(List<String> result, List<String> orcid) {
		return checkContains(result, orcid, true);
	}

	private static boolean checkContains(List<String> result, List<String> orcid, boolean jaro) {
		if (result.size() == 0 || orcid.size() == 0) {
			return true;
		}
		String[][] input = {
			{
				"1", StringUtils.joinWith(" ", result)
			},
			{
				"2", StringUtils.joinWith(" ", orcid)
			}
		};
		// exact match word by word
		Double cosineDistance = new CosineDistance().apply(input[0][1], input[1][1]);
		if (Math.round((1 - cosineDistance) * 100) == 100) {
			return true;
		}
		// check containment one list can be greater than the other, and also composition of words to create the name
		// e.g. pengli yan = li peng yan
		if (orcid.size() < result.size()) {
			if (isIn(orcid, result))
				return true;
		} else {
			if (isIn(result, orcid))
				return true;
		}
		if (jaro) {
			// apply JaroWinkler distance
			double score = new JaroWinkler()
				.score(StringUtils.joinWith(" ", result), StringUtils.joinWith(" ", orcid));
			return score > 0.95;
		}
		return false;
	}

	private static boolean isIn(List<String> lst1, List<String> lst2) {

		int index = 0;
		for (String word : lst1) {
			int i = index;
			boolean found = false;
			while (i < lst2.size()) {
				String wordlist = lst2.get(i);
				if (word.equals(wordlist)) {
					index = i + 1;
					i = lst2.size();
					found = true;
				} else {
					if (word.charAt(0) < wordlist.charAt(0)) {
						if (!checkComposition(word, lst2)) {
							return false;
						} else {
							index = 0;
							i = lst2.size();
							found = true;
						}
					} else {
						if (word.length() == 1 || wordlist.length() == 1) {
							if (word.charAt(0) == wordlist.charAt(0)) {
								index = i + 1;
								i = lst2.size();
								found = true;
							} else {
								i++;
							}
						} else {
							i++;
						}

					}

				}

			}
			if (!found) {
				if (!checkComposition(word, lst2)) {
					return false;
				} else {
					index = 0;
				}
			}
		}

		return true;
	}

	private static boolean checkComposition(String word, List<String> lst2) {
		for (int i = 0; i < lst2.size(); i++) {
			for (int j = 0; j < lst2.size(); j++) {
				if (i != j) {
					String w = lst2.get(i) + lst2.get(j);
					if (word.equals(w)) {
						if (i > j) {
							lst2.remove(i);
							lst2.remove(j);
						} else {
							lst2.remove(j);
							lst2.remove(i);
						}
						return true;
					}
				}
			}
		}
		return false;
	}

}
