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.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 SearchUserStatistics {
	public static final int DATE_TIME_IDX = 0;
	public static final int USERNAME_IDX = 1;
	public static final int FILESYSTEM_IDX = 2;
	public static final int SPACE_ALLOCATED_IDX = 3;
	public static final int SPACE_USED_IDX = 4;
	public static final String FILTERED_USER_NAME = "FILTERED_USER_NAME";
	public static final String INITIAL_DATE = "INITIAL_DATE";
	public static final String FINAL_DATE = "FINAL_DATE";
	
	/* 
	 * Given username, start date, start time, end date, end time
	 * return username as key AND
	 * Log Entry as value 
	 */
	public static void main(String []args) throws Exception{
		Configuration conf = new Configuration();
		String[] programArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if(programArgs.length != 7){
			System.err.println("Usage: SearchUserStatistics <in> <out> <username> <start-date> <start-hour> <end-date> <end-hour>");
			System.exit(1);
		}
		conf.set(FILTERED_USER_NAME, programArgs[2]);
		conf.set(INITIAL_DATE, (programArgs[3] + " " + programArgs[4]));
		conf.set(FINAL_DATE, (programArgs[5] + " " + programArgs[6]));
		@SuppressWarnings("deprecation")
		Job job = new Job(conf, "Search User Statistics");
		job.setJarByClass(SearchUserStatistics.class);
		job.setMapperClass(SearchUserStatisticsMapper.class);
		job.setReducerClass(SearchUserStatisticsReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LogEntryWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(programArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(programArgs[1]));
		
		/* Submit job and wait for completion to finish */
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
	
	public static class SearchUserStatisticsMapper extends Mapper<LongWritable, Text, Text, LogEntryWritable>{
		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(" ", "");
			}
			String userName = logInInfo[USERNAME_IDX];	
			
			Configuration conf = context.getConfiguration();
			String requestedUserName = conf.get(FILTERED_USER_NAME);
			
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			Date startDate = null;
			Date endDate = null;
			Date currentLogInDate = null;
			
			if(requestedUserName.equals(userName)){
				final Text user = new Text(userName);
		
				String dateAndTime = logInInfo[DATE_TIME_IDX];		
				try{
					currentLogInDate = sdf.parse(dateAndTime);
					startDate = sdf.parse(conf.get(INITIAL_DATE));			//start date provided by user
					endDate = sdf.parse(conf.get(FINAL_DATE));		//current log entry date
				}catch(ParseException e){
					System.err.println(e);
					System.exit(1);
				}
				
				if(currentLogInDate.after(startDate) && currentLogInDate.before(endDate)){
					final Text dateTime = new Text(dateAndTime);
						
					String system = logInInfo[FILESYSTEM_IDX];
					final Text fileSystem = new Text(system);
					
					long allocated = Long.parseLong(logInInfo[SPACE_ALLOCATED_IDX]);
					final LongWritable spaceAllocated = new LongWritable(allocated);
					
					long used = Long.parseLong(logInInfo[SPACE_USED_IDX]);
					final LongWritable spaceUsed = new LongWritable(used);
					
					LogEntryWritable logEntry = new LogEntryWritable(dateTime, fileSystem, spaceAllocated, spaceUsed);
					context.write(user, logEntry);
				}
			}
		}
	}
	
	
	public static class SearchUserStatisticsReducer extends Reducer<Text, LogEntryWritable, Text, LogEntryWritable>{
		public void reduce(Text key, Iterable<LogEntryWritable> values, Context context) throws IOException, InterruptedException{
			for(LogEntryWritable logEntry: values){
				context.write(key, logEntry);
			}
		}
	}
	
}
