D-Net

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

import com.google.common.collect.Multimap;

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

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

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

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

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

import org.apache.log4j.Logger;

import org.jsoup.Jsoup;

import java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Date;

import java.util.HashMap;

import java.util.List;

import org.w3c.dom.Element;

import org.w3c.dom.NodeList;

import org.xml.sax.InputSource;

import com.sun.org.apache.xerces.internal.parsers.DOMParser;

import org.w3c.dom.Document;

/**

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

 *         for sqoop

 */

public class Serializer {

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

        switch (oaf.getKind()) {

            case entity:

                OafEntity valueEntity = oaf.getEntity();

                switch (valueEntity.getType()) {

                    case datasource:

                        return buildDatasource(oaf, DELIM, ENCLOSING);

                    case organization:

                        return buildOrganization(oaf, DELIM, ENCLOSING);

                    case project:

                        return buildProject(oaf, DELIM, ENCLOSING);

                    case result:

                        return buildResult(oaf, DELIM, ENCLOSING);

                    /*

                    case person:

                        return buildPerson(oaf, DELIM, ENCLOSING);

                    */

                    default:

                        break;

                }

                break;

            case relation:

                return buildRel(oaf.getRel(), DELIM, ENCLOSING);

        }

        return null;

    }

    public static String serialize(OafRel oaf, String DELIM, String ENCLOSING) {

        switch (oaf.getRelType()) {

            case resultProject:

                return getResultProject(oaf, DELIM, ENCLOSING);

            default:

                return buildRel(oaf, DELIM, ENCLOSING);

        }

    }

    private static String buildRel(OafRel Rel, String DELIM, String ENCLOSING) {

        return cleanId(Rel.getTarget(), DELIM, ENCLOSING) + DELIM;

    }

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

        OafEntity valueEntity = oaf.getEntity();

        getOriginalId(valueEntity, relations, DELIM, ENCLOSING);

        switch (valueEntity.getType()) {

            case datasource:

                getDatasourceLanguages(valueEntity, relations, DELIM, ENCLOSING);

            case result:

                getResultTopics(valueEntity, relations, DELIM, ENCLOSING);

                getResultLanguages(valueEntity, relations, DELIM, ENCLOSING);

                getResultClassifications(valueEntity, relations, DELIM, ENCLOSING);

                getResultDatasources(valueEntity, relations, DELIM, ENCLOSING);

                getResultConcepts(valueEntity, relations, DELIM, ENCLOSING);

                getResultDois(valueEntity, relations, DELIM, ENCLOSING);

                getResultCitations(valueEntity, relations, DELIM, ENCLOSING);

                getResultDescriptions(valueEntity, relations, DELIM, ENCLOSING);

            case project:

                getProjectKeywords(valueEntity, relations, DELIM, ENCLOSING);

                getProjectSubjects(valueEntity, relations, DELIM, ENCLOSING);

            default:

        }

    }

    private static void getOriginalId(OafEntity oafEntity, Multimap<String, String> relations, String DELIM, String ENCLOSING) {

        String relName = oafEntity.getType().toString().toLowerCase() + "Oid";

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

            relations.put(relName, cleanId(oid, DELIM, ENCLOSING));

        }

    }

    private static void getProjectKeywords(OafEntity oafEntity, Multimap<String, String> relations, String DELIM, String ENCLOSING) {

        relations.put("projectKeyword", getStringField(oafEntity.getProject().getMetadata().getKeywords().getValue(), DELIM, ENCLOSING));

    }

    private static void getProjectSubjects(OafEntity oafEntity, Multimap<String, String> relations, String DELIM, String ENCLOSING) {

        for (StructuredProperty subj : oafEntity.getProject().getMetadata().getSubjectsList()) {

            relations.put("projectSubject", getStringField(subj.getValue(), DELIM, ENCLOSING));

        }

    }

    private static String getResultProject(OafRel oaf, String DELIM, String ENCLOSING) {

        StringBuilder buff = new StringBuilder();

        buff.append(cleanId(oaf.getTarget(), DELIM, ENCLOSING) + DELIM);

        // TODO is declared as int!!!

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

        if (diff < 0) {

            diff = 0;

        }

        buff.append(getNumericField(String.valueOf(diff), DELIM, ENCLOSING));

        return buff.toString();

    }

    private static void getDatasourceLanguages(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        Datasource d = valueEntity.getDatasource();

        Metadata metadata = d.getMetadata();

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

            rels.put("datasourceLanguage", getStringField(lang.getValue(), DELIM, ENCLOSING));

        }

    }

    private static void getResultLanguages(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        Result d = valueEntity.getResult();

        Result.Metadata metadata = d.getMetadata();

        if (metadata.getLanguage().getClassname() != null && !metadata.getLanguage().getClassname().isEmpty()) {

            rels.put("resultLanguage", getStringField(metadata.getLanguage().getClassname(), DELIM, ENCLOSING));

        }

    }

    private static void getResultDois(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        for (StructuredProperty pid : valueEntity.getPidList()) {

            rels.put("resultPid",

                    getStringField(pid.getQualifier().getClassname(), DELIM, ENCLOSING) + getStringField(pid.getValue(), DELIM, ENCLOSING));

        }

    }

    private static void getResultClassifications(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        Result result = valueEntity.getResult();

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

            String classification = instance.getInstancetype().getClassname();

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

                rels.put("resultClassification", getStringField(instance.getInstancetype().getClassname(), DELIM, ENCLOSING));

                // TODO HERE KEEP ONLY ONE CLASSIFICATIONS PER RESULT

                break;

            }

        }

    }

    private static void getResultDescriptions(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        Result result = valueEntity.getResult();

        //description

        for (StringField s : result.getMetadata().getDescriptionList()) {

            rels.put("resultDescription", getStringField(Jsoup.parse(s.getValue()).text(), DELIM, ENCLOSING));

        }

    }

    private static void getResultConcepts(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        Result result = valueEntity.getResult();

        for (Result.Context context : result.getMetadata().getContextList()) {

            rels.put("resultConcept", cleanId(context.getId(), DELIM, ENCLOSING));

        }

    }

    private static void getResultDatasources(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        Result result = valueEntity.getResult();

//TODO hosted by

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

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

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

                rels.put("resultDatasource", cleanId(hostedBy, DELIM, ENCLOSING) + DELIM);

            }

        }

