D-Net

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

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.rmi.MSROException;

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

import org.apache.commons.lang3.StringUtils;

import java.util.List;

import java.util.Map;

public class ExtractOutputPathJobNode extends SimpleJobNode {

    private String hdfsOutputPathParam;

    @Override

    protected String execute(NodeToken token) throws Exception {

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

        if (StringUtils.isBlank(sets))

            throw  new MSROException("¯\\\\_(ツ)_/¯ cannot find sets on env");

        final List<Map<String, String>> setsMap = new Gson().fromJson(sets, List.class);

        if (setsMap== null || setsMap.size() !=1)  {

            throw  new MSROException("¯\\\\_(ツ)_/¯ Sets map from json is wrong!");

        }

        final String path = setsMap.get(0).get("path");

        if (StringUtils.isEmpty(path)){

            throw new MSROException("Path is empty");

        }

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

        return Arc.DEFAULT_ARC;

    }

    public String getHdfsOutputPathParam() {

        return hdfsOutputPathParam;

    }

    public void setHdfsOutputPathParam(String hdfsOutputPathParam) {

        this.hdfsOutputPathParam = hdfsOutputPathParam;

    }

}

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

import java.util.List;

import java.util.Map;

import com.google.common.base.Predicate;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.fs.Path;

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

public class PrepareActionSetsJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	private String ACTIONSET_PATH_XQUERY_TEMPLATE =

			"for $x in collection('/db/DRIVER/ActionManagerSetDSResources/ActionManagerSetDSResourceType') "

					+ "return $x/RESOURCE_PROFILE/BODY/SET[@id = '%s']/@directory/string()";

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private String sets;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final List<Map<String, String>> setList = getSetList();

		final String now = DateUtils.now_ISO8601();

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

		final String basePath = lookUpService.getResourceProfileByQuery(

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

		for (Map<String, String> set : setList) {

			set.put("rawset", RawSet.newInstance().getId());

			set.put("creationDate", now);

			set.put("path", getFullPath(basePath, set, lookUpService));

			if (set.get("enabled").equals("true")) {

				log.info("preparing set: " + simplifySetInfo(set));

			}

			// setting the job properties needed to name the rawsets

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

		}

		token.getEnv().setAttribute("sets", new Gson().toJson(setList));

		token.getEnv().setAttribute("actionManagerBasePath", basePath);

		return Arc.DEFAULT_ARC;

	}

	private String getFullPath(final String basePath, final Map<String, String> set, final ISLookUpService lookUpService) throws MSROException {

		return new Path(basePath + "/" + getPath(set.get("set"), lookUpService) + "/" + set.get("rawset")).toString();

	}

	private String getPath(final String setId, final ISLookUpService lookUpService) throws MSROException {

		try {

			return lookUpService.getResourceProfileByQuery(String.format(ACTIONSET_PATH_XQUERY_TEMPLATE, setId));

		} catch (ISLookUpException e) {

			throw new MSROException(String.format("Error obtaining directory from ActionSet profile %s", setId), e);

		}

	}

	private Map<String, String> simplifySetInfo(final Map<String, String> set) {

		return Maps.filterKeys(set, new Predicate<String>() {

			@Override

			public boolean apply(final String k) {

				return k.equals("set") || k.equals("rawset");

			}

		});

	}

	@SuppressWarnings("unchecked")

	protected List<Map<String, String>> getSetList() {

		return new Gson().fromJson(getSets(), List.class);

	}

	public String getSets() {

		return sets;

	}

	public void setSets(final String sets) {

		this.sets = sets;

	}

}

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

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.rmi.ActionManagerService;

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

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

import org.apache.commons.lang.StringUtils;

public class PromoteActionsJobNode extends BlackboardJobNode {

	public static final String ALL_SETS = "ALL SETS";

	private String set;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		job.setAction("PROMOTE");

