D-Net

package eu.dnetlib.msro.workflows.nodes;

import com.googlecode.protobuf.format.JsonFormat;

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

import eu.dnetlib.dli.resolver.PIDResolver;

import eu.dnetlib.dli.resolver.model.*;

import eu.dnetlib.enabling.resultset.ResultSetInfo;

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

import eu.dnetlib.msro.workflows.graph.Arc;

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.msro.workflows.procs.Token;

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

import eu.dnetlib.resolver.parser.DMFResolverParser;

import eu.dnetlib.rmi.common.ResultSet;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.http.converter.StringHttpMessageConverter;

import org.springframework.web.client.RestTemplate;

import java.nio.charset.Charset;

import java.time.LocalDateTime;

import java.util.Arrays;

import java.util.List;

import static eu.dnetlib.data.proto.dli.ScholixObjectProtos.*;

public class ResolveAndIndexJobNode extends SimpleJobNode implements ProgressProvider {

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

    private static final String BASE_CFG_URL = "http://%s:9200/%s/scholix/%s/?pretty";

    private String inputEprParam;

    private String indexHost;

    private String indexName;

    private int counter;

    private int total;

    @Autowired

    private List<PIDResolver> pluginResolver;

    @Autowired

    private ResultSetClient resultSetClient;

    @Override

    protected String execute(Env env) throws Exception {

        final ResultSet<?> rsIn = env.getAttribute(this.inputEprParam, ResultSet.class);

        final Iterable<String> records = resultSetClient.iter(rsIn, String.class);

        ResultSetInfo info = resultSetClient.info(rsIn);

        this.total = info.getTotal();

        final RestTemplate restTemplate = new RestTemplate();

        restTemplate.getMessageConverters()

                .add(0, new StringHttpMessageConverter(Charset.forName("UTF-8")));

        setIndexHost(indexHost);

        setIndexName(indexName);

        DMFResolverParser parser = new DMFResolverParser();

        for (String record : records) {

            this.counter++;

            final ResolvedObject result = parser.parseObject(record);

            if (result == null) {

                log.error("error on parsing " + record);

                continue;

            }

            for (final ObjectRelation rels : result.getRelations()) {

                final ResolvedObject resolvedRelation = resolveRelation(rels.getTargetPID(), result.getDatasourceProvenance().get(0));

                final Scholix.Builder scholix = Scholix.newBuilder();

                scholix.addLinkproviderBuilder()

                        .setName(result.getDatasourceProvenance().get(0).getDatasource())

                        .addIdentifiersBuilder()

                        .setIdentifier(result.getDatasourceProvenance().get(0).getDatasourceId())

                        .setSchema("dnetIdentifier");

                scholix.setRelationship(ScholixRelationship.newBuilder()

                        .setName(rels.getRelationSemantics())

                        .setInverse(rels.getInverseRelation())

                        .setSchema("datacite")

                        .build());

                final ScholixResource source = generateResource(result);

                final ScholixResource target = generateResource(resolvedRelation);

                scholix.setSource(source);

                scholix.setTarget(target);

                scholix.setPublicationDate(LocalDateTime.now().toString());

                restTemplate.postForLocation(String.format(BASE_CFG_URL, indexHost, indexName, generateIdentifier(result, resolvedRelation)), JsonFormat.printToString(scholix.build()));

                scholix.setRelationship(ScholixRelationship.newBuilder()

                        .setInverse(rels.getRelationSemantics())

                        .setName(rels.getInverseRelation())

                        .setSchema("datacite")

                        .build());

                scholix.setTarget(source);

                scholix.setSource(target);

                restTemplate.postForLocation(String.format(BASE_CFG_URL, indexHost, indexName, generateIdentifier(resolvedRelation, result)), JsonFormat.printToString(scholix.build()));

            }

        }

        return Arc.DEFAULT_ARC;

    }

    @Override

    protected void beforeStart(final Token token) {

        token.setProgressProvider(this);

    }

    private String generateIdentifier(final String source, final String target) {

        return AbstractDNetXsltFunctions.md5(String.format("%s::%s", source.toLowerCase().trim(), target.toLowerCase().trim()));

    }

