D-Net

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

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

import eu.dnetlib.data.proto.DatasourceProtos.Datasource;

import eu.dnetlib.data.proto.DatasourceProtos.Datasource.Metadata;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.FieldTypeProtos.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 java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Date;

import java.util.HashMap;

import java.util.List;

/**

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

 *         for sqoop

 */

public class Serializer {

    private static String DELIM;

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

    private String NULL_STRING;

    private String NULL_NUM;

    //TODO no longer used

    private static String ENCLOSED;

    public Serializer() {

    }

    ;

    public String serialize(Oaf oaf) {

        switch (oaf.getKind()) {

            case entity:

                OafEntity valueEntity = oaf.getEntity();

                switch (valueEntity.getType()) {

                    case datasource:

                        return buildDatasource(valueEntity);

                    case organization:

                        return buildOrganization(valueEntity);

                    case project:

                        return buildProject(valueEntity);

                    case result:

                        return buildResult(valueEntity);

//                    case person:

//

//                        return buildPerson(valueEntity);

                    default:

                        log.error("wrong type");

                        break;

                }

                break;

            case relation:

                OafRel valueRel = oaf.getRel();

                return buildRel(valueRel);

        }

        return null;

    }

    public String serialize(OafRel oaf) {

        switch (oaf.getRelType()) {

            case resultProject:

                return getResultProject(oaf);

            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 getResultProject(OafRel oaf) {

        String buff = new String();

        String result = oaf.getTarget();

        buff += getStringField(result);

        // TODO is declared as int!!!

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

        if (diff < 0) {

            diff = 0;

        }

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

        return buff;

    }

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

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

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

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

                // TODO HERE KEEP ONLY ONE CLASSIFICATIONS PER RESULT

                break;

            }

        }

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

//TODO hosted by

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

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

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

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

            }

        }

