D-Net

package eu.dnetlib.data.mapreduce.hbase.broker;

import static eu.dnetlib.data.mapreduce.hbase.broker.mapping.EventFactory.asEvent;

import java.util.LinkedList;

import java.util.List;

import java.util.Queue;

import java.util.function.Predicate;

import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;

import com.google.common.collect.Lists;

import eu.dnetlib.broker.objects.OpenAireEventPayload;

import eu.dnetlib.data.mapreduce.hbase.broker.mapping.HighlightFactory;

import eu.dnetlib.data.mapreduce.hbase.broker.mapping.OpenAireEventPayloadFactory;

import eu.dnetlib.data.mapreduce.hbase.broker.model.EventWrapper;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.miscutils.collections.Pair;

import eu.dnetlib.pace.distance.algo.JaroWinkler;

public class OrcidEventFactory {

	private static final long MAX_AUTHORS = 50;

	private static final Float t = 0.9f;

	public static final String ORCID_TYPE_MARKER = "ORCID";

	public static List<EventWrapper> process(final Oaf current, final Oaf other, final float trust) {

		return new OrcidEventFactory().processOrcid(current, other, trust);

	}

	public List<EventWrapper> processOrcid(final Oaf current, final Oaf other, final float trust) {

		final List<EventWrapper> events = Lists.newArrayList();

		final Queue<FieldTypeProtos.Author> currAuthors = getAuthors(current, noneIsORCID());

		final Queue<FieldTypeProtos.Author> otherAuthors = getAuthors(other, anyIsORCID());

		while (!currAuthors.isEmpty()) {

			final FieldTypeProtos.Author currentAuthor = currAuthors.remove();

			Pair<FieldTypeProtos.Author, Float> bestMatch = null;

			for (final FieldTypeProtos.Author otherAuthor : otherAuthors) {

				final Pair<FieldTypeProtos.Author, Float> pair = new Pair<>(otherAuthor, distance(currentAuthor, otherAuthor));

				if (bestMatch == null || pair.getValue() > bestMatch.getValue()) {

					bestMatch = pair;

				}

			}

			if (bestMatch != null && bestMatch.getValue() >= t) {

				final float authorTrust = trust * bestMatch.getValue(); // adjust it?

				events.add(doProcessOrcid(current, other, new Pair<>(currentAuthor, bestMatch.getKey()), Topic.ENRICH_MISSING_AUTHOR_ORCID, authorTrust));

			}

		}

		return events;

	}

	private LinkedList<FieldTypeProtos.Author> getAuthors(final Oaf oaf, final Predicate<FieldTypeProtos.Author> p) {

		return authors(oaf).stream()

				.filter(p)

				.limit(MAX_AUTHORS)

				.collect(Collectors.toCollection(LinkedList::new));

	}

	private Predicate<FieldTypeProtos.Author> anyIsORCID() {

		return author -> author.getPidList().stream().anyMatch(pid -> ORCID_TYPE_MARKER.equals(pid.getKey()));

	}

	private Predicate<FieldTypeProtos.Author> noneIsORCID() {

		return author -> author.getPidList().stream().noneMatch(pid -> ORCID_TYPE_MARKER.equals(pid.getKey()));

	}

	private EventWrapper doProcessOrcid(final Oaf current,

			final Oaf other,

			final Pair<FieldTypeProtos.Author, FieldTypeProtos.Author> pair,

			final Topic topic,

			final float trust) {

		final Oaf.Builder prototype = Oaf.newBuilder(current);

		for (final FieldTypeProtos.Author.Builder a : prototype.getEntityBuilder().getResultBuilder().getMetadataBuilder().getAuthorBuilderList()) {

			if (a.getFullname().equals(pair.getKey().getFullname())) {

				a.addAllPid(

						pair.getValue().getPidList().stream()

								.filter(p -> ORCID_TYPE_MARKER.equals(p.getKey()))

								.collect(Collectors.toList()));

			}

		}

		final Oaf oaf = prototype.build();

		final OpenAireEventPayload payload =

				HighlightFactory.highlightEnrichOrcidAuthor(OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), other.getEntity(), trust), pair.getValue());

		return EventWrapper.newInstance(

				asEvent(oaf.getEntity(), topic, payload, other.getEntity(), trust),

				payload.getHighlight().getCreators().stream().filter(s -> StringUtils.contains(s, ORCID_TYPE_MARKER)).collect(Collectors.joining(", ")),

				topic.getValue());

	}

	private List<FieldTypeProtos.Author> authors(final Oaf oaf) {

		final List<FieldTypeProtos.Author> authors = oaf.getEntity().getResult().getMetadata().getAuthorList();

		if (authors == null) { return Lists.newLinkedList(); }

		return authors;

	}

	private Float distance(final FieldTypeProtos.Author a, final FieldTypeProtos.Author b) {

		final JaroWinkler jaroWinkler = new JaroWinkler(1.0);

		if (a.hasSurname() && b.hasSurname()) {

			return (float) jaroWinkler.distance(getCanonicalName(a), getCanonicalName(b));

		} else {

			return (float) jaroWinkler.distance(a.getFullname(), b.getFullname());

		}

	}

	// returns the 1st letter of the author name + the author surname, all in lowercase

	// e.g. "pmanghi"

	private String getCanonicalName(final FieldTypeProtos.Author a) {

		return (StringUtils.substring(a.getName(), 0, 1) + a.getSurname()).toLowerCase();

	}

}