D-Net

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

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Iterator;

import java.util.Map;

import java.util.Optional;

import java.util.stream.Collectors;

import eu.dnetlib.common.logging.DnetLogger;

import eu.dnetlib.dhp.collection.ApiDescriptor;

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

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

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

import eu.dnetlib.msro.rmi.MSROException;

import org.apache.commons.lang.math.NumberUtils;

import org.apache.commons.lang3.StringUtils;

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

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

import javax.annotation.Resource;

import static eu.dnetlib.dhp.common.Constants.*;

public class PrepareEnvCollectHadoopJobNode extends SimpleJobNode {

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

    public static final String METADATA_IDENTIFIER_PATH = "metadata_identifier_path";

    public static final String DATE_FORMAT = "yyyy-MM-dd";

    private long ONE_DAY = 1000 * 60 * 60 * 24;

    @Resource(name = "msroWorkflowLogger")

    private DnetLogger dnetLogger;

    @Autowired

    private LocalDatasourceManager<?, ?> dsManager;

    /**

     * MDStore identifier

     */

    private String mdId;

    /**

     * REFRESH | INCREMENTAL

     */

    private String collectionMode;

    /**

     * used in INCREMENTAL mode

     */

    private String fromDateOverride;

    /**

     * used in to set the untilDate

     */

    private String untilDateOverride;

    /**

     * XML | JSON, default = XML

     */

    private String metadataEncoding = "XML";

    /**

     * Maximum number of allowed retires before failing

     */

    private int maxNumberOfRetry = 5;

    /**

     * Delay between request (Milliseconds)

     */

    private int requestDelay = 0;

    /**

     * Time to wait after a failure before retrying (Seconds)

     */

    private int retryDelay = 60;

    /**

     * Connect timeout (Seconds)

     */

    private int connectTimeOut = 30;

    /**

     * Read timeout (Seconds)

     */

    private int readTimeOut = 60;

    @Value("${dnet.openaire.dhp.collection.app.path}")

    private String oozieWfPath;

    @Value("${dnet.openaire.dhp.dnetMessageManagerURL}")

    private String dnetMessageManagerURL;

    @Override

    protected String execute(final NodeToken token) throws Exception {

        final String dsId = token.getEnv().getAttribute("parentDatasourceId");

        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(COLLECTION_MODE, getCollectionMode());

            token.getEnv().setAttribute(METADATA_ENCODING, getMetadataEncoding());

            token.getEnv().setAttribute(OOZIE_WF_PATH, getOozieWfPath());

            token.getEnv().setAttribute(DNET_MESSAGE_MGR_URL, getDnetMessageManagerURL());

            token.getEnv().setAttribute(MAX_NUMBER_OF_RETRY, getMaxNumberOfRetry());

            token.getEnv().setAttribute(REQUEST_DELAY, getRequestDelay());

            token.getEnv().setAttribute(RETRY_DELAY, getRetryDelay());

            token.getEnv().setAttribute(CONNECT_TIMEOUT, getConnectTimeOut());

            token.getEnv().setAttribute(READ_TIMEOUT, getReadTimeOut());

            final ApiDescriptor api = opt.get();

            if ("INCREMENTAL".equals(getCollectionMode())) {

                final String fromDate = calculateFromDate(token);

                log.info("Incremental Harvesting from: " + fromDate);

                if (StringUtils.isNotBlank(fromDate)) {

                   api.getParams().put("fromDate", fromDate);

                }

            }

            if (StringUtils.isNotBlank(getUntilDateOverride())) {

                api.getParams().put("untilDate", getUntilDateOverride());

            }

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

        }

    }

    private String findCurrentWfProfileId(NodeToken token) throws MSROException {

        final String p1 = token.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

        if (p1 != null && !p1.isEmpty()) {

            return p1;

        }

        final String p2 = token.getFullEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

        if (p2 != null && !p2.isEmpty()) {

            return p2;

        }

        final String p3 = token.getProcess().getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

        if (p3 != null && !p3.isEmpty()) {

            return p3;

        }

        throw new MSROException("Missing property in env: " + WorkflowsConstants.SYSTEM_WF_PROFILE_ID);

    }

    private String calculateFromDate(final NodeToken token) throws MSROException {

        if (StringUtils.isNotBlank(getFromDateOverride())) {

            log.info("using override FROM_DATE for incremental harvesting: " + getFromDateOverride());

            return getFromDateOverride();

        }

        final String profId = findCurrentWfProfileId(token);

        final long d = findLastSuccessStartDate(profId);

        return (d > 0) ? (new SimpleDateFormat(DATE_FORMAT)).format(new Date(d - ONE_DAY)) : null;

    }

    private long findLastSuccessStartDate(String profId) {

        long res = -1;

        final Iterator<Map<String, String>> iter = dnetLogger.find(WorkflowsConstants.SYSTEM_WF_PROFILE_ID, profId);

        while (iter.hasNext()) {

            final Map<String, String> map = iter.next();

            if ("true".equalsIgnoreCase(map.get(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY))) {

                final long curr = NumberUtils.toLong(map.get(WorkflowsConstants.SYSTEM_START_DATE), -1);

                if (curr > res) {

                    res = curr;

                }

            }

        }

        return res;

    }

    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;

    }

    public String getOozieWfPath() {

        return oozieWfPath;

    }

    public void setOozieWfPath(String oozieWfPath) {

        this.oozieWfPath = oozieWfPath;

    }

    public String getMetadataEncoding() {

        return metadataEncoding;

    }

    public void setMetadataEncoding(String metadataEncoding) {

        this.metadataEncoding = metadataEncoding;

    }

    public String getFromDateOverride() {

        return fromDateOverride;

    }

    public void setFromDateOverride(String fromDateOverride) {

        this.fromDateOverride = fromDateOverride;

    }

    public String getUntilDateOverride() {

        return untilDateOverride;

    }

    public void setUntilDateOverride(String untilDateOverride) {

        this.untilDateOverride = untilDateOverride;

    }

    public int getMaxNumberOfRetry() {

        return maxNumberOfRetry;

    }

    public void setMaxNumberOfRetry(int maxNumberOfRetry) {

        this.maxNumberOfRetry = maxNumberOfRetry;

    }

    public int getRequestDelay() {

        return requestDelay;

    }

    public void setRequestDelay(int requestDelay) {

        this.requestDelay = requestDelay;

    }

    public int getRetryDelay() {

        return retryDelay;

    }

    public void setRetryDelay(int retryDelay) {

        this.retryDelay = retryDelay;

    }

    public int getConnectTimeOut() {

        return connectTimeOut;

    }

    public void setConnectTimeOut(int connectTimeOut) {

        this.connectTimeOut = connectTimeOut;

    }

    public int getReadTimeOut() {

        return readTimeOut;

    }

    public void setReadTimeOut(int readTimeOut) {

        this.readTimeOut = readTimeOut;

    }

    public String getDnetMessageManagerURL() {

        return dnetMessageManagerURL;

    }

    public void setDnetMessageManagerURL(String dnetMessageManagerURL) {

        this.dnetMessageManagerURL = dnetMessageManagerURL;

    }

}