D-Net

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

import java.io.IOException;

import java.util.List;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;

import eu.dnetlib.data.proto.FieldTypeProtos.DataInfo;

import eu.dnetlib.data.proto.FieldTypeProtos.Qualifier;

import eu.dnetlib.data.proto.KindProtos.Kind;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.OafProtos.OafRel;

import eu.dnetlib.data.proto.RelMetadataProtos.RelMetadata;

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

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

import eu.dnetlib.data.proto.ResultProjectProtos.ResultProject;

import eu.dnetlib.data.proto.ResultProjectProtos.ResultProject.Outcome;

import eu.dnetlib.data.proto.ResultResultProtos.ResultResult;

import eu.dnetlib.data.proto.ResultResultProtos.ResultResult.PublicationDataset;

import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

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

import eu.dnetlib.utils.ontologies.OntologyLoader;

import org.apache.commons.codec.binary.Base64;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.util.StringUtils;

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

/**

 * Created by alessia on 23/10/15.

 */

public class ApplyClaimRelsJobNode extends SimpleJobNode {

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

	private final String SEPARATOR = "_";

	@Autowired

	private UniqueServiceLocator serviceLocator;

	@Autowired

	private ClaimDatabaseUtils claimDatabaseUtils;

	private String sql;

	private String clusterName;

	private String tableName;

	private String fetchSqlAsText(final String path) throws IOException {

		return IOUtils.toString(getClass().getResourceAsStream(path));

	}

	@Override

	protected String execute(NodeToken token) throws Exception {

		//TODO: use claim.claim_date from the claim db

		long timestamp = System.currentTimeMillis();

		List<ClaimRel> claimRels = this.claimDatabaseUtils.query(sql);

		int totalClaims = 0;

		int totalWrites = 0;

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

		for (ClaimRel claim : claimRels) {

			log.debug(claim);

			totalClaims++;

			String sourceId = fullId(claim.getSourceType(), claim.getSource());

			String targetId = fullId(claim.getTargetType(), claim.getTarget());

			String value = getValue(sourceId, targetId, claim.getSemantics(), timestamp);

			/*

			public void addHBaseColumn(final String clusterName,

			final String tableName,

			final String rowKey,

			final String columnFamily,

			final String qualifier,

			final String value)

			 */

			hadoopService.addHBaseColumn(clusterName, tableName, sourceId, claim.getSemantics(), targetId, value);

			totalWrites++;

			String inverseSemantics = OntologyLoader.fetchInverse(claim.getSemantics());

			String inverseValue = getValue(targetId, sourceId, inverseSemantics, timestamp);

			hadoopService.addHBaseColumn(clusterName, tableName, targetId, inverseSemantics, sourceId, inverseValue);

			totalWrites++;

		}

		log.info("totalClaims: " + totalClaims);

		token.getEnv().setAttribute("claimSize", totalClaims);

		log.info("writeOps: " + totalWrites);

		token.getEnv().setAttribute("writeOps", totalWrites);

		return Arc.DEFAULT_ARC;

	}

	private String getValue(final String sourceId, final String semantics, final String targetId, final long timestamp) throws MSROException{

		log.debug(StringUtils.format("%s -- %s -- %s", sourceId, semantics, targetId));

		String[] relInfo = semantics.split(SEPARATOR);

		Qualifier.Builder semanticsBuilder = Qualifier.newBuilder().setClassid(relInfo[2]).setClassname(relInfo[2]);

		Oaf.Builder builder = Oaf.newBuilder().setKind(Kind.relation).setLastupdatetimestamp(timestamp);

		builder.setDataInfo(DataInfo.newBuilder().setTrust("0.91").setInferred(false)

				.setProvenanceaction(

						Qualifier.newBuilder()

								.setClassid("user:claim")

								.setClassname("user:claim")

								.setSchemeid("dnet:provenanceActions")

								.setSchemename("dnet:provenanceActions")

				));

		final SubRelType subRelType = SubRelType.valueOf(relInfo[1]);

		final OafRel.Builder relBuilder = OafRel.newBuilder()

				.setSubRelType(subRelType)

				.setRelClass(relInfo[2])

				.setRelType(RelType.valueOf(relInfo[0]))

				.setSource(sourceId).setTarget(targetId).setChild(false);

		switch (relInfo[0]) {

		case "resultProject":

			relBuilder.setResultProject(ResultProject.newBuilder()

					.setOutcome(Outcome.newBuilder().setRelMetadata(

							RelMetadata.newBuilder().setSemantics(

									semanticsBuilder

											.setSchemeid("dnet:result_project_relations")

											.setSchemename("dnet:result_project_relations")

											.build()

							))));

			break;

		case "resultResult_publicationDataset_isRelatedTo":

			relBuilder.setResultResult(ResultResult.newBuilder()

					.setPublicationDataset(PublicationDataset.newBuilder().setRelMetadata(

							RelMetadata.newBuilder().setSemantics(

									semanticsBuilder

											.setSchemeid("dnet:result_result_relations")

											.setSchemename("dnet:result_result_relations")

											.build()

							))));

			break;

		default:

			throw new MSROException("Semantics "+relInfo[0]+" not supported");

		}

		builder.setRel(relBuilder);

		return Base64.encodeBase64String(builder.build().toByteArray());

	}

