D-Net

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

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

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

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

import org.apache.commons.lang.StringUtils;

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

import org.dom4j.DocumentException;

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

import eu.dnetlib.broker.objects.Software;

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.proto.OafProtos.Oaf;

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

public class SoftwareEventFactory extends ProtoMapping {

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

	private Map<String, String> baseUrlMap;

	public SoftwareEventFactory(Map<String, String> baseUrlMap) {

		this.baseUrlMap = baseUrlMap;

	}

	public static List<EventWrapper> process(final Context context, final Oaf current, final Oaf other, final Float trust, final Map<String, String> baseUrlMap)

			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 SoftwareEventFactory(baseUrlMap).processSoftware(context, current, other, trust);

	}

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

			throws IOException, InterruptedException, DocumentException {

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

	}

	private List<EventWrapper> processSoftware(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 result = oafRel.getRel().getCachedOafTarget().getEntity();

					events.add(doProcessSoftware(context, current, current, result, provenance, Topic.ENRICH_MISSING_SOFTWARE, 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)) {

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

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

						final boolean currentHasSw = Iterables.tryFind(current.getEntity().getCachedOafRelList(), oaf -> {

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

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

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

							return currentSwId.equals(software.getId());

						}).isPresent();

						if (!currentHasSw) {

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

							events.add(doProcessSoftware(context, current, other, software, provenance, Topic.ENRICH_MISSING_PROJECT,

									trust(trust, currentOafRel)));

						}

					}

				}

			}

		}

		return events;

	}

	private EventWrapper doProcessSoftware(final Context context,

			final Oaf current,

			final Oaf other,

			final OafEntity software,

			final String provenance,

			final Topic topic,

			final Float trust)

			throws IOException, InterruptedException, DocumentException {

		final OafEntity currentEntity = current.getEntity();

		final OafEntity otherEntity = other.getEntity();

		final Provenance prov = getProvenance(otherEntity, provenance);

		final OpenAireEventPayload payload = addSoftware(OpenAireEventPayloadFactory.fromOAF(currentEntity, trust, prov), software);

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

		event.setPayload(HighlightFactory.highlightEnrichSoftware(payload, software, provenance).toJSON());

		return EventWrapper.newInstance(event,

				payload.getHighlight().getSoftwares().stream().filter(Objects::nonNull).map(s -> s.getName()).sorted()

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

				topic.getValue());

	}

	private OpenAireEventPayload addSoftware(final OpenAireEventPayload payload, final OafEntity software) {

		final Map<String, Software> swMap = Maps.newHashMap();

		for (final Software s : payload.getPublication().getSoftwares()) {

			swMap.put(s.getLandingPage(), s);

		}

		final Software hlSw = mapRelatedSoftware(software.getResult());

		swMap.put(hlSw.getLandingPage(), hlSw);

		payload.getPublication().setSoftwares(Lists.newArrayList(swMap.values()));

		return payload;

	}

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

		if (inferenceProvenance.contains(provenance)) { return new Provenance()

				.setRepositoryName("OpenAIRE")

				.setUrl(getUrl(entity))

				.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 String getUrl(OafEntity entity) {

		if (entity.getType().equals(TypeProtos.Type.result)) {

			String resulttype = entity.getResult().getMetadata().getResulttype().getClassid();

			return String.format(baseUrlMap.get(resulttype), StringUtils.substringAfter(entity.getId(), "|"));

		}

		return "";

	}

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

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

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

	}

}