package HiBench;

import java.io.IOException;
import java.net.URISyntaxException;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MD5Hash;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapFileOutputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.NLineInputFormat;
import org.apache.nutch.crawl.CrawlDatum;
import org.apache.nutch.crawl.Inlink;
import org.apache.nutch.crawl.Inlinks;
import org.apache.nutch.metadata.Metadata;
import org.apache.nutch.metadata.Nutch;
import org.apache.nutch.parse.Outlink;
import org.apache.nutch.parse.ParseData;
import org.apache.nutch.parse.ParseStatus;
import org.apache.nutch.parse.ParseText;
import org.apache.nutch.util.StringUtil;

public class NutchData {

	private static final Log log = LogFactory.getLog(NutchData.class.getName());

	public static final int CACHE_REF_ITEM_LENGTH = 10;
	public static final int CACHE_REF_SET_SIZE = 200 * 12500;
	public static final int CACHE_REF_THRESHOLD = 4;

	private static final String LINKDB_DIR_NAME = "linkdb";
	private static final String CRAWLDB_DIR_NAME = "crawldb";
	private static final String SEGMENTS_DIR_NAME = "segments";
	
	private DataOptions options;
	private Dummy dummy;

	private static final String NUTCH_WORK_DIR_PARAM_NAME = "nutch.working.dir";
	private static final String URLS_DIR_NAME = "urls";
	private Path segment = null;

	NutchData(DataOptions options) {
		this.options = options;
		parseArgs(options.getRemainArgs());
	}
	
	private void parseArgs(String[] args) {}
	