//TODO  collected froms

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

            String collectedFrom = collectedFromValue.getKey();

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

                buffs.add((getStringField(collectedFrom)));

        }

        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 -> replace them

    private boolean isValidTopic(String t) {

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

            return false;

        }

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

            return false;

        }

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

            return false;

        }

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

            return false;

        }

        // skip dedups

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

            return false;

        }

        return true;

    }

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

        String buff = new String();

        for (StructuredProperty topic : Topics) {

            // TODOs

            if (isValidTopic(topic.getValue())) {

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

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

                    if (isValidTopic(t)) {

                        buff += t + " ";

                    }

                }

            }

        }

        if (!buff.isEmpty()) {

            buff = getStringField(buff);

            buffs.add(buff);

        }

        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;

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

        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;

        //TODO pubtitle

        String titleString = NULL_STRING;

        for (StructuredProperty title : metadata.getTitleList()) {

            titleString = title.getValue();

            break;

        }

        buff += getStringField(titleString);

        // TODO format

        String formatString = NULL_STRING;

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

            formatString = format.getValue();

            break;

        }

        buff += getStringField(formatString);

       //TODO publisher

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

        //TODO journal

        buff += getStringField(metadata.getJournal().getName());

        // year

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

        // date CHANGED THIS TO DATE FORMAT

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

        // access_mode,

        buff += getStringField(getAccessMode(result));

        // bestlicense

        buff += getStringField(getBestLicense(result));

        // type

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

        // embargo_end_date

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

        // `authors`,

        int authors = 0;

        String delayed = "no";

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

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

                authors++;

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

            // TODO remember : in result Project, first id is project, second is

            // result.

            {

                String daysfromend = getYearDifferenceInteger(rel.getResultProject().getOutcome().getRelMetadata().getEnddate(), rel.getResultProject().getOutcome().getRelMetadata().getStartdate());

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

                    delayed = "yes";

                }

            }

        }

        // `delayed`,

        buff += getStringField(delayed);

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

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

        // number??

        buff += getStringField("1");

        if (isValid(buff, full_id)) {

            return buff;

        } else {

            return null;

        }

    }

    //TODO here see if check is ok

    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;

        }

    }

    // TODO here iterate over all values

    private String getAccessMode(Result result) {

        String accessMode = NULL_STRING;

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

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

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

                break;

            }

        }

        return accessMode;

    }

    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`,

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

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

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

        }

        buff += getStringField(acronym);

        //title!

        String title = getStringField(metadata.getTitle().getValue());

        buff += getStringField(title);

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

        }

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

        buff += getStringField(funding_tree_0);

        // `funding_lvl1`,

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

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

        // TODO checking for level 2 value present in other funding tree

        if (funding_tree_2.equals(NULL_STRING)) {

            if (fundList.size() > 1) {

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

            }

        }

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

        // / `funding_lvl2`,

        buff += getStringField(funding_tree_2);

        // `sc39`,

        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 += 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("no");

        // numpubs

        buff += getNumericField("0");

        // enddate

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

        // startdate

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

        // `daysforlastpub`,

        buff += getNumericField("");

        // `delayedpubs`,

        buff += getNumericField("");

        // `number`

        buff += getStringField("1");

        return buff;

    }

    //TODO remove for production - not yet ready

    private String buildPerson(OafEntity data) {

        String buff = new String();

        PersonProtos.Person person = data.getPerson();

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

        person.getCoauthorsCount();

        // `person_id`,

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

        buff += full_id;

        // person_result

        buff += full_id;

        // `fullname`,

        buff += metadata.getFullname();

        // `Nationality`,

        buff += metadata.getNationality();

        // `Email`,

        buff += metadata.getEmail();

        // `Phone`,

        buff += metadata.getPhone();

        // `CoauthorsCount`

        buff += person.getCoauthorsCount();

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

            // if (enddate != null && startdate != null&&) {

            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() || data.equals("-1")) {

            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:,____]", "");

        return number;

    }

    private String getStringField(String data) {

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

            return ENCLOSED + NULL_STRING + ENCLOSED + DELIM;

        } else {

            String field = clean(data);

            if (field == null) {

                return ENCLOSED + NULL_STRING + ENCLOSED + DELIM;

            } else {

                return field + DELIM;

            }

        }

    }

    private String getStringDateField(String data) {

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

            return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

        } else {

            String field = clean(data);

            if (field == null) {

                return ENCLOSED + NULL_NUM + ENCLOSED + DELIM;

            } else {

                return field + DELIM;

            }

        }

    }

    private String getNumericField(String data) {

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

            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(DELIM, "");

            value = value.replaceAll(ENCLOSED, "");

            // 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 long DATEDIFF(String startDate, String endDate) {

        long MILLISECS_PER_DAY = 24 * 60 * 60 * 1000;

        long days = 0l;

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

        Date dateFin = null;

        try {

            dateIni = (Date) format.parse(startDate);

            dateFin = (Date) format.parse(endDate);

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

        } catch (Exception e) {

            e.printStackTrace();

        }

        return days;

    }

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

        return ENCLOSED;

    }

    public void setENCLOSED(String eNCLOSED) {

        ENCLOSED = eNCLOSED;

    }

}

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

import java.util.ArrayList;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.log4j.Logger;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;

public class ContextExporter {

    private ContextTransformer contextTransformer = new ContextTransformer();

    private String outputPath;

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

    private ArrayList<String> context = new ArrayList<String>();

    private ArrayList<String> category = new ArrayList<String>();

    private ArrayList<String> concept = new ArrayList<String>();

    public ContextExporter(String outputPath, String contextMap, boolean readFromUrl) throws Exception {

        if (!outputPath.endsWith("/")) {

            outputPath += "/";

        }

        this.outputPath = outputPath;

        if (readFromUrl) {

            readFromUrl(contextMap);

        } else {

            readFromBuffer(contextMap);

        }

    }

    public void readFromUrl(String url) throws Exception {

        List<String> concepts = getContextResouces(url);

        log.info("Returned concept  " + concepts.size()

        );

        for (String data : concepts) {

            log.info("++++++++++++++ Transforming concept data ");

            String res = contextTransformer.transformXSL(data);

            processData(res);

        }

        writeData(this.context, "context");

        writeData(this.category, "category");

        writeData(this.concept, "concept");

    }

    private void readFromBuffer(String contextMap) throws Exception {

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

            log.error("Context Resources file is empty.");

            throw new Exception("Context Resources file is empty.");

        }

        String data = contextTransformer.transformXSL(contextMap);

        log.info(data);

        processData(data);

    }

    private List<String> getContextResouces(String url) throws ISLookUpException {

        ISLookUpService lookUpService;

        JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();

        factory.setServiceClass(ISLookUpService.class);

        factory.setAddress(url);

        lookUpService = (ISLookUpService) factory.create();

//		for $x in //RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='ContextDSResourceType']

//		[.//RESOURCE_KIND/@value='ContextDSResources'] return  $x

        return lookUpService.quickSearchProfile("//RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='ContextDSResourceType'][.//RESOURCE_KIND/@value='ContextDSResources']");

    }

    private void writeData(ArrayList<String> dataList, String listName) throws Exception {

        log.info(listName + "  size " + dataList.size());

        String data = new String();

        for (int i = 0; i < dataList.size(); i++) {

            data += dataList.get(i);

        }

        data = data.substring(0, data.lastIndexOf("\n"));

        flushString(data, outputPath + listName);

    }

    private void processData(String data) throws Exception {

        try {

            String[] split = data.split("COPY\n");

            context.add(split[0]);

            if (split[1].endsWith("\n")) {

                //  split[1] = split[1].substring(0, split[1].lastIndexOf("\n"));

            }

            category.add(split[1]);

            concept.add(split[2]);

        } catch (Exception e) {

            String msg = " Unable to create file with context, " + "concept and category values in output path " + outputPath + ". Reason: ";

            log.error(msg);

            throw new Exception(msg, e);

        }

    }

    private void flushString(String data, String destination) throws Exception {

        FSDataOutputStream fin = null;

        try {

            log.info("***********************Writing data:***********************\n" + data);

            log.info("***********************  data:***********************\n");

            FileSystem fs = FileSystem.get(new Configuration());

            fin = fs.create(new Path(destination), true);

            fin.write(data.getBytes());

        } catch (Exception e) {

            log.error("Failed  to write exported data to a file : ", e);

            throw new Exception("Failed  to write exported data to a file : " + e.toString(), e);

        } finally {

            fin.close();

        }

    }

    public String getOutputPath() {

        return outputPath;

    }

    public void setOutputPath(String outputPath) {

        this.outputPath = outputPath;

    }

}

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

import java.io.InputStream;

import java.util.Properties;

import org.apache.hadoop.conf.Configuration;

import org.apache.log4j.Logger;

import org.junit.Before;

import org.junit.Test;

import eu.dnetlib.data.mapreduce.hbase.statsExport.daos.SqlDAO;

import eu.dnetlib.data.mapreduce.hbase.statsExport.daos.SqlStore;

import eu.dnetlib.data.mapreduce.hbase.statsExport.drivers.DBDriver;

import eu.dnetlib.data.mapreduce.hbase.statsExport.drivers.SqoopDriver;

import eu.dnetlib.data.mapreduce.hbase.statsExport.drivers.StatsJobDriver;

public class ExportTest {

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

	private SqlDAO statsDao;

	private SqlStore statsStore;

	private SqoopDriver sqoopDriver;

	private Properties props;

	private StatsJobDriver statsJobDriver;

	private DBDriver dbDriver;

//	@Before

	public void init() throws Exception {

		InputStream file = ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/StatsProperties");

		props = new Properties();

		props.load(file);

		file.close();

		statsStore = new SqlStore();

		statsStore.setDbDriver(props.getProperty("Stats_db_Driver"));

		//

		statsStore.setDB_URL(props.getProperty("Stats_db_Url"));

		statsStore.setDbUser(props.getProperty("Stats_db_User"));

		statsStore.setDbPassword(props.getProperty("Stats_db_Pass"));

		statsDao = new SqlDAO();

		statsDao.setStore(statsStore);

		dbDriver = new DBDriver();

		dbDriver.setStatsDao(statsDao);

		sqoopDriver = new SqoopDriver();

		sqoopDriver.setDelim(props.getProperty("Stats_delim_Character"));

		sqoopDriver.setOutputPath(props.getProperty("Stats_outputPath"));

		sqoopDriver.setConnectionUrl(props.getProperty("Stats_db_Url"));

		sqoopDriver.setDbUser(props.getProperty("Stats_db_User"));

		sqoopDriver.setDbPass(props.getProperty("Stats_db_Pass"));

		sqoopDriver.setSqoopReducersCount(props.getProperty("Stats_sqoop_ReducersCount"));

		sqoopDriver.setRecsPerStatement("1");

		sqoopDriver.setStatementPerTrans("1");

		sqoopDriver.setBatch(false);

		sqoopDriver.setVerbose(true);

		sqoopDriver.setConf(new Configuration());

		sqoopDriver.setUseHdfsStore(false);

	}

//	 @Test

	public void parser() {

		String url = "jdbc:postgresql://kiczora.vls.icm.edu.pl:5432/stats";

		if (url.endsWith("/")) {

			url = url.substring(0, url.lastIndexOf('/'));

			System.out.print("url 1 " + url + "\n");

		}

		url = url.substring(0, url.lastIndexOf('/') + 1);

		System.out.print("url 2" + url);

	}

	public void runStatsExport() throws Exception {

		statsJobDriver = new StatsJobDriver();

		statsJobDriver.init();

		statsJobDriver.run();

	}

//	 @Test

	public void runPrepareDB() throws Exception {

		 statsDao.setSearchPathDB(props.getProperty("Stats_db_User"));

//		 dbDriver.prepareDB(props.getProperty("Stats_db_User"));

	}

	// @Test

	public void runSqoopImport() throws Exception {

		// dbDriver.prepareDB(props.getProperty("Stats.dbUser"));

		sqoopDriver.initSqoopJob();

	}

//	@Test

	public void test() throws Exception {

		// InputStream in =

		// ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/"

		// + "ContextResourceProfile.xml");

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

		// in.read(b);

		// String data = new String(b);

//		 in.close();

		// log.info("data" + data);

		// ContextExporter exp= new

		// ContextExporter(props.getProperty("Stats_outputPath"),

		// props.getProperty("ConhaspubstextResourceXML"), false);

		ContextTransformer tr = new ContextTransformer();

		log.info(tr.transformXSL(props.getProperty("ContextResourceXML")));

	}

//	@Test

	public void testExport() throws Exception {

		 ContextExporter exp= new ContextExporter(props.getProperty("Stats_outputPath"),  props.getProperty("isLookupEndpoint"), true);

	}

//	@Test

	public void executeExtraInserts() throws Exception {

//		statsDao.buildViews();

		statsDao.executeExtraInserts();

	}

//	@Test

	public void buildIndexesDb() throws Exception {

		statsDao.executeExtraInserts();

	}

//	@Test

	public void finalizeDb() throws Exception {

		dbDriver.finalizedDB();

	}

//	@Test

	public void clean() throws Exception {

log.info(cleanId("od_______133::000d16306f933e487892fdad65e1fc59"));

	}

	public   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("#", "");

			value = value.replaceAll("!", "");

			value = value.trim();

		}

		if (value == null) {

			return null;

		}

		return     value   ;

	}

}

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet-hadoop-parent</artifactId>

		<version>1.0.0-SNAPSHOT</version>

		<relativePath></relativePath>

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-openaire-stats</artifactId>

	<version>0.0.1-SNAPSHOT</version>

	<build>

		<plugins>

			<!-- <plugin> -->

			<!-- <groupId>org.codehaus.mojo</groupId> -->

			<!-- <artifactId>exec-maven-plugin</artifactId> -->

			<!-- <version>1.2.1</version> -->

			<!-- <configuration> -->

			<!-- <mainClass>eu.dnetlib.data.mapreduce.hbase.statsExport.drivers.SqoopDriver</mainClass> -->

			<!-- </configuration> -->

			<!-- </plugin> -->

			<!-- <plugin> -->

			<!-- <groupId>org.apache.maven.plugins</groupId> -->

			<!-- <artifactId>maven-shade-plugin</artifactId> -->

			<!-- <version>2.1</version> -->

			<!-- <executions> -->

			<!-- <execution> -->

			<!-- <phase>package</phase> -->

			<!-- <goals> -->

			<!-- <goal>shade</goal> -->

			<!-- </goals> -->

			<!-- <configuration> -->

			<!-- <transformers> -->

			<!-- <transformer -->

			<!-- implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> -->

			<!-- </transformer> -->

			<!-- </transformers> -->

			<!-- </configuration> -->

			<!-- </execution> -->

			<!-- </executions> </plugin> -->

		</plugins>

	</build>

	<dependencies>

		<dependency>

			<groupId>javax.xml</groupId>

			<artifactId>jaxp-api</artifactId>

			<version>1.4.2</version>

		</dependency>

		<dependency>

			<groupId>net.sf.opencsv</groupId>

			<artifactId>opencsv</artifactId>

			<version>2.3</version>

		</dependency>

<dependency>

    <groupId>org.apache.poi</groupId>

    <artifactId>poi</artifactId>

    <version>3.8</version>

</dependency>

		<dependency>

			<groupId>jdk.tools</groupId>

			<artifactId>jdk.tools</artifactId>

			<version>1.7.0_05</version>

			<scope>system</scope>

			<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>

		</dependency>

		<dependency>

			<groupId>org.apache.logging.log4j</groupId>

			<artifactId>log4j-api</artifactId>

			<version>2.0-rc2</version>

		</dependency>

		<dependency>

			<groupId>org.apache.logging.log4j</groupId>

			<artifactId>log4j-core</artifactId>

			<version>2.0-rc2</version>

		</dependency>

		<dependency>

			<groupId>log4j</groupId>

			<artifactId>log4j</artifactId>

			<version>1.2.17</version>

		</dependency>

		<dependency>

			<groupId>org.slf4j</groupId>

			<artifactId>slf4j-api</artifactId>

			<version>1.7.6</version>

		</dependency>

		<dependency>

			<groupId>org.slf4j</groupId>

			<artifactId>slf4j-log4j12</artifactId>

			<version>1.7.6</version>

		</dependency>

		<dependency>

			<groupId>com.google.guava</groupId>

			<artifactId>guava-collections</artifactId>

			<version>r03</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-mapreduce-jobs</artifactId>

			<version>0.0.6.2-SNAPSHOT</version>

		</dependency>

        <dependency>

            <groupId>eu.dnetlib</groupId>

            <artifactId>dnet-openaire-data-protos</artifactId>

            <version>[3.0.0,4.0.0)</version>

        </dependency>

		<dependency>

			<groupId>org.apache.httpcomponents</groupId>

			<artifactId>httpclient</artifactId>

			<version>4.0-alpha4</version>

		</dependency>

		<dependency>

			<groupId>commons-httpclient</groupId>

			<artifactId>commons-httpclient</artifactId>

			<version>3.1</version>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>4.8.2</version>

		</dependency>

		<dependency>

			<groupId>org.apache.sqoop</groupId>

			<artifactId>sqoop</artifactId>

			<version>1.4.4</version>

		</dependency>

		<dependency>

			<groupId>postgresql</groupId>

			<artifactId>postgresql</artifactId>

			<version>9.1-901.jdbc4</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-rmi-api</artifactId>

			<version>2.0.0-SNAPSHOT</version>

		</dependency>

		<dependency>

			<groupId>org.apache.cxf</groupId>

			<artifactId>cxf-rt-frontend-jaxws</artifactId>

			<version>2.7.8</version>

		</dependency>

	</dependencies>

	<repositories>

		<repository>

			<id>cloudera</id>

			<name>Cloudera Repository</name>

			<url>https://repository.cloudera.com/artifactory/cloudera-repos</url>

			<releases>

				<enabled>true</enabled>

			</releases>

			<snapshots>

				<enabled>false</enabled>

			</snapshots>

		</repository>

	</repositories>

</project>

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

import java.io.BufferedWriter;

import java.io.OutputStreamWriter;

import java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Date;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.log4j.Logger;

public class HdfsWriter {

    public static void write(String data, String filename) throws Exception {

        Configuration conf = new Configuration();

        FileSystem hdfs = FileSystem.get(conf);

        try {

            Path exportPath = new Path(hdfs.getUri() + filename);

            BufferedWriter br = new BufferedWriter(new OutputStreamWriter(hdfs.create(exportPath, false)));

            // TO append data to a file, use fs.append(Path f)

            br.write(data);

            br.close();

        } catch (Exception e) {

            throw new Exception("Error while writing file ", e);

        }

    }

}