D-Net

dnet.datasourcemanager.db.name=dnet_wds

services.datasourceManager.core=dnetPostgresDSManagerCore

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-wds</artifactId>

			<version>1.0.0-SNAPSHOT</version>

		</dependency>

	/**

	 * endpoint builder.

	 */

	private static final String ENDPOINT_TEMPLATE = "http://%s:%s/%s/services/actionManager";

	private String hostname;

	private String port;

	private String context;

			template.setAttribute("serviceUri", String.format(ENDPOINT_TEMPLATE, hostname, port, context));

	public String getHostname() {

		return hostname;

	public void setHostname(final String hostname) {

		this.hostname = hostname;

	public String getPort() {

		return port;

	}

	@Required

	public void setPort(final String port) {

		this.port = port;

	}

	public String getContext() {

		return context;

	}

	@Required

	public void setContext(final String context) {

		this.context = context;

	}

       xmlns:jaxws="http://cxf.apache.org/jaxws"

       xmlns:p="http://www.springframework.org/schema/p"

       xmlns:http="http://cxf.apache.org/transports/http/configuration"

                            http://cxf.apache.org/jaxws http://cxf.apache.org/schemas/jaxws.xsd">

	      p:endpoint-ref="actionManagerServiceEndpoint" p:hostname="${container.hostname}"

	      p:context="${container.context}"

	      p:port="${container.port}"/>

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

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

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

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * Ask the StatsManagerService to backup the stats cache of the portal specified by the additional BB parameter

 * <code>StatsManagerServiceBBAction.BACKUP_CACHE.getTargetPortalParamName()</code>. The cache can be restored with the last generated

 * backup by sending a "restore" BB message.

 *

 * @author alessia

 * @see RestoreStatsCacheJobNode

 */

public class BackupStatsCacheJobNode extends AbstractStatsJobNode {

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

	@Override

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

		job.setAction(StatsManagerServiceBBAction.BACKUP_CACHE.action());

		final String portal = getPortalName(token.getEnv());

		if (StringUtils.isNotBlank(portal)) {

			job.getParameters().put(StatsManagerServiceBBAction.BACKUP_CACHE.getTargetPortalParamName(), portal);

		} else {

			log.warn(StatsManagerServiceBBAction.BACKUP_CACHE.getTargetPortalParamName() + " not set. The StatsManagerService will use its default.");

		}

	}

}

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

public enum StatsManagerServiceBBAction {

	PROMOTE_SHADOW_STATS {

		@Override

		public String action() {

			return "promoteShadow";

		}

	},

	PROMOTE_SHADOW_CACHE {

		@Override

		public String action() {

			return "promoteCache";

		}

	},

	REFRESH_SHADOW_CACHE {

		@Override

		public String action() {

			return "refreshCache";

		}

	},

	VALIDATE_SHADOW_STATS {

		@Override

		public String action() {

			return "validate";

		}

	},

	BACKUP_CACHE {

		@Override

		public String action() {

			return "backup";

		}

	},

	RESTORE_CACHE {

		@Override

		public String action() {

			return "restore";

		}

	},

	MIGRATE_CACHE {

		@Override

		public String action() {

			return "migrate";

		}

		@Override

		public String getTargetPortalParamName() {

			return "targetCache";

		}

	};

	private static String DEFAULT_PORTAL_PARAM_NAME = "cache";

	public abstract String action();

	public String getSourcePortalParamName() {

		return DEFAULT_PORTAL_PARAM_NAME;

	}

	public String getTargetPortalParamName() {

		return DEFAULT_PORTAL_PARAM_NAME;

	}

}

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

import java.text.Normalizer;

/**

 * Created by claudio on 14/06/16.

 */

public class CleaningXsltFunctions {

	public static String clean(final String s) {

		return Normalizer.normalize(s, Normalizer.Form.NFD)

				.replaceAll("\\(.+\\)", "")

				.replaceAll("(\\W|\\p{InCombiningDiacriticalMarks}|\\p{Punct}|\\n|\\s)+", "")

				.toLowerCase()

				.trim();

	}

}

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

		ClassPathResource resource = new ClassPathResource("eu/dnetlib/wds/collector/plugins/CMR2Datasource.st");

		StringTemplate st = new StringTemplate(IOUtils.toString(resource.getInputStream()));

		datasourcePlugin.setXmlTemplate(st);

		System.out.println(iterator.next());

		System.out.println(iterator.next());

		System.out.println(iterator.next());

		System.out.println(iterator.next());

		System.out.println(iterator.next());

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

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 eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

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

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

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

import eu.dnetlib.rmi.data.hadoop.actionmanager.RawSet;

import eu.dnetlib.rmi.enabling.ISLookUpDocumentNotFoundException;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import eu.dnetlib.rmi.manager.MSROException;

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;

/**

 * 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 dedupConfigSequence;

	/**

	 * The job property.

	 */

	private String jobProperty;

	/**

	 * The action set path param name.

	 */

	private String actionSetPathParam;

	private String setsParam;

	/**

	 * 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 Env env) throws Exception {

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

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

		final String actionSetId = getDedupConfigSequence();

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

		}

		String actionSetDirectory;

		try {

			actionSetDirectory = isLookUpService.getResourceProfileByQuery(

					"/RESOURCE_PROFILE[./HEADER/RESOURCE_TYPE/@value='ActionManagerSetDSResourceType' and .//SET/@id = '" + actionSetId

							+ "']//SET/@ directory/string()");

		} catch (ISLookUpDocumentNotFoundException e) {

			throw new MSROException("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());

		env.setAttribute(set.get("jobProperty"), set.get("rawset"));

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

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

		env.setAttribute(getActionSetPathParam(), path);

		setList.add(set);

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

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

		env.setAttribute(getSetsParam(), sets);

		return Arc.DEFAULT_ARC;

	}

	/**

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

	}

	public String getDedupConfigSequence() {

		return dedupConfigSequence;

	}

	public void setDedupConfigSequence(final String dedupConfigSequence) {

		this.dedupConfigSequence = dedupConfigSequence;

	}

	public String getSetsParam() {

		return setsParam;

	}

	public void setSetsParam(final String setsParam) {

		this.setsParam = setsParam;

	}

}

package eu.dnetlib.msro.workflows.profiles;

/**

 * Created by claudio on 14/06/16.

 */

import java.io.IOException;

import java.io.StringWriter;

import java.util.List;

import java.util.Map;

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang3.StringEscapeUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.core.io.ClassPathResource;

import org.springframework.core.io.Resource;

public class RepositoryXsltFunctions {

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

	private static Resource ifaceTemplate = new ClassPathResource("/eu/dnetlib/msro/openaireplus/workflows/profiles/repo_interface.st");

	private static Resource paramTemplate = new ClassPathResource("/eu/dnetlib/msro/openaireplus/workflows/profiles/repo_param.st");

	public static String buildInterface(final String infopackage) {

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

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

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

		for (String param : parseParams(getValueBetween(infopackage, "===6===", null))) {

			StringTemplate p = getTemplate(paramTemplate);

			String paramName = StringUtils.substringBefore(param, "=").trim();

			String paramValue = StringUtils.substringAfter(param, "=").trim();

			p.setAttribute("param", paramName.toUpperCase());

			p.setAttribute("value", paramValue);

			try {

				switch (Params.valueOf(paramName.toUpperCase())) {

				case FORMAT:

					formats.add(p.toString());

					break;

				case SET:

					sets.add(p.toString());

					break;

				}

			} catch (Exception e) {

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

					otherParams.put(paramName, StringEscapeUtils.escapeXml(paramValue));

				}

			}

		}

		final StringTemplate i = getTemplate(ifaceTemplate);

		i.setAttribute("id", getValueBetween(infopackage, null, "===1==="));

		i.setAttribute("fileMode", getValueBetween(infopackage, "===1===", "===2==="));

		i.setAttribute("fileDesc", getValueBetween(infopackage, "===2===", "===3==="));

		i.setAttribute("iisWf", getValueBetween(infopackage, "===3===", "===4==="));

		i.setAttribute("baseUrl", getBaseUrl(getValueBetween(infopackage, "===4===", "===5===")));

		i.setAttribute("accessProtocol", getAccessProtocol(getValueBetween(infopackage, "===5===", "===6==="), otherParams));

		i.setAttribute("formats", ensureMinOccurs(formats, "<FORMAT/>"));

		i.setAttribute("sets", ensureMinOccurs(sets, "<SET/>"));

		if (!otherParams.isEmpty()) {

			i.setAttribute("otherParams", otherParams);

		}

		final String iface = i.toString();

		return iface;

	}

	;

	private static String getValueBetween(final String s, final String pre, final String post) {

		if ((pre == null) && (post == null)) {

			return s;

		} else if (pre == null) {

			return StringUtils.substringBefore(s, post);

		} else if (post == null) {

			return StringUtils.substringAfter(s, pre);

		} else {

			return StringUtils.substringBetween(s, pre, post);

		}

	}

	private static List<String> ensureMinOccurs(final List<String> list, final String empty) {

		if (list.isEmpty()) {

			list.add(empty);

		}

		return list;

	}

	private static Iterable<String> parseParams(final String s) {

		return Splitter.on("***").omitEmptyStrings().trimResults().split(s);

	}

	private static String getBaseUrl(final String url) {

		final StringTemplate b = getTemplate(paramTemplate);

		b.setAttribute("param", "BASE_URL");

		b.setAttribute("value", url);

		return b.toString();

	}

	private static String getAccessProtocol(final String protocol, final Map<String, String> otherParams) {

		final StringTemplate a = getTemplate(paramTemplate);

		a.setAttribute("param", "ACCESS_PROTOCOL");

		a.setAttribute("value", protocol);

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

			a.setAttribute("attrs", otherParams);

		}

		return a.toString();

	}

	private static StringTemplate getTemplate(final Resource res) {

		final StringWriter body = new StringWriter();

		try {

			IOUtils.copy(res.getInputStream(), body);

			return new StringTemplate(body.toString());

		} catch (IOException e) {

			log.error("unable to get template", e);

			throw new RuntimeException(e);

		}

	}

	private enum Params {

		FORMAT, SET

	}

}

