D-Net

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

import java.io.IOException;

import java.util.Arrays;

import java.util.Map;

import java.util.stream.Stream;

import com.google.common.collect.Lists;

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

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

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

import eu.dnetlib.data.proto.TypeProtos.Type;

import org.apache.commons.collections.MapUtils;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.util.Bytes;

/**

 * Created by claudio on 08/07/16.

 */

public class SoftwareEnrichmentMapper extends AbstractEnrichmentMapper {

	public static final String SOFTWARE = "software";

	public static final String PUBLICATION = "publication";

	public static final String RESULT_RESULT_RELATIONSHIP_IS_RELATED_TO = "resultResult_relationship_isRelatedTo";

	@Override

	protected String counterGroup() {

		return "Broker Enrichment Software";

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {

		final Type type = getEntityType(key);

		final byte[] body = value.getValue(Bytes.toBytes(type.toString()), Bytes.toBytes("body"));

		if (body == null) {

			context.getCounter(counterGroup(), "missing metadata").increment(1);

			return;

		}

		switch (type) {

		case result:

			final Oaf oaf = OafDecoder.decode(body).getOaf();

			if (oaf.getDataInfo().getDeletedbyinference()) {

				context.getCounter(counterGroup(), "result deletedbyinference").increment(1);

				return;

			}

			final Oaf.Builder oafBuilder = Oaf.newBuilder(oaf);

			final String resulttype = oafBuilder.getEntity().getResult().getMetadata().getResulttype().getClassid();

			if (SOFTWARE.equalsIgnoreCase(resulttype)) {

				for (final byte[] publicationId : listRelatedIds(value, RESULT_RESULT_RELATIONSHIP_IS_RELATED_TO)) {

					emit(context, publicationId, body, SOFTWARE);

				}

				break;

			} else if (PUBLICATION.equalsIgnoreCase(resulttype)) {

				for (final String relName : Arrays.asList(RESULT_RESULT_RELATIONSHIP_IS_RELATED_TO)) {

					listRelations(value, relName).forEach(rel -> {

						oafBuilder.getEntityBuilder().addCachedOafRel(rel);

						context.getCounter(counterGroup(), "rel: " + relName).increment(1);

					});

				}

				emit(context, key.copyBytes(), oafBuilder.build().toByteArray(), PUBLICATION);

				break;

			}

			break;

		default:

			throw new IllegalArgumentException("invalid type: " + type);

		}

	}

	private Stream<Oaf> listRelations(final Result value, final String relType) {

		//TODO consider only relationshipt not deletedbyinference

		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));

		return MapUtils.isEmpty(map) ? Stream.empty() : map.values().stream()

				.map(input -> OafDecoder.decode(input).getOaf())

				.filter(rel -> !rel.getRel().getTarget().contains("unidentified"));

	}

	private Iterable<byte[]> listRelatedIds(final Result value, final String relType) {

		//TODO consider only relationshipt not deletedbyinference

		final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(relType));

		return MapUtils.isNotEmpty(map) ? map.keySet() : Lists.newArrayList();

	}

	private Type getEntityType(final ImmutableBytesWritable key) {

		return OafRowKeyDecoder.decode(key.copyBytes()).getType();

	}

}