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

import java.io.IOException;
import java.util.Iterator;
import java.util.Set;

import jp.ac.dendai.cdl.mori.wikie.WikIE;
import jp.ac.dendai.cdl.mori.wikie.util.WNormalizer;
import jp.ac.dendai.cdl.mori.wikie.util.WTitle;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

/**
 * edge処理用Mapper
 * @author Mori
 *
 */
public class EdgeMapper extends WikipediaMapper<LongWritable, Text, Text, Text> {

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException {
        try {
            WPageElement page = parsePageElement(value);
            WTitle entry = normalizer.normalize(page.getTitle());
            String title = entry.toString();
            String id  = page.getPageId();
            String text = WNormalizer.deleteNonPrintingChars(page.getText());
            String kind = linkUtils.getKind(entry, text);
            if (kind != null) {
                if (kind.equals(WikIE.LEAF) || kind.equals(WikIE.NODE)) {
                    Set<WTitle> categoryArrayList = linkUtils.getCategory(text);
                    context.write(new Text(title), new Text(WikIE.ENTRY + "\t" + kind + "\t" + id));
                    Iterator<WTitle> itr = categoryArrayList.iterator();
                    while (itr.hasNext()) {
                        String category = itr.next().toString();
                        try {
                            context.write(new Text(category), new Text(WikIE.EDGE + "\t" + id + "\t" + entry.toString()));
                        }
                        catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                if (kind.equals(WikIE.REDIRECT)) {
                    String redirect = linkUtils.isRedirect(text).toString();
                    context.write(new Text(title), new Text(WikIE.ENTRY + "\t" + WikIE.REDIRECT + "\t" + id));
                    context.write(new Text(redirect), new Text(WikIE.TARGET + "\t" + id + "\t" + entry.toString()));
                }
            }
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }

}