package eu.dnetlib.msro.workflows.nodes.dedup.utils;

import java.util.Queue;

import com.google.gson.Gson;

import com.google.gson.GsonBuilder;

import eu.dnetlib.pace.config.DedupConfig;

/**

 * The Class DedupConfigurationOrchestration.

 */

public class DedupConfigurationOrchestration {

	/**

	 * The entity.

	 */

	private Entity entity;

	/**

	 * The action set id.

	 */

	private String actionSetId;

	/**

	 * The configurations.

	 */

	private Queue<DedupConfig> configurations;

	public DedupConfigurationOrchestration() {

	}

	/**

	 * Instantiates a new dedup configuration orchestration.

	 *

	 * @param entity         the entity

	 * @param actionSetId    the action set id

	 * @param configurations the configurations

	 */

	public DedupConfigurationOrchestration(final Entity entity, final String actionSetId, final Queue<DedupConfig> configurations) {

		super();

		this.setEntity(entity);

		this.setActionSetId(actionSetId);

		this.setConfigurations(configurations);

	}

	/**

	 * From json.

	 *

	 * @param json the json

	 * @return the dedup configuration orchestration

	 */

	public static DedupConfigurationOrchestration fromJSON(final String json) {

		return new Gson().fromJson(json, DedupConfigurationOrchestration.class);

	}

