import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.HashMap;

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 DynamicTrendReport {
	public static final int DATE_TIME_IDX = 0;
	public static final int FILESYSTEM_IDX = 2;
	public static final int SPACE_USED_IDX = 4;
	public static final int PERCENT_IDX = 5;
	public static final String INITIAL_DATE = "INITIAL_DATE";
	public static final String MONTH_INTERVAL = "MONTH_INTERVAL";
	
	/* 
	 * Given start date, start date and month interval
	 * return file system as key AND
	 * average percent and aggregate space used as value  
	 */
	public static void main(String []args) throws Exception{
		Configuration conf = new Configuration();
		String[] programArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if(programArgs.length != 5){
			System.err.println("Usage: DynamicTrendReport <in> <out> <start-date> <time> <month-interval>");
			System.exit(1);
		}
		conf.set(INITIAL_DATE, (programArgs[2] + " " + programArgs[3]));
		conf.set(MONTH_INTERVAL, programArgs[4]);
		@SuppressWarnings("deprecation")
		Job job = new Job(conf, "Dynamic Trend Report");
		job.setJarByClass(DynamicTrendReport.class);
		job.setMapperClass(DynamicTrendReportMapper.class);
		job.setReducerClass(DynamicTrendReportReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(PercentSpaceWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(PercentSpaceWritable.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 DynamicTrendReportMapper extends Mapper<LongWritable, Text, Text, PercentSpaceWritable>{
		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;
			Date currentDate = 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
				currentDate = new Date();								//current date
			}catch(ParseException e){
				System.err.println(e);
				System.exit(1);
			}
			int diffMonth = getMonths(startDate, currentDate);
			if(logEntryDate.after(startDate)){
				int monthInterval;
				try{
					monthInterval = Integer.parseInt(conf.get(MONTH_INTERVAL));		//user provided month interval for analysis
				}catch(Exception e){
					System.err.println(e);
					return;
				}
				
				/* construct map of intervals */
				int idx = 0;
				int intervalIdx = monthInterval;
				HashMap<Integer,Integer> intervals = new HashMap<>();
				while(intervalIdx <= diffMonth){
					intervals.put(idx, intervalIdx);
					idx++;
					intervalIdx += monthInterval;
				}
				if((intervalIdx - monthInterval) != diffMonth){
					intervals.put(idx, diffMonth);
				}
				System.out.println(intervals);
				
				String interval = processEntry(intervals, monthInterval, startDate, logEntryDate, currentDate);	//determine interval for this date
				if(!interval.equals("INVALID")){
					String system = logInInfo[FILESYSTEM_IDX];
					system = (system + " " + interval);
					final Text fileSystem = new Text(system);
					
					int percentUsed = 0;
					if(!(logInInfo[PERCENT_IDX].equals("N/A"))){
						percentUsed = Integer.parseInt(logInInfo[PERCENT_IDX]);
					}
					final DoubleWritable percent = new DoubleWritable(percentUsed);			
					
					String space = logInInfo[SPACE_USED_IDX];
					final LongWritable spaceUsed = new LongWritable(Long.parseLong(space));
					
					PercentSpaceWritable percentAndSpace = new PercentSpaceWritable(percent, spaceUsed);
					context.write(fileSystem, percentAndSpace);
				}
			}
		}
	}
	
	public static class DynamicTrendReportReducer extends Reducer<Text, PercentSpaceWritable, Text, PercentSpaceWritable>{
		public void reduce(Text key, Iterable<PercentSpaceWritable> values, Context context) throws IOException, InterruptedException{
			int percentCount = 0;
			int percentInstances = 0;
			int spaceUsedAggregate = 0;
			
			for(PercentSpaceWritable currentPercentAndSpace: values){
				percentCount += currentPercentAndSpace.getPercent().get();
				percentInstances++;
				spaceUsedAggregate += currentPercentAndSpace.getSpace().get();
			}
			double avg = (double)(percentCount/percentInstances);
			context.write(key, new PercentSpaceWritable(new DoubleWritable(avg), new LongWritable(spaceUsedAggregate)));
		}
	}
	
	public static String processEntry(HashMap<Integer,Integer> intervals, int monthInterval, Date initialDate, Date dateInstance, Date currentDate){
		StringBuilder sb = new StringBuilder();
		if(dateInstance.after(currentDate)){
			sb.append("INVALID");
		} else{
			if(initialDate.after(dateInstance)){
				sb.append("INVALID");
			} else{
				int diffMonth = getMonths(initialDate, dateInstance);
				int idx = (diffMonth/monthInterval);
				System.out.println(idx);
				int endInterval = intervals.get(idx);
				if(idx != 0){
					idx = intervals.get(idx-1);
				}
				sb.append((idx+1) + "-" + endInterval);
			}
		}
		return new String(sb);
	}
	
	public static int getMonths(Date startDate, Date endDate){
		Calendar startCalendar = new GregorianCalendar();
		startCalendar.setTime(startDate);
		Calendar endCalendar = new GregorianCalendar();
		endCalendar.setTime(endDate);
		
		int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);
		int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);
		return diffMonth;
	}
}