		if (!StringUtils.isBlank(getSet()) && !getSet().equals(ALL_SETS)) {

			job.getParameters().put("set", getSet());

			token.getEnv().setAttribute("set", getSet());

		}

	}

	public String getSet() {

		return set;

	}

	public void setSet(final String set) {

		this.set = set;

	}

}

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

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.rmi.ActionManagerService;

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

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

import org.apache.commons.lang.StringUtils;

/**

 * Created by claudio on 14/04/16.

 */

public class PromoteActionsHDFSJobNode extends BlackboardJobNode {

	public static final String ALL_SETS = "ALL SETS";

	private String set;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		job.setAction("PROMOTE_FROM_HDFS");

		if (!StringUtils.isBlank(getSet()) && !getSet().equals(ALL_SETS)) {

			job.getParameters().put("set", getSet());

			token.getEnv().setAttribute("set", getSet());

		}

	}

	public String getSet() {

		return set;

	}

	public void setSet(final String set) {

		this.set = set;

	}

}

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

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

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

import eu.dnetlib.msro.workflows.resultset.ProcessCountingResultSetFactory;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class ProcessContextsJobNode extends SimpleJobNode implements ProgressJobNode {

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

	private String eprParam;

	private String contextObj;

	private ResultsetProgressProvider progressProvider;

	private String contextId;

	private String contextLabel;

	private String contextType;

	private String contextParams;

	private String dashboardVisibility;

	@Resource

	private ResultSetClientFactory resultSetClientFactory;

	@Resource

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String epr = token.getEnv().getAttribute(eprParam);

		this.progressProvider = processCountingResultSetFactory.createProgressProvider(token.getProcess(), epr);

		final Iterable<String> iter = resultSetClientFactory.getClient(progressProvider.getEpr());

		final ContextDesc context = ContextUtils.getContext(iter, getContextId(), getContextLabel(), getContextType(), getContextParams(), getDashboardVisibility());

		token.getEnv().setTransientAttribute(contextObj, context);

		return Arc.DEFAULT_ARC;

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public String getContextObj() {

		return contextObj;

	}

	public void setContextObj(final String contextObj) {

		this.contextObj = contextObj;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

	public String getContextId() {

		return contextId;

	}

	public void setContextId(final String contextId) {

		this.contextId = contextId;

	}

	public String getContextLabel() {

		return contextLabel;

	}

	public void setContextLabel(final String contextLabel) {

		this.contextLabel = contextLabel;

	}

	public String getContextType() {

		return contextType;

	}

	public void setContextType(final String contextType) {

		this.contextType = contextType;

	}

	public String getContextParams() {

		return contextParams;

	}

	public void setContextParams(final String contextParams) {

		this.contextParams = contextParams;

	}

	public String getDashboardVisibility() {

		return dashboardVisibility;

	}

	public void setDashboardVisibility(final String dashboardVisibility) {

		this.dashboardVisibility = dashboardVisibility;

	}

}

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

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import org.dom4j.DocumentHelper;

import org.dom4j.Element;

public class ContextDesc {

	private String id;

	private String label;

	private String type;

	private String xml;

	private Map<String, ContextPart> categories = Maps.newLinkedHashMap();

	private Map<String, String> dbEntries = Maps.newLinkedHashMap();

	private Map<String, String> params;

	public ContextDesc(final String id, final String label, final String type, final Map<String, String> params) {

		this.id = id;

		this.label = label;

		this.type = type;

		this.params = params;

	}

	public String getId() {

		return id;

	}

	public void setId(final String id) {

		this.id = id;

	}

	public String getLabel() {

		return label;

	}

	public void setLabel(final String label) {

		this.label = label;

	}

	public String getType() {

		return type;

	}

	public void setType(final String type) {

		this.type = type;

	}

	public String getXml() {

		return xml;

	}

	public void setXml(final String xml) {

		this.xml = xml;

	}

	public Map<String, ContextPart> getCategories() {

		return categories;

	}

	public void setCategories(final Map<String, ContextPart> categories) {

		this.categories = categories;

	}

	public Map<String, String> getDbEntries() {

		return dbEntries;

	}

	public void setDbEntries(final Map<String, String> dbEntries) {

		this.dbEntries = dbEntries;

	}

	public Element asDomElement() {

		final Element ctxElem = DocumentHelper.createElement("context");

		ctxElem.addAttribute("id", getId());

		ctxElem.addAttribute("label", getLabel());

		ctxElem.addAttribute("type", getType());

		for (Element param : getParamsAsElements()) {

			ctxElem.add(param);

		}

		for (ContextPart part : getCategories().values()) {

			ctxElem.add(part.asDomElement("category"));

		}

		return ctxElem;

	}

	public List<Element> getParamsAsElements() {

		List<Element> elements = Lists.newArrayList();

		for (Entry<String, String> e : params.entrySet()) {

			Element param = DocumentHelper.createElement("param");

			param.addAttribute("name", e.getKey());

			param.setText(e.getValue());

			elements.add(param);

		}

		return elements;

	}

	public Map<String, String> getParams() {

		return params;

	}

	public void setParams(final Map<String, String> params) {

		this.params = params;

	}

}

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

import java.io.StringReader;

import java.lang.reflect.Type;

import java.util.Map;

import java.util.Map.Entry;

import com.google.common.base.Function;

import com.google.common.collect.Maps;

import com.google.common.reflect.TypeToken;

import com.google.gson.Gson;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.*;

import org.dom4j.io.SAXReader;

/**

 * Created by claudio on 01/03/16.

 */

public class ContextUtils {

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

	private static final String DASHBOARD_VISIBILITY = "status";

	public static ContextDesc getContext(final Iterable<String> it,

			final String contextId,

			final String contextLabel,

			final String contextType,

			final String params, final String dashboardVisibility) throws

			DocumentException {

		final Map<String, String> paramMap = getParamMap(params);

		paramMap.put(DASHBOARD_VISIBILITY, dashboardVisibility);

		return new ContextUtils().getContextDesc(it, contextId, contextLabel, contextType, paramMap);

	}

	private static Map<String, String> getParamMap(String jsonMap) {

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

		if (StringUtils.isNotBlank(jsonMap)) {

			Type mapType = new TypeToken<Map<String, String>>() {

			}.getType();

			paramMap = new Gson().fromJson(jsonMap, mapType);

		}

		return paramMap;

	}

	private ContextDesc getContextDesc(final Iterable<String> it,

			final String contextId,

			final String contextLabel,

			final String contextType,

			final Map<String, String> params) throws

			DocumentException {

		final ContextDesc context = new ContextDesc(contextId, contextLabel, contextType, params);

		final SAXReader reader = new SAXReader();

		for (String s : it) {

			populateContext(context, reader, s);

		}

		return context;

	}

	private void populateContext(final ContextDesc context, final SAXReader reader, final String s) throws DocumentException {

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

		for (Object o : doc.selectNodes("//fundingtree")) {

			final Element treeNode = (Element) o;

			final String funder = treeNode.valueOf("./funder/id");

			if (StringUtils.isBlank(funder)) {

				log.debug("No funder found, skipping population from the following XML: \n" + s);

				return;

			}

			if (!context.getDbEntries().containsKey(funder)) {

				log.info("Found funder: " + funder);

				context.getDbEntries().put(funder, "<fundingtree>" + treeNode.selectSingleNode("./funder").asXML() + "</fundingtree>");

				log.debug("db entry: " + context.getDbEntries().get(funder));

			}

			final String openaireId = treeNode.valueOf("./*[starts-with(local-name(),'funding_level_')]/id");

			if (!context.getDbEntries().containsKey(openaireId) && StringUtils.isNotBlank(openaireId)) {

				log.info("Found funding: " + openaireId);

				context.getDbEntries().put(openaireId, treeNode.asXML());

				log.debug("db entry: " + context.getDbEntries().get(openaireId));

				final Node node0 = treeNode.selectSingleNode(".//funding_level_0");

				if (node0 != null) {

					final ContextPart part = calculatePart(node0);

					if (context.getCategories().containsKey(part.getId())) {

						for (ContextPart p : part.getParts().values()) {

							context.getCategories().get(part.getId()).addPart(p);

						}

					} else {

						context.getCategories().put(part.getId(), part);

					}

				}

			}

		}

	}

	private ContextPart calculatePart(final Node node) {

		// final String newId = contextId + "::" + StringUtils.substringAfter(node.valueOf("./id"), "::");

		// ids are built as: nsPrefix :: funderID :: fundingLevel0ID :: etc etc, hence it seems we might not need the contextId parameter.

		final String newId = StringUtils.substringAfter(node.valueOf("./id"), "::");

		final ContextPart part = new ContextPart(newId, node.valueOf("./description"));

		part.getParams().put("name", node.valueOf("./name"));

		part.getParams().put("openaireId", node.valueOf("./id"));

		part.getParams().put("class", node.valueOf("./class"));

		final Element parent = node.getParent() != null ? node.getParent().getParent() : null;

		if (parent != null && parent.getName().startsWith("funding_level_")) {

			final ContextPart p = calculatePart(parent);

			part.getParts().put(p.getId(), p);

		}

		return part;

	}

	public static Function<Entry<String, String>, String> getContextRowTransformer() {

		return new Function<Entry<String, String>, String>() {

			private final SAXReader reader = new SAXReader();

			@Override

			public String apply(final Entry<String, String> e) {

				try {

					final Document docFundingPath = reader.read(new StringReader(e.getValue()));

					final Element root = DocumentHelper.createElement("ROWS");

					final Map<String, String> fundingFields = Maps.newLinkedHashMap();

					fundingFields.put("_dnet_resource_identifier_", e.getKey());

					fundingFields.put("id", e.getKey());

					fundingFields.put("path", e.getValue());

					fundingFields.put("jurisdiction", docFundingPath.valueOf("//funder/jurisdiction"));

					final String desc = docFundingPath.valueOf("(//*[starts-with(local-name(),'funding_level_')])[1]/description");

					if (StringUtils.isNotBlank(desc)) {

						fundingFields.put("description", desc);

					}

					final Map<String, String> orgFields = findFunderInfo(docFundingPath);

					final String orgId = orgFields.get("id");

					if (StringUtils.isNotBlank(orgId)) {

						addRow(root, "dsm_organizations", orgFields);

						fundingFields.put("funder", orgId);

					}

					addRow(root, "fundingpaths", fundingFields);

					if (log.isDebugEnabled()) {

						log.debug("Db entries: " + root.asXML());

					}

					return root.asXML();

				} catch (DocumentException e1) {

					log.error("Error parsing xml", e1);

					throw new RuntimeException("Error parsing xml", e1);

				}

			}

			private Map<String, String> findFunderInfo(final Document doc) {

				final Map<String, String> res = Maps.newLinkedHashMap();

				res.put("_dnet_resource_identifier_", doc.valueOf("//funder/id"));

				res.put("id", doc.valueOf("//funder/id"));

				res.put("legalshortname", doc.valueOf("//funder/shortname"));

				res.put("legalname", doc.valueOf("//funder/name"));

				return res;

			}

			private void addRow(final Element root, final String table, final Map<String, String> fields) {

				final Element row = root.addElement("ROW");

				row.addAttribute("table", table);

				for (Map.Entry<String, String> e : fields.entrySet()) {

					final Element pathField = row.addElement("FIELD");

					pathField.addAttribute("name", e.getKey());

					pathField.setText(e.getValue());

				}

			}

		};

	}

}

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

import java.util.List;

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

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

import eu.dnetlib.msro.workflows.resultset.ProcessCountingResultSetFactory;

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class GenerateFETH2020ContextJobNode extends SimpleJobNode implements ProgressJobNode {

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

	private String eprParam;

	private ResultsetProgressProvider progressProvider;

	@Resource

	private ResultSetClientFactory resultSetClientFactory;

	@Resource

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	public ProgressProvider getProgressProvider() {

		return this.progressProvider;

	}

	@Override

	protected String execute(NodeToken token) throws Exception {

		final String epr = token.getEnv().getAttribute(eprParam);

		this.progressProvider = processCountingResultSetFactory.createProgressProvider(token.getProcess(), epr);

		final Iterable<String> iter = resultSetClientFactory.getClient(progressProvider.getEpr());

		BuildH2020FETTaxonomy builder = new BuildH2020FETTaxonomy();

		builder.setIterator(iter.iterator());

		String taxonomy = builder.parseProjects();

		final String xquery = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType') "

				+ "where $x//CONFIGURATION/context[@id='fet-h2020'] return $x//RESOURCE_IDENTIFIER/@value/string()";

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

		if (list.isEmpty()) {

			registerProfile(taxonomy);

		} else {

			deleteProfile(list.get(0));

			registerProfile(taxonomy);

		}

		return Arc.DEFAULT_ARC;

	}

	private void registerProfile(String profile) throws ISRegistryException {

		log.info("registering fet-h2020 profile");

		ISRegistryService is = serviceLocator.getService(ISRegistryService.class);

		String id = is.registerProfile(profile);

		log.info("Generating profile with id " + id);

	}

	private void deleteProfile(String profId) throws ISRegistryException {

		log.info("deleting fet-h2020 profile with id " + profId);

		ISRegistryService is = serviceLocator.getService(ISRegistryService.class);

		is.deleteProfile(profId);

	}

	public String getEprParam() {

		return eprParam;

	}

	public void setEprParam(final String eprParam) {

		this.eprParam = eprParam;

	}

	public void setProgressProvider(final ResultsetProgressProvider progressProvider) {

		this.progressProvider = progressProvider;

	}

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	public UniqueServiceLocator getServiceLocator() {

		return serviceLocator;

	}

	public void setServiceLocator(final UniqueServiceLocator serviceLocator) {

		this.serviceLocator = serviceLocator;

	}

}

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

import java.io.StringReader;

import java.util.List;

import javax.annotation.Resource;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Element;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

public class SaveContextProfileJobNode extends SimpleJobNode {

	private String contextObj;

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final ContextDesc desc = (ContextDesc) token.getEnv().getTransientAttribute(contextObj);

		final String xquery = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType') " + "where $x//CONFIGURATION/context[@id='"

				+ desc.getId() + "' " + "and @type='" + desc.getType() + "'] " + "return $x";

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

		if (list.isEmpty()) {

			registerNewProfile(desc);

		} else {

			updateProfile(list.get(0), desc);

		}

		return Arc.DEFAULT_ARC;

	}

	private boolean updateProfile(final String profile, final ContextDesc desc) throws DocumentException, ISRegistryException {

		final SAXReader reader = new SAXReader();

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

		final String profId = doc.valueOf("//HEADER/RESOURCE_IDENTIFIER/@value");

		final Element ctxElem = (Element) doc.selectSingleNode("//CONFIGURATION/context[@id='" + desc.getId() + "' and @type='" + desc.getType() + "']");

		updateContextParams(ctxElem, desc);

		for (ContextPart cat : desc.getCategories().values()) {

			final Node catElem = ctxElem.selectSingleNode("./category[@id='" + cat.getId() + "']");

			if (catElem != null) {

				catElem.detach();

			}

			ctxElem.add(cat.asDomElement("category"));

		}

		log.info("registering profile context " + desc.getId());

		return serviceLocator.getService(ISRegistryService.class).updateProfile(profId, doc.asXML(), "ContextDSResourceType");

	}

	private void updateContextParams(final Element ctxElement, ContextDesc desc) {

		//removing old PARAMs

		List<Node> oldParams = ctxElement.selectNodes("param");

		for (Node n : oldParams) {

			n.detach();

		}

		//adding new params

		for (Element param : desc.getParamsAsElements()) {

			ctxElement.add(param);

		}

	}

	private String registerNewProfile(final ContextDesc desc) throws DocumentException, ISRegistryException {

		final SAXReader reader = new SAXReader();

		final Document doc = reader.read(getClass().getResourceAsStream(

				"/eu/dnetlib/msro/openaireplus/workflows/repo-hi/entityreg-contexts/xml/contextProfile.xml"));

		((Element) doc.selectSingleNode("//CONFIGURATION")).add(desc.asDomElement());

		log.info("updating profile context " + desc.getId());

		return serviceLocator.getService(ISRegistryService.class).registerProfile(doc.asXML());

	}

	public String getContextObj() {

		return contextObj;

	}

	public void setContextObj(final String contextObj) {

		this.contextObj = contextObj;

	}

}

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

