D-Net

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

import java.io.IOException;

import com.google.common.base.Predicate;

import com.google.common.collect.Iterables;

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

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.io.Text;

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 {

	protected Text tKey = new Text("");

	public static void process(final Context context, final Oaf current, final Oaf other, final float trust)

			throws IOException, InterruptedException, DocumentException {

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

	}

	public static void process(final Context context, final Oaf current)

			throws IOException, InterruptedException, DocumentException {

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

	}

	private void processSubjectSingle(final Context context, final Oaf current) throws InterruptedException, DocumentException, IOException {

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

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

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

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

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

				}

			}

		}

	}

	private void 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();

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

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

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

					doProcessSubject(context, current, other, subjectOther, getSubjectTopic(subjectOther, false), trust);

				}

			}

		}

	}

	private void 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;

		}

		// 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));

		final EventMessage event = asEvent(oaf.getEntity(), topic, payload, eventEntity, adjustedTrust);

		context.write(tKey, new Text(event.toString()));

		context.getCounter("event", topic.getValue()).increment(1);

	}

	private Topic getSubjectTopic(final StructuredProperty subject, 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 Iterables.any(oaf.getEntity().getResult().getMetadata().getSubjectList(), new Predicate<StructuredProperty>() {

			@Override

			public boolean apply(final StructuredProperty subject) {

				return subject.getValue().equalsIgnoreCase(value) & subject.getQualifier().getClassid().endsWith(classId);

			}

		});

	}

}