D-Net

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

import org.apache.commons.io.IOUtils;

import javax.xml.transform.Transformer;

import javax.xml.transform.TransformerFactory;

import javax.xml.transform.stream.StreamResult;

import javax.xml.transform.stream.StreamSource;

import java.io.ByteArrayInputStream;

import java.io.ByteArrayOutputStream;

import java.io.InputStream;

import java.nio.charset.StandardCharsets;

public class ContextTransformer {

    public String transformXSL(String xml) throws Exception {

        Transformer transformer;

        TransformerFactory tFactory = TransformerFactory.newInstance();

        if (xml == null) {

            throw new IllegalArgumentException("Input xml should not be null!");

        }

        InputStream inputStream = null;

        ByteArrayInputStream readerStream = null;

        ByteArrayOutputStream writerStream = null;

        try {

            inputStream = ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/" + "context.xsl");

            transformer = tFactory.newTransformer(new StreamSource(inputStream));

            readerStream = new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));

            writerStream = new ByteArrayOutputStream();

            transformer.transform(new StreamSource(readerStream), new StreamResult(writerStream));

            return writerStream.toString("UTF8");

        } finally {

            IOUtils.closeQuietly(inputStream);

            IOUtils.closeQuietly(readerStream);

            IOUtils.closeQuietly(writerStream);

        }

    }

}

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

import org.apache.log4j.Logger;

/**

 * Created by envy17 j110ea on 4/5/2015.

 */

//Extract funder info for projects

public class FundingParser {

	private String DELIM;

	private String ENCLOSING;

	public FundingParser(String DELIM, String ENCLOSING) {

		this.DELIM = DELIM;

		this.ENCLOSING = ENCLOSING;

	}

	public String getFundingLevel(String funding_level, int level) {

		if (funding_level.isEmpty()) {

			return ENCLOSING + " " + ENCLOSING + DELIM;

		}

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

			return ENCLOSING + " " + ENCLOSING + DELIM;

		}

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

		funding_level = split[1];

		split = funding_level.split("<name>");

		funding_level = split[1];

		funding_level = funding_level.substring(0, funding_level.indexOf("</name>"));

		funding_level = funding_level.replaceAll("\"", "");

		funding_level = funding_level.replaceAll("/>", "");

		funding_level = funding_level.replaceAll("<", "");

		funding_level = funding_level.replaceAll("&", "");

		if (level == 1) {

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

				funding_level = "SP1-Cooperation";

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

				funding_level = "SP2-Ideas";

			}

			if (funding_level.equalsIgnoreCase("SP3")) {

				funding_level = "SP3-People";

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

				funding_level = "SP4-Capacities";

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

				funding_level = "SP5-Euratom";

			}

		}

		funding_level = funding_level.replaceAll(">", "");

		funding_level = funding_level.replaceAll("</", "");

		funding_level = funding_level.replace(DELIM, " ");

		funding_level = funding_level.replace(ENCLOSING, " ");

		return ENCLOSING + funding_level + ENCLOSING + DELIM;

	}

	public String getFundingInfo(String buff) {

		return getFunder(buff) + getFundingLevel(buff, 0) + (getFundingLevel(buff, 1) + getFundingLevel(buff, 2)

				+ getFundingLevel(buff, 3));

	}

	public String getFunder(String buff) {

		if (buff.isEmpty()) {

			return ENCLOSING + " " + ENCLOSING + DELIM;

		}

		if (!buff.contains("<funder>")) {

			return ENCLOSING + " " + ENCLOSING + DELIM;

		}

		String[] split = buff.split("<funder>");

		String funder = split[1];

		split = funder.split("<name>");

		funder = split[1];

		funder = funder.substring(0, funder.indexOf("</name>"));

		funder = funder.replaceAll(">", "");

		funder = funder.replaceAll("</", "");

		funder = funder.replaceAll("\"", "");

		funder = funder.replaceAll("&", "");

		funder = funder.replace(ENCLOSING, " ");

		return ENCLOSING + funder + ENCLOSING + DELIM;

	}

}

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.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 java.text.DateFormat;

