D-Net

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

import java.util.concurrent.atomic.AtomicInteger;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

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.msro.rmi.MSROException;

import eu.dnetlib.utils.ontologies.OntologyLoader;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.util.StringUtils;

/**

 * Created by alessia on 23/10/15.

 */

public class ApplyClaimRelsJobNode extends AbstractClaimsToHBASE {

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

	private final String SEPARATOR = "_";

	@Override

	protected String execute(NodeToken token) throws Exception {

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

		long timestamp = System.currentTimeMillis();

		setTotal(getClaimDatabaseUtils().count(getCountQuery()));

		final AtomicInteger totalWrites = new AtomicInteger(0);

		final AtomicInteger discardedClaims = new AtomicInteger(0);

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

		getClaimDatabaseUtils().query(getSql()).forEach(claim -> {

			log.debug(claim);

			try {

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

				String targetId = getFullId(getOpenAIREType(claim.getTargetType()), claim.getTarget());

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

				// adding relationship

				hadoopService.addHBaseColumn(getClusterName(), getTableName(), sourceId, claim.getSemantics(), targetId, value);

				totalWrites.incrementAndGet();

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

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

				//adding inverse relationship

				hadoopService.addHBaseColumn(getClusterName(), getTableName(), targetId, inverseSemantics, sourceId, inverseValue);

				totalWrites.incrementAndGet();

				incrementProcessed();

			} catch (Exception e) {

				log.error("Discarding claim " + claim + ". Cause: " + e.getMessage());

				discardedClaims.incrementAndGet();

			}

		});

		log.info("totalClaimRels: " + getTotal());

		token.getEnv().setAttribute("claimRelsSize", getTotal());

		log.info("claim rels writeOps: " + totalWrites.intValue());

		token.getEnv().setAttribute("claimRelsWriteOps", totalWrites.intValue());

		log.info("discardedClaimRels: " + discardedClaims.intValue());

		token.getEnv().setAttribute("discardedClaimRels", discardedClaims.intValue());

		return Arc.DEFAULT_ARC;

	}

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

		if (relInfo.length < 3) {

			throw new MSROException("Semantics " + semantics + " not supported: must be splittable in 3 by '_'");

		}

		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":

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

	}

}