D-Net

package eu.dnetlib.resolver.mdstore.plugin;

import java.util.List;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.Callable;

import com.mongodb.DBObject;

import com.mongodb.client.MongoCollection;

import eu.dnetlib.resolver.PIDResolver;

import eu.dnetlib.resolver.model.CompletionStatus;

import eu.dnetlib.resolver.model.ObjectRelation;

import eu.dnetlib.resolver.model.PID;

import eu.dnetlib.resolver.model.ResolvedObject;

import eu.dnetlib.resolver.model.serializer.ResolverSerializer;

import eu.dnetlib.resolver.parser.DLIParser;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * Created by sandro on 9/22/16.

 */

public class RecordResolver implements Callable<Boolean> {

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

	private final DLIParser parser = new DLIParser();

	private List<PIDResolver> pluginResolver;

	private BlockingQueue<DBObject> inputQueue;

	private MongoCollection<DBObject> outputCollection;

	private ResolverSerializer serializer;

	public String resolve(final String inputRecord) {

		final ResolvedObject inputObject = parser.parse(inputRecord);

		boolean shouldUpdate = false;

		if (inputObject.getCompletionStatus() == null || !inputObject.getCompletionStatus().equals(CompletionStatus.complete.toString())) {

			shouldUpdate = shouldUpdate || tryToResolveRecord(inputObject);

		}

		if (inputObject.getRelations() != null) {

			for (ObjectRelation rel : inputObject.getRelations()) {

				final String dnetId = tryToResolveRelation(rel.getTargetPID());

				if (dnetId != null && !StringUtils.isEmpty(dnetId)) {

					rel.setTargetPID(new PID(dnetId, "dnet"));

					shouldUpdate = true;

				}

			}

			if (shouldUpdate) {

				final String newXML = serializer.serializeReplacingXML(inputRecord, inputObject);

				return newXML;

			}

		}

		return null;

	}

	public void setTemplate(final StringTemplate template) {

		serializer.setDmfTemplate(template);

	}

	private String tryToResolveRelation(final PID currentPid) {

		if (currentPid.getId() != null && currentPid.getId().contains("dli_resolver")) {

			return null;

		}

		for (PIDResolver resolver : pluginResolver) {

			final String currentIdentifier = resolver.retrieveDnetID(currentPid.getId(), currentPid.getType());

			if (currentIdentifier != null && !StringUtils.isBlank(currentIdentifier)) {

				return currentIdentifier;

			}

		}

		return null;

	}

	private boolean tryToResolveRecord(final ResolvedObject object) {

		for (PIDResolver resolver : pluginResolver) {

			final ResolvedObject resolvedObject = resolver.retrievePID(object.getPid(), object.getPidType());

			if (resolvedObject != null &&

					resolvedObject.getCompletionStatus() != null &&

					resolvedObject.getCompletionStatus().toString().equals(CompletionStatus.complete.toString())) {

				{

					object.setAuthors(resolvedObject.getAuthors());

					object.setTitles(resolvedObject.getTitles());

					object.setCompletionStatus(resolvedObject.getCompletionStatus());

					object.setDate(resolvedObject.getDate());

					object.getDatasourceProvenance().addAll(resolvedObject.getDatasourceProvenance());

					object.setDescription(resolvedObject.getDescription());

					object.setSubjects(resolvedObject.getSubjects());

					object.setType(resolvedObject.getType());

					return true;

				}

			}

		}

		return false;

	}

	public void setPluginResolver(final List<PIDResolver> pluginResolver) {

		this.pluginResolver = pluginResolver;

	}

	@Override

	public Boolean call() throws Exception {

		DBObject currentObject = inputQueue.take();

		int i = 0;

		while (currentObject != ResolverMDStorePlugin.DONE) {

			try {

				final String currentRecord = (String) currentObject.get("body");

				final String resolvedRecord = resolve(currentRecord);

				if (resolvedRecord != null) {

					currentObject.put("body", resolvedRecord);

					outputCollection.insertOne(currentObject);

				}

				currentObject = inputQueue.take();

				if (i++ % 100 == 0)

					log.debug(Thread.currentThread().getId() + " total object resolved: " + i);

			} catch (Throwable e) {

				log.error("Error on resolving objects ", e);

				return false;

			}

		}

		if (currentObject == ResolverMDStorePlugin.DONE) {

			inputQueue.put(currentObject);

		}

		return true;

	}

	public BlockingQueue<DBObject> getInputQueue() {

		return inputQueue;

	}

	public void setInputQueue(final BlockingQueue<DBObject> inputQueue) {

		this.inputQueue = inputQueue;

	}

	public MongoCollection<DBObject> getOutputCollection() {

		return outputCollection;

	}

	public void setOutputCollection(final MongoCollection<DBObject> outputCollection) {

		this.outputCollection = outputCollection;

	}

	public void setSerializer(final ResolverSerializer serializer) {

		final ResolverSerializer tmp = new ResolverSerializer();

		tmp.setPmfTemplate(new StringTemplate(serializer.getPmfTemplate().getTemplate()));

		tmp.setDmfTemplate(new StringTemplate(serializer.getDmfTemplate().getTemplate()));

		tmp.setScholixTemplate(new StringTemplate(serializer.getScholixTemplate().getTemplate()));

		this.serializer = tmp;

	}

}