import java.text.ParseException;

import java.text.SimpleDateFormat;

import java.util.Date;

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 {

    private static Logger logger = Logger.getLogger(Serializer.class);

    private String DELIM;

    private String ENCLOSING;

    public Serializer(String DELIM, String ENCLOSING) {

        this.DELIM = DELIM;

        this.ENCLOSING = ENCLOSING;

    }

    public String serialize(Oaf oaf) {

        switch (oaf.getKind()) {

            case entity:

                OafEntity valueEntity = oaf.getEntity();

                switch (valueEntity.getType()) {

                    case datasource:

                        return buildDatasource(oaf);

                    case organization:

                        return buildOrganization(oaf);

                    case project:

                        return buildProject(oaf);

                    case result:

                        return buildResult(oaf);

                    default:

                        break;

                }

                break;

            case relation:

                return buildRel(oaf.getRel());

        }

        return null;

    }

    public String serialize(OafRel oaf) {

        switch (oaf.getRelType()) {

            case resultProject:

                return getResultProject(oaf);

            default:

                return buildRel(oaf);

        }

    }

    private String buildRel(OafRel Rel) {

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

    }

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

        OafEntity valueEntity = oaf.getEntity();

        getOriginalId(valueEntity, relations);

        switch (valueEntity.getType()) {

            case datasource:

                getDatasourceLanguages(valueEntity, relations);

                break;

            case result:

                getResultTopics(valueEntity, relations);

                getResultLanguages(valueEntity, relations);

                getResultClassifications(valueEntity, relations);

                getResultDatasources(valueEntity, relations);

                getResultConcepts(valueEntity, relations);

                getResultDois(valueEntity, relations);

                getResultCitations(valueEntity, relations);

                break;

            case project:

                getProjectKeywords(valueEntity, relations);

                getProjectSubjects(valueEntity, relations);

                break;

        }

    }

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

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

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

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

        }

    }

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

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

    }

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

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

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

        }

    }

    private String getResultProject(OafRel oaf) {

        StringBuilder buff = new StringBuilder();

        buff.append(cleanId(oaf.getTarget())).append(DELIM);

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

        return buff.toString();

    }

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

        Datasource d = valueEntity.getDatasource();

        Metadata metadata = d.getMetadata();

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

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

        }

    }

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

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

        }

    }

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

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

            rels.put("resultPid", getStringField(pid.getQualifier().getClassname()) + getStringField(pid.getValue()));

        }

    }

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

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

            }

        }

    }

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

        Result result = valueEntity.getResult();

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

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

        }

    }

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

        Result result = valueEntity.getResult();

    // hosted by

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

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

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

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

            }

        }

    // collected from

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

            String collectedFrom = collectedFromValue.getKey();

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

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

            }

        }

    }

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

        Result d = valueEntity.getResult();

        Result.Metadata metadata = d.getMetadata();

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

        for (StructuredProperty topic : Topics) {

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

        }

    }

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

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

                    logger.error("Error getting result citations", e);

                }

            }

        }

    }

    private String buildDatasource(Oaf oaf) {

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

        StringBuilder buff = new StringBuilder();

        // name

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

            buff.append(getStringField("Unknown Repository"));

        } else {

            buff.append(getStringField(metadata.getOfficialname().getValue()));

        }

        // type

        if (metadata.hasDatasourcetype()) {

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

        }

        // compatibility,

        buff.append(getStringField(metadata.getOpenairecompatibility().getClassname()));

        // latitude

        buff.append(getLatLongField(metadata.getLatitude().getValue()));

        // longtitude

        buff.append(getLatLongField(metadata.getLongitude().getValue()));

        // dateofvalidation,

        buff.append(getStringDateField(metadata.getDateofvalidation().getValue()));

        // yearofvalidation,

        buff.append(getYearInt(metadata.getDateofvalidation().getValue()));

        //harvested

        buff.append(getStringField("false"));

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

        buff.append(getStringField(metadata.getWebsiteurl().getValue()));

        return buff.toString();

    }

    private String buildOrganization(Oaf oaf) {

        StringBuilder buff = new StringBuilder();

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

        // `name`,

        buff.append(getStringField(metadata.getLegalname().getValue()));

        // `country`,

        buff.append(getStringField(metadata.getCountry().getClassid()));

        return buff.toString();

    }

    private String buildResult(Oaf oaf) {

        StringBuilder buff = new StringBuilder();

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

        // originalId

        buff.append(getId(oaf)).append(DELIM);

        String titleString = "";

        if (metadata.getTitleList().size() > 0) {

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

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

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

        }

        //  pubtitle

        buff.append(getStringField(titleString));

        //  publisher

        buff.append(getStringField(metadata.getPublisher().getValue()));

        //  journal

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

        // year

        buff.append(getYearInt(metadata.getDateofacceptance().getValue()));

        // date

        buff.append(getStringDateField(metadata.getDateofacceptance().getValue()));

        // bestlicense

        buff.append(getStringField(getBestLicense(oaf.getEntity().getResult())));

        // type

        buff.append(getStringField(metadata.getResulttype().getClassname()));

        // embargo_end_date

        buff.append(getStringDateField(metadata.getEmbargoenddate().getValue()));

        // `authors`,

        int authors = metadata.getAuthorCount();

        String delayed = "no";

        for (OafRel rel : oaf.getEntity().getCachedRelList()) {

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

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

                }

            }

        }

        // `delayed`,

        buff.append(getStringField(delayed));

        //authors

        buff.append(getNumericField(String.valueOf(authors)));

        String authorNames = "";

        for (FieldTypeProtos.Author author:metadata.getAuthorList()) {

            authorNames += author.getFullname() + ";";

        }

        buff.append(getStringField(authorNames));

        String sources = "";

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

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

            for (String url : urls) {

                sources += cleanUrl(url) + " ;";

            }

        }

        //sources

        sources = ENCLOSING + sources + ENCLOSING + DELIM;

        buff.append(sources);

        boolean hasAbstract = false;

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

            if (desc != null && desc.getValue() != null && !desc.getValue().trim().isEmpty())

                hasAbstract = true;

        }

        buff.append(getStringField(Boolean.toString(hasAbstract)));

        return buff.toString();

    }

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

        }

    }

    private String buildProject(Oaf oaf) {

        FundingParser fundingParser = new FundingParser(DELIM, ENCLOSING);

        StringBuilder buff = new StringBuilder();

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

        // `acronym`,

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

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

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

        }

        buff.append(getStringField(acronym));

        //title

        buff.append(getStringField(metadata.getTitle().getValue()));

        //funding_lvl

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

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

        {

            //funder + 3 funding levels

            buff.append(fundingParser.getFundingInfo(fundList.get(0).getValue()));

        } else {

            buff.append(fundingParser.getFundingInfo(""));

        }

        //sc39

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

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

        //project_type

        buff.append(getStringField(metadata.getContracttype().getClassid()));

        // start_year

        buff.append(getYearInt(metadata.getStartdate().getValue()));

        // end_year

        buff.append(getYearInt(metadata.getEnddate().getValue()));

        // duration enddate-startdate

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

        // haspubs

        buff.append(getStringField("no"));

        // numpubs

        buff.append(getNumericField("0"));

        // enddate

        buff.append(getStringDateField(metadata.getEnddate().getValue()));

        // startdate

        buff.append(getStringDateField(metadata.getStartdate().getValue()));

        // `daysforlastpub`,

        buff.append(getNumericField(""));

        // `delayedpubs`,

        buff.append(getNumericField(""));

        //call identifier

        buff.append(getStringField(metadata.getCallidentifier().getValue()));

        //code

        buff.append(getStringField(metadata.getCode().getValue()));

        return buff.toString();

    }

    private String getYearDifferenceInteger(String enddate, String startdate) {

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

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

            if (split.length == 0) {

                return ENCLOSING + "0" + ENCLOSING + DELIM;

            }

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

            split = enddate.split("-");

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

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

            return ENCLOSING + "0" + ENCLOSING + DELIM;

        }

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

        if (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 String cleanNumber(String number) {

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

        return number;

    }

    private String getLatLongField(String data) {

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

            return ENCLOSING + "null" + ENCLOSING + DELIM;

        return ENCLOSING + data.replaceAll("[^-0-9.]+", "")  + ENCLOSING + DELIM;

    }

    private String getStringField(String data) {

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

            return ENCLOSING + "null" + ENCLOSING + DELIM;

        return ENCLOSING + clean(data) + ENCLOSING + DELIM;

    }

    private String getStringDateField(String data) {

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

            return ENCLOSING + "0" + ENCLOSING + DELIM;

        } else {

            data = data.replace(DELIM, " ");

            data = data.replace(ENCLOSING, " ");

            data = data.replaceAll("\\r\\n|\\r|\\n", "");

            try {

                DateFormat format = new SimpleDateFormat("yyyy-MM-dd");

                data = format.format(format.parse(data));

                return ENCLOSING + data + ENCLOSING + DELIM;

            } catch (ParseException e) {

                return ENCLOSING + "0" + ENCLOSING + DELIM;

            }

        }

    }

    private String getNumericField(String data) {

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

            return ENCLOSING + "0" + ENCLOSING + DELIM;

        } else {

            return ENCLOSING + data + ENCLOSING + 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());

    }

    private String clean(String value) {

        if (value != null) {

            value = value.replaceAll("[\"\\r\\\\;]", "");

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

            value = value.replace(ENCLOSING, " ");

            value = value.replaceAll("\\r\\n|\\r|\\n", " ");

            return value;

        } else {

            return "";

        }

    }

    private String cleanId(String value) {

        if (value != null) {

            // DO NOT CHANGE THIS: IT REMOVES ID PREFIX ( "5|datacite____::" to "datacite____::")

            // AND REPLACES OCCURRENCES OF DELIM CHARS IN DATA

            value = value.replaceFirst(".*\\|", "");

            value = value.replace("\n", "");

            value = value.replace(ENCLOSING, "");

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

            value = value.replace("\"", "");

            value = value.replace("«", " ");

            value = value.replace("»", " ");

        }

        return ENCLOSING + value + ENCLOSING;

    }

    private String cleanUrl(String value) {

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

        value = value.replace(ENCLOSING, " ");

        value = value.replace(" ", "");

        value = value.replace("\n", "");

        value = value.replace("\r", "");

        value = value.replace("\\n", "");

        value = value.replace("\\r", "");

        return value;

    }

    private long DATEDIFF(String startDate, String endDate) {

        long MILLISECS_PER_DAY = 24 * 60 * 60 * 1000L;

        long days;

        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd"); // "dd/MM/yyyy HH:mm:ss");

        // <startdate>2011-09-01</startdate>

        // <enddate>2015-08-31</enddate>

        Date dateIni;

        Date dateFin;

        if (startDate == null || startDate.isEmpty() || endDate == null || endDate.isEmpty()) {

            return 0;

        }

        try {

            dateIni = format.parse(startDate);

            dateFin = format.parse(endDate);

            days = (dateFin.getTime() - dateIni.getTime()) / MILLISECS_PER_DAY;

        } catch (Exception e) {

            return 0;

        }

        return days;

    }

}

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

