D-Net

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

import eu.dnetlib.data.bulktag.CommunityConfiguration;

import eu.dnetlib.data.bulktag.Pair;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.ResultProtos;

import org.apache.hadoop.mapreduce.Mapper;

import java.util.*;

import java.util.stream.Collectors;

/**

 * Created by miriam on 02/08/2018.

 */

public class ResultTagger {

    private final static String DATA_INFO_TYPE = "bulktagging::community";

    private final static String SCHEMA_NAME = "dnet:provenanceActions";

    private final static String CLASS_ID = "bulktagging::community";

    private final static String SCHEMA_ID = "dnet:provenanceActions";

    private final static String COUNTER_GROUP = "Bulk Tagging";

    public OafProtos.Oaf enrichContext(final OafProtos.Oaf oaf, final CommunityConfiguration conf, final Mapper.Context context) {

        context.getCounter(COUNTER_GROUP, "to enrich").increment(1);

        final OafProtos.Oaf.Builder builder = OafProtos.Oaf.newBuilder(oaf);

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

            context.getCounter(COUNTER_GROUP, "deleted by inference").increment(1);

            return null;

        }

        context.getCounter(COUNTER_GROUP, "not deleted by inference").increment(1);

        final List<ResultProtos.Result.Context> contextList = oaf.getEntity().getResult().getMetadata().getContextList();

        //communities contains all the communities to be added as context for the result

        final Set<String> communities = new HashSet<>();

        List<FieldTypeProtos.StructuredProperty> subjectSet = oaf.getEntity().getResult().getMetadata().getSubjectList();

        communities.addAll(analiseSubjects(subjectSet, conf));

        List<Pair<String, String>> datasourceSet = oaf.getEntity().getResult().getInstanceList()

                .stream()

                .map(i -> new Pair<>(i.getCollectedfrom().getKey(), i.getHostedby().getKey()))

                .collect(Collectors.toList());

        communities.addAll(analiseDatasource(datasourceSet, conf));

        //TODO: add code for Zenodo Communities

        final Map<String,Pair<Integer,ResultProtos.Result.Context>> contexts = new HashMap<>();

        for(int i = 0; i<contextList.size(); i++){

            contexts.put(contextList.get(i).getId(),new Pair<>(i,contextList.get(i)));

        }

        for(String contextId:communities){

            final Pair<Integer,ResultProtos.Result.Context> pair = contexts.get(contextId);

            ResultProtos.Result.Context c;

            if(pair == null)

                c = null;

             else

                 c = pair.getSnd();

            if (c != null) {

                // add new dataInfo if needed

                    Set<String> set = new HashSet<>();

                    set.addAll(c.getDataInfoList().stream().map(datainfo->datainfo.getInferenceprovenance()).collect(Collectors.toList()));

                    if (!set.contains(DATA_INFO_TYPE)) {

                        builder.getEntityBuilder().getResultBuilder().getMetadataBuilder().getContextBuilder(pair.getFst()).addDataInfo(buildDataInfo());

                       context.getCounter(COUNTER_GROUP, "add provenance").increment(1);

                    }

            } else {

                builder.getEntityBuilder().getResultBuilder().getMetadataBuilder().addContext(buildContext(contextId));

                context.getCounter(COUNTER_GROUP, "add context").increment(1);

            }

        }

        return builder.build();

    }

    private Set<String> analiseDatasource(List<Pair<String, String>> datasourceSet, final CommunityConfiguration conf) {

        Set<String> set = new HashSet<>();

        for(Pair<String,String> d: datasourceSet){

            if(!d.getFst().equalsIgnoreCase(d.getSnd())){

                set.addAll(conf.getCommunityForDatasourceValue(d.getFst().substring(d.getFst().indexOf("|")+1)));

            }

            set.addAll((conf.getCommunityForDatasourceValue(d.getSnd().substring(d.getSnd().indexOf("|")+1))));

        }

        return set;

    }

    private Set<String> analiseSubjects(List<FieldTypeProtos.StructuredProperty> subjectList, final CommunityConfiguration conf) {

        Set<String> set = new HashSet<>();

        for(FieldTypeProtos.StructuredProperty s:subjectList){

            set.addAll(conf.getCommunityForSubjectValue( s.getValue()));

        }

        return set;

    }

    private ResultProtos.Result.Context buildContext(final String contextId) {

        return ResultProtos.Result.Context.newBuilder()

                .setId(contextId)

                .addDataInfo(buildDataInfo())

                .build();

    }

    private FieldTypeProtos.DataInfo buildDataInfo() {

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

                .setInferred(true)

                .setProvenanceaction(

                        FieldTypeProtos.Qualifier.newBuilder()

                                .setClassid(CLASS_ID)

                                .setClassname("Bulk Tagging for Communities")

                                .setSchemeid(SCHEMA_ID)

                                .setSchemename(SCHEMA_NAME))

                .setInferenceprovenance(DATA_INFO_TYPE)

                .setTrust("0.85");

        return builder

                .build();

    }

}