	private String fullId(final String type, final String id) {

		final String fullId = AbstractDNetXsltFunctions.oafSimpleId(type, id);

		return OafRowKeyDecoder.decode(fullId).getKey();

	}

	public String getClusterName() {

		return clusterName;

	}

	public void setClusterName(final String clusterName) {

		this.clusterName = clusterName;

	}

	public String getTableName() {

		return tableName;

	}

	public void setTableName(final String tableName) {

		this.tableName = tableName;

	}

}

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

/**

 * Created by Alessia Bardi on 23/06/2017.

 *

 * @author Alessia Bardi

 */

public class ClaimRel {

	private String source, sourceType, target, targetType, semantics;

	public String getSource() {

		return source;

	}

	public ClaimRel setSource(final String source) {

		this.source = source;

		return this;

	}

	public String getSourceType() {

		return sourceType;

	}

	public ClaimRel setSourceType(final String sourceType) {

		this.sourceType = sourceType;

		return this;

	}

	public String getTarget() {

		return target;

	}

	public ClaimRel setTarget(final String target) {

		this.target = target;

		return this;

	}

	public String getTargetType() {

		return targetType;

	}

	public ClaimRel setTargetType(final String targetType) {

		this.targetType = targetType;

		return this;

	}

	public String getSemantics() {

		return semantics;

	}

	public ClaimRel setSemantics(final String semantics) {

		this.semantics = semantics;

		return this;

	}

	@Override

	public String toString() {

		return "ClaimRel{" +

				"source='" + source + '\'' +

				", sourceType='" + sourceType + '\'' +

				", target='" + target + '\'' +

				", targetType='" + targetType + '\'' +

				", semantics='" + semantics + '\'' +

				'}';

	}

}

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

import java.util.List;

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

import org.springframework.jdbc.core.JdbcTemplate;

import org.springframework.stereotype.Component;

/**

 * Created by Alessia Bardi on 23/06/2017.

 *

 * @author Alessia Bardi

 */

@Component

public class ClaimDatabaseUtils {

	@Autowired

	JdbcTemplate claimsJdbcTemplate;

	public List<ClaimRel> query(String sqlQuery){

		return this.claimsJdbcTemplate.query(sqlQuery,

				(rs, rowNum) -> new ClaimRel().setSemantics(rs.getString("semantics"))

						.setSource(rs.getString("source_id")).setSourceType(rs.getString("source_type"))

						.setTarget(rs.getString("target_id")).setTargetType(rs.getString("target_type")));

	}

}

package eu.dnetlib.utils.ontologies;

import javax.annotation.PostConstruct;

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

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import org.springframework.stereotype.Component;

/**

 * Created by claudio on 12/12/2016.

 */

@Component

public class OntologyLoader {

	private static UniqueServiceLocator staticServiceLocator;

	@Autowired

	private UniqueServiceLocator serviceLocator;

// <TERM code="merges" encoding="organizationOrganization_dedup_merges" english_name="merges" native_name="merges">

