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

import java.io.IOException;

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.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.xml.sax.SAXException;

public class NodeMapper extends WikipediaMapper<LongWritable, Text, IntWritable, Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException ,InterruptedException {
        WPageElement page = null;
        try {
            page = parsePageElement(value);
            WTitle title = normalizer.normalize(page.getTitle());
            String text = WNormalizer.deleteNonPrintingChars(page.getText());
            String kind = linkUtils.getKind(title.getNsNumber(), text);

            if (!kind.equals(WikIE.OTHER)) {
                context.write(new IntWritable(Integer.parseInt(page.getPageId())), new Text(page.getTitle() + "\t" + kind));
            }
//            if (kind.equals(WikIE.LEAF)) {
//                context.write(new IntWritable(Integer.parseInt(page.getPageId())), new Text(page.getTitle() + "\t" + WikIE.LEAF));
//            }
//            else if (kind.equals(WikIE.NODE)) {
//                context.write(new IntWritable(Integer.parseInt(page.getPageId())), new Text(page.getTitle() + "\t" + WikIE.NODE));
//            }
//            else if (kind.equals(WikIE.REDIRECT)) {
//                context.write(new IntWritable(Integer.parseInt(page.getPageId())), new Text(page.getTitle() + "\t" + WikIE.REDIRECT));
//            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (SAXException e) {
            e.printStackTrace();
        }
    }
}