import com.google.common.collect.ArrayListMultimap;

import com.google.common.collect.Multimap;

import eu.dnetlib.data.mapreduce.hbase.index.config.EntityConfigTable;

import eu.dnetlib.data.mapreduce.hbase.index.config.IndexConfig;

import eu.dnetlib.data.mapreduce.hbase.index.config.LinkDescriptor;

import eu.dnetlib.data.mapreduce.hbase.statsExport.utils.Serializer;

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

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

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

import eu.dnetlib.data.proto.KindProtos;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.OafProtos.OafRel;

import eu.dnetlib.data.proto.OafProtos.OafRelOrBuilder;

import eu.dnetlib.data.proto.TypeProtos;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import org.apache.log4j.Logger;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

/**

 * Mapper Class that reads HBASE contents and prepares them for the StatsDB

 * export

 */

public class StatsMapper extends TableMapper<Text, ImmutableBytesWritable> {

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

    //Index configuratios for OAFs

    private EntityConfigTable entityConfigTable;

    //Counters for produced entities. Visible in the JobTracker monitoring page.

    private enum STATS_COUNTERS {

        datasource,

        organization,

        result,

        person,

        project

    }

    private enum REL_COUNTERS {

        resultProject,

        datasourceOrganization,

        organizationOrganization,