import java.util.Map;

import java.util.Map.Entry;

import com.google.common.collect.Maps;

import org.dom4j.DocumentHelper;

import org.dom4j.Element;

public class ContextPart {

	private String id;

	private String label;

	private Map<String, String> params = Maps.newLinkedHashMap();

	private Map<String, ContextPart> parts = Maps.newLinkedHashMap();

	public ContextPart() {

	}

	public ContextPart(final String id, final String label) {

		this.id = id;

		this.label = label;

	}

	public String getId() {

		return id;

	}

	public void setId(final String id) {

		this.id = id;

	}

	public String getLabel() {

		return label;

	}

	public void setLabel(final String label) {

		this.label = label;

	}

	public Map<String, String> getParams() {

		return params;

	}

	public void setParams(final Map<String, String> params) {

		this.params = params;

	}

	public Map<String, ContextPart> getParts() {

		return parts;

	}

	public void setParts(final Map<String, ContextPart> parts) {

		this.parts = parts;

	}

	public void addPart(final ContextPart part) {

		if (parts.containsKey(part.getId())) {

			final ContextPart localChild = getParts().get(part.getId());

			for (ContextPart child : part.getParts().values()) {

				localChild.addPart(child);

			}

		} else {

			parts.put(part.getId(), part);

		}

	}

