D-Net

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

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.stream.Collectors;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.datasources.DatasourceManagerClients;

import eu.dnetlib.enabling.datasources.LocalOpenaireDatasourceManager;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

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

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

public class FindInvalidRepoProfilesJobNode extends SimpleJobNode implements ProgressJobNode {

	@Autowired

	private LocalOpenaireDatasourceManager dsManager;

	@Autowired

	private DatasourceManagerClients dsManagerClient;

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private int current = 0;

	private int total = 0;

	private boolean deleteInvalidProfiles = false;

	private boolean registerMissingProfiles = false;

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final Map<String, AtomicInteger> validIds = dsManagerClient.searchSQL("SELECT id FROM dsm_datasources", new HashMap<>())

				.stream()

				.map(m -> (String) m.get("id"))

				.collect(Collectors.toMap(s -> s, s -> new AtomicInteger(0)));

		final List<String> list = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(

				"for $x in collection('/db/DRIVER/RepositoryServiceResources/RepositoryServiceResourceType') return concat($x//DATASOURCE_ORIGINAL_ID, ' @@@ ', $x//RESOURCE_IDENTIFIER/@value)");

		this.current = 0;

		this.total = list.size() + validIds.size();

		int errorInvalidProfiles = 0;

		int errorUnregisteredProfiles = 0;

		int errorTooManyProfiles = 0;

		for (final String s : list) {

			current++;

			final String oid = StringUtils.substringBefore(s, "@@@").trim();

			final String profId = StringUtils.substringAfter(s, "@@@").trim();

			log.info("Evaluating ds: " + oid);

			if (validIds.containsKey(oid)) {

				validIds.get(oid).incrementAndGet();

			} else {

				log.warn("Invalid profile " + profId + ", openaireId " + oid + " not registered properly");

				errorInvalidProfiles++;

				// token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "invalidProfile:" + oid, profId);

				if (deleteInvalidProfiles) {

					log.warn("  - Deleting profile " + profId);

					serviceLocator.getService(ISRegistryService.class).deleteProfile(profId);

				}

			}

		}

		for (final Map.Entry<String, AtomicInteger> e : validIds.entrySet()) {

			current++;

			final String dsId = e.getKey();

			final int n = e.getValue().get();

			if (n == 0) {

				errorUnregisteredProfiles++;

				// token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "unregisterdProfile:" + dsId, dsId);

				log.warn("Missing profile for ds " + dsId);

				if (registerMissingProfiles) {

					dsManager.setManaged(dsId, dsManager.isManaged(dsId)); // This command should regenerate the repo profile

				}

			} else if (n > 1) {

				errorTooManyProfiles++;

				log.error("Too many profiles registerd " + n + " for ds " + dsId);

				// token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "tooManyProfiles:" + dsId, n);

			}

		}

		token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "errorInvalidProfiles", errorInvalidProfiles);

		token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "errorUnregisteredProfiles", errorUnregisteredProfiles);

		token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "errorTooManyProfiles", errorTooManyProfiles);

		return Arc.DEFAULT_ARC;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return new ProgressProvider() {

			@Override

			public int getTotalValue() {

				return total;

			}

			@Override

			public int getCurrentValue() {

				return current;

			}

			@Override

			public boolean isInaccurate() {

				return false;

			}

		};

	}

	public boolean isDeleteInvalidProfiles() {

		return deleteInvalidProfiles;

	}

	public void setDeleteInvalidProfiles(final boolean deleteInvalidProfiles) {

		this.deleteInvalidProfiles = deleteInvalidProfiles;

	}

	public boolean isRegisterMissingProfiles() {

		return registerMissingProfiles;

	}

	public void setRegisterMissingProfiles(final boolean registerMissingProfiles) {

		this.registerMissingProfiles = registerMissingProfiles;

	}

}