package de.tblsoft.solr.pipeline;

import de.tblsoft.solr.pipeline.bean.Document;
import de.tblsoft.solr.util.IOUtils;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;

import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Map;

/**
 * Created by tblsoft on 11.02.16.
 */
public class CSVReader extends AbstractReader {

    @Override
    public void read() {
        try {
        	
        	String charset = getProperty("charset", StandardCharsets.UTF_8.name());
            String filename = getProperty("filename", null);
            String absoluteFilename = IOUtils.getAbsoluteFile(getBaseDir(),filename);

            String delimiter = getProperty("delimiter", ",");
            String[] headers = getPropertyAsArray("headers", null);
            InputStream in = IOUtils.getInputStream(absoluteFilename);
            java.io.Reader reader = new InputStreamReader(in,charset);

            CSVFormat format = CSVFormat.RFC4180;
            if(headers == null) {
                format = format.withHeader();
            } else {
                format = format.withHeader(headers);
            }

            format=format.withDelimiter(delimiter.charAt(0));

            CSVParser parser = format.parse(reader);
            Iterator<CSVRecord> csvIterator = parser.iterator();

            while(csvIterator.hasNext()) {
                CSVRecord record = csvIterator.next();
                Map<String, Integer> header = parser.getHeaderMap();
                Document document = new Document();
                for(Map.Entry<String,Integer> entry : header.entrySet()) {
                    String key = entry.getKey();
                    String value = record.get(key);
                    document.addField(key, value);
                }
                executer.document(document);
            }
            //executer.end();
            in.close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}

@@ -770,24 +770,100 @@
 filename);%0A%0A
+            Long maxRows = getPropertyAsInteger(%22maxRows%22, Long.MAX_VALUE);%0A
             
@@ -912,16 +912,16 @@
 , %22,%22);%0A
-
         
@@ -1523,17 +1523,49 @@
 ator();%0A
+            long rowCounter = 0;
 %0A
-
         
@@ -1595,24 +1595,143 @@
 asNext()) %7B%0A
+                if(rowCounter %3E= maxRows) %7B%0A                    break;%0A                %7D%0A                rowCounter++;%0A
             
@@ -2014,24 +2014,54 @@
 y.getKey();%0A
+                    try %7B%0A    
             
@@ -2120,16 +2120,20 @@
         
+    
 document
@@ -2151,24 +2151,106 @@
 ey, value);%0A
+                    %7D catch (IllegalArgumentException e) %7B%0A%0A                    %7D%0A
             

add the possibility to process only x first rows if a record has not all definined fields, the record is ignored