        resultResult,

        personPerson,

        personResult,

        projectOrganization,

        projectPerson,

        resultOrganization

    }

    //Counter for corrupted records.

    private enum ERROR_COUNTERS {

        rottenRecords,

        rottenRelations

    }

    private Serializer serializer;

    private int id;

    private int increment;

    //Init class: Load Index config and mapping for DB tables.

    @Override

    protected void setup(Context context) {

        loadEntityConfig(context);

        this.serializer = new Serializer(context.getConfiguration().get("stats.delim"), context.getConfiguration().get("stats.enclChar"));

        id = context.getTaskAttemptID().getTaskID().getId();

        increment = context.getConfiguration().getInt("mapred.map.tasks", 0);

        if (increment == 0) {

            throw new IllegalArgumentException("mapred.map.tasks is zero");

        }

    }

    //Read HBASE table and decode Protos to OAF entities.

    @Override

    protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) {

        // generating unique integer if for the entity.

        id += increment;

        try {

            OafRowKeyDecoder keyDecoder = OafRowKeyDecoder.decode(keyIn.copyBytes());

            Oaf oaf = UpdateMerger.mergeBodyUpdates(context, result.getFamilyMap(Bytes.toBytes(keyDecoder.getType().toString())));

            if (isValid(oaf))

                emitProtos(context, result, oaf);

        } catch (Exception e) {

            log.error("Unable to parse proto in row: " + new String(keyIn.copyBytes()), e);

            context.getCounter(ERROR_COUNTERS.rottenRecords).increment(1);

        }

    }

    private boolean isValid(Oaf oaf) {

        return oaf != null && oaf.isInitialized() && !oaf.getDataInfo().getDeletedbyinference() && !oaf.getDataInfo().getInvisible();

    }

    private void emitProtos(Context context, Result result, Oaf oaf) {

        Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(oaf.getKind()).

                setDataInfo(oaf.getDataInfo()).setLastupdatetimestamp(oaf.getLastupdatetimestamp());

        oafBuilder.setEntity(oaf.getEntity());

        // emit relation first so we can cache them to entity protos

        emitRelation(context, result, oaf);

        emitEntity(context, oafBuilder.build());

    }

    private void emitEntity(Context context, Oaf oaf) {

        String serialized = serializer.serialize(oaf);

        if (serialized != null) {

                try {

                    Text TextKeyOut;

                    switch (oaf.getEntity().getType()) {

                        case project:

                        case datasource:

                        case organization:

                            TextKeyOut = new Text(oaf.getEntity().getType().toString() + "," + serializer.getId(oaf));

                            break;

                        case result:

                            default:

                            TextKeyOut = new Text(oaf.getEntity().getType().toString() + "," + id);

                            break;

                    }

                    context.write(TextKeyOut, new ImmutableBytesWritable(serialized.getBytes()));

                    context.getCounter(STATS_COUNTERS.valueOf(oaf.getEntity().getType().toString())).increment(1);

                } catch (Exception e) {

                    log.error("Error writing entity to M/R output", e);

                }

            }

    }

    // may have multiple relations per each field

    private void emitRelation(Context context, Result result, Oaf oaf) {

        try {

            // writing out : personResult,ID,data

            // derived relations; ResultLanguages, resultConcepts etc are

            // created here

            Multimap<String, String> relMap = ArrayListMultimap.create();

            serializer.extractRelations(oaf, relMap);

            if (!relMap.isEmpty()) {

                for (Entry<String, String> rel : relMap.entries()) {

                    Text TextKeyOut;

                    switch (oaf.getEntity().getType()) {

                        case project:

                        case datasource:

                        case organization:

                            TextKeyOut = new Text(rel.getKey() + "," + serializer.getId(oaf));

                            break;

                        case result:

                        default:

                            TextKeyOut = new Text(rel.getKey() + "," + id);

                            break;

                    }

                    context.write((TextKeyOut), new ImmutableBytesWritable(rel.getValue().getBytes()));

                }

            }

        } catch (Throwable e) {

            log.error("Error writing relation to M/R output", e);

        }

        // Existing Hbase relations are generated here

        if (entityConfigTable.getDescriptors(oaf.getEntity().getType()) != null) {

            for (LinkDescriptor ld : entityConfigTable.getDescriptors(oaf.getEntity().getType())) {

                try {

                    ArrayList<OafRel> oafRels = new ArrayList<OafRel>();

                    decodeRelation(oaf, context, result, ld, oafRels);

                    for (OafRel rel : oafRels) {

                        {

                            Text TextKeyOut = new Text(ld.getRelDescriptor().getRelType().toString() + "," + serializer.getId(rel));

                            String buff = serializer.serialize(rel);

                            context.write((TextKeyOut), new ImmutableBytesWritable(buff.getBytes()));

                            context.getCounter(REL_COUNTERS.valueOf(rel.getRelType().toString())).increment(1);

                        }

                    }

                } catch (Throwable e) {

                    log.error("Error for record ", e);

                    context.getCounter(ERROR_COUNTERS.rottenRelations).increment(1);

                }

            }

        }

    }

    private void decodeRelation(final Oaf body, final Context context, Result result, final LinkDescriptor ld, List<OafRel> rels) {

        try {

            final Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(KindProtos.Kind.relation);

            final Map<byte[], byte[]> columnMap = result.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));

            if (hasData(columnMap)) {

                for (Entry<byte[], byte[]> e : columnMap.entrySet()) {

                    Oaf decodedOaf = decodeProto(e.getValue(), context);

                    OafRel.Builder relBuilder = OafRel.newBuilder(decodedOaf.getRel());

                    if (ld.isSymmetric()) {

                        RelDescriptor rd = ld.getRelDescriptor();

                        relBuilder.setCachedTarget(body.getEntity()).setRelType(rd.getRelType()).setSubRelType(rd.getSubRelType());

                    }

                    OafRel oafRel = relBuilder.setChild(ld.isChild()).build();

                    rels.add(oafBuilder.setDataInfo(decodedOaf.getDataInfo()).setRel(oafRel).build().getRel());

                }

            }

        } catch (Throwable throwable) {

            log.error("Error Decoding relation for: " + body.getRel().getRelType() + " " + body.getEntity().getId() + " ", throwable);

            context.getCounter(ERROR_COUNTERS.rottenRelations).increment(1);

        }

    }

    private Oaf decodeProto(final byte[] body, Context context) {

        try {

            return Oaf.parseFrom(body);

        } catch (Exception e) {

            log.error(e);

            context.getCounter(ERROR_COUNTERS.rottenRecords).increment(1);

        }

        return null;

    }

    private void loadEntityConfig(Context context) {

        String indexConf = context.getConfiguration().get("stats.indexConf");

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

            log.error("NULL ENTITY CONFIGURATION TABLE IN MAPPER  : ");

        }

        entityConfigTable = IndexConfig.load(indexConf).getConfigMap();

    }

    private boolean isDedupSelf(final OafRelOrBuilder rel) {

        return rel.getSource().contains(rel.getTarget());

    }

    private boolean hasData(final Map<byte[], byte[]> columnMap) {

        return columnMap != null && !columnMap.isEmpty();

    }

    public EntityConfigTable getEntityConfigTable() {

        return entityConfigTable;

    }

    public void setEntityConfigTable(EntityConfigTable entityConfigTable) {

        this.entityConfigTable = entityConfigTable;

    }

}

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

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.TaskInputOutputContext;