	public Element asDomElement(final String name) {

		final Element elem = DocumentHelper.createElement(name);

		elem.addAttribute("id", id);

		elem.addAttribute("label", label);

		elem.addAttribute("claim", "false");

		for (Entry<String, String> e : params.entrySet()) {

			final Element p = elem.addElement("param");

			p.addAttribute("name", e.getKey());

			p.setText(e.getValue());

		}

		for (ContextPart child : parts.values()) {

			elem.add(child.asDomElement("concept"));

		}

		return elem;

	}

}

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

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

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

import eu.dnetlib.enabling.datasources.LocalOpenaireDatasourceManager;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.workflows.nodes.repohi.UpdateMetaWfStatusJobNode;

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

public class UpdateOpenaireMetaWfStatusJobNode extends UpdateMetaWfStatusJobNode {

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Autowired

	private LocalDatasourceManager dsManager;

	@Override

	protected void updateDatasource(final String dsId, final String ifaceId) throws Exception {

		final String openaireDsId = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(

				"/*[.//RESOURCE_IDENTIFIER/@value = '" + dsId + "']//FIELD/value[../key='OpenAireDataSourceId']/text()");

		if (openaireDsId.equals("openaire____::bootstrap")) {

			super.updateDatasource(dsId, ifaceId);

		} else {

			dsManager.setManaged(openaireDsId, true);

			dsManager.setActive(openaireDsId, ifaceId, true);

		}

	}

}

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

