D-Net

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet45-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-deduplication</artifactId>

	<packaging>jar</packaging>

	<version>1.5.8</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-deduplication/tags/dnet-deduplication-1.5.8</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-msro-service</artifactId>

			<version>[3.0.0,4.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-hadoop-service-rmi</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-actionmanager-api</artifactId>

			<version>[4.0.0,5.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-modular-ui</artifactId>

			<version>[3.0.0,4.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-index-solr-client</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaireplus-mapping-utils</artifactId>

			<version>[6.0.0,7.0.0)</version>

		</dependency>

		<dependency>

			<groupId>javax.servlet</groupId>

			<artifactId>javax.servlet-api</artifactId>

			<version>${javax.servlet.version}</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>com.fasterxml.jackson.core</groupId>

			<artifactId>jackson-databind</artifactId>

			<version>${jackson.version}</version>

		</dependency>

		<dependency>

			<groupId>com.google.guava</groupId>

			<artifactId>guava</artifactId>

			<version>${google.guava.version}</version>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.version}</version>

			<scope>test</scope>

		</dependency>

	</dependencies>

</project>

package eu.dnetlib.msro.workflows.dedup;

import java.nio.file.FileSystems;

import java.nio.file.Path;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.hadoop.rmi.HadoopService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

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;

/**

 * Created by claudio on 14/10/15.

 */

public class MinDistSearchHadoopJobNode extends DedupConfigurationAwareJobNode {

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

	private final static String StatusParam = "MinDistSearchHadoopJobNode.status";

	private final static String DepthParam = "mindist_recursion_depth";

	private final static String UpdateCounterParam = "UpdateCounter.UPDATED";

	private final static String DebugParam = "mindist_DEBUG";

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private boolean debug = false;

	private String outPathParam;

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		String depthString = token.getFullEnv().getAttribute(DepthParam);

		log.debug(String.format("found depthParam: '%s'", depthString));

		if (StringUtils.isBlank(depthString)) {

			depthString = "0";

		}

		int depth = Integer.valueOf(depthString);

		final String cluster = token.getEnv().getAttribute("cluster");

		final String outputPath = getPath(token.getEnv().getAttribute("workDir"), depth);

		final HadoopService hadoopService = serviceLocator.getService(HadoopService.class);

		switch (getStatusFromEnv(token)) {

		case DATALOAD:

			setHadoopJob("dedupSimilarity2GraphJob");

			job.getParameters().put("mapred.output.dir", getPath(token.getEnv().getAttribute("workDir"), depth) + "/out");

			hadoopService.createHdfsDirectory(cluster, outputPath, true);

			break;

		case DEPTH_N:

			setHadoopJob("dedupMinDistGraphJob");

			final String newOutputPath = getPath(token.getEnv().getAttribute("workDir"), depth + 1);

			hadoopService.createHdfsDirectory(cluster, newOutputPath, true);

			job.getParameters().put(DepthParam, String.valueOf(depth));

			job.getParameters().put(DebugParam, String.valueOf(isDebug()));

			job.getParameters().put("mapred.input.dir", outputPath + "/out");

			job.getParameters().put("mapred.output.dir", newOutputPath + "/out");

			if (log.isDebugEnabled()) {

				log.debug(String.format("input job parameters: %s", job.getParameters()));

			}

			token.getFullEnv().setAttribute(DepthParam, String.valueOf(depth + 1));

			token.getFullEnv().setAttribute(getOutPathParam(), newOutputPath + "/out");

			break;

		}

		super.prepareJob(job, token);

	}

	private String getPath(final String basePath, final int depth) {

		Path path = FileSystems.getDefault().getPath(basePath, "depth_" + depth);

		return path.toAbsolutePath().toString();

	}

	private STATUS getStatusFromEnv(final NodeToken token) {

		if(StringUtils.isBlank(token.getEnv().getAttribute(StatusParam))) {

			return STATUS.DATALOAD;

		}

		STATUS current = STATUS.DATALOAD;

		try {

			current = STATUS.valueOf(token.getEnv().getAttribute(StatusParam));

			log.debug("found status: " + current.toString());

		} catch (IllegalArgumentException e) {}

		return current;

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Engine engine, final NodeToken token) {

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void onDone(final BlackboardJob job) {

				final STATUS status = getStatusFromEnv(token);

				log.debug("complete phase: " + status);

				switch (status) {

				case DATALOAD:

					token.getFullEnv().setAttribute(StatusParam, STATUS.DEPTH_N.toString());

					token.getFullEnv().setAttribute(DepthParam, "0");

					complete("depth_n");

					break;

				case DEPTH_N:

					if (log.isDebugEnabled()) {

						log.debug(String.format("return job parameters: %s=%s, %s=%s", DepthParam, job.getParameters().get(DepthParam), UpdateCounterParam, job.getParameters().get(UpdateCounterParam)));

					}

					final String counter = job.getParameters().get(UpdateCounterParam);

					if (StringUtils.isBlank(counter)) {

						token.getFullEnv().removeAttribute(StatusParam);

						token.getFullEnv().removeAttribute(DepthParam);

						log.info(String.format("done iteration %s:%s", UpdateCounterParam, 0));

						complete(Arc.DEFAULT_ARC);

					} else {

						log.info(String.format("continue with next iteration %s:%s", UpdateCounterParam, counter));

						complete("depth_n");

					}

					break;

				}

			}

			private void complete(final String arc) {

				engine.complete(token, arc);

				engine.executeQueuedArcTokens(token.getProcess());

			}

		};

	}

	public boolean isDebug() {

		return debug;

	}

	public void setDebug(boolean debug) {

		this.debug = debug;

	}

	public String getOutPathParam() {

		return outPathParam;

	}

	public void setOutPathParam(String outPathParam) {

		this.outPathParam = outPathParam;

	}

	enum STATUS {DATALOAD, DEPTH_N}

}

