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

            Long maxRows = getPropertyAsInteger("maxRows", Long.MAX_VALUE);
            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();
            long rowCounter = 0;
            while(csvIterator.hasNext()) {
                if(rowCounter >= maxRows) {
                    break;
                }
                rowCounter++;
                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();
                    try {
                        String value = record.get(key);
                        document.addField(key, value);
                    } catch (IllegalArgumentException e) {

                    }
                }
                executer.document(document);
            }
            //executer.end();
            in.close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}

@@ -517,24 +517,84 @@
 id read() %7B%0A
+    %09String absoluteFilename;%0A    %09boolean addMeta = false;%0A
         try 
@@ -759,23 +759,16 @@
         
-String 
 absolute
@@ -823,24 +823,86 @@
 filename);%0A%0A
+            addMeta = getPropertyAsBoolean(%22addMeta%22, false);%0A
             
@@ -1658,21 +1658,20 @@
 long row
-Count
+Numb
 er = 0;%0A
@@ -1735,21 +1735,20 @@
   if(row
-Count
+Numb
 er %3E= ma
@@ -1824,13 +1824,12 @@
  row
-Count
+Numb
 er++
@@ -2359,32 +2359,222 @@
               %7D%0A
+                if(addMeta) %7B%0A                %09document.addField(%22rowNumber%22, String.valueOf(rowNumber));%0A                %09document.addField(%22fileName%22, absoluteFilename);%0A                %7D%0A
                 

add metadata to the csv reader