	private static class CreateUrlHash extends MapReduceBase implements
	Mapper<LongWritable, Text, LongWritable, Text> {

		private static final Log log = LogFactory.getLog(CreateUrlHash.class.getName());
		
		private Path workdir;
		private HtmlCore generator;
		private JobConf job;

		public void configure(JobConf job) {

			try {
				workdir = new Path(job.get(NUTCH_WORK_DIR_PARAM_NAME));
				generator = new HtmlCore(job);
				this.job = job;
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<LongWritable, Text> output, Reporter reporter)
				throws IOException {
			
			int slotId = Integer.parseInt(value.toString().trim());
			generator.fireRandom(slotId);
			
			try {
				DecimalFormat df = new DecimalFormat("00000");
				String name = "part-" + df.format(slotId);
				FileSystem fs = workdir.getFileSystem(job);

				Path crawldb =
						new Path(new Path (new Path(workdir, CRAWLDB_DIR_NAME), "current"), name);
				SequenceFile.Writer crawldbOut =
						new SequenceFile.Writer(fs, job, crawldb, Text.class, CrawlDatum.class);
				
				Path crawl =
						new Path(new Path(workdir, CrawlDatum.PARSE_DIR_NAME), name);
				SequenceFile.Writer crawlOut =
						new SequenceFile.Writer(fs, job, crawl, Text.class, CrawlDatum.class);
				
				Path generate =
						new Path(new Path(workdir, CrawlDatum.FETCH_DIR_NAME), name);
				SequenceFile.Writer generateOut =
						new SequenceFile.Writer(fs, job, generate, Text.class, CrawlDatum.class);
				
				CrawlDatum datum = new CrawlDatum();
				
				long i = slotId - 1;
				while (i < generator.totalpages) {
					key.set(i);
					Text textUrl = generator.nextUrlText();
					
					if (i < generator.pages) {
						datum.setStatus(CrawlDatum.STATUS_FETCH_SUCCESS);
						crawlOut.append(textUrl, datum);
						
						datum.setStatus(CrawlDatum.STATUS_FETCH_SUCCESS);
						generateOut.append(textUrl, datum);

						datum.setStatus(CrawlDatum.STATUS_DB_FETCHED);
						crawldbOut.append(textUrl, datum);
					} else {
						datum.setStatus(CrawlDatum.STATUS_LINKED);
						crawldbOut.append(textUrl, datum);
					}
					output.collect(key, textUrl);

					if (0==((i / generator.slots) % 10000)) {
						log.info("still running: " + i + " of <" + generator.pages + ", " + generator.totalpages +">");
					}
					i = i + generator.slots;
				}

				crawlOut.close();
				crawldbOut.close();
				generateOut.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	private static class CombineReferences extends MapReduceBase implements
	Reducer<Text, References, Text, References> {

		@Override
		public void reduce(Text key, Iterator<References> values,
				OutputCollector<Text, References> output, Reporter reporter)
				throws IOException {
			
			References sum = null;
			while (values.hasNext()) {
				References links = values.next();
				if (links.len > 0) {
					if (null == sum) {
						sum = links;
					} else {
						long[] ids = new long[links.len + sum.len];
						int j = 0;
						for (int i=0; i<sum.len; i++) ids[j++] = sum.refs[i];
						for (int i=0; i<links.len; i++) ids[j++] = links.refs[i];
						sum = new References(ids.length, ids);
					}
				} else {
					output.collect(key, links);
				}
			}
		}
	}
	
	private static class CreateLinks extends MapReduceBase implements
	Reducer<Text, References, Text, NutchParse> {

//		private static final Log log = LogFactory.getLog(CreateLinks.class.getName());
		
		private HtmlCore generator;
		private IndexedMapFile indexedUrls;
		private String segName;
		private long[] cost;

		public void configure(JobConf job) {

			try {
				generator = new HtmlCore(job);
				
				indexedUrls = Utils.getSharedMapFile(URLS_DIR_NAME, job);

				int pid = job.getInt("mapred.task.partition", 0);
				generator.fireRandom(pid * 1000 + 1);
				segName = job.get(Nutch.SEGMENT_NAME_KEY);
				
				cost = new long[6];
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		@Override
		public void reduce(Text key, Iterator<References> values,
				OutputCollector<Text, NutchParse> output, Reporter reporter)
				throws IOException {

//			log.info("0> one new value ...");
			Date d = new Date();
			long[] start = new long[7];
			
			start[0] = d.getTime();
			References olinks = null;
			References ilinks = null;
			while (values.hasNext()) {
				References links = values.next();
				if (links.len > 0) {
					if (null == ilinks) {
						ilinks = links;
					} else {
						long[] ids = new long[links.len + ilinks.len];
						int j = 0;
						for (int i=0; i<ilinks.len; i++) ids[j++] = ilinks.refs[i];
						for (int i=0; i<links.len; i++) ids[j++] = links.refs[i];
						ilinks = new References(ids.length, ids);
					}
				} else {
					olinks = links;
				}
			}
//			log.info("1> olinks & ilinks ready ...");
			d = new Date();
			start[1] = d.getTime();
			
			String[] txtContent = generator.genPageWordsAndTitls();
			ParseText text = new ParseText(txtContent[0]);
//			log.info("2> text & titles ready ...");
			d = new Date();
			start[2] = d.getTime();
			
			Outlink[] outlinks = new Outlink[-olinks.len];
			for (int i=0; i<-olinks.len; i++) {
				outlinks[i] = new Outlink(indexedUrls.get(olinks.refs[i]).toString());
			}
			d = new Date();
			start[3] = d.getTime();
			
			Metadata contentMeta = new Metadata();
			contentMeta.add(Nutch.SEGMENT_NAME_KEY, segName);
		    contentMeta.add(Nutch.SIGNATURE_KEY,
		    		StringUtil.toHexString(MD5Hash.digest(txtContent[0].getBytes()).getDigest()));

			ParseData data = new ParseData(new ParseStatus(ParseStatus.SUCCESS), txtContent[1], outlinks, contentMeta, new Metadata());
//			log.info("3> outlinks ready ...");
			d = new Date();
			start[4] = d.getTime();
			
			Inlinks inlinks = new Inlinks();
			if (null != ilinks) {
				for (int i=0; i<ilinks.len; i++) {
					inlinks.add(new Inlink(indexedUrls.get(ilinks.refs[i]).toString()));
				}
			}			
//			log.info("4> inlinks ready ...");
			d = new Date();
			start[5] = d.getTime();
			
			NutchParse parse = new NutchParse(inlinks, text, data);
			output.collect(key, parse);

//			log.info("5> output finished ...");
			d = new Date();
			start[6] = d.getTime();
			for (int i=0; i<cost.length; i++) {
				cost[i] = cost[i] + start[i+1] - start[i];
			}
		}
		
		public void close() {
//			log.info("5>>>>>>>> all finished ...");
			log.info("<<Time>> o/ilinks: " + cost[0] + ", text: " + cost[1]
					+ ", outlinks: " + cost[2] + ", md5: " + cost[3]
					+ ", inlinks: " + cost[4] + ", write: " + cost[5]);
		}
	}
	
	private static class CreateNutchPages extends MapReduceBase implements
	Mapper<LongWritable, Text, Text, References> {

		private static final Log log = LogFactory.getLog(CreateNutchPages.class.getName());
		
		private HtmlCore generator;
		IndexedMapFile indexedUrls;

		public void configure(JobConf job) {
			try {
				generator = new HtmlCore(job);
				indexedUrls = Utils.getSharedMapFile(URLS_DIR_NAME, job);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, References> output, Reporter reporter)
				throws IOException {
			
			int slotId = Integer.parseInt(value.toString().trim());
			generator.fireRandom(slotId);
			
			HashMap<Long, References> hash = new HashMap<Long, References>();
			try {
				long i = slotId - 1;
				while (i < generator.pages) {
					
					References olinks = generator.genPageLinks();
					
					for (int j=0; j<-olinks.len; j++) {
						long to = olinks.refs[j];
						References froms = hash.get(to);
						if (null != froms) {
							if (froms.len == froms.refs.length) {
								output.collect(indexedUrls.get(to), froms);
								froms.len = 0;
							}
						} else {
							if (hash.size() > CACHE_REF_SET_SIZE) {
								for (Entry<Long, References> entry : hash.entrySet()) {
									output.collect(indexedUrls.get(entry.getKey()), entry.getValue());
								}
								hash.clear();
							}
							froms = new References(0, new long[CACHE_REF_ITEM_LENGTH]);
							hash.put(to, froms);
						}
						froms.refs[froms.len++] = i;
					}
					output.collect(indexedUrls.get(i), olinks);
					
					if (0==((i / generator.slots) % 10000)) {
						log.info("still running: " + i + " of " + generator.pages);
					}

					i = i + generator.slots;
				}
				
				for (Entry<Long, References> entry : hash.entrySet()) {
					output.collect(indexedUrls.get(entry.getKey()), entry.getValue());
				}
				hash.clear();
				indexedUrls.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
	}

	private void setNutchOptions(JobConf job) throws URISyntaxException {

		job.setLong("pages", options.getNumPages() );
		job.setLong("slotpages", options.getNumSlotPages());

		Utils.shareLinkZipfCore(options, job);
		Utils.shareWordZipfCore(options, job);
		
		job.set(NUTCH_WORK_DIR_PARAM_NAME , options.getResultPath().toString());
		
//		job.set(DataOptions.URL_MAP_NAME, paths.getPath(DataPaths.URLS).toString());
	}
	
	public String generateSegmentName() {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		return sdf.format(new Date(System.currentTimeMillis()));
	}

	private void createNutchUrls() throws IOException, URISyntaxException {

		log.info("Creating nutch urls ...");
		
		JobConf job = new JobConf(NutchData.class);
		Path urls = new Path(options.getWorkPath(), URLS_DIR_NAME);
		Utils.checkHdfsPath(urls);
		
		String jobname = "Create nutch urls";
		job.setJobName(jobname);

		setNutchOptions(job);
		
		FileInputFormat.setInputPaths(job, dummy.getPath());
		job.setInputFormat(NLineInputFormat.class);

		job.setMapperClass(CreateUrlHash.class);
		job.setNumReduceTasks(0);

		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputFormat(MapFileOutputFormat.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
		MapFileOutputFormat.setOutputPath(job, urls);
		
//		SequenceFileOutputFormat.setOutputPath(job, fout);
/*		
		SequenceFileOutputFormat.setCompressOutput(job, true);
		SequenceFileOutputFormat.setOutputCompressorClass(job, DefaultCodec.class);
		SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);
*/		

		log.info("Running Job: " +jobname);
		log.info("Pages file " + dummy.getPath() + " as input");
		log.info("Rankings file " + urls + " as output");
		JobClient.runJob(job);
		log.info("Finished Running Job: " + jobname);

		log.info("Cleaning temp files...");
		Utils.cleanTempFiles(urls);
	}
	
	private void createNutchIndexData() throws IOException, URISyntaxException {

		log.info("creating nutch index files ... ");

		JobConf job = new JobConf(NutchData.class);
		
		Utils.shareUrls(URLS_DIR_NAME, options, job);
		Utils.shareDict(options, job);
		
		setNutchOptions(job);
		
		Path fsegments = new Path(options.getResultPath(), SEGMENTS_DIR_NAME);
		Utils.checkHdfsPath(fsegments, true);
		
		segment = new Path(fsegments, generateSegmentName());
		Utils.checkHdfsPath(segment, true);
		
		String jobname = "Create nutch index data";
		job.setJobName(jobname);
		
		job.set(Nutch.SEGMENT_NAME_KEY, segment.getName());
		
		FileInputFormat.setInputPaths(job, dummy.getPath());
		job.setInputFormat(NLineInputFormat.class);

		job.setMapperClass(CreateNutchPages.class);
		job.setCombinerClass(CombineReferences.class);
		job.setReducerClass(CreateLinks.class);
		
		if (options.getNumReds() > 0) {
			job.setNumReduceTasks(options.getNumReds());
		} else {
			job.setNumReduceTasks(Utils.getMaxNumMaps());
		}
		
		FileOutputFormat.setOutputPath(job, segment);
		job.setOutputFormat(NutchOutputFormat.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(References.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NutchParse.class);

		log.info("Running Job: " + jobname);
		log.info("Pages file " + dummy.getPath() + " as input");
		log.info("Rankings file " + segment + " as output");
		JobClient.runJob(job);
		log.info("Finished Running Job: " + jobname);

		log.info("Cleaning temp files...");
		Utils.cleanTempFiles(segment);
	}
	
	private void init() throws IOException {

		log.info("Initializing Nutch data generator...");

		Utils.checkHdfsPath(options.getResultPath(), true);
		Utils.checkHdfsPath(options.getWorkPath(), true);
		
		int words = RawData.putDictToHdfs(new Path(options.getWorkPath(), HtmlCore.getDictName()), options.getNumWords());
		options.setNumWords(words);
		
		Utils.serialLinkZipf(options);
		Utils.serialWordZipf(options);

		dummy = new Dummy(options.getWorkPath(), options.getNumMaps());
	}
/*	
	private void test2LevelMapFile(Path furl) throws IOException {

		JobConf job = new JobConf();
		FileSystem fs = FileSystem.get(job);
		MapFile.Reader reader = new MapFile.Reader(fs, furl.toString(), job);
		Text value = new Text();
		reader.get(new LongWritable(1000), value);
		if (null != value) {
			log.info("---Find it: <1000, " + value + ">");
		}
	}
*/	
	public void generate() throws Exception {
		
		init();
		createNutchUrls();
		createNutchIndexData();
		
		Path ffetch = new Path(options.getResultPath(), CrawlDatum.FETCH_DIR_NAME);
		Path fparse = new Path(options.getResultPath(), CrawlDatum.PARSE_DIR_NAME);
		Path linkdb = new Path(segment, LINKDB_DIR_NAME);
		
		FileSystem fs = ffetch.getFileSystem(new Configuration());
		fs.rename(ffetch, new Path(segment, CrawlDatum.FETCH_DIR_NAME));
		fs.rename(fparse, new Path(segment, CrawlDatum.PARSE_DIR_NAME));
		fs.rename(linkdb, new Path(options.getResultPath(), LINKDB_DIR_NAME));
		fs.close();
		
		close();
	}
	
	private void close() throws IOException {
		log.info("Closing nutch data generator...");
		Utils.checkHdfsPath(options.getWorkPath());
	}
	
	public static final String getLinkDbName() {
		return LINKDB_DIR_NAME;
	}
}
