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

    public Serializer(String delim, String enclosing) {

        this.setDELIM(delim);

        this.setENCLOSED(enclosing);

        this.fundingParser = new FundingParser(delim, enclosing);

    }

    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;

            case organization:

                relations.addAll(getDedups(valueEntity));

                return relations;

            default:

                return relations;

        }

    }

    private String buildRel(OafRel Rel) {

        return getStringField(Rel.getRelType().name()) + 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);

        //Datasourcetype

        if (metadata.hasDatasourcetype()) {

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

        } else {

            buff += getStringField(null);

        }

        //Openairecompatibility

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

        //OfficialName

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

        //  Englishname

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

        //Websiteurl

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

        //LogoURL

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

        //Contactemail

        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() + ';';

        }

        //subjectList

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

        String trust = " ";

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

            trust = getStringDateField(info.getTrust());

            break;

        }

        buff += getStringField(trust);

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

        String trust = " ";

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

            trust = getStringDateField(info.getTrust());

            break;

        }

        buff += getStringField(trust);

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

        // 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 ExternalReference

        buff += getStringField(" ");

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

        //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#!

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

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

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

        String trust = " ";

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

            trust = getStringDateField(info.getTrust());

            break;

        }

        buff += getStringField(trust);

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

        // `Websiteurl`,

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

        // `Acronym`,

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

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

        // buff += getStringField("");

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

        }

        String trust = " ";

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

            trust = getStringDateField(info.getTrust());

            break;

        }

        buff += getStringField(trust);

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

        String trust = " ";

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

            trust = getStringDateField(info.getTrust());

            break;

        }

        buff += getStringField(trust);

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

        if (!valueEntity.getChildrenList().isEmpty()) {

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

            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;

    }

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