	/**

	 * Gets the entity.

	 *

	 * @return the entity

	 */

	public Entity getEntity() {

		return entity;

	}

	public void setEntity(final Entity entity) {

		this.entity = entity;

	}

	/**

	 * Gets the action set id.

	 *

	 * @return the action set id

	 */

	public String getActionSetId() {

		return actionSetId;

	}

	public void setActionSetId(final String actionSetId) {

		this.actionSetId = actionSetId;

	}

	/**

	 * Gets the configurations.

	 *

	 * @return the configurations

	 */

	public Queue<DedupConfig> getConfigurations() {

		return configurations;

	}

	public void setConfigurations(final Queue<DedupConfig> configurations) {

		this.configurations = configurations;

	}

	/*

	 * (non-Javadoc)

	 * 

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

	 */

	@Override

	public String toString() {

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

	}

}

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

import java.io.StringReader;

import java.util.Map;

import java.util.function.UnaryOperator;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.Element;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

/**

 * The Class PatchHostedBy.

 */

public class PatchHostedBy implements UnaryOperator<String> {

	/**

	 * The Constant log.

	 */

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

	/**

	 * The set spec hosted by map.

	 */

	private final Map<String, HostedByEntry> setSpecHostedByMap;

	/**

	 * The counters.

	 */

