D-Net

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

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 java.io.IOException;

import java.util.Map;

import java.util.stream.Collectors;

/**

 * Created by miriam on 17/08/2018.

 */

public class PropagationCountryInstitutionalOrganizationMapper extends TableMapper<Key, ImmutableBytesWritable> {

    private ImmutableBytesWritable valueOut;

    @Override

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

        super.setup(context);

        valueOut = new ImmutableBytesWritable();

    }

    @Override

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

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

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

        if (body != null){

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

            OafProtos.OafEntity entity = oaf.getEntity();

            switch (OafRowKeyDecoder.decode(keyIn.copyBytes()).getType()) {

                case datasource:

                    DatasourceProtos.Datasource datasource = entity.getDatasource();

                    if (datasource == null){

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

                    }

                    final Key datasource1 = Key.datasource(entity.getId());

                    if(datasource.getMetadata().getDatasourcetype().getClassid().equals("pubsrepository::institutional")){

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

                        valueOut.set("1".getBytes());

                        context.write(datasource1,valueOut);

                    }else{

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

                        valueOut.set("0".getBytes());

                        context.write(datasource1,valueOut);

                    }

                    break;

                case organization:

                    OrganizationProtos.Organization organization = entity.getOrganization();

                    if (organization == null){

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

                    }

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

                    if (country == null){

                        context.getCounter("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(country.getClassid().getBytes());

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

                                context.getCounter("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(entity.getId().getBytes());

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

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

                        if (!hostedBy.equals(collectedFrom)) {

                            context.write(Key.publication(collectedFrom),valueOut);

                        }

                    }

                    break;

            }

        }

//        if (body != null) {

//            context.getCounter("Bulk Tagging", "not null body ").increment(1);

//            final OafProtos.Oaf oaf = tagger.enrichContext(, cc, context);

//            if (oaf == null) {

//                //context.getCounter("In mapper", " null oaf ").increment(1);

//                return;

//            }

//

//            long tagged = oaf.getEntity().getResult().getMetadata().getContextList().stream()

//                    .flatMap(c -> c.getDataInfoList().stream())

//                    .map(FieldTypeProtos.DataInfo::getInferenceprovenance)

//                    .filter(infProv -> "bulktagging::community".equals(infProv))

//                    .count();

//            context.getCounter("Bulk Tagging", " bulktagged ").increment(tagged);

//

//

//            final Put put = new Put(key.copyBytes()).add(Bytes.toBytes("update"), Bytes.toBytes("update_"+System.nanoTime()), oaf.toByteArray());

//

//            if(tagged > 0){

//                context.write(key, put);

//                context.getCounter("Bulk Tagging", " write op ").increment(1);

//            }

//

//        }

//        else{

//            context.getCounter("Bulk Tagging", " null body ").increment(1);

//        }

    }

}