D-Net

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

import java.io.IOException;

import java.util.Map;

import java.util.Set;

import com.google.common.base.Predicate;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import eu.dnetlib.broker.objects.Instance;

import eu.dnetlib.broker.objects.OpenAireEventPayload;

import eu.dnetlib.broker.objects.Project;

import eu.dnetlib.broker.objects.Provenance;

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.mapping.ProtoMapping;

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

import eu.dnetlib.data.mapreduce.util.DedupUtils;

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

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

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;

import static eu.dnetlib.data.mapreduce.util.OafHbaseUtils.getValue;

@SuppressWarnings("rawtypes")

public class ProjectEventFactory extends ProtoMapping {

	private static String COUNTER_GROUP = "Broker Enrichment projects";

	private Set<String> inferenceProvenance = Sets.newHashSet("sysimport:mining:repository", "iis");

	private Set<String> aggregationProvenance = Sets.newHashSet("sysimport:crosswalk:repository");

	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 {

		/*

		if (!current.getRel().hasCachedOafTarget() || (other != null && !other.getRel().hasCachedOafTarget())) {

			context.getCounter(COUNTER_GROUP, "events skipped: missing project 2nd step").increment(1);

			return;

		}

		*/

		new ProjectEventFactory().processProject(context, current, other, trust);

	}

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

			throws IOException, InterruptedException, DocumentException {

		process(context, current, null, null);

	}

	private void processProject(final Context context, final Oaf current, final Oaf other, final Float trust)

			throws IOException, InterruptedException, DocumentException {

		if (other == null) {

			for (final Oaf oafRel : current.getEntity().getCachedOafRelList()) {

				final String provenance = oafRel.getDataInfo().getProvenanceaction().getClassid();

				if (inferenceProvenance.contains(provenance)) {

					final OafEntity project = oafRel.getRel().getCachedOafTarget().getEntity();

					doProcessProject(context, current, current, project, provenance, Topic.ENRICH_MISSING_PROJECT, trust(trust, oafRel));

				}

			}

		} else {

			for (final Oaf currentOafRel : current.getEntity().getCachedOafRelList()) {

				for (final Oaf otherOafRel : other.getEntity().getCachedOafRelList()) {

					final String currentTarget = currentOafRel.getRel().getTarget();

					final String otherTarget = otherOafRel.getRel().getTarget();

					if (!currentTarget.equals(otherTarget) && !DedupUtils.isRoot(current.getEntity().getId())) {

						final String provenance = otherOafRel.getDataInfo().getProvenanceaction().getClassid();

						final OafEntity project = otherOafRel.getRel().getCachedOafTarget().getEntity();

						boolean currentHasProject = Iterables.tryFind(current.getEntity().getCachedOafRelList(), new Predicate<Oaf>() {

							@Override

							public boolean apply(final Oaf oaf) {

								final String currentProjectId = oaf.getRel().getCachedOafTarget().getEntity().getId();

								//System.out.println(String.format("%s = %s ? %s", currentProjectId, project.getId(), currentProjectId.equals(project.getId())));

								return currentProjectId.equals(project.getId());

							}

						}).isPresent();

						if (!currentHasProject) {

							//System.out.println(String.format("generating event for other = %s\n\nproject = %s", other, project));

							doProcessProject(context, current, other, project, provenance, Topic.ENRICH_MISSING_PROJECT, trust(trust, currentOafRel));

						}

					}

				}

			}

		}

	}

	private void doProcessProject(final Context context,

			final Oaf current,

			final Oaf other,

			final OafEntity project,

			final String provenance,

			final Topic topic,

			final Float trust)

			throws IOException, InterruptedException, DocumentException {

		final OafEntity currentEntity = current.getEntity();

		final OafEntity otherEntity = other.getEntity();

		//System.out.println("ProjectEventFactory.doProcessProject = " + entity);

		final Provenance prov = getProvenance(otherEntity, provenance);

		final OpenAireEventPayload payload = addProject(OpenAireEventPayloadFactory.fromOAF(currentEntity, trust, prov), project);

		final EventMessage event = asEvent(currentEntity, topic, payload, otherEntity, trust);

		event.setPayload(HighlightFactory.highlightEnrichProject(payload, project, provenance).toJSON());

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

		context.getCounter(COUNTER_GROUP, "Event:  " + topic.getValue()).increment(1);

	}

	private OpenAireEventPayload addProject(final OpenAireEventPayload payload, final OafEntity project) {

		final Map<String, Project> projects = Maps.newHashMap();

		for(Project prj : payload.getPublication().getProjects()) {

			projects.put(prj.getCode() + prj.getTitle(), prj);

		}

		final Project hlProject = mapRelatedProject(project.getProject());

		projects.put(hlProject.getCode() + hlProject.getTitle(), hlProject);

		payload.getPublication().setProjects(Lists.newArrayList(projects.values()));

		return payload;

	}

	private Provenance getProvenance(final OafEntity entity, final String provenance) {

		if (inferenceProvenance.contains(provenance)) {

			return new Provenance()

					.setRepositoryName("OpenAIRE")

					.setUrl("https://beta.openaire.eu/search/publication?articleId=" + StringUtils.substringAfter(entity.getId(), "|"))

					.setId(Iterables.getFirst(entity.getOriginalIdList(), ""));

		}

		return new Provenance()

				.setRepositoryName(getValue(entity.getCollectedfromList()))

				.setUrl(Iterables.getFirst(mapInstances(entity.getResult().getInstanceList()), new Instance()).getUrl())

				.setId(getValue(entity.getOriginalIdList()));

	}

	private Float trust(final Float trust, final Oaf oaf) {

		final Float provenanceTrust = Float.valueOf(oaf.getDataInfo().getTrust());

		return trust != null ? trust * provenanceTrust : provenanceTrust;

	}

}