import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import org.apache.log4j.Logger;

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.Iterator;

public class StatsReducer extends Reducer<Text, ImmutableBytesWritable, Text, Text> {

	private Logger log = Logger.getLogger(StatsReducer.class);

	private MultipleOutputs MultipleOutputWriter;

	/**

	 * Reducer that splits input according to their Type ( datasource, results

	 * etc..) and writes each kind to a seperate output

	 */

	@Override

	protected void setup(Context context) throws IOException, InterruptedException {

		MultipleOutputWriter = new MultipleOutputs((TaskInputOutputContext) context);

	};

	@Override

	protected void reduce(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {

		Iterator<ImmutableBytesWritable> it = values.iterator();

		while (it.hasNext()) {

			String[] split = key.toString().split(",");

			String type = split[0];

			String id = split[1];

			String value = new String(it.next().copyBytes(),Charset.forName("UTF-8"));

            value = value.trim();

			MultipleOutputWriter.write(type, new Text(id.getBytes(Charset.forName("UTF-8"))), new Text(value.getBytes(Charset.forName("UTF-8"))), type);

		}

	}

	@Override

	protected void cleanup(Context context) throws IOException, InterruptedException {

		log.info("Cleaning up reducer...");

		MultipleOutputWriter.close();

	}

}

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

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.log4j.Logger;

import org.junit.Test;

import org.postgresql.copy.CopyManager;

import org.postgresql.core.BaseConnection;

import org.springframework.jdbc.core.JdbcTemplate;

import org.springframework.jdbc.core.simple.SimpleJdbcCall;

import org.springframework.jdbc.datasource.DriverManagerDataSource;

import java.io.FileInputStream;

import java.io.InputStream;

import java.sql.CallableStatement;

import java.sql.Connection;

import java.sql.ResultSet;

import java.sql.ResultSetMetaData;

/**

 * @author eri

 *         <p/>

 *         DB DAO with methods for creating the temporary shadow schema of the

 *         statsDB, executing extra inserts, creating views and indexes .

 */

public class StatsDAO {

