D-Net

package eu.dnetlib.data.mapreduce.hbase.dataimport;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import java.util.Objects;

import com.google.gson.JsonArray;

import com.google.gson.JsonElement;

import com.google.gson.JsonObject;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.actionmanager.actions.ActionFactory;

import eu.dnetlib.actionmanager.actions.AtomicAction;

import eu.dnetlib.actionmanager.common.Agent;

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

import eu.dnetlib.data.proto.RelMetadataProtos;

import eu.dnetlib.data.proto.RelTypeProtos;

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

import org.apache.commons.lang3.StringUtils;

import static eu.dnetlib.data.proto.FieldTypeProtos.*;

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

import static eu.dnetlib.data.proto.OafProtos.*;

import static eu.dnetlib.data.proto.ResultProtos.Result;

import static eu.dnetlib.data.proto.ResultProtos.Result.*;

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

import static eu.dnetlib.data.proto.TypeProtos.Type;

public class ScholixToActions {

    public static List<AtomicAction> generateActionsFromScholix(final JsonObject rootElement, final Map<String, ScholExplorerConfiguration> conf,

                                                                final String setName, final Agent agent, ActionFactory factory, String nsPrefix, final String dsName,

                                                                final String dsId, String dateOfCollection) {

        final List<AtomicAction> actions = new ArrayList<>();

        final String typology = getStringValue(rootElement, "typology");

        final List<String> publisher = getArrayValues(rootElement, "publisher");

        final String abstractValue = getStringValue(rootElement, "abstract");

        final List<String> authors = getArrayValues(rootElement, "author");

        final List<String> dates = getArrayValues(rootElement, "date");

        final JsonArray localIdentifiers = rootElement.getAsJsonArray("localIdentifier");

        final String dnetId = getStringValue(rootElement, "id").substring(17);

        String title = "";

        if (rootElement.has("title") && rootElement.get("title").isJsonArray()) {

            StringBuilder ttl = new StringBuilder();

            getArrayValues(rootElement, "title").forEach(ttl::append);

            title = ttl.toString();

        } else {

            title = getStringValue(rootElement, "title");

        }

        if (title != null && title.charAt(0) == '"' && title.charAt(title.length() - 1) == '"') {

            title = title.substring(1, title.length() - 1);

        }

        final Oaf.Builder oafBuilder = Oaf.newBuilder();

        final boolean isVisible = StringUtils.isNotBlank(title) && StreamUtils.toStream(localIdentifiers.iterator())

                .map(JsonElement::getAsJsonObject)

                .anyMatch(o -> {

                    final String type = getStringValue(o, "type");

                    return StringUtils.isNotBlank(type) && conf.containsKey(type) && conf.get(type).isVisible();

                });

        oafBuilder.setDataInfo(

                DataInfo.newBuilder()

                        .setInvisible(!isVisible)

                        .setDeletedbyinference(false)

                        .setInferred(false)

                        .setTrust("0.9")

                        .setProvenanceaction(getQualifier("sysimport:actionset", "dnet:provenanceActions"))

                        .build());

        oafBuilder.setKind(Kind.entity);

        final String sourceId = String.format("50|%s::%s", nsPrefix, dnetId);

        final KeyValue collectedFrom = KeyValue.newBuilder()

                .setValue(dsName)

                .setKey("10|openaire____::" + AbstractDNetXsltFunctions.md5(dsId))

                .build();

        final OafEntity.Builder oafEntityBuilder = OafEntity.newBuilder()

                .setType(Type.result)

                .setDateofcollection(dateOfCollection)

                .addCollectedfrom(collectedFrom)

                .setId(sourceId);

        StreamUtils.toStream(localIdentifiers.iterator())

                .map(JsonElement::getAsJsonObject)

                .map(localIdentifier -> getPid(localIdentifier, conf))

                .filter(Objects::nonNull)

                .forEach(oafEntityBuilder::addPid);

        final Result.Builder result = Result.newBuilder();

        final Metadata.Builder metadata = Metadata.newBuilder()

            .setResulttype(getQualifier(typology, "dnet:result_typologies"))

            .setLanguage(Qualifier.newBuilder()

                .setClassid("und")

                .setClassname("Undetermined")

                .setSchemeid("dent:languages")

                .setSchemename("dent:languages")

                .build());

        if (StringUtils.isNotBlank(title)) {

            metadata.addTitle(StructuredProperty.newBuilder()

                    .setValue(title)

                    .setQualifier(getQualifier("main title", "dnet:dataCite_title"))

                    .build());

        }

        if (publisher.size() > 0)

            metadata.setPublisher(StringField.newBuilder().setValue(publisher.get(0)).build());

        if (StringUtils.isNotEmpty(abstractValue)) {

            metadata.addDescription(StringField.newBuilder().setValue(abstractValue).build());

        }

        dates.stream().map(it -> {

           if (it.length() == 4) {

               return it+"-01-01";

           }

           else if (it.length() > 10) {

               return it.substring(0,10);

           }

           else

               return it;

        }).forEach(it -> metadata.addRelevantdate(StructuredProperty.newBuilder()

                .setValue(it)

                .setQualifier(getQualifier("dnet:date", "dnet:date"))

                .build()));

        if (rootElement.has("subject")) {

            JsonArray subject = rootElement.getAsJsonArray("subject");

            subject.forEach(it -> {

                    final JsonObject item = it.getAsJsonObject();

                    final String scheme = getStringValue(item, "scheme");

                    metadata.addSubject(StructuredProperty.newBuilder()

                            .setValue(Objects.requireNonNull(getStringValue(item, "value")))

                            .setQualifier(getQualifier(scheme, "dnet:subject"))

                            .build());

                }

            );

        }

        int i = 1;

        for (String it : authors) {

            metadata.addAuthor(Author.newBuilder()

                    .setFullname(it)

                    .setRank(i++)

                    .build());

        }

        result.setMetadata(metadata.build());

        localIdentifiers.forEach(it -> {

            final JsonObject localIdentifier = it.getAsJsonObject();

            final String pidType = getStringValue(localIdentifier, "type");

            final ScholExplorerConfiguration currentConf = conf.get(pidType);

            if (currentConf.getGeneratedUrl() != null) {

                final Instance.Builder instance = Instance.newBuilder();

                final String pidValue = getStringValue(localIdentifier, "id");

                instance.addUrl(String.format(currentConf.getGeneratedUrl(), pidValue));

                instance.setAccessright(Qualifier.newBuilder()

                        .setClassid("UNKNOWN")

                        .setClassname("not available")

                        .setSchemeid("dnet:access_modes")

                        .setSchemename("dnet:access_modes")

                        .build());

                instance.setInstancetype(Qualifier.newBuilder()

                        .setClassid("0000")

                        .setClassname("Unknown")

                        .setSchemeid("dnet:publication_resource")

                        .setSchemename("dnet:publication_resource")

                        .build());

                instance.setHostedby(KeyValue.newBuilder()

                        .setKey("10|openaire____::55045bd2a65019fd8e6741a755395c8c")

                        .setValue("Unknown Repository")

                        .build());

                instance.setCollectedfrom(collectedFrom);

                result.addInstance(instance);

            }

        });

        generateExternalReference(extractRelations(rootElement, "externalRels"))

                .forEach(result::addExternalReference);

        oafEntityBuilder.setResult(result.build());

        oafBuilder.setEntity(oafEntityBuilder.build());

        System.out.println(JsonFormat.printToString(oafBuilder.build()));

        actions.add(factory.createAtomicAction(setName, agent, oafEntityBuilder.getId(), "result", "body", oafBuilder.build().toByteArray()));

        final List<JsonObject> doiRels = extractRelations(rootElement, "doiRels");

        doiRels.stream().map(it -> convertDoiRelations(it, factory, sourceId, nsPrefix, collectedFrom, setName, agent)).forEach(actions::addAll);

        return actions;

    }

