D-Net

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

import com.google.gson.JsonObject;

import com.google.gson.JsonParser;

import eu.dnetlib.actionmanager.actions.ActionFactory;

import eu.dnetlib.actionmanager.actions.AtomicAction;

import eu.dnetlib.actionmanager.common.Agent;

import eu.dnetlib.miscutils.datetime.DateUtils;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

public class ScholexplorerMapper extends Mapper<LongWritable, Text, Text, Text> {

    private ActionFactory factory;

    private JsonParser parser;

    private String setName;

    private Agent agent;

    private String nsPrefix;

    private String dsName;

    private String dsId;

    private String dateOfCollection;

    private Text keyout;

    private Text valueOut;

    private Map<String, ScholExplorerConfiguration> configurationMap= new HashMap<>();

    @Override

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

        factory = new ActionFactory();

        parser = new JsonParser();

        configurationMap.put("issn", new ScholExplorerConfiguration(null, false));

        configurationMap.put("pmid", new ScholExplorerConfiguration("pmid", true,"https://www.ncbi.nlm.nih.gov/pubmed/%s"));

        configurationMap.put("doi", new ScholExplorerConfiguration("doi", true,"http://dx.doi.org/%s"));

        configurationMap.put("pbmid", new ScholExplorerConfiguration("pmid", true,"https://www.ncbi.nlm.nih.gov/pubmed/%s"));

        configurationMap.put("openaire ", new ScholExplorerConfiguration(null, false));

        configurationMap.put("pmcid", new ScholExplorerConfiguration("pmc", true,"https://europepmc.org/articles/%s"));

        configurationMap.put("pubmedid", new ScholExplorerConfiguration("pmid", true,"https://www.ncbi.nlm.nih.gov/pubmed/%s"));

        configurationMap.put("icpsr", new ScholExplorerConfiguration(null, false));

        configurationMap.put("dnet", new ScholExplorerConfiguration(null, false));

        configurationMap.put("url ", new ScholExplorerConfiguration(null, true,"%s"));

        configurationMap.put("openaire", new ScholExplorerConfiguration(null, false));

        setName = context.getConfiguration().get("setName");

        agent= new Agent(context.getConfiguration().get("agentId"), context.getConfiguration().get("agentName"), Agent.AGENT_TYPE.service);

        nsPrefix = context.getConfiguration().get("ns_prefix");

        dsName = context.getConfiguration().get("dsName");

        dsId = context.getConfiguration().get("dsId");

        dateOfCollection = context.getConfiguration().get("dateOfCollection", DateUtils.now_ISO8601());

        keyout = new Text("");

        valueOut = new Text("");

    }

    @Override

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        final String inputJson =value.toString();

        final JsonObject rootElement = parser.parse(inputJson).getAsJsonObject();

        final List<AtomicAction> actions = ScholixToActions.generateActionsFromScholix(rootElement, configurationMap, setName,agent, factory, nsPrefix, dsName, dsId, dateOfCollection);

        for(final AtomicAction action : actions) {

            keyout.set(action.getRowKey());

            valueOut.set(action.toJSON());

            context.write(keyout, valueOut);

        }

    }

}