    private org.springframework.jdbc.datasource.DriverManagerDataSource statsDatasource;

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

    private long startTime;

    private long endtime;

    public StatsDAO(String dbUrl, String dbUser, String dbPassword, String dbDriver) {

        statsDatasource = new DriverManagerDataSource(dbUrl, dbUser, dbPassword);

        statsDatasource.setDriverClassName(dbDriver);

    }

    /**

     * Reads create schema script from a file and executes it. Used to create a

     * temporary Shadow schema for data import

     *

     * @throws Exception

     */

    public void createSchema() throws Exception {

        InputStream in = ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/stats_db_schema.sql");

        byte[] b = new byte[in.available()];

        in.read(b);

        String q = new String(b);

        in.close();

        Connection con = null;

        log.info(" Creating Shadow Schema Database   ....");

        try {

            con = statsDatasource.getConnection();

            CallableStatement st = con.prepareCall(q);

            st.execute();

            st.close();

        } catch (Exception e) {

            log.error("Could not   Create Tables in  Temporary database  :" + e.getMessage());

            throw new Exception("Could not Create Tables in Temporary database  :", e);

        } finally {

            con.close();

        }

        log.info("Shadow Schema created.");

    }

    /**

     * Sets the default schema (search path ) for the sqoop user.

     *

     * @param user

     * @throws Exception

     */