    private static AtomicAction createResultResultRelation(final String source, final String target,

                                                           final KeyValue collectedFrom, final ResultResult resultResultRel, final String relClass, final String cfRelation, final ActionFactory factory, final String setName, final Agent agent) {

        final Oaf.Builder oaf = Oaf.newBuilder();

        oaf.setDataInfo(

                DataInfo.newBuilder()

                        .setDeletedbyinference(false)

                        .setInferred(false)

                        .setTrust("0.9")

                        .setProvenanceaction(getQualifier("sysimport:actionset", "dnet:provenanceActions"))

                        .build());

        oaf.setKind(Kind.relation);

        final OafRel.Builder relation = OafRel.newBuilder();

        relation.setSource(source);

        relation.setTarget(target);

        relation.setRelType(RelTypeProtos.RelType.resultResult);

        relation.setSubRelType(RelTypeProtos.SubRelType.publicationDataset);

        relation.setChild(false);

        relation.setResultResult(resultResultRel);

        relation.setRelClass(relClass);

        relation.addCollectedfrom(collectedFrom);

        oaf.setRel(relation.build());

        //System.out.println(JsonFormat.printToString(oaf.build()));

        return factory.createAtomicAction(setName, agent, source, cfRelation, target, oaf.build().toByteArray());

    }

