D-Net

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

import java.io.IOException;

import java.util.List;

import java.util.stream.Collectors;

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 eu.dnetlib.data.proto.OafProtos.OafEntity;

import eu.dnetlib.data.proto.ResultProtos.Result.Metadata;

import org.apache.commons.lang.StringUtils;

import org.apache.hadoop.mapreduce.Reducer.Context;

import org.dom4j.DocumentException;

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

/**

 * Created by claudio on 16/11/2016.

 */

public class SubjectEventFactory {

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

			throws IOException, InterruptedException, DocumentException {

		return new SubjectEventFactory().processSubjectPair(context, current, other, trust);

	}

	public static List<EventWrapper> process(final Context context, final Oaf current)

			throws IOException, InterruptedException, DocumentException {

		return new SubjectEventFactory().processSubjectSingle(context, current);

	}

	private List<EventWrapper> processSubjectSingle(final Context context, final Oaf current) throws InterruptedException, DocumentException, IOException {

		final Metadata mCurrent = current.getEntity().getResult().getMetadata();

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

		if (mCurrent.getSubjectList().isEmpty()) {

			for (final StructuredProperty subject : mCurrent.getSubjectList()) {

				if (subject.hasDataInfo() && subject.getDataInfo().getInferred()) {

					events.add(

							doProcessSubject(context, current, null, subject, getSubjectTopic(subject, true), Float.valueOf(subject.getDataInfo().getTrust())));

				}

			}

		}

		return events;

	}

	private List<EventWrapper> processSubjectPair(final Context context, final Oaf current, final Oaf other, final Float trust)

			throws IOException, InterruptedException {

		final Metadata mCurrent = current.getEntity().getResult().getMetadata();

		final Metadata mOther = other.getEntity().getResult().getMetadata();

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

		if (mCurrent.getSubjectList().isEmpty()) {

			for (final StructuredProperty subjectOther : mOther.getSubjectList()) {

				events.add(doProcessSubject(context, current, other, subjectOther, getSubjectTopic(subjectOther, true), trust));

			}

		}

		if (other != null) {

			for (final StructuredProperty subjectOther : mOther.getSubjectList()) {

				if (!hasSubjectValue(current, subjectOther.getQualifier().getClassid(), subjectOther.getValue())) {

					final EventWrapper e = doProcessSubject(context, current, other, subjectOther, getSubjectTopic(subjectOther, false), trust);

					if (e != null) {

						events.add(e);

					}

				}

			}

		}

		return events;

	}

	private EventWrapper doProcessSubject(final Context context,

			final Oaf current,

			final Oaf other,

			final StructuredProperty subjectOther,

			final Topic topic,

			final Float trust) throws IOException, InterruptedException {

		if (topic == null) {

			context.getCounter("events skipped", "SUBJECT not mapped").increment(1);

			return null;

		}

		// we got java.lang.NumberFormatException: empty String, because of empty trust values. Default to 1.0 in such cases

		final String t = subjectOther.getDataInfo().getTrust();

		final float adjustedTrust = trust * Float.valueOf(StringUtils.isBlank(t) ? "1.0" : t);

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

		prototype.getEntityBuilder().getResultBuilder().getMetadataBuilder().addSubject(subjectOther);

		final Oaf oaf = prototype.build();

		final OafEntity eventEntity = other != null ? other.getEntity() : current.getEntity();

		final OpenAireEventPayload payload =

				HighlightFactory.highlightEnrichSubject(OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), eventEntity, adjustedTrust),

						Lists.newArrayList(subjectOther));

		return EventWrapper.newInstance(

				asEvent(oaf.getEntity(), topic, payload, eventEntity, adjustedTrust),

				payload.getHighlight().getSubjects().stream().filter(StringUtils::isNotBlank).sorted().collect(Collectors.joining(", ")),

				topic.getValue());

	}

	private Topic getSubjectTopic(final StructuredProperty subject, final boolean missingOrMore) {

		switch (subject.getQualifier().getClassid()) {

		case "mesheuropmc":

			return missingOrMore ? Topic.ENRICH_MISSING_SUBJECT_MESHEUROPMC : Topic.ENRICH_MORE_SUBJECT_MESHEUROPMC;

		case "arxiv":

			return missingOrMore ? Topic.ENRICH_MISSING_SUBJECT_ARXIV : Topic.ENRICH_MORE_SUBJECT_ARXIV;

		case "jel":

			return missingOrMore ? Topic.ENRICH_MISSING_SUBJECT_JEL : Topic.ENRICH_MORE_SUBJECT_JEL;

		case "ddc":

			return missingOrMore ? Topic.ENRICH_MISSING_SUBJECT_DDC : Topic.ENRICH_MORE_SUBJECT_DDC;

		case "acm":

			return missingOrMore ? Topic.ENRICH_MISSING_SUBJECT_ACM : Topic.ENRICH_MORE_SUBJECT_ACM;

		case "rvk":

			return missingOrMore ? Topic.ENRICH_MISSING_SUBJECT_RVK : Topic.ENRICH_MORE_SUBJECT_RVK;

		default:

			return null;

		}

	}

	private boolean hasSubjectValue(final Oaf oaf, final String classId, final String value) {

		return oaf.getEntity().getResult().getMetadata().getSubjectList()

				.stream()

				.anyMatch(s -> s.getValue().equalsIgnoreCase(value) & s.getQualifier().getClassid().endsWith(classId));

	}

}