D-Net

package eu.dnetlib.msro.workflows.nodes.collect;

import java.text.SimpleDateFormat;

import java.time.Instant;

import java.time.LocalDateTime;

import java.time.ZoneId;

import java.time.ZonedDateTime;

import java.time.format.DateTimeFormatter;

import java.util.*;

import javax.annotation.Resource;

import eu.dnetlib.enabling.tools.DnetStreamSupport;

import eu.dnetlib.miscutils.datetime.DateUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import eu.dnetlib.msro.logging.DnetLogger;

import eu.dnetlib.msro.workflows.graph.Arc;

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

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.msro.workflows.procs.ProcessAware;

import eu.dnetlib.msro.workflows.procs.WorkflowProcess;

public class FindDateRangeForIncrementalHarvestingJobNode extends SimpleJobNode implements ProcessAware {

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

	private String fromDateParam;

	private String untilDateParam;

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

	private WorkflowProcess process;

	@Resource(name = "msroWorkflowLogger")

	private DnetLogger dnetLogger;

	@Override

	protected String execute(final Env env) throws Exception {

		final String fromDate = calculateFromDate();

		final String untilDate = null;

		log.info("Incremental Harv Details - from: " + fromDate + " - until: " + untilDate);

		if (fromDate != null) {

			env.setAttribute(getFromDateParam(), fromDate);

		}

		// if (untilDate != null) {

		// env.setAttribute(getUntilDateParam(), untilDate);

		// }

		return Arc.DEFAULT_ARC;

	}

	private String calculateFromDate() {

		final long d = findLastSuccessStartDate();

        LocalDateTime zdt = LocalDateTime.ofInstant(Instant.ofEpochMilli(d), ZoneId.of("Etc/UTC"));

        return DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'").format(zdt);

//		return (d > 0) ? (new SimpleDateFormat("yyyy-MM-dd")).format(new Date(d - ONE_DAY)) : null;

    }

	private long findLastSuccessStartDate() {

		final long res = -1;

		// TODO find the right condition to obtain the list of the previous executions

        //dnetLogger.

        final Map<String, Object> query = new HashMap<>();

        query.put("system:profileTemplateId", process.getProfileId());

        query.put("system:parentProfileId", process.getParentProfileId());

        query.put("system:processStatus", "SUCCESS");

        Iterator<Map<String, String>> mapIterator = dnetLogger.find(query);

        Optional<Long> maxDate = DnetStreamSupport.generateStreamFromIterator(mapIterator).map(it -> it.get("system:startDate")).map(Long::parseLong).max(Long::compare);

        if (maxDate.isPresent())

            return maxDate.get();

        else

            return -1;

    }

	public String getFromDateParam() {

		return this.fromDateParam;

	}

	public void setFromDateParam(final String fromDateParam) {

		this.fromDateParam = fromDateParam;

	}

	public String getUntilDateParam() {

		return this.untilDateParam;

	}

	public void setUntilDateParam(final String untilDateParam) {

		this.untilDateParam = untilDateParam;

	}

	public WorkflowProcess getProcess() {

		return this.process;

	}

	@Override

	public void setProcess(final WorkflowProcess process) {

		this.process = process;

	}

}