    private static List<AtomicAction> convertDoiRelations(final JsonObject doiRel, final ActionFactory factory, final String sourceId, final String nsPrefix, final KeyValue collectedFrom, final String setName, final Agent agent) {

        final String target = Objects.requireNonNull(getStringValue(doiRel, "dnetId")).substring(17);

        final String targetId = String.format("50|%s::%s", nsPrefix, target);

        final String relationSemantic = getStringValue(doiRel, "relationSemantic");

        String cfRelation;

        String cfInverseRelation;

        ResultResult.Builder resultRel = ResultResult.newBuilder();

        ResultResult.Builder resultInverseRel = ResultResult.newBuilder();

        String relClass;

        String inverseRelClass;

        switch (relationSemantic) {

            case "isSupplementedBy": {

                cfRelation = "resultResult_supplement_isSupplementedBy";

                cfInverseRelation = "resultResult_supplement_isSupplementTo";

                relClass = ResultResult.Supplement.RelName.isSupplementedBy.toString();

                inverseRelClass = ResultResult.Supplement.RelName.isSupplementTo.toString();

                resultRel.setSupplement(ResultResult.Supplement.newBuilder()

                        .setRelMetadata(RelMetadataProtos.RelMetadata.newBuilder()

                                .setSemantics(getQualifier(relClass, "dnet:result_result_relations"))

                                .build())

                        .build());

                resultInverseRel.setSupplement(ResultResult.Supplement.newBuilder()

                        .setRelMetadata(RelMetadataProtos.RelMetadata.newBuilder()

                                .setSemantics(getQualifier(inverseRelClass, "dnet:result_result_relations"))

                                .build())

                        .build());

                break;

            }

            case "isSupplementTo": {

                cfRelation = "resultResult_supplement_isSupplementTo";

                cfInverseRelation = "resultResult_supplement_isSupplementedBy";

                inverseRelClass = ResultResult.Supplement.RelName.isSupplementedBy.toString();

                relClass = ResultResult.Supplement.RelName.isSupplementTo.toString();

                resultInverseRel.setSupplement(ResultResult.Supplement.newBuilder()

                        .setRelMetadata(RelMetadataProtos.RelMetadata.newBuilder()

                                .setSemantics(getQualifier(inverseRelClass, "dnet:result_result_relations"))

                                .build())

                        .build());

                resultRel.setSupplement(ResultResult.Supplement.newBuilder()

                        .setRelMetadata(RelMetadataProtos.RelMetadata.newBuilder()

                                .setSemantics(getQualifier(relClass, "dnet:result_result_relations"))

                                .build())

                        .build());

                break;

            }

            default: {

                cfRelation = "resultResult_publicationDataset_isRelatedTo";

                cfInverseRelation = "resultResult_publicationDataset_isRelatedTo";

                relClass = ResultResult.PublicationDataset.RelName.isRelatedTo.toString();

                inverseRelClass = relClass;

                resultInverseRel.setPublicationDataset(ResultResult.PublicationDataset.newBuilder()

                        .setRelMetadata(RelMetadataProtos.RelMetadata.newBuilder()

                                .setSemantics(getQualifier(relClass, "dnet:result_result_relations"))

                                .build())

                        .build());

                resultRel = resultInverseRel;

            }

        }

        final List<AtomicAction> actions = new ArrayList<>();

        actions.add(createResultResultRelation(sourceId, targetId, collectedFrom, resultRel.build(), relClass, cfRelation, factory, setName, agent));

        actions.add(createResultResultRelation(targetId, sourceId, collectedFrom, resultInverseRel.build(), inverseRelClass, cfInverseRelation, factory, setName, agent));

        return actions;

    }

