D-Net

import com.google.common.collect.Maps;

import eu.dnetlib.data.proto.OafProtos;

import org.apache.commons.lang3.StringUtils;

import org.dom4j.DocumentException;

import java.io.IOException;

import java.util.*;

import java.util.function.Predicate;

import java.util.stream.Collectors;

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

    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(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) {

                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(Oaf oaf, 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(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) {

        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(FieldTypeProtos.Author a) {

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

    }