D-Net

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

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import com.google.common.collect.Sets;

import com.google.protobuf.InvalidProtocolBufferException;

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

import eu.dnetlib.data.mapreduce.hbase.propagation.compositekeys.InstOrgKey;

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

import eu.dnetlib.data.proto.*;

import org.apache.commons.collections.MapUtils;

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.Map;

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.Utils.*;

/**

 * Created by miriam on 17/08/2018.

 */

public class PropagationCountryFromDsOrgResultMapper extends TableMapper<InstOrgKey, Text> {

    private Text valueOut;

    private Set<String> datasourceTypes = Sets.newHashSet("pubsrepository::institutional");

    private Set<String> whiteList = Sets.newHashSet("10|opendoar____::300891a62162b960cf02ce3827bb363c");

    private Set<String> blackList = Sets.newHashSet("");

    @Override

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

        super.setup(context);

        valueOut = new Text();

        datasourceTypes.addAll(getParam(context, "datasource.types"));

        whiteList.addAll(getParam(context, "datasource.whitelist"));

    }

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

        if (entity != null) {

            switch (type) {

                case datasource:

                    final DatasourceProtos.Datasource datasource = entity.getDatasource();

                    final String id = entity.getId();

                    if (datasource == null) {

                        throw new RuntimeException("oaf type is datasource, but datasource proto is not found in oafproto");

                    }

                    String dsType = datasource.getMetadata().getDatasourcetype().getClassid();

                    if (datasourceTypes.contains(dsType)) {

                        // verify datasource is in blacklist

                        if (blackList.contains(id)){

                            context.getCounter(COUNTER_PROPAGATION,"blacklisted").increment(1);

                            emitNotAllowedDatasource(context,entity.getId());

                        } else {

                            emitAllowedDatasource(value, context, entity.getId(), dsType);

                        }

                    } else {

                        // verify datasource is in whiteList

                        if (whiteList.contains(id)) {

                            context.getCounter(COUNTER_PROPAGATION,"whitelisted " + id).increment(1);

                            emitAllowedDatasource(value,context,entity.getId(),dsType);

                        } else {

                            emitNotAllowedDatasource(context, entity.getId());

                        }

                    }

                    break;

                case organization:

                    OrganizationProtos.Organization organization = entity.getOrganization();

                    if (organization == null) {

                        throw new RuntimeException("oaf type is organizationtoresult, but organizationtoresult proto is not found in oafproto");

                    }

                    FieldTypeProtos.Qualifier country = organization.getMetadata().getCountry();

                    if (country == null) {

                        context.getCounter(COUNTER_PROPAGATION, "country elem does not exists").increment(1);

                    } else {

                        final Map<byte[], byte[]> ds_org = value.getFamilyMap("datasourceOrganization_provision_isProvidedBy".getBytes());

                        if (MapUtils.isNotEmpty(ds_org)) {

                            for (String dsId : ds_org.keySet().stream().map(String::new).collect(Collectors.toList())) {

                                valueOut.set(Value.newInstance(country.getClassid()).toJson());

                                context.write( InstOrgKey.organization(dsId), valueOut);

                                context.getCounter(COUNTER_PROPAGATION, "country ").increment(1);

                            }

                        }

                    }

                    break;

                case result:

                    ResultProtos.Result result = entity.getResult();

                    for (ResultProtos.Result.Instance instance : result.getInstanceList()) {

                        //todo add check if key is not empty and field is not null

                        String hostedBy = instance.getHostedby().getKey();

                        valueOut.set(Value.newInstance(entity.getId()).toJson());

                        context.write( InstOrgKey.publication(hostedBy),valueOut);

                        context.getCounter(COUNTER_PROPAGATION, "emit hostedby | collectedfrom for publication ").increment(1);

                        String collectedFrom = instance.getCollectedfrom().getKey();

                        if (!hostedBy.equals(collectedFrom)) {

                            context.write(eu.dnetlib.data.mapreduce.hbase.propagation.compositekeys.InstOrgKey.publication(collectedFrom), valueOut);

                            context.getCounter(COUNTER_PROPAGATION, "emit hostedby | collectedfrom for publication ").increment(1);

                        }

                    }

                    break;

            }

        }

    }

    private void emitNotAllowedDatasource(Context context, String id) throws IOException, InterruptedException {

        valueOut.set(Value.newInstance(ZERO).toJson());

        context.write( InstOrgKey.datasource(id), valueOut);

        context.getCounter(COUNTER_PROPAGATION, "ds Type not in propagation allowed list").increment(1);

    }

    private void emitAllowedDatasource(Result value, Context context, String id, String dsType) throws IOException, InterruptedException {

        valueOut.set(Value.newInstance(ONE, getTrust(value)).toJson());

        context.write( InstOrgKey.datasource(id), valueOut);

        context.getCounter(COUNTER_PROPAGATION, String.format("%s in propagation allowed list", dsType)).increment(1);

    }

    private String getTrust(Result value) throws InvalidProtocolBufferException {

        final Map<byte[],byte[]> map = value.getFamilyMap(Bytes.toBytes("datasource"));

        final byte[] body = map.get(Bytes.toBytes("body"));

        if (body != null){

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

            return oaf.getDataInfo().getTrust();

        }

        return null;

    }

    private ArrayList<String> getParam(Context context, String s) {

        return Lists.newArrayList(Splitter.on(",").omitEmptyStrings().split(context.getConfiguration().get(s, "")));

    }

}