D-Net

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

import com.google.gson.Gson;

import com.googlecode.protobuf.format.JsonFormat;

import com.sun.org.apache.commons.logging.Log;

import com.sun.org.apache.commons.logging.LogFactory;

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

import eu.dnetlib.data.proto.*;

import org.apache.commons.lang3.StringUtils;

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

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

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

import java.util.Iterator;

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

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

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

public class PropagationCountryFromDsOrgResultFileReducer extends Reducer<InstOrgKey, ImmutableBytesWritable,Text,Text> {

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

    private Text keyOut;

    private Text outValue;

    final static String DNETCOUNTRYSCHEMA = "dnet:countries";

    private final static String DATA_INFO_TYPE = "propagation";

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

    private final static String CLASS_ID = "propagation::country::instrepos";

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

    @Override

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

        super.setup(context);

        keyOut = new Text("");

        outValue = new Text();

    }

    private void emit(final Context context, final String key, final String data) {

        keyOut.set(key);

        outValue.set(data.getBytes());

        try {

            context.write(keyOut, outValue);

        } catch (Exception e) {

            e.printStackTrace();

            throw new RuntimeException(e);

        }

    }

    @Override

    protected void reduce(InstOrgKey dsId, Iterable<ImmutableBytesWritable> values, Context context)  {

        final Iterator<ImmutableBytesWritable> it = values.iterator();

        if(!it.hasNext()){

            context.getCounter(COUNTER_PROPAGATION,"empty information for key").increment(1);

            return;

        }

        Gson gson = new Gson();

        Value v = gson.fromJson(Bytes.toString(it.next().get()),Value.class);

        final String first  = Bytes.toString(v.getValue().get());

        final String trust = v.getTrust();

        if (!(first.equals(ZERO) || first.equals(ONE))) {

            if (dsId.getKeyType().get() == TypeProtos.Type.organization.getNumber())

                context.getCounter(COUNTER_PROPAGATION,"First Element in reducer is not type of datasource,  but the organization exists").increment(1);

            else {

               // context.getCounter(COUNTER_PROPAGATION,String.format("WARNINGS: First element value is  %s " , first));

                while (it.hasNext()) {

                    byte[] targetRowKey = gson.fromJson(Bytes.toString(it.next().get()), Value.class).getValue().get();

                    String resultId = Bytes.toString(targetRowKey);

                    if (!resultId.startsWith("50|"))

                        context.getCounter(COUNTER_PROPAGATION,"ERROR ORDERING CHECK").increment(1);

                }

            }

        }

//ensure we are dealing with an institutional repository

        if (first.equals(ONE)) {

            context.getCounter(COUNTER_PROPAGATION,"institutional datasource").increment(1);

            if(!it.hasNext()){

                context.getCounter(COUNTER_PROPAGATION,"no information apart of type of datasource").increment(1);

                return;

            }

            final String country = Bytes.toString(gson.fromJson(Bytes.toString(it.next().get()),Value.class).getValue().get()); // need to update the information for the country to each element in the iterator

            if(country.trim().length() != 2){

                context.getCounter(COUNTER_PROPAGATION,"second element in reducer is not country").increment(1);

                return;

            }

            boolean propagate = true;

            while(it.hasNext()) {

                byte[] targetRowKey = gson.fromJson(Bytes.toString(it.next().get()),Value.class).getValue().get();

                String resultId = Bytes.toString(targetRowKey);

                if (!resultId.startsWith(("50|"))) {

                    if(!resultId.equalsIgnoreCase(country)) {

                        context.getCounter(COUNTER_PROPAGATION, String.format("resultId expected in ordering was not found. First country %s, second country %s", country, resultId)).increment(1);

                        //throw new RuntimeException("Problem in country values of institutional repositories" + targetRowKey);

                        propagate = false;

                    }

                }else{

                    if (propagate){

                        OafProtos.Oaf oafUpdate = getOafCountry(resultId, country,trust);

                        emit(context, resultId, JsonFormat.printToString(oafUpdate));

                        context.getCounter(COUNTER_PROPAGATION, String.format(" added country %s to product from repo %s",country, StringUtils.substringBefore( resultId,"::"))).increment(1);

                    }

                }

            }

        } else {

            context.getCounter(COUNTER_PROPAGATION,"not allowed dsType institutional datasource").increment(1);

        }

    }

    private FieldTypeProtos.DataInfo.Builder getDataInfo(String trust) {

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

                .setInferred(true)

                .setProvenanceaction(

                        FieldTypeProtos.Qualifier.newBuilder()

                                .setClassid(CLASS_ID)

                                .setClassname("Propagation of country information from datasources belonging to institutional repositories")

                                .setSchemeid(SCHEMA_ID)

                                .setSchemename(SCHEMA_NAME))

                .setInferenceprovenance(DATA_INFO_TYPE)

                .setTrust(trust);

        return builder;

    }

    private OafProtos.Oaf getOafCountry(String resultId, String countryValue, String trust) {

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

                .setClassid(countryValue)

                .setClassname(countryValue)

                .setSchemeid(DNETCOUNTRYSCHEMA)

                .setSchemename(DNETCOUNTRYSCHEMA);

        country.setDataInfo(getDataInfo(trust));

        final ResultProtos.Result.Metadata.Builder metadata = ResultProtos.Result.Metadata.newBuilder().addCountry(country);

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

    }

}