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<Claim> claimRels = this.claimDatabaseUtils.query(sql);

		int totalClaims = 0;

		int totalWrites = 0;

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

		for (Claim claim : claimRels) {

			log.debug(claim);

			totalClaims++;

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

			String targetId = fullId(getOpenAIREType(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;

	}

	protected String getOpenAIREType(final String type){

		switch(type){

		case "publication":

		case "dataset":

			return "result";

		default:

			return type;

		}

	}

	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;

	}

	public UniqueServiceLocator getServiceLocator() {

		return serviceLocator;

	}

	public void setServiceLocator(final UniqueServiceLocator serviceLocator) {

		this.serviceLocator = serviceLocator;

	}

	public ClaimDatabaseUtils getClaimDatabaseUtils() {

		return claimDatabaseUtils;

	}

	public void setClaimDatabaseUtils(final ClaimDatabaseUtils claimDatabaseUtils) {

		this.claimDatabaseUtils = claimDatabaseUtils;

	}

	public String getSql() {

		return sql;

	}

	public void setSql(final String sql) {

		this.sql = sql;

	}

}