D-Net

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

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.proto.*;

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

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

import org.apache.hadoop.mapreduce.Mapper;

import java.util.HashSet;

import java.util.Map;

import java.util.Objects;

import java.util.Set;

import java.util.stream.Collectors;

import java.util.stream.Stream;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.DATA_INFO_TYPE;

public final class Utils {

    public static final int MAX_RELATIONS = 10000;

    public static  OafProtos.OafEntity getEntity(Result value, TypeProtos.Type type) throws InvalidProtocolBufferException {

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

        if (body != null){

            OafProtos.Oaf oaf = OafProtos.Oaf.parseFrom(body);

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

                return null;

            return oaf.getEntity();

        }

        return null;

    }

    public static FieldTypeProtos.DataInfo.Builder getDataInfo(String trust, String class_id,String schema_id,String schema_name, String data_info_type,String class_name) {

        FieldTypeProtos.DataInfo.Builder builder = FieldTypeProtos.DataInfo.newBuilder()

                .setInferred(true)

                .setProvenanceaction(getQualifier(class_id,schema_id,schema_name,class_name)                        )

                .setInferenceprovenance(data_info_type)

                .setTrust(trust);

        return builder;

    }

    public static FieldTypeProtos.Qualifier.Builder getQualifier(String class_id, String schema_id, String schema_name, String class_name){

        return FieldTypeProtos.Qualifier.newBuilder()

                .setClassid(class_id)

                .setClassname(class_name)

                .setSchemeid(schema_id)

                .setSchemename(schema_name);

    }

    public static FieldTypeProtos.Qualifier.Builder getCountry(String countryValue, String trust,String dnet_schema, String class_id,String schema_id,String schema_name, String data_info_type, String class_name) {

        final FieldTypeProtos.Qualifier.Builder country = FieldTypeProtos.Qualifier.newBuilder()

                .setClassid(countryValue)

                .setClassname(countryValue)

                .setSchemeid(dnet_schema)

                .setSchemename(dnet_schema);

        country.setDataInfo(getDataInfo(trust,class_id,schema_id,schema_name,data_info_type,class_name));//"Propagation of country information from datasources belonging to institutional repositories"));

        return country;

    }

    public static  ResultProtos.Result.Context getContext(String contextid, String trust) {

        final ResultProtos.Result.Context.Builder cBuilder = ResultProtos.Result.Context.newBuilder()

                .setId(contextid)

                .addDataInfo(getDataInfo(trust, CLASS_COMMUNITY_ID, SCHEMA_ID, SCHEMA_NAME, DATA_INFO_TYPE, DATA_INFO_TYPE));

        return cBuilder.build();

    }

    public static  ResultProtos.Result.Context getContext(String contextid, String trust, String class_id, String dataInfoType, String class_name) {

        final ResultProtos.Result.Context.Builder cBuilder = ResultProtos.Result.Context.newBuilder()

                .setId(contextid)

                .addDataInfo(getDataInfo(trust, class_id, SCHEMA_ID, SCHEMA_NAME, dataInfoType, class_name));

        return cBuilder.build();

    }

    public static Set<String> getRelationTarget(Result value, String sem_rel, final Mapper.Context context, String counter) {

        HashSet<String> valid_relation = Stream.of(value.raw())

                .map(kv -> kv.split())

                .filter(s -> sem_rel.equals(new String(s.getFamily())))

                .map(s -> asOaf(s.getValue()))

                .filter(Objects::nonNull)

                .filter(o -> isValid(o))

                .filter(o -> !o.getDataInfo().getDeletedbyinference())

                .map(o -> o.getRel().getTarget())

                .limit(MAX_RELATIONS)

                .collect(Collectors.toCollection(HashSet::new));

        /*

        we could extract the target qualifiers from the familyMap's keyset, but we also need to check the relationship is not deletedbyinference

        return relationMap.keySet().stream()

                .map(String::new)

                .collect(Collectors.toCollection(HashSet::new));

        */

        context.getCounter(counter, sem_rel).increment(valid_relation.size());

        return valid_relation;

    }

    private static OafProtos.Oaf asOaf(byte[] r) {

        try {

            return OafProtos.Oaf.parseFrom(r);

        } catch (InvalidProtocolBufferException e) {

            return null;

        }

    }

    public static OafProtos.Oaf getUpdate(ResultProtos.Result.Metadata.Builder metadata, String resultId) {

        final ResultProtos.Result.Builder result = ResultProtos.Result.newBuilder().setMetadata(metadata);

        final OafProtos.OafEntity.Builder entity = OafProtos.OafEntity.newBuilder()

                .setType(TypeProtos.Type.result)

                .setId(resultId)

                .setResult(result);

        return OafProtos.Oaf.newBuilder()

                .setKind(KindProtos.Kind.entity)

                .setEntity(entity)

                .build();

    }

    private static boolean isValid(final OafProtos.Oaf oaf) {

        return (oaf != null) && oaf.isInitialized();

    }

}