import java.util.Iterator;

import java.util.Map;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.common.logging.DnetLogger;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

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

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

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public class IncrementalOperationJobNode extends SimpleJobNode {

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

    //incremental or refresh

    private String operationType;

    @Autowired

    private DnetLogger dnetLogger;

    @Override

    protected String execute(NodeToken nodeToken) throws Exception {

        if ("incremental".equalsIgnoreCase(operationType)) {

            final String currentWfProfileId = findCurrentWfProfileId(nodeToken);

            final Long lastSuccessEndDate = findLastSuccessEndDate(currentWfProfileId);

            if(forceRefresh(nodeToken, lastSuccessEndDate, currentWfProfileId)) return Arc.DEFAULT_ARC;

            log.info("Last success date "+ lastSuccessEndDate);

            nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Operation type set to INCREMENTAL with date "+DateUtils.calculate_ISO8601(lastSuccessEndDate));

            nodeToken.getFullEnv().setAttribute("operationType", "INCREMENTAL");

            nodeToken.getFullEnv().setAttribute("DateFromFilter", lastSuccessEndDate);            

            return Arc.DEFAULT_ARC;

        }

        nodeToken.getFullEnv().setAttribute("operationType", "REFRESH");

        nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Operation type manually set to REFRESH");

        return Arc.DEFAULT_ARC;

    }

    protected boolean forceRefresh(final NodeToken nodeToken, final Long lastSuccessEndDate, final String currentWfProfileId) throws Exception {

        if (lastSuccessEndDate < 0) {

            nodeToken.getFullEnv().setAttribute("OperationTypeInfo", "Last success date < 0, operation forced to REFRESH");

            nodeToken.getFullEnv().setAttribute("operationType", "REFRESH");

            return true;

        }

        return false;

    }

    private Long findLastSuccessEndDate(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();

            log.debug("Iterating on the logs");

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

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

                if (curr > res) {

                    res = curr;

                }

            }

        }

        return res;

    }

    private String findCurrentWfProfileId(NodeToken token) throws MSROException {

        log.debug("Start to find the current profile Id");

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

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

            log.debug("The profile Id found is "+p1);

            return p1;

        }

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

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

            log.debug("The profile Id found is "+p2);

            return p2;

        }

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

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

            log.debug("The profile Id found is "+p3);

            return p3;

        }

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

    }

    public String getOperationType() {

        return operationType;

    }

    public void setOperationType(final String operationType) {

        this.operationType = operationType;

    }

}

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