    private static List<ExternalReference> generateExternalReference(final List<JsonObject> jsonRels) {

        final List<ExternalReference> result = new ArrayList<>();

        jsonRels.forEach(it -> {

            ExternalReference.Builder builder = ExternalReference.newBuilder();

            if("url".equals(getStringValue(it.getAsJsonObject("id"), "schema"))) {

                builder.setUrl(Objects.requireNonNull(getStringValue(it.getAsJsonObject("id"), "identifier")));

            }

            result.add(builder

                    .setRefidentifier(Objects.requireNonNull(getStringValue(it.getAsJsonObject("id"), "identifier")))

                    .setSitename(Objects.requireNonNull(getStringValue(it, "collectedFrom")))

                    .setQualifier(Qualifier.newBuilder()

                            .setClassid(Objects.requireNonNull(getStringValue(it.getAsJsonObject("id"), "schema")))

                            .setClassname(Objects.requireNonNull(getStringValue(it.getAsJsonObject("id"), "schema")))

                            .setSchemename("dnet:externalReference_typologies")

                            .setSchemeid("dnet:externalReference_typologies")

                            .build())

                    .build());

        });

        return result;

    }

    private static List<JsonObject> extractRelations(final JsonObject rootElement, final String fieldType) {

        final List<JsonObject> result = new ArrayList<>();

        if (rootElement.has(fieldType) && rootElement.get(fieldType).isJsonArray()) {

            final JsonArray asJsonArray = rootElement.getAsJsonArray(fieldType);

            asJsonArray.forEach(it -> result.add(it.getAsJsonObject()));

        }

        return result;

    }

    public static Qualifier getQualifier(final String classValue, final String schemeValue) {

        return Qualifier.newBuilder()

                .setSchemeid(schemeValue)

                .setSchemename(schemeValue)

                .setClassname(classValue)

                .setClassid(classValue)

                .build();

    }

    private static StructuredProperty getPid(final JsonObject localIdentifier, final Map<String, ScholExplorerConfiguration> conf) {

        final String pidType = getStringValue(localIdentifier, "type");

        final ScholExplorerConfiguration configuration = conf.get(pidType);

        if (configuration.getCleandPidType() == null) {

            return null;

        }

        final String pid = getStringValue(localIdentifier, "id");

        return StructuredProperty.newBuilder()

                .setValue(pid)

                .setQualifier(getQualifier(configuration.getCleandPidType(), "dnet:pid_types"))

                .build();

    }

    public static String getStringValue(final JsonObject root, final String key) {

        if (root.has(key) && !root.get(key).isJsonNull())

            return root.get(key).getAsString();

        return null;

    }

    public static List<String> getArrayValues(final JsonObject root, final String key) {

        if (root.has(key) && root.get(key).isJsonArray()) {

            final JsonArray asJsonArray = root.get(key).getAsJsonArray();

            final List<String> result = new ArrayList<>();

            asJsonArray.forEach(it -> {

                if (StringUtils.isNotBlank(it.getAsString())) {

                    result.add(it.getAsString());

                }

            });

            return result;

        }

        return new ArrayList<>();

    }

}