D-Net

package eu.dnetlib.msro.workflows.nodes.merge;

public enum MergeAlgorithm {

	REPLACE, APPEND

}

package eu.dnetlib.msro.workflows.nodes.merge;

public class FieldRule {

	private String name;

	private MergeAlgorithm algo;

	private String parentXpath;

	private String xpath;

	public FieldRule() {}

	public FieldRule(final String name, final MergeAlgorithm algo, final String parentXpath, final String xpath) {

		this.name = name;

		this.algo = algo;

		this.parentXpath = parentXpath;

		this.xpath = xpath;

	}

	public String getName() {

		return name;

	}

	public void setName(final String name) {

		this.name = name;

	}

	public MergeAlgorithm getAlgo() {

		return algo;

	}

	public void setAlgo(final MergeAlgorithm algo) {

		this.algo = algo;

	}

	public String getParentXpath() {

		return parentXpath;

	}

	public void setParentXpath(final String parentXpath) {

		this.parentXpath = parentXpath;

	}

	public String getXpath() {

		return xpath;

	}

	public void setXpath(final String xpath) {

		this.xpath = xpath;

	}

}

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.Node;

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);

				}

			}

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

				((Node) o).detach();

			}

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

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

			}

			pkgNode.getParent().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 other 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;

	}

}

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

	<!-- Wf Nodes -->

	<bean id="wfNodeMergeRecords"

		class="eu.dnetlib.msro.workflows.nodes.MergeRecordsJobNode"

		scope="prototype" />

	<!-- MDStore plugins -->

[ {

	"name"       : "title",

	"algo"       : "REPLACE",

	"parentXpath": "./*[local-name()='titles']",

	"xpath"      : "./*[local-name()='title']"

},{

	"name"       : "author",

	"algo"       : "REPLACE",

	"parentXpath": "./*[local-name()='creators']",

	"xpath"      : "./*[local-name()='creatorName']"

},{

	"name"       : "id",

	"algo"       : "APPEND",

	"parentXpath": ".",

	"xpath"      : "./*[local-name()='identifier']"

},{

	"name"       : "url",

	"algo"       : "APPEND",

	"parentXpath": "./*[local-name()='alternateIdentifiers']",

	"xpath"      : "./*[local-name()='alternateIdentifier' and @alternateIdentifierType='url']"

} ]

				<NODE name="merge" type="MergeRecords">

						<PARAM name="rule" value="/eu/dnetlib/isti/workflows/merge/unpack.json" />