	private final HostedByCounters counters;

	/**

	 * The xpath.

	 */

	private final String xpath;

	/**

	 * The reader.

	 */

	private final SAXReader reader = new SAXReader();

	/**

	 * Instantiates a new patch hosted by.

	 *

	 * @param setSpecHostedByMap the set spec hosted by map

	 * @param xpath              the xpath

	 * @param counters           the counters

	 */

	public PatchHostedBy(final Map<String, HostedByEntry> setSpecHostedByMap, final String xpath, final HostedByCounters counters) {

		this.setSpecHostedByMap = setSpecHostedByMap;

		this.xpath = xpath;

		this.counters = counters;

		if (log.isDebugEnabled()) {

			log.debug("****************************************");

			log.debug("SetSpec/hostedBy map:");

			for (final Map.Entry<String, HostedByEntry> e : setSpecHostedByMap.entrySet()) {

				log.debug("   " + e.getKey() + " -> " + e.getValue());

			}

			log.debug("****************************************");

		}

	}

	/**

	 * Evaluate.

	 *

	 * @param record the record

	 * @return the string

	 */

	@Override

	public String apply(final String record) {

		try {

			final Document doc = this.reader.read(new StringReader(record));

			final Element node = (Element) doc.selectSingleNode("//*[local-name()='hostedBy']");

			if (node != null) {

				final HostedByEntry ds = findHostedBy(doc);

				if (ds != null) {

					node.addAttribute("id", ds.getId());

					node.addAttribute("name", ds.getName());

					this.counters.increaseCounter(ds.getId());

				}

				return doc.asXML();

			} else if (log.isDebugEnabled()) {

				log.debug(" -- Missing hostedBy --");

			}

		} catch (final Throwable e) {

			log.error("Error adding hosted by to " + record);

		}

		return record;

	}

	/**

	 * Find hosted by.

	 *

	 * @param doc the doc

	 * @return the hosted by entry

	 */

	private HostedByEntry findHostedBy(final Document doc) {

		for (final Object o : doc.selectNodes(this.xpath)) {

			final String set = ((Node) o).getText().trim();

			if (this.setSpecHostedByMap.containsKey(set)) {

				if (log.isDebugEnabled()) {

					log.debug(set + " -> " + this.setSpecHostedByMap.get(set));

				}

				return this.setSpecHostedByMap.get(set);

			} else if (log.isDebugEnabled()) {

				log.debug(set + " -> UNKNOWN REPO");

			}

		}

		return null;

	}

}

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

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

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

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

import eu.dnetlib.msro.workflows.nodes.dedup.utils.DedupConfigurationOrchestration;

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

import org.apache.commons.collections.CollectionUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class DedupDuplicateScanJobNode extends DedupConfigurationAwareJobNode {

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

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Token token) {

		return new DedupBlackboardWorkflowJobListener(token);

	}

	private class DedupBlackboardWorkflowJobListener extends BlackboardWorkflowJobListener {

		public DedupBlackboardWorkflowJobListener(final Token token) {

			super(token);

		}

		@Override

		protected void onDone(final BlackboardJob job) {

			final DedupConfigurationOrchestration confs = getDedupConfigurationOrchestration();

			confs.getConfigurations().poll();

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

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

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

				getToken().release("done");

			} else {

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

				getToken().release(Arc.DEFAULT_ARC);

			}

		}

	}

}

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

