D-Net

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

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

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

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

import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;

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 java.util.ArrayList;

import java.util.List;

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 {

        if (dataList.isEmpty()) return;

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

            if (split.length > 0) {

                context.add(split[0]);

            }

            if (split.length > 1) {

                category.add(split[1]);

            }

            if (split.length > 2) {

                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;

    }

}