package eu.dnetlib.msro.workflows.hadoop;

import java.util.List;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Iterables;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.hadoop.rmi.HadoopBlackboardActions;

import eu.dnetlib.data.hadoop.rmi.HadoopJobType;

import eu.dnetlib.data.hadoop.rmi.HadoopService;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

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

public class SubmitHadoopJobNode extends BlackboardJobNode {

	/**

	 * logger.

	 */

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

	public static final String OOZIE_REPORT_ACTIONS = "oozie.report.actions.csv";

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String hadoopJob;

	private String cluster;

	private boolean simulation = false;

	private String oozieReportActionsCsv;

	@Override

	protected String obtainServiceId(final NodeToken token) {

		return getServiceLocator().getServiceId(HadoopService.class);

	}

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		String type = getJobType(getHadoopJob());

		log.info("submitting job " + getHadoopJob() + " type: " + type);

		job.setAction(type);

		job.getParameters().put("job.name", getHadoopJob());

		job.getParameters().put("cluster", cluster(token));

		job.getParameters().put("simulation", String.valueOf(isSimulation()));

		if (StringUtils.isNotBlank(getOozieReportActionsCsv())) {

			job.getParameters().put(OOZIE_REPORT_ACTIONS, getOozieReportActionsCsv());

		}

		job.getParameters().putAll(parseJsonParameters(token));

	}

	private String cluster(final NodeToken token) {

		if (token.getEnv().hasAttribute("cluster")) {

			String cluster = token.getEnv().getAttribute("cluster");

			log.info("found override value in wfEnv for 'cluster' param: " + cluster);

			return cluster;

		}

		return getCluster();

	}

	/**

	 * reads the job type for the given job name

	 *

	 * @param jobName

	 * @return

	 * @throws ISLookUpException

	 */

	private String getJobType(final String jobName) throws ISLookUpException {

		List<String> res =

				serviceLocator.getService(ISLookUpService.class).quickSearchProfile(

						"/RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'HadoopJobConfigurationDSResourceType']//HADOOP_JOB[./@name='" + jobName

						+ "']/@type/string()");

		if (res.isEmpty()) { throw new IllegalStateException("unable to find job type for job: " + jobName); }

		final HadoopJobType type = HadoopJobType.valueOf(Iterables.getOnlyElement(res));

		switch (type) {

		case mapreduce:

			return HadoopBlackboardActions.SUBMIT_MAPREDUCE_JOB.toString();

		case admin:

			return HadoopBlackboardActions.SUBMIT_ADMIN_JOB.toString();

		case oozie:

			return HadoopBlackboardActions.SUBMIT_OOZIE_JOB.toString();

		default:

			throw new IllegalStateException("undefined job type: " + type.toString());

		}

	}

	public String getHadoopJob() {

		return hadoopJob;

	}

	public void setHadoopJob(final String hadoopJob) {

		this.hadoopJob = hadoopJob;

	}

	public String getCluster() {

		return cluster;

	}

	public void setCluster(final String cluster) {

		this.cluster = cluster;

	}

	public boolean isSimulation() {

		return simulation;

	}

	public void setSimulation(final boolean simulation) {

		this.simulation = simulation;

	}

	public String getOozieReportActionsCsv() {

		return oozieReportActionsCsv;

	}

	public void setOozieReportActionsCsv(final String oozieReportActionsCsv) {

		this.oozieReportActionsCsv = oozieReportActionsCsv;

	}

}