    private String generateIdentifier(final ResolvedObject source, ResolvedObject target) {

        return AbstractDNetXsltFunctions.md5(String.format("%s::%s", source.getPid().toLowerCase().trim(), target.getPid().toLowerCase().trim()));

    }

    private ScholixResource generateResource(ResolvedObject result) {

        final ScholixResource.Builder builder = ScholixResource.newBuilder();

        if (result.getDatasourceProvenance() != null)

            result.getDatasourceProvenance().forEach(

                    objectProvenance -> {

                        builder.addCollectedFrom(ScholixCollectedFrom.newBuilder()

                                .setProvisionMode(objectProvenance.getProvisionMode())

                                .setCompletionStatus(objectProvenance.getCompletionStatus())

                                .setProvider(ScholixEntityId.newBuilder()

                                        .setName(objectProvenance.getDatasource())

                                        .addIdentifiers(ScholixIdentifier.newBuilder().setIdentifier(objectProvenance.getDatasourceId())

                                                .setSchema("dnetIdentifier").build())

                                        .build()));

                        if (StringUtils.isNotEmpty(objectProvenance.getPublisher())) {

                            builder.addPublisher(ScholixEntityId.newBuilder()

                                    .setName(objectProvenance.getPublisher())

                                    .build());

                        }

                    });

        builder.addIdentifier(ScholixIdentifier.newBuilder().

                setIdentifier(result.getPid())

                .setSchema(result.getPidType())

                .build());

        builder.setObjectType(result.getType().toString());

        if (result.getTitles() != null && result.getTitles().size() > 0)

            builder.setTitle(result.getTitles().get(0));

        if (result.getAuthors() != null)

            result.getAuthors().forEach(author -> builder.addCreator(

                    ScholixEntityId.newBuilder()

                            .setName(author)

                            .build()));

        if (StringUtils.isNotBlank(result.getDate())) {

            builder.setPublicationDate(result.getDate());

        }

        String tp = null;

        switch (result.getType()) {

            case dataset:

                tp = "60";

                break;

            case unknown:

                tp = "70";

                break;

            case publication:

                tp = "50";

                break;

        }

        builder.setDnetIdentifier(tp + "|dnet________::" + result.getIdentifier());

        return builder.build();

    }

    private ResolvedObject resolveRelation(final PID currentPid, final ObjectProvenance provenance) {

        for (PIDResolver resolver : pluginResolver) {

            final ResolvedObject currentIdentifier = resolver.retrievePID(currentPid.getId(), currentPid.getType());

            if (currentIdentifier != null &&

                    !StringUtils.isBlank(currentIdentifier.getPid()) &&

                    currentIdentifier.getPid().toLowerCase().equals(currentPid.getId().toLowerCase())) {

                return currentIdentifier;

            }

        }

        final ResolvedObject resolvedObject = new ResolvedObject();

        resolvedObject.setPid(currentPid.getId());

        resolvedObject.setPidType(currentPid.getType());

        ObjectProvenance resultProvenance = new ObjectProvenance();

        resultProvenance.setDatasource(provenance.getDatasource());

        resultProvenance.setDatasourceId(provenance.getDatasourceId());

        resultProvenance.setCompletionStatus(CompletionStatus.incomplete.toString());

        resultProvenance.setProvisionMode(ObjectProvisionMode.collected.toString());

        resolvedObject.setDatasourceProvenance(Arrays.asList(resultProvenance));

        return resolvedObject;

    }

    public String getInputEprParam() {

        return inputEprParam;

    }

    public void setInputEprParam(String inputEprParam) {

        this.inputEprParam = inputEprParam;

    }

    public String getIndexHost() {

        return indexHost;

    }

    public void setIndexHost(String indexHost) {

        this.indexHost = indexHost;

    }

    public String getIndexName() {

        return indexName;

    }

    public void setIndexName(String indexName) {

        this.indexName = indexName;

    }

    @Override

    public String getProgressDescription() {

        return this.counter < 0 ? "-" : String.format("%d / %d", this.counter, this.total);

    }

}