import java.io.StringReader;

import java.util.Iterator;

import java.util.List;

import java.util.Queue;

import com.google.common.collect.Lists;

import com.google.common.collect.Queues;

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

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

import eu.dnetlib.enabling.resultset.factory.ResultSetFactory;

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

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

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

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

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

import eu.dnetlib.rmi.common.ResultSet;

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

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

public class BuildSimilarityMeshJobNode extends AsyncJobNode {

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

	/**

	 * The result set client factory.

	 */

	@Autowired

	private ResultSetClient resultSetClient;

	@Autowired

	private ResultSetFactory resultSetFactory;

	private StringTemplate similarity;

	private String inputEprParam;

	private String outputEprParam;

	@Override

	protected String execute(final Env env) throws Exception {

		final ResultSet<?> inputRs = env.getAttribute(getInputEprParam(), ResultSet.class);

		final Iterator<String> it = resultSetClient.iter(inputRs, String.class).iterator();

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

		final SAXReader reader = new SAXReader();

		if (it.hasNext()) {

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

		}

		final ResultSet<String> rsOut = resultSetFactory.createResultSet(() -> {

			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() && it.hasNext()) {

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

						}

						return buildSimilarity((Similarity) o);

					}

				}

				@Override

				public void remove() {

					throw new UnsupportedOperationException();

				}

			};

		});

		env.setAttribute(getOutputEprParam(), rsOut);

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

import java.io.StringReader;

import java.util.Map;

import java.util.function.UnaryOperator;

import com.google.common.collect.Maps;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import eu.dnetlib.enabling.resultset.factory.ResultSetFactory;

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

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

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

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.data.DatabaseException;

import eu.dnetlib.rmi.data.DatabaseService;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

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

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

public class PatchHostedByJobNode extends SimpleJobNode {

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

	private String inputEprParam;

	private String outputEprParam;

	@Value("${dnet.openaire.db.name}")

	private String dbName;

	private String countersParam;

	private String hostedbyMapTable;

	private String xpathEntry;

	private String overrideDataSourceId;

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Autowired

	private ResultSetFactory resultSetFactory;

	@Autowired

	private ResultSetClient resultSetClient;

	/**

	 * {@inheritDoc}

	 */

	@Override

	protected String execute(final Env env) throws Exception {

		final ResultSet<?> inputEpr = env.getAttribute(this.inputEprParam, ResultSet.class);

		final HostedByCounters counters = new HostedByCounters();

		String datasourceId;

		if (StringUtils.isEmpty(this.overrideDataSourceId)) {

			datasourceId = env.getAttribute("parentDatasourceId", String.class);

		} else {

			datasourceId = getOverrideDataSourceId();

		}

		final UnaryOperator<String> mapper = new PatchHostedBy(loadHostedByMap(datasourceId), getXpathEntry(), counters);

		final ResultSet<String> epr = this.resultSetFactory.map(inputEpr, String.class, mapper);

		env.setAttribute(this.outputEprParam, epr);

		env.setAttribute(this.countersParam, counters);

		return Arc.DEFAULT_ARC;

	}

	private Map<String, HostedByEntry> loadHostedByMap(final String datasourceId) throws DocumentException, DatabaseException {

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

		final String sql = "SELECT d.id, d.officialname, p.entry from %s p JOIN datasources d ON (p.datasourceid = d.id) WHERE p.oa_source_id= '%s'";

		final ResultSet<String> epr = this.serviceLocator.getService(DatabaseService.class).searchSQL(getDbName(),

				String.format(sql, getHostedbyMapTable(), datasourceId));

		final SAXReader reader = new SAXReader();

		for (final String s : this.resultSetClient.iter(epr, String.class)) {

			final Document doc = reader.read(new StringReader(s));

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

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

			final String dsName = doc.valueOf("//FIELD[@name='officialname']");

			map.put(entry, new HostedByEntry(dsId, dsName));

		}

		log.info(String.format("built hostedByMap from dsId '%s', size: '%s'", datasourceId, map.size()));

		return map;

	}

