D-Net

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

import com.google.common.base.Functions;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

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 eu.dnetlib.data.proto.ResultProtos.Result.Context;

import org.apache.commons.lang3.StringUtils;

import org.apache.hadoop.mapreduce.Mapper;

import java.util.*;

import java.util.stream.Collectors;

import java.util.stream.Stream;

/**

 * Created by miriam on 02/08/2018.

 */

public class ResultTagger {

    private final static String DATA_INFO_TYPE = "bulktagging";

    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";

    private final static String ZENODO_COMMUNITY_INDICATOR = "https://zenodo.org/communities/";

    private String trust = "0.8";

    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();

        if(contextList.size()>0){

            context.getCounter(COUNTER_GROUP, "exist context list").increment(1);

        }else{

            context.getCounter(COUNTER_GROUP, "not exist context list").increment(1);

        }

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

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

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

        oaf.getEntity().getResult().getMetadata().getSubjectList().stream()

                .map(subject -> subject.getValue())

                .filter(StringUtils::isNotBlank)

                .map(String::toLowerCase)

                .map(String::trim)

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

                .forEach(s -> subjects.addAll(conf.getCommunityForSubjectValue(s)));

        //updating counters for subject matching

        subjects.forEach(c->context.getCounter(COUNTER_GROUP, "Matching subject for community " + c).increment(1));

        communities.addAll(subjects);

        context.getCounter(COUNTER_GROUP,"match found for subjects ").increment(subjects.size());

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

        oaf.getEntity().getResult().getInstanceList()

                .stream()

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

                .flatMap(p -> Stream.of(p.getFst(), p.getSnd()))

                .map(s -> StringUtils.substringAfter(s, "|"))

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

                .forEach(dsId -> datasources.addAll(conf.getCommunityForDatasourceValue(dsId)));

        datasources.forEach(c->context.getCounter(COUNTER_GROUP,"Matching datasource for community " + c).increment(1));

        communities.addAll(datasources);

        context.getCounter(COUNTER_GROUP,"Match found for content providers " ).increment(datasources.size());

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

        final ResultProtos.Result.Metadata.Builder mBuilder = builder.getEntityBuilder().getResultBuilder().getMetadataBuilder();

        mBuilder.getContextBuilderList().stream().filter(cBuilder -> cBuilder.getId().contains(ZENODO_COMMUNITY_INDICATOR))

                .collect(Collectors.toList())

                .forEach(c->czenodo.addAll(conf.getCommunityForZenodoCommunityValue(c.getId().substring(c.getId().lastIndexOf("/")+1).trim())));

        czenodo.forEach(c->context.getCounter(COUNTER_GROUP,"Matching Zenodo community for community " + c).increment(1));

        context.getCounter(COUNTER_GROUP,"Match found for Zenodo communities " ).increment(czenodo.size());

        communities.addAll(czenodo);

        List<Context.Builder> clist = mBuilder.getContextBuilderList().stream()

                .filter(c -> (!c.getId().contains(ZENODO_COMMUNITY_INDICATOR))).collect(Collectors.toList());

        mBuilder.clearContext();

        clist.forEach(c->mBuilder.addContext(c));

        final Map<String, ResultProtos.Result.Context.Builder> cBuilders = Maps.newHashMap();

        if(communities.isEmpty()){

            context.getCounter(COUNTER_GROUP, "list of communities empty").increment(1);

        }else{

            context.getCounter(COUNTER_GROUP, "list of communities has values!").increment(1);

        }

        mBuilder.getContextBuilderList().forEach(cBuilder -> {

            cBuilders.put(cBuilder.getId(), cBuilder);

        });

        for(String contextId:communities){

            final ResultProtos.Result.Context.Builder cBuilder = cBuilders.get(contextId);

            if (cBuilder != null) {

                if (!cBuilder.getDataInfoBuilderList().stream()

                        .map(di -> di.getInferenceprovenance())

                        .anyMatch(s -> DATA_INFO_TYPE.equals(s))) {

                    cBuilder.addDataInfo(buildDataInfo());

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

                } else {

                    context.getCounter(COUNTER_GROUP, "provenance already bulk tagged").increment(1);

                }

            } else {

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

                mBuilder.addContext(buildContext(contextId));

            }

        }

        return builder.build();

    }

    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(trust);

        return builder

                .build();

    }

    public void setTrust(String s) {

        trust = s;

    }

}