D-Net

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

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import org.apache.log4j.Logger;

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

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

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

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

/**

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

	private String NULL_NUM;

	private static String ENCLOSED;

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

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

		default:

			return buildRel(oaf);

		}

	}

	private String buildRel(OafRel Rel) {

		return getStringField(Rel.getTarget());

	}

	public HashMap<String, List<String>> extractRelations(Oaf oaf) {

		OafEntity valueEntity = oaf.getEntity();

		switch (valueEntity.getType()) {

		case datasource:

			return getDatasourceLanguages(valueEntity);

		case result:

			HashMap<String, List<String>> relations = new HashMap<String, List<String>>();

			// relations.putAll(getResultLanguages(valueEntity));

			relations.putAll(getResultTopics(valueEntity));

			// relations.putAll(getResultClassifications(valueEntity));

			// relations.putAll(getResultDatasources(valueEntity));

			relations.putAll(getResultConcepts(valueEntity));

			return relations;

		default:

			return null;

		}

	}

	private String buildresultProject(OafRel oaf) {

		String buff = new String();

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

		// TODO is declared as int!!!

		buff += getYearDifferenceInteger(oaf.getResultProject().getOutcome().getRelMetadata().getEnddate(), oaf.getResultProject().getOutcome().getRelMetadata().getStartdate());

		return buff;

	}

	private HashMap<String, List<String>> getDatasourceLanguages(OafEntity valueEntity) {

		HashMap<String, List<String>> rels = new HashMap<String, List<String>>();

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

		Datasource d = valueEntity.getDatasource();

		Metadata metadata = d.getMetadata();

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

			buffs.add(getStringField(lang.getValue()));

		}

		rels.put("datasourceLanguage", buffs);

		return rels;

	}

	private HashMap<String, List<String>> getResultLanguages(OafEntity valueEntity) {

		HashMap<String, List<String>> rels = new HashMap<String, List<String>>();

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

		Result d = valueEntity.getResult();

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

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

			buffs.add(getStringField(metadata.getLanguage().getClassname()));

		}

		rels.put("resultLanguage", buffs);

		return rels;

	}

	private HashMap<String, List<String>> getResultClassifications(OafEntity valueEntity) {

		HashMap<String, List<String>> rels = new HashMap<String, List<String>>();

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

		Result result = valueEntity.getResult();

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

			buffs.add(getStringField(instance.getInstancetype().getClassname()));

		}

		rels.put("resultClassification", buffs);

		return rels;

	}

	private HashMap<String, List<String>> getResultConcepts(OafEntity valueEntity) {

		HashMap<String, List<String>> rels = new HashMap<String, List<String>>();

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

		Result result = valueEntity.getResult();

		for (eu.dnetlib.data.proto.ResultProtos.Result.Context context : result.getMetadata().getContextList()) {

			buffs.add(getStringField(context.getId()));

		}

		rels.put("resultConcept", buffs);

		return rels;

	}

	private HashMap<String, List<String>> getResultDatasources(OafEntity valueEntity) {

		HashMap<String, List<String>> rels = new HashMap<String, List<String>>();

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

		Result result = valueEntity.getResult();

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

			String hostedBy = null;

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

			buffs.add((getStringField(hostedBy)));

		}

		rels.put("resultDatasource", buffs);

		return rels;

	}

	public static 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.

	}

	// TODO there are topics with "null" as value -> repalce 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 HashMap<String, List<String>> getResultTopics(OafEntity valueEntity) {

		HashMap<String, List<String>> rels = new HashMap<String, List<String>>();

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

		Result d = valueEntity.getResult();

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

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

		for (StructuredProperty topic : Topics) {

			// TODOs

			if (topic.getValue() != null && !topic.getValue().isEmpty() && !topic.getValue().equals("") && !topic.getValue().equals(" ") && !topic.getValue().equals("null")) {

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

					String t = getStringField(topic.getValue());

					if (isValidTopic(t)) {

						buffs.add(t);

					}

				}

			}

		}

		rels.put("resultTopic", buffs);

		return rels;

	}

	private String buildDatasource(OafEntity data) {

		String buff = new String();

		Datasource d = data.getDatasource();

		Metadata metadata = d.getMetadata();

		String full_id = getStringField(data.getId());

		buff += full_id;

		buff += full_id;

		buff += full_id;

		buff += full_id;

		// TODO move this here???

		// UPDATE "shadow".datasource SET name='Other' where name='Unknown

		// Repository';

		// name

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

			buff += getStringField("Unknown Repository");

		} else {

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

		}

		// type

		if (metadata.hasDatasourcetype())

		{

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

		} else {

			buff += getStringField(null);

		}

		// compatibility,

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

		// latitude

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

		// longtitude

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

		// dateofvalidation,

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

		// yearofvalidation,

		// parse year of validation

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

		// number??

		buff += getStringField("1");

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

		// `organization_datasources`,

		String full_id = getStringField(data.getId());

		buff += full_id;

		// organization_projects

		buff += full_id;

		// `name`,

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

		// `country`,

		if (metadata.getCountry().getClassname().equals("UNITED KINGDOM"))

		{

			buff += getStringField("United Kingdom");

		} else if (metadata.getCountry().getClassname().equals("GREECE")) {

			buff += getStringField("Greece");

		} else

		{

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

		}

		// `number`,

		buff += getStringField("1");

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

		// result_topics/

		String full_id = getStringField(data.getId());

		buff += full_id;

		// result_languages

		buff += full_id;

		// `result_projects`,

		buff += full_id;

		// `result_datasources`,

		buff += full_id;

		// `result_classifications`,

		buff += full_id;

		// / `result_infrastructures`,

		buff += full_id;

		// `result_claims`,

		buff += full_id;

		// `result_results`,

		buff += full_id;

		// year

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

		// date

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

		// access_mode,

		buff += getStringField(getAcessMode(result));

		// bestlicense

		buff += getStringField(getBestLicense(result));

		// type

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

		// embargo_end_date

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

		// `delayed`,

		buff += getStringField("no");

		// `authors`,

		int authors = 0;

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

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

				authors++;

			}

		}

		// log.info("Result " + full_id +"Author count : " + authors );

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

		// number??

		buff += getStringField("1");

		// TODO this is a rotten record.//

		// if

		// (full_id.contentEquals("datacite____::5903fa1fba477116aba4a922cc9eb56f"))

		// {

		// log.error("ROTTEN RECORD " + buff);

		// return null;

		// }

		// TODO check if valid or empty protobuff

		// TODO do it in all protos?

		if (isValid(buff, full_id)) {

			return buff;

		} else {

			return null;

		}

	}

	private boolean isValid(String buff, String id) {

		if (buff.endsWith(ENCLOSED)) {

			log.error("Empty Result with  " + id + " with body: \n" + buff);

			return false;

		}

		return true;

	}

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

		}

	}

	private String getAcessMode(Result result) {

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

			return instance.getLicence().getClassname();

		}

		return NULL_STRING;

	}

	private String buildProject(OafEntity data) {

		String buff = new String();

		Project project = data.getProject();

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

		// project_organizations

		String full_id = getStringField(data.getId());

		buff += full_id;

		// project_results

		buff += full_id;

		// `acronym`,

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

		// `funding_lvl0`,

		String funding_tree_0 = NULL_STRING;

		String funding_tree_1 = NULL_STRING;

		String funding_tree_2 = NULL_STRING;

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

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

		{

			funding_tree_0 = getFundingLevel(fundList.get(0).getValue(), 0);

			funding_tree_1 = getFundingLevel(fundList.get(0).getValue(), 1);

			// log.info(funding_tree_1);

			funding_tree_2 = getFundingLevel(fundList.get(0).getValue(), 2);

			// log.info(funding_tree_2);

		}

		buff += getStringField(funding_tree_0);

		// `funding_lvl1`,

		if (funding_tree_1.equalsIgnoreCase("SP1")) {

			funding_tree_1 = "SP1-Cooperation";

		} else if (funding_tree_1.equalsIgnoreCase("SP2")) {

			funding_tree_1 = "SP2-Ideas";

		} else if (funding_tree_1.equalsIgnoreCase("SP3")) {

			funding_tree_1 = "SP3-People";

		} else if (funding_tree_1.equalsIgnoreCase("SP4")) {

			funding_tree_1 = "SP4-Capacities";

		} else if (funding_tree_1.equalsIgnoreCase("SP5")) {

			funding_tree_1 = "SP5-Euratom";

		}

		buff += getStringField(funding_tree_1);

		// / `funding_lvl2`,

		buff += getStringField(funding_tree_2);

		// `sc39`,

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

		if (sc39.contains("true")) {

			sc39 = "yes";

		} else if (sc39.contains("false")) {

			sc39 = "no";

		}

		buff += getStringField(sc39);

		// `url`,

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

		// start_year

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

		// end_year

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

		// duration enddate-startdate

		buff += getYearDifferenceInteger(metadata.getEnddate().getValue(), metadata.getStartdate().getValue());

		// haspubs

		buff += getStringField("");

		// numpubs

		buff += getNumericField("");

		// enddate

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

		// startdate

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

		// `daysforlastpub`,

		buff += getNumericField("");

		// `delayedpubs`,

		buff += getNumericField("");

		// `number`

		buff += getStringField("1");

		return buff;

	}

	private String getFundingLevel(String funding_level, int level) {

		if (funding_level.isEmpty()) {

			return NULL_STRING;

		}

		if (!funding_level.contains("funding_level_" + level)) {

			return NULL_STRING;

		}

		String[] split = funding_level.split("funding_level_" + level);

		funding_level = split[1];

		split = funding_level.split("name");

		split = split[1].split(",");

		funding_level = split[0].replaceAll(".*:\"", "");

		funding_level = funding_level.replaceFirst(ENCLOSED, "");

		funding_level = funding_level.trim();

		return funding_level;

	}

	private String getYearDifferenceInteger(String enddate, String startdate) {

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

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

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

			split = enddate.split("-");

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

			return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

		}

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

		if (split != null) {

			String year = split[0];

			year = cleanNumber(year);

			return ENCLOSED + year + ENCLOSED + DELIM;

		} else {

			return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

		}

	}

	private String cleanNumber(String number) {

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

		number.trim();

		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 getNumericField(String data) {

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

			return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

		} else {

			return ENCLOSED + data + ENCLOSED + DELIM;

		}

	}

	public String getId(Oaf oaf) {

		switch (oaf.getKind()) {

		case entity:

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

		case relation:

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

		}

		return null;

	}

	public String getId(OafRel relOaf) {

		return cleanId(relOaf.getSource());

	}

	public static String clean(String value) {

		if (value != null) {

			// TODO 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("[^A-Za-z0-9:,____]", " ");

			value = value.trim();

		}

		if (value == null) {

			return null;

		}

		return ENCLOSED + value + ENCLOSED;

	}

	public static String cleanId(String value) {

		if (value != null) {

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

	}

	public String getName(Oaf oaf) {

		switch (oaf.getKind()) {

		case entity:

			return oaf.getEntity().getType().name();

		case relation:

			return oaf.getRel().getSource() + oaf.getRel().getTarget();

		}

		return null;

	}

	public String getDELIM() {

		return DELIM;

	}

	public void setDELIM(String dELIM) {

		DELIM = dELIM;

	}

	public String getNULL_STRING() {

		return NULL_STRING;

	}

	public void setNULL_STRING(String nULL_STRING) {

		NULL_STRING = nULL_STRING;

	}

	public String getNULL_NUM() {

		return NULL_NUM;

	}

	public void setNULL_NUM(String nULL_NUM) {

		NULL_NUM = nULL_NUM;

	}

	public static String getENCLOSED() {

		return ENCLOSED;

	}

	public void setENCLOSED(String eNCLOSED) {

		ENCLOSED = eNCLOSED;

	}

}