D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.dhp;

import java.util.Map;

import java.util.Optional;

import java.util.stream.Collectors;

import eu.dnetlib.dhp.model.mdstore.Provenance;

import eu.dnetlib.enabling.datasources.common.Datasource;

import eu.dnetlib.enabling.datasources.common.DsmNotFoundException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Autowired;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.collector.worker.model.ApiDescriptor;

import eu.dnetlib.enabling.datasources.common.ApiParam;

import eu.dnetlib.enabling.datasources.common.LocalDatasourceManager;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

public class PrepareEnvCollectHadoopJobNode extends SimpleJobNode {

    private static final Log log = LogFactory.getLog(PrepareEnvCollectHadoopJobNode.class);

    public static final String METADATA_IDENTIFIER_PATH = "metadata_identifier_path";

    @Autowired

    private LocalDatasourceManager<?, ?> dsManager;

    /**

     * MDStore identifier

     */

    private String mdId;

    /**

     * REFRESH | INCREMENTAL

     */

    private String collectionMode;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        // param 1 : hdfs path

        // param 2 : api descriptor (json)

        // param 3 : nameNode

        final String dsId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_ID);

        log.info("dsId: " + dsId);

        final String apiId = token.getEnv().getAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE);

        log.info("apiId: " + apiId);

        final Optional<ApiDescriptor> opt = dsManager.getApis(dsId)

                .stream()

                .filter(a -> a.getId().equals(apiId))

                .map(a -> {

                    final ApiDescriptor res = new ApiDescriptor();

                    res.setBaseUrl(a.getBaseurl());

                    res.setId(a.getId());

                    res.setProtocol(a.getProtocol());

                    res.getParams().put(METADATA_IDENTIFIER_PATH, a.getMetadataIdentifierPath());

                    res.getParams().putAll(a.getApiParams()

                            .stream()

                            .map(o -> (ApiParam) o)

                            .collect(Collectors.toMap(ApiParam::getParam, ApiParam::getValue)));

                    return res;

                })

                .findFirst();

        if (opt.isPresent()) {

            token.getEnv().setAttribute("mdId", getMdId());

            token.getEnv().setAttribute("collectionMode", getCollectionMode());

            final ApiDescriptor api = opt.get();

            token.getEnv().setAttribute("apiDescription", new Gson().toJson(api));

            final Provenance provenance = new Provenance();

            provenance.setDatasourceId(dsId);

            final Datasource<?, ?> ds = dsManager.getDs(dsId);

            provenance.setDatasourceName(ds.getOfficialname());

            provenance.setNsPrefix(ds.getNamespaceprefix());

            final String dsProvenance = new Gson().toJson(provenance);

            log.info("datasource provenance: " + dsProvenance);

            token.getEnv().setAttribute("dataSourceInfo", dsProvenance);

            token.getEnv().setAttribute("timestamp", ""+System.currentTimeMillis());

            token.getEnv().setAttribute("identifierPath",api.getParams().get(METADATA_IDENTIFIER_PATH));

            token.getEnv().setAttribute("workflowId",token.getProcess().getEnv().getAttribute("system:processId"));

            token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_INTERFACE_BASEURL, api.getBaseUrl());

            token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_PREFIX + "protocol", api.getProtocol());

            final Map<String, String> params = api.getParams();

            if (params != null) {

                for(Map.Entry<String, String> e : params.entrySet()) {

                    token.getEnv().setAttribute(WorkflowsConstants.DATAPROVIDER_PREFIX + e.getKey(), e.getValue());

                }

            }

            return Arc.DEFAULT_ARC;

        } else {

            throw new DsmNotFoundException("cannot find ds interface: " + apiId);

        }

    }

    public String getMdId() {

        return mdId;

    }

    public void setMdId(String mdId) {

        this.mdId = mdId;

    }

    public String getCollectionMode() {

        return collectionMode;

    }

    public void setCollectionMode(String collectionMode) {

        this.collectionMode = collectionMode;

    }

}