    public void setSearchPathDB(String user) throws Exception {

        String q = "ALTER ROLE " + user + " SET search_path TO shadow,public; ";

        Connection con = null;

        try {

            con = statsDatasource.getConnection();

            CallableStatement st = con.prepareCall(q);

            st.execute();

            st.close();

        } catch (Exception e) {

            log.error("Could not change default schema for user :" + e.getMessage());

            throw new Exception("Could not change default schema for user ", e);

        } finally {

            con.close();

        }

        log.info(" Default user schema set to shadow");

    }

    /**

     * Keeps a backup of the current public schema and then replaces it with the new  Shadow one.

     *

     * @throws Exception

     */

    public void renameSchema() throws Exception {

        Connection con = statsDatasource.getConnection();

        String q = "DROP SCHEMA IF EXISTS backup CASCADE;" +

                "DROP SCHEMA IF EXISTS public CASCADE;" +

                " ALTER SCHEMA shadow RENAME TO public; ";

        try {

            CallableStatement st = con.prepareCall(q);

            log.info("Renaming  Shadow  schema to  Public,,,");

            st.execute();

            st.close();

            con.close();

            log.info("Renaming done");

        } catch (Exception e) {

            log.error("Could not rename schema " + e.getMessage());

            throw new Exception("Could not rename schema  ", e);

        } finally {

            con.close();

        }

        log.info("Schema renamed");

    }