	/**

	 * Getter for property 'inputEprParam'.

	 *

	 * @return Value for property 'inputEprParam'.

	 */

	public String getInputEprParam() {

		return this.inputEprParam;

	}

	/**

	 * Setter for property 'inputEprParam'.

	 *

	 * @param inputEprParam Value to set for property 'inputEprParam'.

	 */

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	/**

	 * Getter for property 'outputEprParam'.

	 *

	 * @return Value for property 'outputEprParam'.

	 */

	public String getOutputEprParam() {

		return this.outputEprParam;

	}

	/**

	 * Setter for property 'outputEprParam'.

	 *

	 * @param outputEprParam Value to set for property 'outputEprParam'.

	 */

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	/**

	 * Getter for property 'dbName'.

	 *

	 * @return Value for property 'dbName'.

	 */

	public String getDbName() {

		return this.dbName;

	}

	/**

	 * Setter for property 'dbName'.

	 *

	 * @param dbName Value to set for property 'dbName'.

	 */

	public void setDbName(final String dbName) {

		this.dbName = dbName;

	}

	/**

	 * Getter for property 'countersParam'.

	 *

	 * @return Value for property 'countersParam'.

	 */

	public String getCountersParam() {

		return this.countersParam;

	}

	/**

	 * Setter for property 'countersParam'.

	 *

	 * @param countersParam Value to set for property 'countersParam'.

	 */

	public void setCountersParam(final String countersParam) {

		this.countersParam = countersParam;

	}

	/**

	 * @return the hostedbyMapTable

	 */

	public String getHostedbyMapTable() {

		return this.hostedbyMapTable;

	}

	/**

	 * @param hostedbyMapTable the hostedbyMapTable to set

	 */

	public void setHostedbyMapTable(final String hostedbyMapTable) {

		this.hostedbyMapTable = hostedbyMapTable;

	}

	public String getXpathEntry() {

		return this.xpathEntry;

	}

	public void setXpathEntry(final String xpathEntry) {

		this.xpathEntry = xpathEntry;

	}

	/**

	 * @return the overrideDataSourceId

	 */

	public String getOverrideDataSourceId() {

		return this.overrideDataSourceId;

	}

	/**

	 * @param overrideDataSourceId the overrideDataSourceId to set

	 */

	public void setOverrideDataSourceId(final String overrideDataSourceId) {

		this.overrideDataSourceId = overrideDataSourceId;

	}

}

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

import java.util.Map.Entry;

import javax.annotation.Resource;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

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

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

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

import eu.dnetlib.rmi.datasource.DatasourceDesc;

import eu.dnetlib.rmi.datasource.DatasourceManagerService;

import eu.dnetlib.rmi.datasource.DatasourceManagerServiceException;

