D-Net

package eu.dnetlib.data.mapreduce.util;

import java.util.*;

import java.util.function.Predicate;

import java.util.stream.Collectors;

import java.util.stream.Stream;

import com.google.common.base.Splitter;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.common.collect.Sets;

import eu.dnetlib.data.proto.FieldTypeProtos.*;

import static eu.dnetlib.openaire.hadoop.utils.HBaseTableUtils.*;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.TypeProtos;

import eu.dnetlib.openaire.hadoop.utils.HBaseTableUtils;

import org.apache.commons.lang3.StringUtils;

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

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

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

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

import com.google.common.base.Function;

import com.google.protobuf.InvalidProtocolBufferException;

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

import eu.dnetlib.data.transform.OafUtils;

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

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Reducer;

public class OafHbaseUtils extends OafUtils {

	public static OafDecoder decode(final ImmutableBytesWritable oaf) {

		return new OafDecoder(oaf.copyBytes());

	}

	public static Function<ImmutableBytesWritable, OafDecoder> decoder() {

		return input -> OafDecoder.decode(input.copyBytes());

	}

	public static Iterable<Oaf> asOaf(final Iterable<ImmutableBytesWritable> in) {

		return Iterables.transform(in, oafDecoder());

	}

	public static Function<ImmutableBytesWritable, Oaf> oafDecoder() {

		return input -> parse(input);

	}

	public static Oaf parse(final ImmutableBytesWritable input) {

		try {

			return Oaf.parseFrom(input.copyBytes());

		} catch (final InvalidProtocolBufferException e) {

			throw new IllegalArgumentException(e);

		}

	}

	public static <T> String getValue(T t) {

		return mapValue(t);

	}

	public static <T> String getKey(T t) {

		return mapKey(t);

	}

	public static <T> String getValue(Iterable<T> ts) {

		return Iterables.getFirst(listValues(ts), "");

	}

	public static <T> Set<String> hashSetValues(Iterable<T> ts) {

		return Sets.newHashSet(Iterables.transform(ts, t -> mapValue(t)));

	}

	public static <T> List<String> listValues(Iterable<T> ts) {

		return Lists.newArrayList(Iterables.transform(ts, t -> mapValue(t)));

	}

	public static <T> String getKey(Iterable<T> ts) {

		return Iterables.getFirst(listKeys(ts), "");

	}

	public static <T> List<String> listKeys(Iterable<T> ts) {

		return Lists.newArrayList(Iterables.transform(ts, t -> mapKey(t)));

	}

	public static <T> Set<String> hashSetKeys(Iterable<T> ts) {

		return Sets.newHashSet(Iterables.transform(ts, t -> mapKey(t)));

	}

	private static <T> String mapKey(final T t) {

		if (t instanceof KeyValue) return ((KeyValue) t).getKey();

		if (t instanceof String) return (String) t;

		if (t instanceof Qualifier) return ((Qualifier) t).getClassid();

		throw new IllegalArgumentException(String.format("type %s not mapped", t.getClass()));

	}

	private static <T> String mapValue(final T t) {

		if (t instanceof StructuredProperty) return ((StructuredProperty) t).getValue();

		if (t instanceof KeyValue) return ((KeyValue) t).getValue();

		if (t instanceof String) return (String) t;

		if (t instanceof StringField) return ((StringField) t).getValue();

		if (t instanceof Qualifier) return ((Qualifier) t).getClassname();

		if (t instanceof Author) return ((Author) t).getFullname();

		throw new IllegalArgumentException(String.format("type %s not mapped", t.getClass()));

	}

	public static List<String> getPropertyValues(final Reducer.Context context, final String name) {

		return doGetPropertyValues(context.getConfiguration().get(name, ""));

	}

	public static List<String> getPropertyValues(final Mapper.Context context, final String name) {

		return doGetPropertyValues(context.getConfiguration().get(name, ""));

	}

	private static List<String> doGetPropertyValues(final String s) {

		return Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(s));

	}

	public static List<Oaf> rel(final Result value) {

		return value.list().stream()

				.filter(kv -> {

					final String q = new String(kv.getQualifier());

					return q.matches(OafRowKeyDecoder.ID_REGEX);

				})

				.filter(kv -> kv.getValue() != null && kv.getValue().length > 0)

				.map(kv -> parseProto(kv.getValue()))

				.collect(Collectors.toList());

	}

	public static Oaf parseProto(final byte[] value) {

		final OafDecoder d = OafDecoder.decode(value);

		return d.getOaf();

	}

	public static Put asPut(final Oaf oaf) {

		switch (oaf.getKind()) {

			case entity:

				final Put entity = getPut(oaf.getEntity().getId());

				byte[] cf = Bytes.toBytes(oaf.getEntity().getType().toString());

				return entity.add(cf, DedupUtils.BODY_B, oaf.toByteArray());

			case relation:

				final OafProtos.OafRel rel = oaf.getRel();

				final Put putRel = getPut(rel.getSource());

				final OafRelDecoder relDecoder = OafRelDecoder.decode(rel);

				final byte[] cfRel = Bytes.toBytes(relDecoder.getCFQ());

				final byte[] qualifier = Bytes.toBytes(rel.getTarget());

				return putRel.add(cfRel, qualifier, oaf.toByteArray());

			default:

				throw new IllegalArgumentException("invalid kind");

		}

	}

	private static Put getPut(final String rowkey) {

		final Put put = new Put(Bytes.toBytes(rowkey));

		put.setDurability(Durability.SKIP_WAL);

		return put;

	}

	public static byte[] getBodyB(final Result value, final TypeProtos.Type type) {

		return value.getValue(Bytes.toBytes(type.toString()), DedupUtils.BODY_B);

	}

	public static Oaf getBody(final Result value, final TypeProtos.Type type) throws InvalidProtocolBufferException {

		final byte[] body = getBodyB(value, type);

		return body != null ? parseProto(body) : null;

	}

}