D-Net

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

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

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

import org.apache.commons.lang3.StringUtils;

import java.util.Collection;

import java.util.List;

import java.util.Objects;

import java.util.function.Function;

import java.util.stream.Collectors;

import java.util.stream.Stream;

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

/**

 * Created by claudio on 26/07/16.

 */

public class PIDEventFactory {

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

		return new PIDEventFactory().processPID(current, other, trust);

	}

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

		final Collection<StructuredProperty> pids = other.getEntity().getPidList()

				.stream()

				.collect(Collectors.toMap(

						p -> StringUtils.lowerCase(p.getQualifier().getClassid()) + ":" +StringUtils.lowerCase(p.getValue()),

						Function.identity(),

						(p1, p2) -> p1))

				.values();

		final Stream<EventWrapper> more = pids.stream()

				.filter(p -> !hasPidValue(current, p.getValue()))

				.map(p -> doProcessPID(current, other, p, Topic.ENRICH_MORE_PID, trust));

		if (current.getEntity().getPidList().isEmpty()) { return Stream.concat(

				more,

				pids.stream().map(p -> doProcessPID(current, other, p, Topic.ENRICH_MISSING_PID, trust)))

				.collect(Collectors.toList()); }

		return more.collect(Collectors.toList());

	}

	private EventWrapper doProcessPID(

			final Oaf current,

			final Oaf other,

			final StructuredProperty pidOther,

			final Topic topic,

			final float trust) {

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

		prototype.getEntityBuilder().addPid(pidOther);

		final Oaf oaf = prototype.build();

		final OpenAireEventPayload payload =

				HighlightFactory.highlightEnrichPid(OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), other.getEntity(), trust),

						Lists.newArrayList(pidOther));

		return EventWrapper.newInstance(

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

				payload.getHighlight().getPids().stream().filter(Objects::nonNull).map(p -> p.getType() + ":" + p.getValue()).sorted()

						.collect(Collectors.joining(", ")),

				topic.getValue());

	}

	private boolean hasPidValue(final Oaf oaf, final String value) {

		return oaf.getEntity().getPidList()

				.stream()

				.anyMatch(pid -> pid.getValue().equalsIgnoreCase(value));

	}

}