D-Net

package eu.dnetlib.data.mdstore.plugins;

import java.io.StringReader;

import java.util.Collection;

import java.util.Map;

import java.util.Objects;

import java.util.Set;

import java.util.stream.Collectors;

import java.util.stream.StreamSupport;

import org.apache.commons.lang3.StringUtils;

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

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

import org.dom4j.Element;

import org.dom4j.Namespace;

import org.dom4j.QName;

import org.dom4j.io.SAXReader;

import com.google.common.base.Joiner;

import com.mongodb.DBObject;

import com.mongodb.client.MongoCollection;

import eu.dnetlib.data.mdstore.modular.mongodb.MongoMDStore;

import eu.dnetlib.miscutils.functional.hash.Hashing;

import eu.dnetlib.rmi.data.MDStoreServiceException;

public class DedupPublicationsPlugin extends AbstractIstiMDStorePlugin {

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

	private static final Namespace OAI_NS = new Namespace("oai", "http://www.openarchives.org/OAI/2.0/");

	private static final Namespace DRI_NS = new Namespace("dri", "http://www.driver-repository.eu/namespace/dri");

	@Override

	public void process(final Map<String, String> params) throws MDStoreServiceException {

		// TODO Auto-generated method stub

		log.info("*****************************************");

		log.info("* EXECUTING DEDUP...");

		params.entrySet().forEach(e -> log.info("*   PARAM " + e.getKey() + " = " + e.getValue()));

		log.info("*****************************************");

		final MongoMDStore sourceStore = resolveStore(params, "sourceMdId");

		final MongoMDStore targetStore = resolveStore(params, "targetMdId");

		final MongoCollection<DBObject> coll = sourceStore.getCollection();

		final Collection<Set<String>> groups = StreamSupport.stream(coll.find().spliterator(), false)

				.map(this::createDedupEntry)

				.filter(Objects::nonNull)

				.collect(Collectors.groupingBy(Pair::getRight, Collectors.mapping(Pair::getLeft, Collectors.toSet())))

				.values();

		targetStore.truncate();

		targetStore.feed(() -> groups.stream()

				.map(ids -> pack(ids, sourceStore))

				.iterator(), false);

		touch(targetStore);

	}

	private Pair<String, String> createDedupEntry(final DBObject obj) {

		try {

			final Document doc = (new SAXReader()).read(new StringReader(obj.get("body").toString()));

			final String id = (obj.get("id").toString());

			final String title = doc.valueOf("//*[local-name() = 'title' and position() = 1]/text()");

			final String type = doc.valueOf("//*[local-name() = 'resourceType']/text()");

			if (StringUtils.isNotBlank(id) && StringUtils.isNotBlank(title)) { return ImmutablePair.of(id, clean(title) + "@@@" + clean(type)); }

		} catch (final DocumentException e) {

			log.warn("Problem parsing a mdstore record");

		}

		return null;

	}

	private String clean(final String s) {

		return s == null ? "" : s.replaceAll("\\W", "").toLowerCase();

	}

	private String pack(final Set<String> ids, final MongoMDStore store) {

		final String newId = "dedup_______::" + Hashing.md5(Joiner.on(",").join(ids));

		final Document doc = DocumentHelper.createDocument();

		final Element root = doc.addElement(new QName("record", OAI_NS));

		root.addElement(new QName("header", OAI_NS)).addElement(new QName("objIdentifier", DRI_NS)).setText(newId);

		final Element pkg = root.addElement("package");

		ids.forEach(id -> {

			try {

				final Document innerDoc = (new SAXReader()).read(new StringReader(store.getRecord(id)));

				final Element record = pkg.addElement("record");

				record.addAttribute("provenance", id.startsWith("people") ? "people" : id.startsWith("puma") ? "puma" : "unknown");

				record.addAttribute("trust", id.startsWith("people") ? "0.7" : id.startsWith("puma") ? "0.8" : "0.6");

				record.add(innerDoc.selectSingleNode("//*[local-name() = 'metadata']/*[local-name() = 'resource']").detach());

			} catch (final Exception e) {

				log.error("Error generating package: " + ids, e);

			}

		});

		return doc.asXML();

	}

}