D-Net

package eu.dnetlib.data.mapreduce.hbase.dataexport.linkinganalysis.publicationsoftware;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.proto.OafProtos;

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

import java.util.stream.Collectors;

import eu.dnetlib.data.mapreduce.hbase.dataexport.linkinganalysis.publicationsoftware.Constants.Type;

public class PubSfwLinkMapper extends TableMapper<ImmutableBytesWritable, Text> {

    private ImmutableBytesWritable keyOut;

    private Text valueOut;

    @Override

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

        super.setup(context);

        keyOut = new ImmutableBytesWritable();

        valueOut = new Text();

    }

    @Override

    protected void map(final ImmutableBytesWritable key, 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) {

            context.getCounter("Find Link Pub Software", "not null body ").increment(1);

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

            if (oaf == null) {

                return;

            }

            if(oaf.getDataInfo().getDeletedbyinference()){

                context.getCounter("Find Link Pub Software","deleted by inference").increment(1);

                return;

            }

            List<String> lst = oaf.getEntity().getPidList().stream()

                    .map(pid -> pid.getQualifier().getClassid() + "," + pid.getValue())

                    .collect(Collectors.toList());

            switch (oaf.getEntity().getResult().getMetadata().getResulttype().getClassid()) {

                case "publication":

                    context.getCounter("Find Link Pub Software", "publication").increment(1);

                    emitI(oaf.getEntity().getId(), lst, Constants.Type.publication, context);

                    break;

                case "software":

                    context.getCounter("Find Link Pub Software", "software").increment(1);

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

                    List<String> targets = relationMap.values().stream()

                            .map(this::asOaf)

                            .filter(Objects::nonNull)

                            .filter(o -> isValid(o))

                            .filter(o -> !o.getDataInfo().getDeletedbyinference())

                            .map(o -> o.getRel().getTarget())

                            .collect(Collectors.toList());

                    List<String> urls = oaf.getEntity().getResult().getInstanceList().stream().map(i -> oaf.getEntity().getId() + ";" +

                            String.join(",", i.getUrlList().stream().collect(Collectors.toCollection(HashSet::new)))

                    ).collect(Collectors.toList());

                    for(String target:targets)

                        emit(target,lst,urls, Type.software, context);

                    break;

            }

        }

    }

    private Identifier getIdentifier(String pid){

        String[] tmp = pid.split(",");

        return new Identifier().setPidtype(tmp[0].trim()).setPid(tmp[1].trim());

    }

    private List<Identifier> getIdentifiers(List<String> pids){

        List<Identifier> ret = new ArrayList<>();

        for(String pid : pids)    {

            ret.add(getIdentifier(pid));

        }

        return ret;

    }

    private void emit(String target, List<String> lst, List<String> urls, Type type, Context context) throws IOException, InterruptedException {

        keyOut.set(target.getBytes());

        List<Identifier> sfwPids = getIdentifiers(lst);

        for(String url:urls){

            String[] tmp= url.split(";");

            final String softwareID = tmp[0].trim();//softwareId

            final List<String> webresource = Arrays.asList(tmp[1].split(","));//urls

            SoftwareResource sr = new SoftwareResource().setOpenAireId(softwareID).setUrls(webresource).setSoftwarePIDs(sfwPids);

            valueOut.set(Value.newInstance(sr.toJson(),type).toJson());

            context.write(keyOut, valueOut);

        }

    }

    private OafProtos.Oaf asOaf(byte[] r) {

        try {

            return OafProtos.Oaf.parseFrom(r);

        } catch (InvalidProtocolBufferException e) {

            return null;

        }

    }

    private boolean isValid(final OafProtos.Oaf oaf) {

        return (oaf != null) && oaf.isInitialized();

    }

    private void emitI (String identifier, List<String> values, Type type, Context context) throws IOException, InterruptedException {

        keyOut.set(identifier.getBytes());

        for(String value : values){

                valueOut.set(Value.newInstance(getIdentifier(value).toJson(), type).toJson());

                context.write(keyOut, valueOut);

            }

        }

    }