package jp.ac.dendai.cdl.mori.wikie.mapred;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import jp.ac.dendai.cdl.mori.wikie.WikIE;

import org.apache.hadoop.io.Text;

/**
 * edge処理用Reducer
 * @author Mori
 *
 */
public class EdgeReducer extends WikipediaReducer<Text, Text, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
    throws IOException ,InterruptedException {
        String entryId = null;

        List<String> valueList = new ArrayList<String>();
        for (Text inputValue  : values) {
            String[] status = inputValue.toString().split("\t");
            if (status[0].equals(WikIE.ENTRY)) {
                entryId = status[2];
            }
            else {
                valueList.add(inputValue.toString());
            }
        }

        if (entryId == null) {
            return;
        }

        Iterator<String> inputValueIterator = valueList.iterator();
        while (inputValueIterator.hasNext()) {
            String inputValue = inputValueIterator.next();
            String[] status = inputValue.split("\t");
            String type = status[0];
            if (type.equals(WikIE.EDGE)) {
                context.write(new Text(status[1]), new Text(entryId + "\t" + WikIE.HYPERNYM));
            }
            else if (type.equals(WikIE.TARGET)) {
                context.write(new Text(status[1]), new Text(entryId + "\t" + WikIE.TARGET));
            }
        }
    };

    //    @Override
    //    protected void reduce(Text key, Iterable<Text> values, Reducer.Context context) throws IOException, InterruptedIOException {
    //        String entryId = null;
    //
    //        List<String> valueList = new ArrayList<String>();
    //        while (inputValueItr.hasNext()) {
    //            String inputValue = inputValueItr.next().toString();
    //            String[] status = inputValue.split("\t");
    //            if (status[0].equals(WikIE.ENTRY)) {
    //                entryId = status[2];
    //            }
    //            else {
    //                valueList.add(inputValue);
    //            }
    //        }
    //        Iterator<String> inputValueIterator = valueList.iterator();
    //        while (inputValueIterator.hasNext()) {
    //            String inputValue = inputValueIterator.next();
    //            String[] status = inputValue.split("\t");
    //            String type = status[0];
    //            String currentTitle = status[2];
    //            String targetTitle = key.toString();
    //                if (type.equals(WikIE.EDGE)) {
    //                    output.collect(new Text(status[1]), new Text(entryId + "\t" + WikIE.HYPERNYM + "\t" + currentTitle + "\t" + targetTitle));
    //                }
    //                else if (type.equals(WikIE.TARGET)) {
    //                    output.collect(new Text(status[1]), new Text(entryId + "\t" + WikIE.TARGET + "\t" + currentTitle + "\t" + targetTitle));
    //                }
    //        }
    //    }
}