//TODO  collected froms

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

            String collectedFrom = collectedFromValue.getKey();

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

                rels.put("resultDatasource", cleanId(collectedFrom, DELIM, ENCLOSING) + DELIM);

            }

        }

    }

    private static void getResultTopics(OafEntity valueEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        Result d = valueEntity.getResult();

        Result.Metadata metadata = d.getMetadata();

        List<StructuredProperty> Topics = metadata.getSubjectList();

        for (StructuredProperty topic : Topics) {

            // TODO result topics

            rels.put("resultTopic", getStringField(topic.getValue(), DELIM, ENCLOSING));

        }

    }

    private static void getResultCitations(OafEntity oafEntity, Multimap<String, String> rels, String DELIM, String ENCLOSING) {

        for (FieldTypeProtos.ExtraInfo extraInfo : oafEntity.getExtraInfoList()) {

            if (extraInfo.getName().equals("result citations")) {

                DOMParser parser = new DOMParser();

                try {

                    parser.parse(new InputSource(new java.io.StringReader(extraInfo.getValue())));

                    Document doc = parser.getDocument();

                    doc.getDocumentElement().normalize();

                    NodeList citations = doc.getElementsByTagName("citation");

                    for (int temp = 0; temp < citations.getLength(); temp++) {

                        Element citation = (Element) citations.item(temp);

                        NodeList ids = citation.getElementsByTagName("id");

                        for(int temp1 = 0; temp1 < ids.getLength(); temp1++){

                            Element id = (Element) ids.item(temp1);

                            if(id.getAttribute("type").equals("openaire")){

                                //System.out.println(id.getAttribute("value"));

                                rels.put("resultCitation", id.getAttribute("value"));

                            }

                        }

                    }

                } catch (Exception e) {

                }

                /*

                rels.put("resultCitation", getStringField(extraInfo.getTrust(), DELIM, ENCLOSING) +

                        getStringField(extraInfo.getProvenance(), DELIM, ENCLOSING) + getStringField(extraInfo.getValue(), DELIM, ENCLOSING));

                        */

            }

        }

    }

    private static String buildDatasource(Oaf oaf, String DELIM, String ENCLOSING) {

        Metadata metadata = oaf.getEntity().getDatasource().getMetadata();

        StringBuilder buff = new StringBuilder();

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // name

        if (metadata.getOfficialname().getValue().equalsIgnoreCase("unknown")) {

            buff.append(getStringField("Unknown Repository", DELIM, ENCLOSING));

        } else {

            buff.append(getStringField(metadata.getOfficialname().getValue(), DELIM, ENCLOSING));

        }

        // type

        if (metadata.hasDatasourcetype()) {

            buff.append(getStringField(metadata.getDatasourcetype().getClassname().replaceFirst(".*::", ""), DELIM, ENCLOSING));

        }

        // compatibility,

        buff.append(getStringField(metadata.getOpenairecompatibility().getClassname(), DELIM, ENCLOSING));

        // latitude

        buff.append(getLatLongField(metadata.getLatitude().getValue(), DELIM, ENCLOSING));

        // longtitude

        buff.append(getLatLongField(metadata.getLongitude().getValue(), DELIM, ENCLOSING));

        // dateofvalidation,

        buff.append(getStringDateField(metadata.getDateofvalidation().getValue(), DELIM, ENCLOSING));

        // yearofvalidation,

        buff.append(getYearInt(metadata.getDateofvalidation().getValue(), DELIM, ENCLOSING));

        //website

        buff.append(getStringField(metadata.getWebsiteurl().getValue(), DELIM, ENCLOSING));

        //piwik_id

        String piwik_id = "";

        for (String oid : oaf.getEntity().getOriginalIdList()) {

            if (oid.contains("piwik")) {

                piwik_id = oid.split(":")[1];

                break;

            }

        }

        buff.append(getStringField(cleanNumber(piwik_id), DELIM, ENCLOSING));

        //harvested

        buff.append(getStringField("false", DELIM, ENCLOSING));

        // deletedByInference

        buff.append(getStringField(String.valueOf(oaf.getDataInfo().getDeletedbyinference()), DELIM, ENCLOSING));

        // number??

        buff.append(getStringField("1", DELIM, ENCLOSING));

        return buff.toString();

    }

    private static String buildOrganization(Oaf oaf, String DELIM, String ENCLOSING) {

        StringBuilder buff = new StringBuilder();

        Organization.Metadata metadata = oaf.getEntity().getOrganization().getMetadata();

        // `organization_datasources`,

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // organization_projects

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // `name`,

        buff.append(getStringField(metadata.getLegalname().getValue(), DELIM, ENCLOSING));

        // `country`,

        buff.append(getStringField(metadata.getCountry().getClassid(), DELIM, ENCLOSING));

        //website

        buff.append(getStringField(metadata.getWebsiteurl().getValue(), DELIM, ENCLOSING));

        // deletedByInference

        buff.append(getStringField(String.valueOf(oaf.getDataInfo().getDeletedbyinference()), DELIM, ENCLOSING));

        // number

        buff.append(getStringField("1", DELIM, ENCLOSING));

        return buff.toString();

    }

    private static String buildResult(Oaf oaf, String DELIM, String ENCLOSING) {

        StringBuilder buff = new StringBuilder();

        Result.Metadata metadata = oaf.getEntity().getResult().getMetadata();

        // result_topics/

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // result_languages

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // `result_projects`,

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // `result_datasources`,

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // `result_classifications`,

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        /// `result_infrastructures`,

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // `result_claims`,

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // `result_results`,

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // pubtitle

        String titleString = 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", " ");

            }

            break;

        }

        //  pubtitle

        buff.append(getStringField(titleString, DELIM, ENCLOSING));

        //   format

        String formatString = new String();

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

            formatString += format.getValue() + ";";

        }

        buff.append(getStringField(formatString, DELIM, ENCLOSING));

        //  publisher

        buff.append(getStringField(metadata.getPublisher().getValue(), DELIM, ENCLOSING));

        //  journal

        buff.append(getStringField(metadata.getJournal().getName(), DELIM, ENCLOSING));  //#null#!

        // year

        buff.append(getYearInt(metadata.getDateofacceptance().getValue(), DELIM, ENCLOSING));

        // date CHANGED THIS TO DATE FORMAT

        buff.append(getStringDateField(metadata.getDateofacceptance().getValue(), DELIM, ENCLOSING));

        // access_mode,

        buff.append(getStringField(getAccessMode(oaf.getEntity().getResult()), DELIM, ENCLOSING));

        // bestlicense

        buff.append(getStringField(getBestLicense(oaf.getEntity().getResult()), DELIM, ENCLOSING));

        // type

        buff.append(getStringField(metadata.getResulttype().getClassname(), DELIM, ENCLOSING));

        // embargo_end_date

        buff.append(getStringDateField(metadata.getEmbargoenddate().getValue(), DELIM, ENCLOSING));

        // `authors`,

        int authors = 0;

        String delayed = "no";

        for (OafRel rel : oaf.getEntity().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(), DELIM, ENCLOSING);

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

                    delayed = "yes";

                }

            }

        }

        // `delayed`,

        buff.append(getStringField(delayed, DELIM, ENCLOSING));

        //authors

        buff.append(getNumericField(String.valueOf(authors), DELIM, ENCLOSING));

        String sources = new String();

        for (Instance instance : (oaf.getEntity().getResult().getInstanceList())) {

            List<String> urls = instance.getUrlList();

            for (String url : urls) {

                sources += cleanUrl(url, DELIM, ENCLOSING) + " ;";

            }

        }

        //sources

        sources = ENCLOSING + sources + ENCLOSING + DELIM;

        buff.append(sources);

        // deletedByInference

        buff.append(getStringField(String.valueOf(oaf.getDataInfo().getDeletedbyinference()), DELIM, ENCLOSING));

        // number??

        buff.append(getStringField("1", DELIM, ENCLOSING));

        return buff.toString();

    }

    /*

    private static String getBestLicense(Result result) {

        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;

        }

    }

    // TODO here iterate over all values

    private static String getAccessMode(Result result) {

        String accessMode = " ";

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

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

                accessMode = instance.getLicence().getClassname();

                break;

            }

        }

        return accessMode;

    }

    */

    private static String getBestLicense(Result result) {

        Qualifier bestLicense = null;

        LicenseComparator lc = new LicenseComparator();

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

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

                bestLicense = instance.getAccessright();

            }

        }

        if (bestLicense != null) {

            return bestLicense.getClassname();

        } else {

            return null;

        }

    }

    // TODO here iterate over all values

    private static String getAccessMode(Result result) {

        String accessMode = " ";

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

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

                accessMode = instance.getAccessright().getClassname();

                break;

            }

        }

        return accessMode;

    }

    private static String buildProject(Oaf oaf, String DELIM, String ENCLOSING) {

        StringBuilder buff = new StringBuilder();

        Project.Metadata metadata = oaf.getEntity().getProject().getMetadata();

        // project_organizations

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // project_results

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // `acronym`,

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

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

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

        }

        buff.append(getStringField(acronym, DELIM, ENCLOSING));

        //title!

        buff.append(getStringField(metadata.getTitle().getValue(), DELIM, ENCLOSING));

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

        } else {

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

        }

        String sc39 = metadata.getEcsc39().getValue().toString();

        if (sc39.equalsIgnoreCase("true") || sc39.equalsIgnoreCase("t") || sc39.contains("yes")) {

            sc39 = "yes";

        } else if (sc39.equalsIgnoreCase("false") || sc39.equalsIgnoreCase("f") || sc39.contains("no")) {

            sc39 = "no";

        }

        buff.append(getStringField(sc39, DELIM, ENCLOSING));

        //project_type

        buff.append(getStringField(metadata.getContracttype().getClassid(),DELIM, ENCLOSING));

        // `url`,

        buff.append(getStringField(metadata.getWebsiteurl().getValue(), DELIM, ENCLOSING));

        // start_year

        buff.append(getYearInt(metadata.getStartdate().getValue(), DELIM, ENCLOSING));

        // end_year

        buff.append(getYearInt(metadata.getEnddate().getValue(), DELIM, ENCLOSING));

        // duration enddate-startdate

        buff.append(getYearDifferenceInteger(metadata.getEnddate().getValue(), metadata.getStartdate().getValue(), DELIM, ENCLOSING));

        // haspubs

        buff.append(getStringField("no", DELIM, ENCLOSING));

        // numpubs

        buff.append(getNumericField("0", DELIM, ENCLOSING));

        // enddate

        buff.append(getNumericField(metadata.getEnddate().getValue(), DELIM, ENCLOSING));

        // startdate

        buff.append(getNumericField(metadata.getStartdate().getValue(), DELIM, ENCLOSING));

        // `daysforlastpub`,

        buff.append(getNumericField("", DELIM, ENCLOSING));

        // `delayedpubs`,

        buff.append(getNumericField("", DELIM, ENCLOSING));

        //call identifier

        buff.append(getStringField(metadata.getCallidentifier().getValue(), DELIM, ENCLOSING));

        //code

        buff.append(getStringField(metadata.getCode().getValue(), DELIM, ENCLOSING));

        //esc39

        buff.append(getStringField(metadata.getEcsc39().getValue(), DELIM, ENCLOSING));

        //getUrl

        String sources = new String();

        for (Instance instance : (oaf.getEntity().getResult().getInstanceList())) {

            List<String> urls = instance.getUrlList();

            for (String u : urls) {

                sources += u + ";";

            }

        }

        sources = cleanUrl(sources, DELIM, ENCLOSING);

        sources = ENCLOSING + sources + ENCLOSING + DELIM;

        buff.append(sources);

        // deletedByInference

        buff.append(getStringField(String.valueOf(oaf.getDataInfo().getDeletedbyinference()), DELIM, ENCLOSING));

        // `number`

        buff.append(getStringField("1", DELIM, ENCLOSING));

        return buff.toString();

    }

    /*

    private static String buildPerson(Oaf oaf, String DELIM, String ENCLOSING) {

        StringBuilder buff = new StringBuilder();

        PersonProtos.Person.Metadata metadata = oaf.getEntity().getPerson().getMetadata();

        // `person_id`,

        //buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        // person_result

        buff.append(cleanId(oaf.getEntity().getId(), DELIM, ENCLOSING) + DELIM);

        //firstName

        buff.append(getStringField(metadata.getFirstname().getValue(), DELIM, ENCLOSING));

        //LastName

        String secondName = new String();

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

            secondName += s.getValue().replace("\n", " ").replace(",", " ") + " ";

        }

        buff.append(getStringField(secondName, DELIM, ENCLOSING));

        // `fullname`,

        buff.append(getStringField(metadata.getFullname().getValue(), DELIM, ENCLOSING));

        // `Nationality`,

        buff.append(getStringField(metadata.getNationality().getClassid(), DELIM, ENCLOSING));

        // `Email`,

        buff.append(getStringField(metadata.getEmail().getValue(), DELIM, ENCLOSING));

        // `Phone`,

        buff.append(getStringField(metadata.getPhone().getValue(), DELIM, ENCLOSING));

        //deletedByInference

        buff.append(getStringField(String.valueOf(oaf.getDataInfo().getDeletedbyinference()), DELIM, ENCLOSING));

        // `number`,

        buff.append(getStringField("1", DELIM, ENCLOSING));

        return buff.toString();

    }

    */

    private static String getYearDifferenceInteger(String enddate, String startdate, String DELIM, String ENCLOSING) {

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

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

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

                return ENCLOSING + "0" + ENCLOSING + DELIM;

            }

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

            split = enddate.split("-");

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

                return ENCLOSING + "0" + ENCLOSING + DELIM;

            }

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

            int diff = Enddate - Startdate;

            return ENCLOSING + diff + ENCLOSING + DELIM;

        }

        return ENCLOSING + "0" + ENCLOSING + DELIM;

    }

    private static String getYearInt(String data, String DELIM, String ENCLOSING) {

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

            return ENCLOSING + "0" + ENCLOSING + DELIM;

        }

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

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

            return ENCLOSING + "0" + ENCLOSING + DELIM;

        }

        String year = split[0];

        year = cleanNumber(year);

        if (year == null || year.isEmpty()) year = "0";

        return ENCLOSING + year + ENCLOSING + DELIM;

    }

    private static String cleanNumber(String number) {

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

        return number;

    }

    private static String getLatLongField(String data, String DELIM, String ENCLOSING) {

        if (data == null || data.isEmpty()) return ENCLOSING + "null" + ENCLOSING + DELIM;