	public static String fetchInverse(final String relType) throws ISLookUpException {

		final String xquery = "let $x:= /RESOURCE_PROFILE["

				+ " .//RESOURCE_TYPE/@value = 'OntologyDSResourceType' and "

				+ " .//TERM/@encoding='"+relType+"']"

				+ "let $y:= $x//TERM[./@encoding='"+relType+"']//RELATION[./@type='inverseOf']/@code/string() "

				+ "return $x//TERM[./@code = $y]/@encoding/string()";

		return staticServiceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xquery);

	}

	@PostConstruct

	public void init() {

		OntologyLoader.staticServiceLocator = serviceLocator;

	}

}

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="9bd3c6d5-4be1-4209-967c-02371827b119_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl"/>

        <RESOURCE_TYPE value="WorkflowDSResourceType"/>

        <RESOURCE_KIND value="WorkflowDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2006-05-04T18:13:51.0Z"/>

    </HEADER>

    <BODY>

        <WORKFLOW_NAME>[TMP] Claim to HBase -- merge the nodes in db2hbase</WORKFLOW_NAME>

        <WORKFLOW_TYPE>Data Load</WORKFLOW_TYPE>

        <WORKFLOW_PRIORITY>30</WORKFLOW_PRIORITY>

        <CONFIGURATION start="manual">

            <NODE isStart="true" name="start">

                <DESCRIPTION>start</DESCRIPTION>

                <PARAMETERS/>

                <ARCS>

                    <ARC to="queryClaimRels"/>

                </ARCS>

            </NODE>

            <NODE name="applyClaimRels" type="ApplyClaimRels">

                <DESCRIPTION>Apply Claim Rels</DESCRIPTION>

                <PARAMETERS>

                    <PARAM managedBy="system" name="clusterName" required="true" type="string">DM</PARAM>

                    <PARAM managedBy="system" name="tableName" required="true" type="string">db_openaireplus_services_beta</PARAM>

                    <PARAM managedBy="system" name="sql" required="true" type="string">/eu/dnetlib/msro/openaireplus/workflows/hbase/queryClaimsRel.sql</PARAM>

                </PARAMETERS>

                <ARCS>

                    <ARC to="applyClaimUpdates"/>

                </ARCS>

            </NODE>

            <NODE name="applyClaimUpdates" type="ApplyClaimUpdates">

                <DESCRIPTION>query Claim updates</DESCRIPTION>

                <PARAMETERS>

	                <PARAM managedBy="system" name="clusterName" required="true" type="string">DM</PARAM>

	                <PARAM managedBy="system" name="tableName" required="true" type="string">db_openaireplus_services_beta</PARAM>

                    <PARAM managedBy="system" name="sql" required="true" type="string">/eu/dnetlib/msro/openaireplus/workflows/hbase/queryClaimsUpdate.sql</PARAM>

                </PARAMETERS>

                <ARCS>

                    <ARC to="success"/>

                </ARCS>

            </NODE>

        </CONFIGURATION>

        <STATUS>

        </STATUS>

    </BODY>

</RESOURCE_PROFILE>

--- SELECT xml, provenance FROM claims WHERE type = 'rels2actions' and set = 'userclaim_result_project'	or set = 'userclaim_result_result'

SELECT source_type, source_id, target_type, target_id, semantics from claim where approved=true and source_type !='context';

-- SELECT

-- 	regexp_replace(xml, '<\?xml version="1\.0" encoding="UTF-8"\?>', '', 'i') AS xml,

-- 	provenance

-- FROM claims

-- WHERE type = 'updates2actions' AND set = 'userclaim_dmf'

SELECT source_type, source_id, target_type, target_id, semantics from claim where approved=true and source_type ='context';

dnet.openaire.claims.db.url=jdbc:postgresql://localhost:5432/dnet_openaireplus

dnet.openaire.claims.db.username=dnet

dnet.openaire.claims.db.password=dnetPwd

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

<beans xmlns="http://www.springframework.org/schema/beans"

	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"

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

	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd

		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd

		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">

	<bean id="claimsJdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate"

		p:dataSource-ref="claimsDatasource"/>

	<bean id="claimsDatasource" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">

		<property name="driverClassName" value="org.postgresql.Driver"/>

		<property name="url" value="${dnet.openaire.claims.db.url}"/>

		<property name="username" value="${dnet.openaire.claims.db.username}"/>

		<property name="password" value="${dnet.openaire.claims.db.password}"/>

	</bean>

</beans>

    <bean id="wfNodeApplyClaimRels"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.claims.ApplyClaimRelsJobNode"

          scope="prototype"/>

        <dependency>

            <groupId>org.springframework</groupId>

            <artifactId>spring-jdbc</artifactId>

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

        </dependency>