D-Net

package eu.dnetlib.msro.workers.aggregation.collect.plugins.csv;

import java.io.InputStreamReader;

import java.io.Reader;

import java.net.URL;

import java.util.Iterator;

import java.util.Set;

import java.util.stream.Stream;

import org.apache.commons.csv.CSVFormat;

import org.apache.commons.csv.CSVParser;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentHelper;

import org.dom4j.Element;

import com.google.common.collect.Iterators;

import eu.dnetlib.enabling.tools.DnetStreamSupport;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.rmi.data.InterfaceDescriptor;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.CollectorPlugin;

/**

 * The Class HttpCSVCollectorPlugin.

 */

public class HttpCSVCollectorPlugin implements CollectorPlugin {

	private static final Log log = LogFactory.getLog(HttpCSVCollectorPlugin.class);

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.msro.workers.aggregation.collect.plugin.CollectorPlugin#collect(eu.dnetlib.msro.workers.aggregation.collect.rmi.

	 * InterfaceDescriptor, java.lang.String, java.lang.String)

	 */

	@Override

	public Stream<String> collect(final InterfaceDescriptor descriptor, final String fromDate, final String untilDate) throws CollectException {

		return DnetStreamSupport.generateStreamFromIterator(getIterator(descriptor));

	}

	private Iterator<String> getIterator(final InterfaceDescriptor descriptor) {

		try {

			final URL url = new URL(descriptor.getBaseUrl());

			url.openConnection();

			final String separatorString = descriptor.getParams().get("separator");

			final String identifier = descriptor.getParams().get("identifier");

			final String quote = descriptor.getParams().get("quote");

			final char separator = separatorString.equals("\\t") || StringUtils.isBlank(separatorString) ? '\t' : separatorString.charAt(0);

			final CSVFormat format = StringUtils.isBlank(quote) ? CSVFormat.EXCEL.withHeader().withDelimiter(separator)

					: CSVFormat.EXCEL.withHeader().withDelimiter(separator).withQuote(quote.charAt(0));

			try (final Reader reader = new InputStreamReader(url.openStream());

					final CSVParser parser = new CSVParser(reader, format)) {

				final Set<String> headers = parser.getHeaderMap().keySet();

				return Iterators.transform(parser.iterator(), input -> {

					final Document document = DocumentHelper.createDocument();

					final Element root = document.addElement("csvRecord");

					for (final String key : headers) {

						final Element row = root.addElement("column");

						row.addAttribute("name", key).addText(input.get(key));

						if (key.equals(identifier)) {

							row.addAttribute("isID", "true");

						}

					}

					return document.asXML();

				});

			} catch (final Exception e) {

				log.error("Error iterating csv lines", e);

				return null;

			}

		} catch (final Exception e) {

			log.error("Error iterating csv lines", e);

			return null;

		}

	}

}