D-Net

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

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

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

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;

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.ProjectProtos.Project;

import eu.dnetlib.data.proto.RelTypeProtos.RelType;

import eu.dnetlib.data.proto.ResultProtos;

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

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

import org.apache.log4j.Logger;

import java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Date;

import java.util.List;

/**

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

 *         for sqoop

 */

public class Serializer {

    private static String DELIM;

    private Logger log = Logger.getLogger(this.getClass());

    private String NULL_STRING = null;

    private String NULL_NUM = "0";

    private static String ENCLOSED;

    private FundingParser fundingParser = new FundingParser();

    public Serializer() {

    }

    public String serialize(Oaf oaf) {

        switch (oaf.getKind()) {

            case entity:

                OafEntity valueEntity = oaf.getEntity();

                switch (valueEntity.getType()) {

                    case datasource:

                        return buildDatasource(valueEntity);

                    case organization:

                        return buildOrganization(valueEntity);

                    case project:

                        return buildProject(valueEntity);

                    case result:

                        return buildResult(valueEntity);

                    case person:

                        return buildPerson(valueEntity);

                    default:

                        log.error("wrong type");

                        break;

                }

                break;

            case relation:

                OafRel valueRel = oaf.getRel();

                return buildRel(valueRel);

        }

        return null;

    }

    public String serialize(OafRel oaf) {

        switch (oaf.getRelType()) {

            default:

                return buildRel(oaf);

        }

    }

    public ArrayList<String> extractRelations(Oaf oaf) {

        OafEntity valueEntity = oaf.getEntity();

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

        switch (valueEntity.getType()) {

            case result:

                relations.addAll(getResultDatasources(valueEntity));

                relations.addAll(getDedups(valueEntity));

                return relations;

            case datasource:

                relations.addAll(getDedups(valueEntity));

                return relations;

            case person:

                relations.addAll(getDedups(valueEntity));

                return relations;

            default:

                return null;

        }

    }

    private String buildRel(OafRel Rel) {

        return getStringField(Rel.getRelType().getDescriptorForType().getName()) + getStringField(Rel.getSource()) + getStringField(Rel.getTarget());

    }

    private String getResultResult(OafRel oaf) {

        String buff = new String();

        buff += getStringField(oaf.getTarget());

        buff += getStringField(String.valueOf(oaf.getResultResult().getSimilarity().getSimilarity()));

      /* •	hasAmongTopNSimilarDocuments: r1 hasAmongTopNSimilarDocuments r2 means that Results r1 and r2 are similar, and that we also have r2 isAmongTopNSimilarDocuments of r1.

        In OpenAIRE, N so far always equals 20.

        •	isAmongTopNSimilarDocuments: r1 isAmongTopNSimilarDocuments r2 means that Results r1 and r2 are similar and that r2 hasAmongTopNSimilarDocuments of r1;

        •	isRelatedTo: two results are somehow related to each other. OpenAIRE may further refine the semantics of possible types of "relatedness" by adding new classes in the Qualifier.

                Scheme "dnet:result_result_relations";

oaf. */

        return buff;

    }

    private String getHeader(OafEntity data) {

        String buff = new String();

        //  EntityType

        buff += getStringField(data.getType().name());

        // OpenaireID

        buff += getStringField(getStringDateField(cleanId(data.getId())));

        //  dateOfTransformation

        // TODO change to dateOfTransformation here when released

        buff += getStringField(data.getDateofcollection());

        //    dateOfCollection

        buff += getStringField(data.getDateofcollection());

        String originalId = new String();

// originalId

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

            originalId += oid + ";";

        }

        buff += getStringField(originalId);