package eu.dnetlib.msro.workflows.dedup.conf;

import com.google.gson.GsonBuilder;

/**

 * The Class Entity.

 */

public class Entity {

	/** The name. */

	private String name;

	/** The code. */

	private String code;

	/** The label. */

	private String label;

	public Entity() {}

	/**

	 * Instantiates a new entity.

	 *

	 * @param name

	 *            the name

	 * @param code

	 *            the code

	 * @param label

	 *            the label

	 */

	public Entity(final String name, final String code, final String label) {

		super();

		this.setName(name);

		this.setCode(code);

		this.setLabel(label);

	}

	/**

	 * Gets the name.

	 *

	 * @return the name

	 */

	public String getName() {

		return name;

	}

	/**

	 * Gets the code.

	 *

	 * @return the code

	 */

	public String getCode() {

		return code;

	}

	/**

	 * Gets the label.

	 *

	 * @return the label

	 */

	public String getLabel() {

		return label;

	}

	public void setName(final String name) {

		this.name = name;

	}

	public void setCode(final String code) {

		this.code = code;

	}

	public void setLabel(final String label) {

		this.label = label;

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see java.lang.Object#toString()

	 */

	@Override

	public String toString() {

		return new GsonBuilder().setPrettyPrinting().create().toJson(this);

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.io.StringReader;

import java.util.Iterator;

import java.util.List;

import java.util.Queue;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import com.google.common.collect.Lists;

import com.google.common.collect.Queues;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

import eu.dnetlib.msro.workflows.hadoop.utils.Similarity;

import eu.dnetlib.msro.workflows.hadoop.utils.SimilarityMeshBuilder;

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

public class BuildSimilarityMeshJobNode extends AsyncJobNode {

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

	/** The result set factory. */

	@Resource(name = "iterableResultSetFactory")

	private IterableResultSetFactory resultSetFactory;

	/** The result set client factory. */

	@Resource(name = "resultSetClientFactory")

	private ResultSetClientFactory resultSetClientFactory;

	private StringTemplate similarity;

	private String inputEprParam;

	private String outputEprParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String inputEpr = token.getEnv().getAttribute(getInputEprParam());

		final Iterator<String> rsClient = resultSetClientFactory.getClient(inputEpr).iterator();

		final Queue<Object> queue = Queues.newLinkedBlockingQueue();

		final SAXReader reader = new SAXReader();

		if (rsClient.hasNext()) {

			populateQueue(queue, reader, rsClient.next());

		}

		final W3CEndpointReference eprOut = resultSetFactory.createIterableResultSet(new Iterable<String>() {

			@Override

			public Iterator<String> iterator() {

				return new Iterator<String>() {

					@Override

					public boolean hasNext() {

						synchronized (queue) {

							return !queue.isEmpty();

						}

					}

					@Override

					public String next() {

						synchronized (queue) {

							final Object o = queue.poll();

							while (queue.isEmpty() && rsClient.hasNext()) {

								populateQueue(queue, reader, rsClient.next());

							}

							return buildSimilarity((Similarity) o);

						}

					}

					@Override

					public void remove() {

						throw new UnsupportedOperationException();

					}

				};

			}

		});

		token.getEnv().setAttribute(getOutputEprParam(), eprOut.toString());

		return Arc.DEFAULT_ARC;

	}

	private void populateQueue(final Queue<Object> q, final SAXReader r, final String xml) {

		try {

			final Document d = r.read(new StringReader(xml));

			final String groupid = d.valueOf("//FIELD[@name='id']");

			final List<?> items = d.selectNodes("//FIELD[@name='group']/ITEM");

			final String entitytype = d.valueOf("//FIELD[@name='entitytype']");

			final List<String> group = Lists.newArrayList();

			for (final Object id : items) {

				group.add(((Node) id).getText());

			}

			// compute the full mesh

			final Type type = Type.valueOf(entitytype);

			final List<Similarity> mesh = SimilarityMeshBuilder.build(type, group);

			// total += mesh.size();

			if (log.isDebugEnabled()) {

				log.debug(String.format("built mesh for group '%s', size %d", groupid, mesh.size()));

			}

			for (final Similarity s : mesh) {

				if (log.isDebugEnabled()) {

					log.debug(String.format("adding to queue: %s", s.toString()));

				}

				q.add(s);

			}

		} catch (final DocumentException e) {

			log.error("invalid document: " + xml);

		}

	}

	private String buildSimilarity(final Similarity s) {

		final StringTemplate template = new StringTemplate(getSimilarity().getTemplate());

		template.setAttribute("source", s.getPair().getKey());

		template.setAttribute("target", s.getPair().getValue());

		template.setAttribute("type", s.getType().toString());

		final String res = template.toString();

		return res;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public StringTemplate getSimilarity() {

		return similarity;

	}

	@Required

	public void setSimilarity(final StringTemplate similarity) {

		this.similarity = similarity;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

import eu.dnetlib.msro.workflows.hadoop.SubmitHadoopJobNode;

import eu.dnetlib.pace.config.DedupConfig;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class DedupConfigurationAwareJobNode extends SubmitHadoopJobNode {

	private static final Log log = LogFactory.getLog(DedupConfigurationAwareJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private String dedupConfigSequenceParam;

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		super.prepareJob(job, token);

		final DedupConfigurationOrchestration dedupConfigurations = dedupConfigurations(token);

		final DedupConfig currentConf = dedupConfigurations.getConfigurations().peek();

		log.debug("using dedup configuration: '" + currentConf + "'");

		job.getParameters().put("dedup.conf", currentConf.toString());

		token.getEnv().setAttribute("dedup.conf", currentConf.toString());

	}

	protected DedupConfigurationOrchestration dedupConfigurations(final NodeToken token) {

		final String configs = token.getFullEnv().getAttribute(getDedupConfigSequenceParam());

		if ((configs == null) || configs.trim().isEmpty())

			throw new IllegalStateException("Cannot find dedup configurations in workflow env: '" + getDedupConfigSequenceParam() + "'");

		return DedupConfigurationOrchestration.fromJSON(configs);

	}

	protected String getEntityType(final NodeToken token) {

		final String entityType = token.getEnv().getAttribute("entityType");

		if (StringUtils.isBlank(entityType)) throw new IllegalStateException("Cannot find 'entityType' parameter in workflow env.");

		return entityType;

	}

	// //////////

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.util.Queue;

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.google.common.base.Function;

import com.google.common.base.Splitter;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestrationLoader;

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

public class DedupCheckEntitySequenceJobNode extends AsyncJobNode {

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

	@Autowired

	private DedupConfigurationOrchestrationLoader dedupOrchestrationLoader;

	private String dedupConfigSequenceParam;

	private String entitySequence;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (StringUtils.isBlank(getEntitySequence())) throw new MSROException("missing entity sequence, e.g. a csv: organization,person,result");

		if (token.getFullEnv().hasAttribute(DedupGrouperJobNode.DEDUP_GROUPER_LOOPER)) {

			log.info("reset env variable: " + DedupGrouperJobNode.DEDUP_GROUPER_LOOPER + " to zero");

			token.getFullEnv().setAttribute(DedupGrouperJobNode.DEDUP_GROUPER_LOOPER, 0);

		}

		if (!token.getEnv().hasAttribute("entitySequence")) {

			log.info("parsing config sequence: " + getEntitySequence());

			token.getEnv().setAttribute("entitySequence", getEntitySequence());

			final Iterable<String> sequence = Splitter.on(",").omitEmptyStrings().split(getEntitySequence());

			final Queue<DedupConfigurationOrchestration> q =

					Lists.newLinkedList(Iterables.transform(sequence, new Function<String, DedupConfigurationOrchestration>() {

						@Override

						public DedupConfigurationOrchestration apply(final String entityName) {

							try {

								final DedupConfigurationOrchestration dco = Iterables.getFirst(dedupOrchestrationLoader.loadByEntityName(entityName), null);

								if (dco == null) throw new RuntimeException("unable to find DedupOrchestration profile for entity type: " + entityName);

								return dco;

							} catch (final Throwable e) {

								throw new RuntimeException("", e);

							}

						}

					}));

			log.info("built sequence of dedup orchestration profiles, size: " + q.size());

			final DedupConfigurationOrchestration dco = q.remove();

			log.info("closing mesh for entity: " + dco.getEntity().getName());

			setDedupConfParams(token, dco);

			token.getEnv().setTransientAttribute("entitySequenceQueue", q);

			return Arc.DEFAULT_ARC;

		}

		@SuppressWarnings("unchecked")

		final Queue<DedupConfigurationOrchestration> q = (Queue<DedupConfigurationOrchestration>) token.getEnv().getTransientAttribute("entitySequenceQueue");

		if (!q.isEmpty()) {

			log.info("remaining dedup orchestration profiles: " + q.size());

			final DedupConfigurationOrchestration dco = q.remove();

			log.info("closing mesh for entity: " + dco.getEntity().getName());

			setDedupConfParams(token, dco);

			return Arc.DEFAULT_ARC;

		}

		log.info("completed closing mesh for entities: " + getEntitySequence());

		return "done";

	}

	private void setDedupConfParams(final NodeToken token, final DedupConfigurationOrchestration dco) {

		token.getEnv().setAttribute("entityType", dco.getEntity().getName());

		token.getEnv().setAttribute("entityTypeId", dco.getEntity().getCode());

		token.getEnv().setAttribute(getDedupConfigSequenceParam(), dco.toString());

	}

	public String getEntitySequence() {

		return entitySequence;

	}

	public void setEntitySequence(final String entitySequence) {

		this.entitySequence = entitySequence;

	}

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.util.List;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.base.Splitter;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

public class ResetCountersJobNode extends SimpleJobNode {

	private static final Log log = LogFactory.getLog(ResetCountersJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private String attributesCSV;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (StringUtils.isNotBlank(getAttributesCSV())) {

			log.info("got wf attributes CSV: " + getAttributesCSV());

			final Splitter splitter = Splitter.on(",").trimResults().omitEmptyStrings();

			final List<String> wfAttrs = splitter.splitToList(getAttributesCSV());

			for (final String attr : wfAttrs) {

				resetWorkflowParam(token, attr);

			}

		} else {

			log.info("attribute list is empty, nothing to do here.");

		}

		return Arc.DEFAULT_ARC;

	}

	private void resetWorkflowParam(final NodeToken token, final String attribute) {

		final String count = token.getFullEnv().getAttribute(attribute);

		if (StringUtils.isNotBlank(count)) {

			log.info(String.format("found loop counter '%s', value '%s'", attribute, count));

			token.getFullEnv().setAttribute(attribute, 0);

			log.info(String.format("set '%s', to 0", attribute));

		} else {

			log.info("loop counter was not found in workflow env, nothing to do here.");

		}

	}

	public String getAttributesCSV() {

		return attributesCSV;

	}

	public void setAttributesCSV(final String attributesCSV) {

		this.attributesCSV = attributesCSV;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.util.List;

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.is.lookup.rmi.ISLookUpException;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

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

public class DedupCheckConfigurationJobNode extends AsyncJobNode {

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

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private String dedupConfigSequenceParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String dcoJson = token.getEnv().getAttribute(getDedupConfigSequenceParam());

		final DedupConfigurationOrchestration dco = DedupConfigurationOrchestration.fromJSON(dcoJson);

		if (!existActionSetProfile(dco)) throw new MSROException("missing action set profile: " + dco.getActionSetId());

		log.info("found action set profile: " + dco.getActionSetId());

		return Arc.DEFAULT_ARC;

	}

	private boolean existActionSetProfile(final DedupConfigurationOrchestration dco) throws ISLookUpException {

		final String xquery = "for $x in //RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'DedupOrchestrationDSResourceType' and .//ACTION_SET/@id='%s'] return 1";

		log.info("looking for action set profile, id: " + dco.getActionSetId());

		final List<String> actionSets = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(String.format(xquery, dco.getActionSetId()));

		return !actionSets.isEmpty();

	}

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.functionality.modular.ui.workflows.values;

import java.util.List;

import java.util.Map;

import javax.annotation.Resource;

import com.google.common.collect.Lists;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class ListDedupOrchestrationValues extends ValidNodeValuesFetcher {

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected List<DnetParamValue> obtainValues(final Map<String, String> params) throws Exception {

		final String xquery =

				"for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='DedupOrchestrationDSResourceType'] return $x//ACTION_SET/@id/string()";

		final List<String> result = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

		final List<DnetParamValue> values = Lists.newArrayList();

		for (final String s : result) {

			values.add(new DnetParamValue(s, s));

		}

		return values;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import static java.lang.String.format;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.provision.index.rmi.IndexService;

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

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

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

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.functionality.index.solr.feed.InputDocumentFactory;

import eu.dnetlib.msro.rmi.MSROException;

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

public class FinalizeDedupIndexJobNode extends BlackboardJobNode {

	private static final Log log = LogFactory.getLog(FinalizeDedupIndexJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Override

	protected String obtainServiceId(final NodeToken token) {

		return getServiceLocator().getServiceId(IndexService.class);

	}

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		final String indexDsId = getEnvParam(token, "index_id");

		log.info("preparing blackboard job DELETE_BY_QUERY index: " + indexDsId);

		final String backendId = getBackendId(indexDsId);

		if (StringUtils.isBlank(backendId))

			throw new MSROException("empty index backend Id");

		job.setAction("DELETE_BY_QUERY");

		job.getParameters().put("id", indexDsId);

		job.getParameters().put("backend_Id", backendId);

		job.getParameters().put("query",

				buildQuery(getEnvParam(token, "entityType"), getEnvParam(token, "index.feed.timestamp"), getEnvParam(token, "actionset")));

	}

	private String buildQuery(final String entityType, final String version, final String actionset) {

		final String query =

				String.format("__dsversion:{* TO %s} AND oaftype:%s AND actionset:%s", InputDocumentFactory.getParsedDateField(version), entityType, actionset);

		log.info("delete by query: " + query);

		return query;

	}

	private String getEnvParam(final NodeToken token, final String name) throws MSROException {

		final String value = token.getEnv().getAttribute(name);

		if (StringUtils.isBlank(value))

			throw new MSROException(format("unable to finalize index feeding, cannot find property '%s' in the workflow env.", name));

		return value;

	}

	public String getBackendId(final String indexDsId) throws ISLookUpDocumentNotFoundException, ISLookUpException {

		return getServiceLocator().getService(ISLookUpService.class).getResourceProfileByQuery(

				"//RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='" + indexDsId + "']//BACKEND/text()");

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.lang.StringUtils;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestrationLoader;

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

public class DedupConfigurationSetterJobNode extends AsyncJobNode {

	private String dedupConfigSequence;

	private String dedupConfigSequenceParam;

	@Autowired

	private DedupConfigurationOrchestrationLoader dedupOrchestrationLoader;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (StringUtils.isBlank(getDedupConfigSequence())) throw new IllegalArgumentException("missing configuration sequence");

		final DedupConfigurationOrchestration dedupOrchestration = dedupOrchestrationLoader.loadByActionSetId(getDedupConfigSequence());

		token.getEnv().setAttribute("entityType", dedupOrchestration.getEntity().getName());

		token.getEnv().setAttribute("entityTypeId", dedupOrchestration.getEntity().getCode());

		token.getEnv().setAttribute(getDedupConfigSequenceParam(), dedupOrchestration.toString());

		return Arc.DEFAULT_ARC;

	}

	public String getDedupConfigSequence() {

		return dedupConfigSequence;

	}

	public void setDedupConfigSequence(final String dedupConfigSequence) {

		this.dedupConfigSequence = dedupConfigSequence;

	}

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.msro.workflows.actions;

import java.util.List;

import java.util.Map;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.set.RawSet;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

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

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;

/**

 * The Class PrepareConfiguredActionSetJobNode.

 */

public class PrepareConfiguredActionSetJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	/**

	 * The dedup config sequence param.

	 */

	private String dedupConfigSequenceParam;

	/**

	 * The job property.

	 */

	private String jobProperty;

	/**

	 * The action set path param name.

	 */

	private String actionSetPathParam;

	/**

	 * The service locator.

	 */

	@Autowired

	private UniqueServiceLocator serviceLocator;

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.msro.workflows.nodes.SimpleJobNode#execute(com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final List<Map<String, String>> setList = Lists.newArrayList();

		final Map<String, String> set = Maps.newHashMap();

		final String actionSetId = getActionSetId(token);

		final ISLookUpService isLookUpService = serviceLocator.getService(ISLookUpService.class);

		final String basePath = isLookUpService.getResourceProfileByQuery(

				"/RESOURCE_PROFILE[./HEADER/RESOURCE_TYPE/@value='ActionManagerServiceResourceType']//SERVICE_PROPERTIES/PROPERTY[@key='basePath']/@value/string()");

		if (StringUtils.isBlank(basePath)) {

			throw new IllegalStateException("missing basePath in ActionManagerService");

		}

		final String actionSetDirectory = isLookUpService.getResourceProfileByQuery(

				"/RESOURCE_PROFILE[./HEADER/RESOURCE_TYPE/@value='ActionManagerSetDSResourceType' and .//SET/@id = '"+actionSetId+"']//SET/@ directory/string()");

		if (StringUtils.isBlank(actionSetDirectory)) {

			throw new IllegalStateException("missing directory in ActionSet profile: " + actionSetId);

		}

		final String rawSetId = RawSet.newInstance().getId();

		set.put("rawset", rawSetId);

		set.put("creationDate", DateUtils.now_ISO8601());

		set.put("set", actionSetId);

		set.put("enabled", "true");

		set.put("jobProperty", getJobProperty());

		token.getEnv().setAttribute(set.get("jobProperty"), set.get("rawset"));

		final String path = basePath + "/" + actionSetDirectory + "/" + rawSetId;

		log.info("using action set path: " + path);

		token.getEnv().setAttribute(getActionSetPathParam(), path);

		setList.add(set);

		final String sets = new Gson().toJson(setList);

		log.debug("built set: " + sets);

		token.getEnv().setAttribute("sets", sets);

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Gets the action set id.

	 *

	 * @param token the token

	 * @return the action set id

	 */

	private String getActionSetId(final NodeToken token) {

		final String json = token.getEnv().getAttribute(getDedupConfigSequenceParam());

		final DedupConfigurationOrchestration dco = DedupConfigurationOrchestration.fromJSON(json);

		final String actionSetId = dco.getActionSetId();

		log.info("found actionSetId in workflow env: " + actionSetId);

		return actionSetId;

	}

	/**

	 * Gets the dedup config sequence param.

	 *

	 * @return the dedup config sequence param

	 */

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	/**

	 * Sets the dedup config sequence param.

	 *

	 * @param dedupConfigSequenceParam the new dedup config sequence param

	 */

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

	/**

	 * Gets the job property.

	 *

	 * @return the job property

	 */

	public String getJobProperty() {

		return jobProperty;

	}

	/**

	 * Sets the job property.

	 *

	 * @param jobProperty the new job property

	 */

	public void setJobProperty(final String jobProperty) {

		this.jobProperty = jobProperty;

	}

	public String getActionSetPathParam() {

		return actionSetPathParam;

	}

	public void setActionSetPathParam(final String actionSetPathParam) {

		this.actionSetPathParam = actionSetPathParam;

	}

}

package eu.dnetlib.functionality.modular.ui.dedup;

/**

 * Utility Class EntityType, helps the user interface to display the different entity types.

 */

public class EntityType {

	/** The id. */

	private String id;

	/** The type. */

	private String type;

	/** The label. */

	private String label;

	/**

	 * Instantiates a new entity type.

	 */

	public EntityType() {};

	/**

	 * Instantiates a new entity type.

	 *

	 * @param id

	 *            the id

	 * @param type

	 *            the type

	 * @param label

	 *            the label

	 */

	public EntityType(final String id, final String type, final String label) {

		super();

		setId(id);

		setType(type);

		setLabel(label);

	}

	/**

	 * Gets the id.

	 *

	 * @return the id

	 */

	public String getId() {

		return id;

	}

	/**

	 * Sets the id.

	 *

	 * @param id

	 *            the new id

	 */

	public void setId(final String id) {

		this.id = id;

	}

	/**

	 * Gets the type.

	 *

	 * @return the type

	 */

	public String getType() {

		return type;

	}

	/**

	 * Sets the type.

	 *

	 * @param type

	 *            the new type

	 */

	public void setType(final String type) {

		this.type = type;

	}

	/**

	 * Gets the label.

	 *

	 * @return the label

	 */

	public String getLabel() {

		return label;

	}

	/**

	 * Sets the label.

	 *

	 * @param label

	 *            the new label

	 */

	public void setLabel(final String label) {

		this.label = label;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.collections.CollectionUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

public class DedupDuplicateScanJobNode extends DedupConfigurationAwareJobNode {

	private static final Log log = LogFactory.getLog(DedupDuplicateScanJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private class DedupBlackboardWorkflowJobListener extends BlackboardWorkflowJobListener {

		public DedupBlackboardWorkflowJobListener(final Engine engine, final NodeToken token) {

			super(engine, token);

		}

		@Override

		protected void onDone(final BlackboardJob job) {

			final DedupConfigurationOrchestration confs = dedupConfigurations(getToken());

			confs.getConfigurations().poll();

			log.info("checking dedup configs queue, size: " + confs.getConfigurations().size());

			if (CollectionUtils.isEmpty(confs.getConfigurations())) {

				log.info("dedup similarity scan done");

				super.complete(job, "done");

			} else {

				log.debug("remaining confs: " + confs);

				getToken().getEnv().setAttribute(getDedupConfigSequenceParam(), confs.toString());

				super.complete(job, Arc.DEFAULT_ARC);

			}

		}

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Engine engine, final NodeToken token) {

		return new DedupBlackboardWorkflowJobListener(engine, token);

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.proto.DedupSimilarityProtos.DedupSimilarity;

import eu.dnetlib.data.proto.RelTypeProtos.RelType;

import eu.dnetlib.data.proto.RelTypeProtos.SubRelType;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.msro.rmi.MSROException;

public class DedupSimilarityToActionsJobNode extends DedupConfigurationAwareJobNode {

	private static final Log log = LogFactory.getLog(DedupSimilarityToActionsJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken token) throws Exception {

		super.prepareJob(job, token);

		final String entityType = token.getEnv().getAttribute("entityType");

		if (StringUtils.isBlank(entityType))

			throw new MSROException("unable to find wf param: entityType");

		String cf = "_" + SubRelType.dedupSimilarity + "_" + DedupSimilarity.RelName.isSimilarTo;

		switch (Type.valueOf(entityType)) {

		case organization:

			cf = RelType.organizationOrganization + cf;

			break;

		case person:

			cf = RelType.personPerson + cf;

			break;

		case result:

			cf = RelType.resultResult + cf;

			break;

		default:

			throw new MSROException("invalid parameter entityType: " + entityType);

		}