D-Net

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

import com.google.gson.Gson;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.mapreduce.hbase.dedup.fixrelation.Key;

import eu.dnetlib.data.mapreduce.hbase.propagation.Value;

import eu.dnetlib.data.mapreduce.hbase.propagation.communitythroughorganization.DedupedList;

import eu.dnetlib.data.mapreduce.hbase.propagation.communitythroughorganization.OrganizationMap;

import eu.dnetlib.data.mapreduce.hbase.propagation.projecttoresult.PropagationProjectToResultReducer;

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

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.TypeProtos;

import org.apache.avro.generic.GenericData;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

import org.apache.hadoop.hbase.mapreduce.TableMapper;

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

import org.apache.hadoop.io.Text;

import java.io.IOException;

import java.util.ArrayList;

import java.util.HashSet;

import java.util.List;

import java.util.Set;

import java.util.stream.Collectors;

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

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

import static eu.dnetlib.data.mapreduce.hbase.propagation.Utils.getEntity;

import static eu.dnetlib.data.mapreduce.hbase.propagation.Utils.getRelationTarget;

public class PropagationOrcidToResultMapper extends TableMapper<ImmutableBytesWritable, Text> {

    private static final Log log = LogFactory.getLog(PropagationOrcidToResultMapper.class); // NOPMD by marko on 11/24/08 5:02 PM

    private Text valueOut;

    private ImmutableBytesWritable keyOut;

    private String[] sem_rels;

    private String trust;

    @Override

    protected void setup(final Context context) throws IOException, InterruptedException {

        super.setup(context);

        valueOut = new Text();

        keyOut = new ImmutableBytesWritable();

        sem_rels = context.getConfiguration().getStrings("propagatetoorcid.semanticrelations", DEFAULT_RESULT_RELATION_SET);

        trust = context.getConfiguration().get("propagatetoorcid.trust","0.85");

    }

    @Override

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

        final TypeProtos.Type type = OafRowKeyDecoder.decode(keyIn.copyBytes()).getType();

        final OafProtos.OafEntity entity = getEntity(value, type);//getEntity already verified that it is not delByInference

        if (entity != null) {

            if (type == TypeProtos.Type.result){

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

                //verifico se il risultato ha una relazione semantica verso uno o piu' risultati.

                //per ogni risultato linkato con issupplementto o issupplementedby emetto:

                // id risultato linkato come chiave,

                // id risultato oggetto del mapping e lista degli autori del risultato oggetto del mapper come value

                for(String sem : sem_rels){

                     result_result.addAll(getRelationTarget(value, sem, context, COUNTER_PROPAGATION));

                }

                if(!result_result.isEmpty()){

                    List<String> authorlist = getAuthorList(entity.getResult().getMetadata().getAuthorList());

                    Emit e = new Emit();

                    e.setId(Bytes.toString(keyIn.get()));

                    e.setAuthor_list(authorlist);

                    valueOut.set(Value.newInstance(new Gson().toJson(e, Emit.class),

                            trust,

                            Type.fromsemrel).toJson());

                    for (String result: result_result){

                        keyOut.set(Bytes.toBytes(result));

                        context.write(keyOut,valueOut);

                        context.getCounter(COUNTER_PROPAGATION,"emit for sem_rel").increment(1);

                    }

                    //emetto anche id dell'oggetto del mapper come chiave e lista degli autori come valore

                        e.setId(keyIn.toString());

                        e.setAuthor_list(authorlist);

                        valueOut.set(Value.newInstance(new Gson().toJson(e, Emit.class), trust, Type.fromresult).toJson());

                        context.write(keyIn, valueOut);

                        context.getCounter(COUNTER_PROPAGATION,"emit for result with orcid").increment(1);

                }

            }

        }

    }

    private List<String> getAuthorList(List<FieldTypeProtos.Author> author_list){

        return author_list.stream().map(a -> new JsonFormat().printToString(a)).collect(Collectors.toList());

    }

}