D-Net

package eu.dnetlib.data.mapreduce.hbase.lodExport.utils;

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

import eu.dnetlib.data.proto.DatasourceProtos.Datasource.Metadata;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.FieldTypeProtos.StringField;

import eu.dnetlib.data.proto.FieldTypeProtos.StructuredProperty;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.OafProtos.OafEntity;

import eu.dnetlib.data.proto.OafProtos.OafRel;

import eu.dnetlib.data.proto.OrganizationProtos.Organization;

//import eu.dnetlib.data.proto.PersonProtos;

import eu.dnetlib.data.proto.PersonProtos;

import eu.dnetlib.data.proto.ProjectProtos.Project;

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

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

import java.util.ArrayList;

import java.util.List;

import java.util.Set;

/**

 * @author eri Simple serializer that parses input Oaf Protos and prepares them

 *         for sqoop

 */

public class Serializer {

    private static final String TEXT_PATTERN = "[^a-zA-Z0-9 .-_:/@+=-]";

    private static final String ID_PATTERN = "[^a-zA-Z0-9.:/_-]";

    private static final String DOI_PATTERN = ".*\\/(10.)";

    public static String serialize(Oaf oaf, String DELIM) {

        switch (oaf.getKind()) {

            case entity:

                OafEntity valueEntity = oaf.getEntity();

                switch (valueEntity.getType()) {

                    case datasource:

                        return buildDatasource(valueEntity, DELIM);

                    case organization:

                        return buildOrganization(valueEntity, DELIM);

                    case project:

                        return buildProject(valueEntity, DELIM);

                    case result:

                        return buildResult(valueEntity, DELIM);

                        /*

                    case person:

                        return buildPerson(valueEntity, DELIM);

                        */

                    default:

                        break;

                }

                break;

            case relation:

                OafRel valueRel = oaf.getRel();

                return serialize(valueRel, DELIM);

        }

        return null;

    }

    public static void extractRelations(Oaf oaf, String DELIM, Set<String> relations) {

        OafEntity valueEntity = oaf.getEntity();

        switch (valueEntity.getType()) {

            case result:

                getResultDatasources(valueEntity, DELIM, relations);

                getDedups(valueEntity, DELIM, relations);

            case datasource:

                getDedups(valueEntity, DELIM, relations);

                /*

            case person:

                getDedups(valueEntity, DELIM, relations);

                */

            case organization:

                getDedups(valueEntity, DELIM, relations);

            default:

        }

    }

    public static String serialize(OafRel Rel, String DELIM) {

        StringBuilder buff;

        switch (Rel.getRelType()) {

            case datasourceOrganization:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("datasource").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM).append("\n");

                return buff.toString();

            case resultResult:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);

                return buff.toString();

                /*

            case personPerson:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("person").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("person").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);

                return buff.toString();

                */

            case organizationOrganization:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("organization").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);

                return buff.toString();

                /*

            case personResult:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("person").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);

                return buff.toString();

                */

            case projectOrganization:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);

                return buff.toString();

                /*

            case projectPerson:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("person").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);

                return buff.toString();

                */

            case resultOrganization:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)

                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);

                return buff.toString();

