package de.tuberlin.dima.schubotz.common.utils;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;

import de.tuberlin.dima.schubotz.wiki.WikiProgram;
import de.tuberlin.dima.schubotz.wiki.types.WikiTuple;
import eu.stratosphere.api.java.DataSet;
import eu.stratosphere.api.java.ExecutionEnvironment;
import eu.stratosphere.api.java.io.CsvInputFormat;
import eu.stratosphere.api.java.io.CsvReader;
import eu.stratosphere.api.java.operators.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.common.collect.HashMultiset;

/**
 * Helper for preprocessed generated CSV files
 */
public class CSVHelper {
	public static Log LOG = LogFactory.getLog(CSVHelper.class);
    public static final String CSV_LINE_SEPARATOR = WikiProgram.CSV_LINE_SEPARATOR;
    public static final char CSV_FIELD_SEPARATOR = WikiProgram.CSV_FIELD_SEPARATOR.charAt(0);
	/**
	 * @param in path + name of file to read from
	 * @return HashMultiset, where index is first CSV field, count is second CSV field
	 * @throws FileNotFoundException
	 * @throws IOException
	 */
	public static HashMultiset<String> csvToMultiset(String in) throws FileNotFoundException, IOException {
		HashMultiset<String> out = HashMultiset.create();
		BufferedReader br = new BufferedReader(new FileReader(in));
        String line = "";
        while ((line = br.readLine()) != null) {
        	String[] parts = line.split(String.valueOf(CSV_FIELD_SEPARATOR));
        	try {
        		out.add(parts[0], Integer.valueOf(parts[1]).intValue());
        	} catch (NullPointerException e) {
        		if (LOG.isWarnEnabled()) {
        			LOG.warn("Non integer in CSV!");
        		}
        		continue;
        	}
        }
        br.close();
        return out;
	}

    /**
     * Must be run in stratosphere environment
     * @param in filename
     * @return dataset
     */
    public static DataSet<WikiTuple> csvToWikiTuple(ExecutionEnvironment env, String in) {
        CsvReader reader = env.readCsvFile(in);
        reader.fieldDelimiter(CSV_FIELD_SEPARATOR);
        reader.lineDelimiter(CSV_LINE_SEPARATOR);
        return reader.tupleType(WikiTuple.class);
    }
}

@@ -61,115 +61,35 @@
 .io.
-BufferedReader;%0Aimport java.io.FileNotFoundException;%0Aimport java.io.FileReader;%0Aimport java.io.IOException
+*;%0Aimport java.util.Scanner
 ;%0A%0Ai
@@ -1252,141 +1252,327 @@
 ();%0A
-%09%09BufferedReader br = new BufferedReader(new FileReader(in));%0A        String line = %22%22;%0A        while ((
+        final InputStream is = new FileInputStream(in);%0A        try %7B%0A            final Scanner s = new Scanner(is, %22UTF-8%22);%0A            s.useDelimiter(%22%5C%5CA%22);%0A            //TODO more hacks, make this more efficient?%0A            String%5B%5D 
 line
+s
  = 
-br.readLine()) != null
+s.next().split(CSV_LINE_SEPARATOR);%0A            for (String line : lines
 ) %7B%0A
@@ -1579,17 +1579,24 @@
         
+ 
-%09
+       
 String%5B%5D
@@ -1661,17 +1661,24 @@
         
-%09
+        
 try %7B%0A  
@@ -1683,18 +1683,28 @@
         
+   
-%09%09
+         
 out.add(
@@ -1760,17 +1760,24 @@
         
-%09
+        
 %7D catch 
@@ -1781,19 +1781,29 @@
 ch (
-NullPointer
+ArrayIndexOutOfBounds
 Exce
@@ -1821,18 +1821,28 @@
         
-%09%09
+            
 if (LOG.
@@ -1872,64 +1872,143 @@
     
-%09%09%09LOG.warn(%22Non integer in CSV!%22);%0A        %09%09%7D%0A
+                LOG.warn(%22Bad format in CSV!%22);%0A                        LOG.warn(%22Line: %22 + line);%0A                    %7D%0A      
         
 %09%09co
@@ -2003,18 +2003,22 @@
         
-%09%09
+      
 continue
@@ -2027,19 +2027,30 @@
         
-%09%7D%0A
+        %7D%0A    
         
@@ -2063,19 +2063,53 @@
     
-br.close();
+%7D finally %7B%0A            is.close();%0A        %7D
 %0A   

rewrite CSVhelper to use custom delimiters