        return buff;

    }

    private String buildDatasource(OafEntity data) {

        Datasource d = data.getDatasource();

        Metadata metadata = d.getMetadata();

        String buff = getHeader(data);

/*

        OfficialName

                Englishname

        Websiteurl

                Logourl

        Contactemail

                Namespaceprefix

        Latitude

                Longitude

        Dateofvalidation

                Description

        SubjectsList

                Odnumberofitems

        Odnumberofitemsdate

                Odpolicies

        OdlanguagesList

                OdcontenttypesList

        Accessinfopackage

                Releasestartdate

        Releaseenddate

                Missionstatementurl

        Dataprovider

                Serviceprovider

        Databaseaccessrestriction

                Datauploadtype

        Datauploadrestriction

                Versioning

        Citationguidelineurl

                Qualitymanagementkind

        Pidsystems

                Certificates

        Policies

                Trust

*/

        //Datasourcetype

        if (metadata.hasDatasourcetype()) {

            buff += getStringField(metadata.getDatasourcetype().getClassname().replaceFirst(".*::", ""));

        } else {

            buff += getStringField(null);

        }

        //Openairecompatibility

        buff += getStringField(metadata.getOpenairecompatibility().getClassname());

        //Official Name

        buff += getStringField(metadata.getOfficialname().getValue());

        // English Name

        buff += getStringField(metadata.getEnglishname().getValue());

        //Website URL

        buff += getStringField(metadata.getWebsiteurl().getValue());

        //LogoURL

        buff += getStringField(metadata.getLogourl().getValue());

        //Email

        buff += getStringField(metadata.getContactemail().getValue());

        //Namespaceprefix

        buff += getStringField(metadata.getNamespaceprefix().getValue());

        // latitude

        buff += getStringField(metadata.getLatitude().getValue());

        // longtitude

        buff += getStringField(metadata.getLongitude().getValue());

        // dateofvalidation,

        buff += getStringField(metadata.getDateofvalidation().getValue());

        //Description

        buff += getStringField(metadata.getDescription().getValue());

        //subjects

        String subj = new String();

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

            subj += s.getValue() + ';';

        }

        //subjects

        buff += getStringField(subj);

        //Number of items

        buff += getStringField(metadata.getOdnumberofitems().getValue());

        //Date of number of items

        buff += getStringField(metadata.getOdnumberofitemsdate().getValue());

        // Policies

        buff += getStringField(metadata.getOdpolicies().getValue());

        //languages

        String languages = new String();

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

            languages += lang.getValue() + ";";

        }

        buff += getStringField(languages);

        // Content type

        String contentType = new String();

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

            contentType += c.getValue() + ";";

        }

        buff += getStringField(contentType);

        //Access info package

        String accessInfo = new String();

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

            accessInfo += c.getValue() + ";";

        }

        buff += getStringField(accessInfo);

        //Release start date

        buff += getStringField(metadata.getReleasestartdate().getValue());

        //Release end date

        buff += getStringField(metadata.getReleaseenddate().getValue());

        //Mission statement url

        buff += getStringField(metadata.getMissionstatementurl().getValue());

        //Data provider

        buff += getStringField(String.valueOf(metadata.getDataprovider().getValue()));

        //Service provider

        buff += getStringField(String.valueOf(metadata.getServiceprovider().getValue()));

        //Database access type

        buff += getStringField(metadata.getDatabaseaccessrestriction().getValue());

        //Data upload type

        buff += getStringField(metadata.getDatauploadtype().getValue());

        //Data upload restrictions

        buff += getStringField(metadata.getDatauploadrestriction().getValue());

        //Versioning

        buff += getStringField(String.valueOf(metadata.getVersioning().getValue()));

        //Citation guideline url

        buff += getStringField(String.valueOf(metadata.getCitationguidelineurl().getValue()));

        //Quality management kind

        buff += getStringField(String.valueOf(metadata.getQualitymanagementkind().getValue()));

        //PID systems

        buff += getStringField(metadata.getPidsystems().getValue());

        //Certificates

        buff += getStringField(metadata.getCertificates().getValue());

        //Policies

        String policies = new String();

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

            policies += property.getValue() + ";";

        }

        buff += getStringField(policies);

        return buff;

    }

    private String buildOrganization(OafEntity data) {

        Organization organization = data.getOrganization();

        eu.dnetlib.data.proto.OrganizationProtos.Organization.Metadata metadata = organization.getMetadata();

        String buff = getHeader(data);

        //getLegalshortname

        buff += getStringField(metadata.getLegalshortname().getValue());

        // `name`,

        buff += getStringField(metadata.getLegalname().getValue());

        //website URL

        buff += getStringField(metadata.getWebsiteurl().getValue());

        //logourl

        buff += getStringField(metadata.getLogourl().getValue());

        // `country`,

        buff += getStringField(metadata.getCountry().getClassname());

        return buff;

    }

    private String buildResult(OafEntity data) {

        Result result = data.getResult();

        eu.dnetlib.data.proto.ResultProtos.Result.Metadata metadata = result.getMetadata();

        String buff = getHeader(data);

        //   titleString

        String titleString = new String();

        String alternativeTitles = new String();

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

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

            if (i == 0) {

                titleString = title.getValue().replaceAll("\\s+", " ");

                titleString = titleString.replaceAll("\n", " ");

            } else {

                alternativeTitles += title.getValue().replaceAll("\\s+", " ") + " ; ";

                alternativeTitles = alternativeTitles.replaceAll("\n", " ");

            }

            break;

        }

        //  pubtitle

        buff += getStringField(titleString);

        // alternative titles

        //buff += getStringField(alternativeTitles);    //  null#!

        // date of acceptance CHANGED THIS TO DATE FORMAT

        buff += getStringDateField(metadata.getDateofacceptance().getValue());

        // publisher

        buff += getStringField(metadata.getPublisher().getValue());

        //PID

        String pids = new String();

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

            pids += p.getValue() + ";";

        }

        buff += getStringField(pids);

        //language

        buff += getStringField(metadata.getLanguage().getClassname());

        // RelevantDate

        String reldate = new String();

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

            reldate += p.getValue();

            break;

        }

        buff += getStringField(reldate);

        //Subject

        String subjects = new String();

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

            if (isValidTopic(subj.getValue())) {

                if (!isNumeric(subj.getValue())) {

                    subjects += subj.getValue() + ";";

                }

            }

        }

        buff += subjects;

        // TODO Instance

        // buff += getStringField();

        //TODO ExternalReference

        //Source

        String source = new String();

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

            source += s.getValue() + ";";

        }

        buff += getStringField(source);

        //TODO Format     

        buff += getStringField("");

        //DOES NOT EXIST

          /*String formatString = new String();

        for (StringField format : metadata.getFormatList()) {

            formatString = format.getValue();

            break;}

               buff += getStringField(formatString);

        }*/

        //Context

        String context = new String();

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

            context += s.getId() + ";";

        }

        buff += getStringField(context);

        //country TODO does not exist; throws error

        String country = new String();

        // for (Qualifier s : metadata.getCountryList()) {

        //  country += s.getClassname() + ";";

        // }

        buff += getStringField(country);

        // access_mode,

        buff += getStringField(getAccessMode(result));

        //Best License

        buff += getStringField(getBestLicense(result));

        //Description

        String description = new String();

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

            description += desc;

            break;

        }

        buff += getStringField(description);

        //Journal  

        buff += getStringField(metadata.getJournal().getName().replaceAll("\n", " "));  //#null#!

        //journalName                     

        buff += getStringField(metadata.getJournal().getName().replaceAll("\n", " "));  //#null#!

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

        //ISSN                                  

        buff += getStringField(metadata.getJournal().getIssnLinking());

        //embargoEndDate    

        buff += getStringField(metadata.getEmbargoenddate().getValue());

        // `authors`,

        int authors = 0;

        String delayed = "no";

        for (OafRel rel : data.getCachedRelList()) {

            if (rel.getRelType().equals(RelType.personResult)) {

                authors++;

            } else if (rel.getRelType().equals(RelType.resultProject))

            // TODO remember : in result Project, first id is project, second is  result.

            {

                String daysfromend = getYearDifferenceInteger(rel.getResultProject().getOutcome().getRelMetadata().getEnddate(), rel.getResultProject().getOutcome().getRelMetadata().getStartdate());

                if (Integer.parseInt(daysfromend) > 0) {

                    delayed = "yes";

                }

            }

        }

        buff += getNumericField(String.valueOf(authors));

        // TODO isRelatedTo

        //   resource type

        buff += getStringField(metadata.getResourcetype().getClassname());

        //   device

        buff += getStringField(metadata.getDevice().getValue());

        //   size

        buff += getStringField(metadata.getSize().getValue());

        //     version

        buff += getStringField(metadata.getVersion().getValue());

        //   metadata update

        buff += getStringField(metadata.getLastmetadataupdate().getValue());

        //   metadata version

        buff += getStringField(metadata.getMetadataversionnumber().getValue());

        // `delayed`,

        buff += getStringField(delayed);

        // year

        buff += getYearInt(metadata.getDateofacceptance().getValue());

        // type

        buff += getStringField(metadata.getResulttype().getClassname());

        //classification

        String classification = new String();

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

            if (instance.getInstancetype().getClassname() != null && !instance.getInstancetype().getClassname().isEmpty()) {

                classification += instance.getInstancetype().getClassname() + ';';

            }

        }

        buff += getStringField(classification);

        //  hosted by

        String hostedBy = new String();

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

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

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

                hostedBy += host + ";";

            }

        }

        buff += getStringField(hostedBy);

        //   collectedfrom

        String collectedFrom = new String();

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

            String host = collectedFromValue.getKey();

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

                collectedFrom += host + ';';

            }

        }

        buff += getStringField(collectedFrom);

        return buff;

    }

    private String buildProject(OafEntity data) {

        Project project = data.getProject();

        eu.dnetlib.data.proto.ProjectProtos.Project.Metadata metadata = project.getMetadata();

        String buff = getHeader(data);

        //Code

        buff += getStringField(metadata.getCode().getValue());

        // `url`,

        buff += getStringField(metadata.getWebsiteurl().getValue());

        // `acronym`,

        String acronym = metadata.getAcronym().getValue();

        if (acronym.equalsIgnoreCase("UNKNOWN")) {

            acronym = metadata.getTitle().getValue();

        }

        buff += getStringField(acronym);

        //title!

        String title = getStringField(metadata.getTitle().getValue());

        buff += getStringField(title);

        // startdate

        buff += getNumericField(metadata.getStartdate().getValue());

        // enddate

        buff += getNumericField(metadata.getEnddate().getValue());

        //`Call identifer`

        buff += getStringField(metadata.getCallidentifier().getValue());

        //`KeyWords`

        buff += getStringField(metadata.getKeywords().getValue());

        //`Duration`

        buff += getStringField(metadata.getDuration().getValue());

        //esc39

        buff += getStringField(metadata.getEcsc39().getValue().toString());

        //`Contracttype`

        buff += getStringField(metadata.getContracttype().getClassname());

        //`OA mandate pubs`  TODO DOES NOT EXIST

        buff += getStringField(metadata.getOamandatepublications().getValue());

        //`Subjects` TODO DOES NOT EXIST EITHER

        String subjects = new String();

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

            subjects += s.getValue() + ';';

        }

        buff += getStringField(subjects);

        //`EC293`

        buff += getStringField(metadata.getEcarticle293().getValue());

        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 += this.fundingParser.getFundingInfo(fundList.get(0).getValue());

        } else {

            buff += this.fundingParser.getFundingInfo("");

        }

        return buff;

    }

    private String buildPerson(OafEntity data) {

        PersonProtos.Person person = data.getPerson();

        eu.dnetlib.data.proto.PersonProtos.Person.Metadata metadata = person.getMetadata();

        String buff = getHeader(data);

        // `firstname`,

        buff += metadata.getFirstname();

        // `secondNames`,

        String secondNames = new String();

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

            secondNames += s.getValue() + ' ';

        }

        buff += getStringField(secondNames);

        // `fullname`,

        buff += getStringField(metadata.getFullname().getValue());

        // `Fax`,

        buff += getStringField(metadata.getFax().getValue());

        // `Email`,

        buff += getStringField(metadata.getEmail().getValue());

        // `Phone`,

        buff += getStringField(metadata.getPhone().getValue());

        // `Nationality`,

        buff += getStringField(metadata.getNationality().getClassname());

        // `PIDS`,

        String pids = new String();

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

            pids += s.getValue() + ";";

        }

        buff += getStringField(pids);

        // `collected from`,

        String collectedFrom = new String();

        for (FieldTypeProtos.KeyValue s : data.getCollectedfromList()) {

            collectedFrom += s.getValue() + ";";

        }

        buff += getStringField(collectedFrom);

        return buff;

    }

    private ArrayList<String> getResultDatasources(OafEntity valueEntity) {

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

        Result result = valueEntity.getResult();

        //TODO hosted by

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

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

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

                buffs.add(getStringField("resultDatasource") + getStringField(hostedBy));

            }

        }

        //TODO  collected froms

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

            String collectedFrom = collectedFromValue.getKey();

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

                buffs.add((getStringField("resultDatasource") + getStringField(collectedFrom)));

        }

        return buffs;

    }

    public String cleanId(String value) {

        if (value != null) {

            //   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("\n", "");

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

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

            value = value.trim();

        }

        if (value == null) {

            return null;

        }

        return ENCLOSED + value + ENCLOSED;

    }

    private ArrayList<String> getDedups(OafEntity valueEntity) {

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

        String buff = getStringField("Dedup") + getStringField(cleanId(valueEntity.getId()));

        //TODO hosted by

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

            {

                if (child.getType() == valueEntity.getType()) { // if it is result, then its the deduplication

                    buff += getStringField(cleanId(child.getId()));

                }

            }

        }

        buffs.add(buff);

        return buffs;

    }

    private String getNumericField(String data) {

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

            return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

        } else {

            return ENCLOSED + data + ENCLOSED + DELIM;

        }

    }

    private String getYearDifferenceInteger(String enddate, String startdate) {

        if (enddate != null && !enddate.isEmpty() && startdate != null && !startdate.isEmpty()) {

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

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

                return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

            }

            int Startdate = Integer.parseInt(split[0]);

            split = enddate.split("-");

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

                return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

            }

            int Enddate = Integer.parseInt(split[0]);

            int diff = Enddate - Startdate;

            return ENCLOSED + diff + ENCLOSED + DELIM;

        }

        return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

    }

    private String getYearInt(String data) {

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

            return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

        }

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

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

            return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

        }

        String year = split[0];

        year = cleanNumber(year);

        return ENCLOSED + year + ENCLOSED + DELIM;

    }

    private static String cleanNumber(String number) {

        number = number.replaceAll("[^A-Za-z0-9:,____]", "");

        return number;

    }

    private String getStringField(String data) {

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

            return ENCLOSED + NULL_STRING + ENCLOSED + DELIM;

        } else {

            String field = clean(data);

            if (field == null) {

                return ENCLOSED + NULL_STRING + ENCLOSED + DELIM;

            } else {

                return field + DELIM;

            }

        }

    }

    private String getStringDateField(String data) {

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

            return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

        } else {

            String field = clean(data);

            if (field == null) {

                return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

            } else {

                return field + DELIM;

            }

        }

    }

    public String getId(OafProtos.Oaf oaf) {

        switch (oaf.getKind()) {

            case entity:

                return cleanId(oaf.getEntity().getId());

            case relation:

                return cleanId(oaf.getRel().getSource());

        }

        return null;