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.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()) {

            case resultProject:

                return getResultProject(oaf);

            case resultResult:

                return getResultResult(oaf);

            default:

                return buildRel(oaf);

        }

    }

    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 buildRel(OafRel Rel) {

        return getStringField(Rel.getTarget());

    }

    private String getResultProject(OafRel oaf) {

        String buff = new String();

        String result = oaf.getTarget();

        buff += getStringField(result);

        //   is declared as int!!!

        long diff = DATEDIFF(oaf.getResultProject().getOutcome().getRelMetadata().getEnddate(), oaf.getResultProject().getOutcome().getRelMetadata().getStartdate());

        if (diff < 0) {

            diff = 0;

        }

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

        return buff;

    }

    private String buildDatasource(OafEntity data) {

        String buff = new String();

        Datasource d = data.getDatasource();

        Metadata metadata = d.getMetadata();

        //`original Id`

        String originalId = new String();

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

            originalId += oid + ";";

        }

        buff += getStringField(originalId);

        //dateOfCollection

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

        //Provider Type

        if (metadata.hasDatasourcetype()) {

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

        } else {

            buff += getStringField(null);

        }

        //Compatibility

        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) {

        String buff = new String();

        Organization organization = data.getOrganization();

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

        //`original Id`

        String originalId = new String();

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

            originalId += oid + ";";

        }

        buff += getStringField(originalId);

        //dateOfCollection

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

        //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) {

        String buff = new String();

        Result result = data.getResult();

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

        //`original Id`

        String originalId = new String();

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

            originalId += oid + ";";

        }

        buff += getStringField(originalId);

        //dateOfCollection

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

        //   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) {

        String buff = new String();

        Project project = data.getProject();

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

        //`original Id`

        String originalId = new String();

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

            originalId += oid + ";";

        }

        buff += getStringField(originalId);

        //dateOfCollection

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

        //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();

//TODO possible to have multiple funders per project?

        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) {

        String buff = new String();

        PersonProtos.Person person = data.getPerson();

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

        //`original Id`

        String originalId = new String();

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

            originalId += oid + ";";

        }

        buff += getStringField(originalId);

        //dateOfCollection

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

        // `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;

    }

    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 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;

    }

    private boolean isNumeric(String str) {

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

        str = str.replaceAll(" ", "");

        return str.matches("-?\\d+(\\.\\d+)?"); // match a number with optional

        // '-' and decimal.

    }

    //   there are topics with "null" as value -> replace them

    private boolean isValidTopic(String t) {

        if (t == null || t.isEmpty()) {

            return false;

        }

        if (t.equals("") || t.equals(" ")) {

            return false;

        }

        if (t.equals("null") || t.equals("Null") || t.equals("NULL")) {

            return false;

        }

        if (t.equals(ENCLOSED + ENCLOSED + DELIM) || t.equals(ENCLOSED + NULL_STRING + ENCLOSED + DELIM)) {

            return false;

        }

        // skip dedups

        if (t.contains("ddc:")) {

            return false;

        }

        return true;

    }

    private String getBestLicense(ResultProtos.Result result) {

        FieldTypeProtos.Qualifier bestLicense = null;

        LicenseComparator lc = new LicenseComparator();

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

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

                bestLicense = instance.getLicence();

            }

        }

        if (bestLicense != null) {

            return bestLicense.getClassname();

        } else {

            return null;