//TODO - SOS RELATION RESULT PROJECT IS INVERTED! SOURCE IS PROJECT, TARGET IS RESULT

            case resultProject:

                buff = new StringBuilder();

                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM)

                        .append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM);

                return buff.toString();

            default:

        }

        return "";

    }

    private static String getHeader(OafEntity data, String DELIM) {

        String SEPERATOR = ";";

        StringBuilder buff = new StringBuilder();

        //  EntityType

        buff.append(data.getType().name()).append(DELIM);

        // OpenaireID

        buff.append(cleanId(data.getId())).append(DELIM);

        //  dateOfTransformation

        // TODO  CRITERIA FOR IDENTIFYING UPDATED RECORDS

        buff.append(cleanId(data.getDateoftransformation())).append(DELIM);

        //    dateOfCollection

        buff.append(clean(data.getDateofcollection())).append(DELIM);

        String dataStr = new String();

        // originalId

        for (String oid : data.getOriginalIdList()) {

            dataStr += cleanDoi(oid) + SEPERATOR; //oid is not openaireid, clean instead of cleanid

        }

        buff.append(dataStr).append(DELIM);

        return buff.toString();

    }

    private static String buildDatasource(OafEntity data, String DELIM) {

        String SEPERATOR = ";";

        StringBuilder buff = new StringBuilder();

        buff.append(getHeader(data, DELIM));

        Metadata metadata = data.getDatasource().getMetadata();

        //Datasourcetype

        buff.append(clean(metadata.getDatasourcetype().getClassname())).append(DELIM);

        //Openairecompatibility

        buff.append(clean(metadata.getOpenairecompatibility().getClassname())).append(DELIM);

        //OfficialName

        buff.append(clean(metadata.getOfficialname().getValue())).append(DELIM);

        //  Englishname

        buff.append(clean(metadata.getEnglishname().getValue())).append(DELIM);

        //Websiteurl

        buff.append(clean(metadata.getWebsiteurl().getValue())).append(DELIM);

        //LogoURL

        buff.append(clean(metadata.getLogourl().getValue())).append(DELIM);

        //Contactemail

        buff.append(clean(metadata.getContactemail().getValue())).append(DELIM);

        //Namespaceprefix

        buff.append(clean(metadata.getNamespaceprefix().getValue())).append(DELIM);

        // latitude

        buff.append(clean(metadata.getLatitude().getValue())).append(DELIM);

        // longtitude

        buff.append(clean(metadata.getLongitude().getValue())).append(DELIM);

        // dateofvalidation,

        buff.append(clean(metadata.getDateofvalidation().getValue())).append(DELIM);

        //Description

        buff.append(clean(metadata.getDescription().getValue())).append(DELIM);

        //subjects

        String subj = new String();

        for (StructuredProperty s : metadata.getSubjectsList()) {

            subj += clean(s.getValue()) + SEPERATOR;

        }

        //subjectList

        buff.append(clean(subj)).append(DELIM);

        //Number of items

        buff.append(clean(metadata.getOdnumberofitems().getValue())).append(DELIM);

        //Date of number of items

        buff.append(clean(metadata.getOdnumberofitemsdate().getValue())).append(DELIM);

        // Policies

        buff.append(clean(metadata.getOdpolicies().getValue())).append(DELIM);

        //languages

        String dataStr = new String();

        for (StringField lang : metadata.getOdlanguagesList()) {

            dataStr += clean(lang.getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        ;

        // Content type

        dataStr = " ";

        for (StringField c : metadata.getOdcontenttypesList()) {

            dataStr += clean(c.getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        //Access info package

        dataStr = " ";

        for (StringField c : metadata.getAccessinfopackageList()) {

            dataStr += clean(c.getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        //Release start date

        buff.append(clean(metadata.getReleasestartdate().getValue())).append(DELIM);

        //Release end date

        buff.append(clean(metadata.getReleaseenddate().getValue())).append(DELIM);

        //Mission statement url

        buff.append(clean(metadata.getMissionstatementurl().getValue())).append(DELIM);

        //Data provider

        buff.append(clean(String.valueOf(metadata.getDataprovider().getValue()))).append(DELIM);

        //Service provider

        buff.append(clean(String.valueOf(metadata.getServiceprovider().getValue()))).append(DELIM);

        //Database access type

        buff.append(clean(metadata.getDatabaseaccessrestriction().getValue())).append(DELIM);

        //Data upload type

        buff.append(clean(metadata.getDatauploadtype().getValue())).append(DELIM);

        //Data upload restrictions

        buff.append(clean(metadata.getDatauploadrestriction().getValue())).append(DELIM);

        //Versioning

        buff.append(clean(String.valueOf(metadata.getVersioning().getValue()))).append(DELIM);

        //Citation guideline url

        buff.append(clean(metadata.getCitationguidelineurl().getValue())).append(DELIM);

        //Quality management kind

        buff.append(clean(metadata.getQualitymanagementkind().getValue())).append(DELIM);

        //PID systems

        buff.append(clean(metadata.getPidsystems().getValue())).append(DELIM);

        //Certificates

        buff.append(clean(metadata.getCertificates().getValue())).append(DELIM);

        //Policies

        dataStr = " ";

        for (FieldTypeProtos.KeyValue property : metadata.getPoliciesList()) {

            dataStr += clean(property.getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        buff.append(getTrust(data)).append(DELIM);

        return buff.toString();

    }

    private static String buildOrganization(OafEntity data, String DELIM) {

        String SEPERATOR = ";";

        StringBuilder buff = new StringBuilder();

        buff.append(getHeader(data, DELIM));

        Organization organization = data.getOrganization();

        Organization.Metadata metadata = organization.getMetadata();

        //getLegalshortname

        buff.append(clean(metadata.getLegalshortname().getValue())).append(DELIM);

        // `name`,

        buff.append(clean(metadata.getLegalname().getValue())).append(DELIM);

        //website URL

        String[] split = metadata.getWebsiteurl().getValue().split(",");

        String dataStr = new String();

        for (String s : split) {

            dataStr += s.replace(DELIM, " ") + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        //logourl

        buff.append(clean(metadata.getLogourl().getValue())).append(DELIM);

        // `country`,

        buff.append(clean(metadata.getCountry().getClassid())).append(DELIM);

        buff.append(getTrust(data)).append(DELIM);

        return buff.toString();

    }

    static String getTrust(OafEntity data) {

        for (FieldTypeProtos.ExtraInfo info : data.getExtraInfoList()) {

            return (info.getTrust());

        }

        return " ";

    }

    private static String buildResult(OafEntity data, String DELIM) {

        String SEPERATOR = ";";

        Result.Metadata metadata = data.getResult().getMetadata();

        StringBuilder buff = new StringBuilder();

        buff.append(getHeader(data, DELIM));

        //   titleString

        String dataStr = new String();

        for (int i = 0; i < metadata.getTitleList().size(); i++) {

            StructuredProperty title = metadata.getTitleList().get(i);

            dataStr = clean(title.getValue());

            break;

        }

        //  pubtitle

        buff.append(clean(dataStr)).append(DELIM);

        // date of acceptance CHANGED THIS TO DATE FORMAT

        buff.append(clean(metadata.getDateofacceptance().getValue())).append(DELIM);

        // publisher

        buff.append(clean(metadata.getPublisher().getValue())).append(DELIM);

        //PID

        dataStr = " ";

        for (StructuredProperty p : data.getPidList()) {

            dataStr += clean(p.getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        //language

        buff.append(clean(metadata.getLanguage().getClassid())).append(DELIM);

        // RelevantDate

        dataStr = " ";

        for (StructuredProperty p : metadata.getRelevantdateList()) {

            dataStr += clean(p.getValue());

            break;

        }

        buff.append(dataStr).append(DELIM);

        //Subject

        dataStr = " ";

        for (StructuredProperty subj : metadata.getSubjectList()) {

            if (subj.getValue() != null && !subj.getValue().isEmpty()) {

                dataStr += clean(subj.getValue()) + SEPERATOR;

            }

        }

        buff.append(dataStr).append(DELIM);

        //TODO ExternalReference

        buff.append(" ").append(DELIM);

        //Source

        dataStr = " ";

        for (StringField s : metadata.getSourceList()) {

            dataStr += clean(s.getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        //TODO Format     

        buff.append(" ").append(DELIM);

        //Context

        dataStr = " ";

        for (Result.Context s : metadata.getContextList()) {

            dataStr += clean(s.getId()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        //country

        String country = " ";

        for (FieldTypeProtos.Qualifier c : metadata.getCountryList()) {

            country += clean(c.getClassid()) + SEPERATOR;

        }

        buff.append(country).append(DELIM);

        //Best License

        buff.append(getBestLicense(data.getResult())).append(DELIM);

        //Description

        dataStr = " ";

        for (StringField desc : metadata.getDescriptionList()) {

            dataStr += clean(desc.getValue());

            break;

        }

        buff.append(dataStr).append(DELIM);

        //Journal  

        buff.append(clean(metadata.getJournal().getName())).append(DELIM);  //#null#!

        // TODO ERI SOS : HERE IN GET JOUTNAL. GET DATA INFO I CAN FIND PROVENANCE AND SIMILARITY

        // TODO isRelatedTo

        //   resource type

        buff.append(clean(metadata.getResourcetype().getClassname())).append(DELIM);

        //   device

        buff.append(clean(metadata.getDevice().getValue())).append(DELIM);

        //   size

        buff.append(clean(metadata.getSize().getValue())).append(DELIM);

        //     version

        buff.append(clean(metadata.getVersion().getValue())).append(DELIM);

        //   metadata update

        buff.append(clean(metadata.getLastmetadataupdate().getValue())).append(DELIM);

        //   metadata version

        buff.append(clean(metadata.getMetadataversionnumber().getValue())).append(DELIM);

        // year

        buff.append(clean(getYearInt(metadata.getDateofacceptance().getValue()))).append(DELIM);

        // type

        buff.append(clean(metadata.getResulttype().getClassname())).append(DELIM);

        buff.append(getTrust(data)).append(DELIM);

        //Authors

        dataStr = " ";

        for(PersonProtos.Person p : data.getResult().getAuthorList()){

            dataStr += clean(p.getMetadata().getFullname().getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        return buff.toString();

    }

    private static String buildProject(OafEntity data, String DELIM) {

        String SEPERATOR = ";";

        StringBuilder buff = new StringBuilder();

        buff.append(getHeader(data, DELIM));

        Project.Metadata metadata = data.getProject().getMetadata();

        //Code

        buff.append(metadata.getCode().getValue()).append(DELIM);

        // `Websiteurl`,

        buff.append(clean(metadata.getWebsiteurl().getValue())).append(DELIM);

        //TODO here

        // `Acronym`,

        buff.append(clean(metadata.getAcronym().getValue())).append(DELIM);

        //Title

        buff.append(clean(metadata.getTitle().getValue())).append(DELIM);

        // Startdate

        buff.append(clean(metadata.getStartdate().getValue())).append(DELIM);

        // Enddate

        buff.append(clean(metadata.getEnddate().getValue())).append(DELIM);

        //`Call identifer`

        buff.append(clean(metadata.getCallidentifier().getValue())).append(DELIM);

        //`KeyWords`

        buff.append(clean(metadata.getKeywords().getValue())).append(DELIM);

        //`Duration`

        buff.append(clean(metadata.getDuration().getValue())).append(DELIM);

        //esc39

        buff.append(clean(metadata.getEcsc39().getValue())).append(DELIM);

        //`Contracttype`

        buff.append(clean(metadata.getContracttype().getClassname())).append(DELIM);

        //`OA mandate pubs`  TODO DOES NOT EXIST

        buff.append(clean(metadata.getOamandatepublications().getValue())).append(DELIM);

        //`Subjects`

        String dataStr = new String();

        for (StructuredProperty s : metadata.getSubjectsList()) {

            dataStr += clean(s.getValue()) + SEPERATOR;

        }

        buff.append(dataStr).append(DELIM);

        //`EC293`

        buff.append(clean(metadata.getEcarticle293().getValue())).append(DELIM);

        List<StringField> fundList = metadata.getFundingtreeList();

        if (!fundList.isEmpty()) // `funding_lvl0`,

        {//TODO funder + 3 funding levels

           /* funder text,

            funding_lvl0 text,

	        funding_lvl1 text,

	        funding_lvl2 text,

	        funding_lvl3 text,*/

            buff.append(FundingParser.getFundingInfo(fundList.get(0).getValue(), DELIM));

        } else {

            buff.append(FundingParser.getFundingInfo("", DELIM));

        }

        buff.append(getTrust(data)).append(DELIM);

        return buff.toString();

    }

    /*

    private static String buildPerson(OafEntity data, String DELIM) {

        String SEPERATOR = ";";

        PersonProtos.Person person = data.getPerson();

        PersonProtos.Person.Metadata metadata = person.getMetadata();

        StringBuilder buff = new StringBuilder();

        buff.append(getHeader(data, DELIM));

        // `firstname`,

        buff.append(clean(metadata.getFirstname().getValue())).append(DELIM);

        // `secondNames`,

        String dataStr = new String();

        for (StringField s : metadata.getSecondnamesList()) {

            dataStr += clean(s.getValue()) + ' ';

        }

        buff.append(dataStr).append(DELIM);

        // `fullname`,

        buff.append(clean(metadata.getFullname().getValue())).append(DELIM);

        // `Fax`,

        buff.append(clean(metadata.getFax().getValue())).append(DELIM);

        // `Email`,

        buff.append(clean(metadata.getEmail().getValue())).append(DELIM);

        // `Phone`,

        buff.append(clean(metadata.getPhone().getValue())).append(DELIM);

        // `Nationality`,

        buff.append(clean(metadata.getNationality().getClassid())).append(DELIM);

        // `PIDS`,

        dataStr = " ";

        for (StructuredProperty s : data.getPidList()) {

            dataStr += clean(s.getValue()) + ";";

        }

        buff.append(dataStr).append(DELIM);

        buff.append(getTrust(data)).append(DELIM);

        return buff.toString();

    }

    */

    private static void getResultDatasources(OafEntity valueEntity, String DELIM, Set<String> returnList) {

        String SEPERATOR = ";";

        Result result = valueEntity.getResult();

        if (valueEntity.getId().contains("dedup")) return;

        //TODO hosted by

        for (Instance instance : (result.getInstanceList())) {

            String hostedBy = instance.getHostedby().getKey();

            if (hostedBy != null && !hostedBy.isEmpty()) {

                returnList.add("resultDatasource" + DELIM + "result" + DELIM +

                        cleanId(valueEntity.getId()) + DELIM + "datasource" + DELIM + cleanId(hostedBy) + DELIM);

            }

        }

        //TODO  collected froms

        for (FieldTypeProtos.KeyValue collectedFromValue : (valueEntity.getCollectedfromList())) {

            String collectedFrom = collectedFromValue.getKey();

            if (collectedFrom != null && !collectedFrom.isEmpty())

                returnList.add(("resultDatasource" + DELIM

                        + "result" + DELIM + cleanId(valueEntity.getId()) + DELIM

                        + "datasource" + DELIM + cleanId(collectedFrom) + DELIM));

        }

    }

    public static String cleanId(String value) {

        if (value == null) {

            return " ";

        }

        //   DO NOT CHANGE THIS: IT REMOVES ID PREFIX ( 5|datacite____::

        // to datacite____:: )

        // AND REPLACES OCCURENCES OF DELIM CHARS IN DATA

        value = value.replaceFirst(".*\\|", "");

        value = value.replaceAll(ID_PATTERN, "");

        value = value.replace("#", " ");

        value = value.replace("\n", " ");

        return value;

    }

    private static String clean(String value) {

        if (value == null) {

            return " ";

        }

        // TODO DO NOT CHANGE THIS: IT REMOVES ID PREFIX ( 5|datacite____::

        value = value.replaceAll("[\"\\r\\\\;]", "");

        value = value.replace("\\", "");

        value = value.replaceAll(TEXT_PATTERN, " ");

        value = value.replace(">", " ");

        value = value.replace("<", " ");

        value = value.replace("\"", " ");

        value = value.replace("\\", " ");

        value = value.replace("'", " ");

        value = value.replace("«", " ");

        value = value.replace("»", " ");

        value = value.replace("#", " ");

        value = value.replace("\\", " ");

        value = value.replace("\n", " ");

        value = value.replace(",", "");

        value = value.replace("#", "");

        value = value.replace(";", "");

        return value;

    }

    private static String cleanDoi(String value) {

        value = value.replaceAll(DOI_PATTERN, "");

        value = value.replace("#", " ");

        value = value.replace("\n", " ");

        return value;

    }

    //TODO make them in pairs

    private static void getDedups(OafEntity valueEntity, String DELIM, Set<String> returnList) {

        if (!valueEntity.getChildrenList().isEmpty() && valueEntity.getId().contains("dedup")) {

            ArrayList<String> entries = new ArrayList<String>();

            for (OafEntity child : valueEntity.getChildrenList()) {

                if (child.getType() == valueEntity.getType() && !child.getId().contains("dedup")) {

                    // if it is result, then its the deduplication

                    entries.add(cleanId(child.getId()));

                }

            }

            for (int i = 0; i < entries.size() - 1; i++) {

                for (int j = i + 1; j < entries.size(); j++) {

                    returnList.add("dedup" + DELIM +

                            valueEntity.getType().name() + DELIM +

                            entries.get(i) + DELIM +

                            valueEntity.getType().name() + DELIM +

                            entries.get(j) + DELIM);

                }

            }

        }

    }

    private static String getYearInt(String data) {

        if (data == null || data.isEmpty() || data.equals("-1")) {

            return " ";

        }

        String[] split = data.split("-");

        if (split == null || split.length == 0) {

            return " ";

        }

        return split[0];

    }

    private static String getBestLicense(Result result) {

        FieldTypeProtos.Qualifier bestLicense = null;

        LicenseComparator lc = new LicenseComparator();

        for (Instance instance : (result.getInstanceList())) {

            if (lc.compare(bestLicense, instance.getLicence()) > 0) {

                bestLicense = instance.getLicence();

            }

        }

        if (bestLicense != null) {

            return bestLicense.getClassname();

        } else {

            return null;

        }

    }

}