    public void fix_integer_ids() throws Exception {

        log.info("fixing integer ids...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.fix_result_ids()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to fix integer ids: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void addArrayColumns() throws Exception {

        log.info("Adding new columns ...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_arrays()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to add columns: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    public void cleanTables() throws Exception {

        log.info(" Cleaning Tables...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "CREATE TABLE \"shadow\".rd_distinct AS SELECT DISTINCT * FROM \"shadow\".result_datasources;";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        q = "TRUNCATE \"shadow\".result_datasources;";

        st = con.prepareCall(q);

        st.execute();

        st.close();

        q = "INSERT INTO \"shadow\".result_datasources SELECT * FROM \"shadow\".rd_distinct;";

        st = con.prepareCall(q);

        st.execute();

        st.close();

        q = "DROP TABLE \"shadow\".rd_distinct;";

        st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to clean tables : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Create charts.

     *

     * @throws Exception

     */

    public void createCharts() throws Exception {

        log.info(" Creating Chart Tables...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_charts()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to create chart tables: " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Create chart indexes.

     *

     * @throws Exception

     */

    public void createChartIndexes() throws Exception {

        log.info(" Create Chart Indexes...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_chart_indexes()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to create chart indexes : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Builds indexes.

     *

     * @throws Exception

     */

    public void buildIndexes() throws Exception {

        log.info(" Building Database Indexes...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_indexes()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to build indexes : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Builds views.

     *

     * @throws Exception

     */

    public void buildViews() throws Exception {

        log.info(" Building Database Views...");

        Connection con = statsDatasource.getConnection();

        startTime = System.currentTimeMillis();

        String q = "{call shadow.create_views()}";

        CallableStatement st = con.prepareCall(q);

        st.execute();

        st.close();

        con.close();

        endtime = System.currentTimeMillis();

        log.info("Time to build Views : " + ((endtime - startTime) / 60000) + " minutes ");

    }

    /**

     * Executes extra inserts ( table updates )

     *

     * @throws Exception

     */

    public void executeExtraInserts() throws Exception {

        log.info("  Executing Extra Inserts...");

        startTime = System.currentTimeMillis();

        populateDefaults();

        //update_project_results();

        update_project_has_pubs();

        update_project_pubs_count();

        update_project_delated_pubs();

        update_project_daysforlastpub();

        update_project_delayed_pubs();

        cleanUp();

        endtime = System.currentTimeMillis();

        log.info("Total time for Extra Inserts : " + ((endtime - startTime) / 60000) + " minutes ");

    }