D-Net

package eu.dnetlib.msro.workflows.nodes;

import java.io.StringReader;

import java.util.ArrayList;

import java.util.List;

import java.util.stream.Collectors;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang3.math.NumberUtils;

import org.apache.commons.lang3.tuple.Pair;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Element;

import org.dom4j.Namespace;

import org.dom4j.Node;

import org.dom4j.QName;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.core.io.ClassPathResource;

import com.google.gson.Gson;

import com.google.gson.reflect.TypeToken;

import eu.dnetlib.enabling.resultset.factory.ResultSetFactory;

import eu.dnetlib.msro.workflows.graph.Arc;

import eu.dnetlib.msro.workflows.nodes.merge.FieldRule;

import eu.dnetlib.msro.workflows.nodes.merge.MergeAlgorithm;

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.manager.MSROException;

public class MergeRecordsJobNode extends SimpleJobNode {

	private String inputEprParam;

	private String outputEprParam;

	private String rule;

	private List<FieldRule> fields = new ArrayList<>();

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

	@Autowired

	private ResultSetFactory resultSetFactory;

	@Override

	protected String execute(final Env env) throws Exception {

		final ResultSet<?> rsIn = env.getAttribute(inputEprParam, ResultSet.class);

		if ((rsIn == null)) { throw new MSROException("InputEprParam (" + inputEprParam + ") not found in ENV"); }

		fields = new Gson().fromJson(

				IOUtils.toString(new ClassPathResource(rule).getInputStream()),

				new TypeToken<List<FieldRule>>() {}.getType());

		final ResultSet<String> rsOut = resultSetFactory.map(rsIn, String.class, this::merge);

		env.setAttribute(outputEprParam, rsOut);

		return Arc.DEFAULT_ARC;

	}

	@SuppressWarnings("unchecked")

	private String merge(final String pkg) {

		try {

			final Document doc = new SAXReader().read(new StringReader(pkg));

			final Element pkgNode = (Element) doc.selectSingleNode("//package");

			final List<Element> records = ((List<Element>) pkgNode.elements())

					.stream()

					.filter(e -> e.getName() == "record")

					.map(e -> {

						final Element record = (Element) e.selectSingleNode("./*[local-name() = 'resource']");

						final float trust = NumberUtils.toFloat(e.valueOf("@trust"), 0.1f);

						return Pair.of(record, trust);

					})

					.sorted((x, y) -> Float.compare(y.getRight(), x.getRight()))

					.map(Pair::getLeft)

					.collect(Collectors.toList());

			if (records.isEmpty()) {

				log.error("Empty package");

				throw new RuntimeException("Empty package");

			}

			final Element master = records.get(0);

			for (final FieldRule rule : fields) {

				for (int i = 1; (i < records.size()) && isNotComplete(master, rule); i++) {

					merge(master, records.get(i), rule);

				}

			}

			final Element projectsNode = (Element) master.selectSingleNode("./*[local-name()='projects']");

			for (final Object o : projectsNode.selectNodes("./*")) {

				((Node) o).detach();

			}

			for (final Object o : pkgNode.selectNodes("./*[local-name()='projects']/*")) {

				projectsNode.add(((Node) o).detach());

			}

			pkgNode.getParent()

					.addElement(new QName("metadata", new Namespace("oai", "http://www.openarchives.org/OAI/2.0/")))

					.add(master.detach());

			pkgNode.detach();

			return doc.asXML();

		} catch (final DocumentException e) {

			log.error("Error processing package: " + pkg, e);

			throw new RuntimeException("Error processing package: " + pkg, e);

		}

	}

	@SuppressWarnings("unchecked")

	private void merge(final Element master, final Element other, final FieldRule rule) {

		final Element nodeMaster = (Element) master.selectSingleNode(rule.getParentXpath());

		if (nodeMaster == null) {

			log.warn("Missing parent node in MASTER record, xpath: " + rule.getParentXpath());

			return;

		}

		final Element nodeOther = (Element) other.selectSingleNode(rule.getParentXpath());

		if (nodeOther == null) {

			log.warn("Missing parent node in SECOND record, xpath: " + rule.getParentXpath());

			return;

		}

		nodeOther.selectNodes(rule.getXpath()).forEach(o -> {

			nodeMaster.add(((Node) o).detach());

		});

	}

	private boolean isNotComplete(final Element master, final FieldRule rule) {

		return ((rule.getAlgo() == MergeAlgorithm.APPEND) || master.selectSingleNode(rule.getParentXpath()).selectNodes(rule.getXpath()).isEmpty());

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public String getRule() {

		return rule;

	}

	public void setRule(final String rule) {

		this.rule = rule;

	}

}