import java.io.StringReader;

import java.util.Map;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.lang.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.Required;

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

import com.google.common.collect.Maps;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.database.rmi.DatabaseException;

import eu.dnetlib.enabling.database.rmi.DatabaseService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.resultset.MappedResultSetFactory;

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

import eu.dnetlib.enabling.resultset.client.utils.EPRUtils;

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

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;

	private String keyTypeFilter;

	@Resource

	private UniqueServiceLocator serviceLocator;

	private MappedResultSetFactory mappedResultSetFactory;

	private ResultSetClientFactory resultSetClientFactory;

	/**

	 * {@inheritDoc}

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final W3CEndpointReference inputEpr = new EPRUtils().getEpr(token.getEnv().getAttribute(inputEprParam));

		final HostedByCounters counters = new HostedByCounters();

		String datasourceId;

		if (StringUtils.isEmpty(overrideDataSourceId)) {

			datasourceId = token.getEnv().getAttribute("parentDatasourceId");

		} else {

			datasourceId = getOverrideDataSourceId();

		}

		Map<String, HostedByEntry> hostedByEntryMap;

		if (!StringUtils.isBlank(keyTypeFilter)) {

			hostedByEntryMap = loadHostedByMapFromType(keyTypeFilter);

		} else {

			hostedByEntryMap = loadHostedByMap(datasourceId);

		}

		final W3CEndpointReference epr = mappedResultSetFactory.createMappedResultSet(inputEpr, new PatchHostedBy(hostedByEntryMap,

				getXpathEntry(), counters));

		token.getEnv().setAttribute(outputEprParam, epr.toString());

		token.getEnv().setTransientAttribute(countersParam, counters);

		return Arc.DEFAULT_ARC;

	}

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

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

		return getHostedByEntryMapByQuery(datasourceId, sql);

	}

	private Map<String, HostedByEntry> loadHostedByMapFromType(final String keyType) throws DocumentException, DatabaseException {

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

		return getHostedByEntryMapByQuery(keyType, sql);

	}

	private Map<String, HostedByEntry> getHostedByEntryMapByQuery(final String datasourceId, final String sql)

			throws DatabaseException, DocumentException {

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

		final W3CEndpointReference epr = serviceLocator.getService(DatabaseService.class).searchSQL(getDbName(),

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

		final SAXReader reader = new SAXReader();

		for (final String s : resultSetClientFactory.getClient(epr)) {

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

	}

	/**

	 * Setter for property 'dbName'.

	 *

	 * @param dbName

	 *            Value to set for property 'dbName'.

	 */

	public void setDbName(final String dbName) {

		this.dbName = dbName;

	}

	public String getKeyTypeFilter() {

		return keyTypeFilter;

	}

	public void setKeyTypeFilter(final String keyTypeFilter) {

		this.keyTypeFilter = keyTypeFilter;

	}

	/**

	 * Getter for property 'mappedResultSetFactory'.

	 *

	 * @return Value for property 'mappedResultSetFactory'.

	 */

	public MappedResultSetFactory getMappedResultSetFactory() {

		return mappedResultSetFactory;

	}

	/**

	 * Setter for property 'mappedResultSetFactory'.

	 *

	 * @param mappedResultSetFactory

	 *            Value to set for property 'mappedResultSetFactory'.

	 */

	@Required

	public void setMappedResultSetFactory(final MappedResultSetFactory mappedResultSetFactory) {

		this.mappedResultSetFactory = mappedResultSetFactory;

	}

	/**

	 * Getter for property 'resultSetClientFactory'.

	 *

	 * @return Value for property 'resultSetClientFactory'.

	 */

	public ResultSetClientFactory getResultSetClientFactory() {

		return resultSetClientFactory;

	}

	/**

	 * Setter for property 'resultSetClientFactory'.

	 *

	 * @param resultSetClientFactory

	 *            Value to set for property 'resultSetClientFactory'.

	 */

	@Required

	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {

		this.resultSetClientFactory = resultSetClientFactory;

	}

	/**

	 * Getter for property 'countersParam'.

	 *

	 * @return Value for property 'countersParam'.

	 */

	public String getCountersParam() {

		return 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 hostedbyMapTable;

	}

	/**

	 * @param hostedbyMapTable

	 *            the hostedbyMapTable to set

	 */

	public void setHostedbyMapTable(final String hostedbyMapTable) {

		this.hostedbyMapTable = hostedbyMapTable;

	}

	public String getXpathEntry() {

		return xpathEntry;

	}

	public void setXpathEntry(final String xpathEntry) {

		this.xpathEntry = xpathEntry;

	}

	/**

	 * @return the overrideDataSourceId

	 */

	public String getOverrideDataSourceId() {

		return overrideDataSourceId;

	}

	/**

	 * @param overrideDataSourceId

	 *            the overrideDataSourceId to set

	 */

	public void setOverrideDataSourceId(final String overrideDataSourceId) {

		this.overrideDataSourceId = overrideDataSourceId;

	}

}

<?xml version="1.0" ?>

<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-openaireplus-workflows</artifactId>

	<packaging>jar</packaging>

	<version>6.3.27</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-openaireplus-workflows/tags/dnet-openaireplus-workflows-6.3.27</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-data-flow-monitoring-core</artifactId>

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

			<exclusions>

				<exclusion>

					<artifactId>wstx-asl</artifactId>

					<groupId>org.codehaus.woodstox</groupId>

				</exclusion>

			</exclusions>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-enabling-database-api</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-enabling-database-service</artifactId>

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

		</dependency>

		<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>cnr-resultset-service</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaire-datasource-manager</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

			<version>[6.2.26]</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-index-solr-common</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-collector-plugins</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-oai-common-workflows</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-mongo-mdstore</artifactId>

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

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-modular-objectstore-service</artifactId>

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

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

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

		</dependency>