package HiBench;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.MapFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.InvalidJobConfException;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.util.Progressable;
import org.apache.nutch.crawl.CrawlDatum;
import org.apache.nutch.crawl.Inlinks;
import org.apache.nutch.parse.ParseData;
import org.apache.nutch.parse.ParseText;


/* Parse content in a segment. */
public class NutchOutputFormat implements OutputFormat<Text, NutchParse> {
//	private static final Log log = LogFactory.getLog(NutchOutputFormat.class);

	public void checkOutputSpecs(FileSystem fs, JobConf job) throws IOException {
		Path out = FileOutputFormat.getOutputPath(job);
		if ((out == null) && (job.getNumReduceTasks() != 0)) {
			throw new InvalidJobConfException(
					"Output directory not set in JobConf.");
		}
		if (fs == null) {
			fs = out.getFileSystem(job);
		}

		if (fs.exists(new Path(out, CrawlDatum.PARSE_DIR_NAME)))
			throw new IOException("Segment already parsed!");
		
		
	}

	public RecordWriter<Text, NutchParse> getRecordWriter(FileSystem fs, JobConf job,
			String name, Progressable progress) throws IOException {

		final CompressionType compType = SequenceFileOutputFormat.getOutputCompressionType(job);
		Path out = FileOutputFormat.getOutputPath(job);
		
		Path text = new Path(new Path(out, ParseText.DIR_NAME), name);
		Path data = new Path(new Path(out, ParseData.DIR_NAME), name);
		Path linkdb = new Path(new Path(new Path(out, NutchData.getLinkDbName()), "current"), name);
//		Path crawl = new Path(new Path(out, CrawlDatum.PARSE_DIR_NAME), name);
//		Path crawldb = new Path(out.getParent(), "/crawldb/current");
//		Path crawlgenerate = new Path(new Path(out, CrawlDatum.GENERATE_DIR_NAME), name);

//		final String[] parseMDtoCrawlDB = job.get("db.parsemeta.to.crawldb","").split(" *, *");

		final MapFile.Writer textOut =
				new MapFile.Writer(job, fs, text.toString(), Text.class, ParseText.class,
						CompressionType.RECORD, progress);

		final MapFile.Writer dataOut =
				new MapFile.Writer(job, fs, data.toString(), Text.class, ParseData.class,
						compType, progress);

		final MapFile.Writer linkdbOut =
				new MapFile.Writer(job, fs, linkdb.toString(), Text.class, Inlinks.class,
						compType, progress);
/*
		final SequenceFile.Writer crawlOut =
				SequenceFile.createWriter(fs, job, crawl, Text.class, CrawlDatum.class,
						compType, progress);

		final SequenceFile.Writer crawlDbOut =
				SequenceFile.createWriter(fs, job, crawldb, Text.class, CrawlDatum.class,
						compType, progress);

		final SequenceFile.Writer generateOut =
				SequenceFile.createWriter(fs, job, crawlgenerate, Text.class, CrawlDatum.class,
						compType, progress);
*/		
		return new RecordWriter<Text, NutchParse>() {

			public void write(Text key, NutchParse parse)
					throws IOException {

//				CrawlDatum crawldatum = new CrawlDatum();
/*				
				crawldatum.setStatus(CrawlDatum.STATUS_FETCH_SUCCESS);
				crawlOut.append(parse.url, crawldatum);
				
				crawldatum.setStatus(CrawlDatum.STATUS_FETCH_SUCCESS);
				generateOut.append(parse.url, crawldatum);

				crawldatum.setStatus(CrawlDatum.STATUS_FETCH_SUCCESS);
				crawlDbOut.append(parse.url, crawldatum);
*/				
				if (null != parse.text) {
					textOut.append(key, parse.text);
				}
				
				if (null != parse.data) {
					dataOut.append(key, parse.data);
				}
				
				if (null != parse.inlinks) {
					linkdbOut.append(key, parse.inlinks);
				}
				
				// TODO: remember to append real page text

				/**
				 *  TODO: remember to append correct crawldatum
					1.	CrawlDatum d = new CrawlDatum(CrawlDatum.STATUS_SIGNATURE, 0);
						String sig = parseData.getContentMeta().get(Nutch.SIGNATURE_KEY);
					2.	parseMDCrawlDatum = new CrawlDatum(CrawlDatum.STATUS_PARSE_META, 0);
						if (parseMDCrawlDatum != null) crawlOut.append(key, parseMDCrawlDatum);
					3.	newDatum.setStatus(CrawlDatum.STATUS_LINKED);
						newDatum.getMetaData().put(Nutch.WRITABLE_REPR_URL_KEY,	new Text(reprUrl));
					4.	CrawlDatum target = new CrawlDatum(CrawlDatum.STATUS_LINKED, interval);
						for (Entry<Text, CrawlDatum> target : targets) {
							crawlOut.append(target.getKey(), target.getValue());
						}
					5.	adjust, STATUS_LINKED
					6.	
					CrawlDatum datum = new CrawlDatum();
					datum.setStatus(CrawlDatum.STATUS_FETCH_SUCCESS);
					String timeString = parse.getData().getContentMeta().get(Nutch.FETCH_TIME_KEY);
					try {
						datum.setFetchTime(Long.parseLong(timeString));
					} catch (Exception e) {
						LOG.warn("Can't read fetch time for: " + key);
						datum.setFetchTime(System.currentTimeMillis());
					}
					crawlOut.append(key, datum);
				 */
			}

			public void close(Reporter reporter) throws IOException {
				textOut.close();
				dataOut.close();
				linkdbOut.close();
//				crawlOut.close();
//				crawlDbOut.close();
//				generateOut.close();
//				linkdbOut.close();
			}
		};
	}
}