import eu.dnetlib.rmi.datasource.IfaceDesc;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class UpsertHostedByApisJobNode extends SimpleJobNode {

	public static final String extraFieldsForTotal = "last_aggregation_total";

	public static final String extraFieldForDate = "last_aggregation_date";

	public static final String extraFieldForMdId = "last_aggregation_mdId";

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

	private static final String HOSTED_BY_COMPLIANCE = "hostedBy";

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String countersParam;

	private String mdId;

	@Override

	protected String execute(final Env env) throws Exception {

		final HostedByCounters counters = env.getAttribute(this.countersParam, HostedByCounters.class);

		final DatasourceManagerService dsManager = this.serviceLocator.getService(DatasourceManagerService.class);

		final String date = DateUtils.now_ISO8601();

		final String namespacePrefix = env.getAttribute("namespacePrefix", String.class);

		log.info(counters);

		for (final Entry<String, Integer> e : counters.getCounters().entrySet()) {

			updateHostedByApi(dsManager, e.getKey(), namespacePrefix, date, e.getValue());

		}

		return Arc.DEFAULT_ARC;

	}

	private void updateHostedByApi(final DatasourceManagerService dsManager,

			final String dsId,

			final String namepsacePrefix,

			final String date,

			final int size) {

		log.info("Verifying hostedBy api in ds: " + dsId);

		try {

			final DatasourceDesc ds = dsManager.getDatasource(dsId);

			for (final IfaceDesc iface : ds.getInterfaces()) {

				if (HOSTED_BY_COMPLIANCE.equals(iface.getCompliance())) {

					// dsManager.updateExtraField(dsId, iface.getId(), namepsacePrefix + ":" + extraFieldForDate, date, false);

					// dsManager.updateExtraField(dsId, iface.getId(), namepsacePrefix + ":" + extraFieldsForTotal, Integer.toString(size),

					// false);

					// dsManager.updateExtraField(dsId, iface.getId(), namepsacePrefix + ":" + extraFieldForMdId, mdId, false);

					return;

				}

			}

			final IfaceDesc iface = new IfaceDesc();

			iface.setId("api_________::" + dsId + "::hostedBy");

			iface.setTypology(ds.getDatasourceClass());

			iface.setCompliance(HOSTED_BY_COMPLIANCE);

			iface.setAccessProtocol("UNKNOWN");

			iface.setContentDescription("metadata");

			iface.setBaseUrl("");

			iface.setActive(false);

			iface.setRemovable(true);

			iface.getExtraFields().put(namepsacePrefix + ":" + extraFieldForDate, date);

			iface.getExtraFields().put(namepsacePrefix + ":" + extraFieldsForTotal, Integer.toString(size));

			iface.getExtraFields().put(namepsacePrefix + ":" + extraFieldForMdId, this.mdId);

			dsManager.addInterface(dsId, iface);

		} catch (final DatasourceManagerServiceException e) {

			log.warn("Error setting hostedBy api of ds: " + dsId, e);

		}

	}

	public String getMdId() {

		return this.mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

	public String getCountersParam() {

		return this.countersParam;

	}

	public void setCountersParam(final String countersParam) {

		this.countersParam = countersParam;

	}

}

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

import java.nio.file.FileSystems;

import java.nio.file.Path;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

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

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

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

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

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;

/**

 * Created by claudio on 14/10/15.

 */

public class MinDistSearchHadoopJobNode extends DedupConfigurationLoaderJobNode {

	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;

	private String workDir;

	@Override

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

		String depthString = token.getEnv().getAttribute(DepthParam, String.class);

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

		if (StringUtils.isBlank(depthString)) {

			depthString = "0";

		}

		int depth = Integer.valueOf(depthString);

		final String outputPath = getPath(getWorkDir(), depth);

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

		switch (getStatusFromEnv(token.getEnv())) {

		case DATALOAD:

			setHadoopJob("dedupSimilarity2GraphJob");

			job.getParameters().put("mapred.output.dir", getPath(getWorkDir(), depth) + "/out");

			hadoopService.createHdfsDirectory(getCluster(), outputPath, true);

			break;

		case DEPTH_N:

			setHadoopJob("dedupMinDistGraphJob");

			final String newOutputPath = getPath(getWorkDir(), depth + 1);

			hadoopService.createHdfsDirectory(getCluster(), 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.getEnv().setAttribute(DepthParam, String.valueOf(depth + 1));

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

			break;

		}

		super.prepareJob(job, token);

	}

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

		log.info("got basePath: " + basePath);

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

		final String path = fsPath.toAbsolutePath().toString();

		log.info("built outputPath: " + path);

		return path;

	}

	private STATUS getStatusFromEnv(final Env env) {

		if (StringUtils.isBlank(env.getAttribute(StatusParam, String.class))) {

			return STATUS.DATALOAD;

		}

		STATUS current = STATUS.DATALOAD;

		try {

			current = STATUS.valueOf(env.getAttribute(StatusParam, String.class));

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

		} catch (IllegalArgumentException e) {}

		return current;

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Token token) {

		return new BlackboardWorkflowJobListener(token) {

			@Override

			protected void onDone(final BlackboardJob job) {