package org.iidp.ostmap.batch_processing;

import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.hadoop.io.Text;
import org.codehaus.jettison.json.JSONException;
import org.iidp.ostmap.commons.Tokenizer;
import org.codehaus.jettison.json.JSONObject;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;


/**
 * maps RawTwitterData rows to mutations to insert into TermIndex
 */
public class ConverterFlatMap implements FlatMapFunction<Tuple2<Key, Value>, Tuple2<Text,Mutation>>, Serializable {

    private Tokenizer tokenizer;
    private String outputTableName;

    public ConverterFlatMap(Tokenizer tokenizer, String outputTableName){
        this.tokenizer = tokenizer;
        this.outputTableName = outputTableName;
    }

    @Override
    public void flatMap(Tuple2<Key, Value> in, Collector<Tuple2<Text, Mutation>> out) {

        JSONObject obj = null;
        String user = "";
        String text = "";

        try {
            obj = new JSONObject(in.f1.toString());
            text = obj.getString("text");
            user = obj.getJSONObject("user").getString("screen_name");

        } catch (JSONException e) {
            e.printStackTrace();
        }


        Map<String, Integer> tokenCount= new HashMap<>();

        //count token occurences
        for(String token: tokenizer.tokenizeString(text)){

            if(tokenCount.containsKey(token)){

                tokenCount.put(token,tokenCount.get(token)+1);

            }else{
                tokenCount.put(token,1);
            }
        }

        //create mutations for tokens
        for(Map.Entry<String, Integer> kv: tokenCount.entrySet()){
            Mutation m = new Mutation(kv.getKey());
            m.put("text", in.f0.getRow().toString(), kv.getValue().toString());
            out.collect(new Tuple2<>(new Text(outputTableName), m));
        }


        //create mutations for username
        Mutation m = new Mutation(user);
        m.put("user", in.f0.getRow().toString(), "1");
        out.collect(new Tuple2<>(new Text(outputTableName), m));

    }
}

@@ -1148,23 +1148,63 @@
 ing user
+Name = %22%22;%0A        String userScreenName
  = %22%22;%0A
-
         
@@ -1346,16 +1346,26 @@
     user
+ScreenName
  = obj.g
@@ -1410,16 +1410,84 @@
 _name%22);
+%0A            userName = obj.getJSONObject(%22user%22).getString(%22name%22);
 %0A%0A      
@@ -2267,16 +2267,32 @@
 username
+ and screen name
 %0A       
@@ -2322,16 +2322,20 @@
 ion(user
+Name
 );%0A     
@@ -2344,20 +2344,190 @@
  m.put(%22
-user
+name%22, in.f0.getRow().toString(), %221%22);%0A        out.collect(new Tuple2%3C%3E(new Text(outputTableName), m));%0A%0A        m = new Mutation(userScreenName);%0A        m.put(%22screen_name
 %22, in.f0

use user.name and user.screen_name for converter #19