D-Net

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

import java.io.IOException;

import java.util.List;

import java.util.Map;

import java.util.Objects;

import java.util.Set;

import java.util.stream.Collectors;

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.hbase.broker.model.EventWrapper;

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 final Set<String> inferenceProvenance = Sets.newHashSet("sysimport:mining:repository", "iis");

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

	protected Text tKey = new Text("");

	public static List<EventWrapper> 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; }

		 */

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

	}

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

			throws IOException, InterruptedException, DocumentException {

		return process(context, current, null, null);

	}

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

			throws IOException, InterruptedException, DocumentException {

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

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

					events.add(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();

						final boolean currentHasProject = Iterables.tryFind(current.getEntity().getCachedOafRelList(), 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));

							events.add(

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

						}

					}

				}

			}

		}

		return events;

	}

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

		return EventWrapper.newInstance(event,

				payload.getHighlight().getProjects()

						.stream()

						.filter(Objects::nonNull)

						.map(p -> p.getFunder() + ":" + p.getCode())

						.sorted()

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

				topic.getValue());

	}

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

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

		for (final 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;

	}

}