import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class TopTenQuotaReport {
	public static final int DATE_TIME_IDX = 0;
	public static final int USERNAME_IDX = 1;
	public static final int PERCENT_IDX = 5;
	public static final String INITIAL_DATE = "INITIAL_DATE";
	
	public static void main(String []args) throws Exception{
		Configuration conf = new Configuration();
		String[] programArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if(programArgs.length != 4){
			System.err.println("Usage: DynamicTrendReport <in> <out> <start-date> <start-time>");
			System.exit(1);
		}
		conf.set(INITIAL_DATE, (programArgs[2] + " " + programArgs[3]));
		@SuppressWarnings("deprecation")
		Job job = new Job(conf, "Dynamic Trend Report");
		job.setJarByClass(TopTenQuotaReport.class);
		job.setMapperClass(TopTenQuotaReportMapper.class);
		job.setReducerClass(TopTenQuotaReportReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(programArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(programArgs[1]));
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
	
	public static class TopTenQuotaReportMapper extends Mapper<LongWritable, Text, Text, DoubleWritable>{
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
			String[] logInInfo = value.toString().split(",");
			for(int i = 1; i < logInInfo.length; i++){
				logInInfo[i] = logInInfo[i].replace(" ", "").replace("%", "");
			}
			Configuration conf = context.getConfiguration();
			
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			Date startDate = null;
			Date logEntryDate = null;
			try{
				startDate = sdf.parse(conf.get(INITIAL_DATE));			//start date provided by user
				logEntryDate = sdf.parse(logInInfo[DATE_TIME_IDX]);		//current log entry date
			}catch(ParseException e){
				System.err.println(e);
				System.exit(1);
			}
			if(logEntryDate.after(startDate)){
				String user = logInInfo[USERNAME_IDX];
				final Text userName = new Text(user);
				
				int percentUsed = 0;
				if(!(logInInfo[PERCENT_IDX].equals("N/A"))){
					percentUsed = Integer.parseInt(logInInfo[PERCENT_IDX]);
				}
				final DoubleWritable percent = new DoubleWritable(percentUsed);			
				
				context.write(userName, percent);
			}
		}
	}
	
	public static class TopTenQuotaReportReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable>{
		public void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException{
			double percentCount = 0;
			int percentInstances = 0;
			
			for(DoubleWritable currentPercent: values){
				percentCount += currentPercent.get();
				percentInstances++;
			}
			double avg = (double)(percentCount/percentInstances);
			context.write(key, new DoubleWritable(avg));
		}
	}
}
