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 SearchSpaceUsed {
	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_USED_IDX = 4;
	public static final String DATE_WAS_FILTERED = "DATE_WAS_FILTERED";
	public static final String USER_NAME_WAS_FILTERED = "USER_NAME_WAS_FILTERED";
	public static final String FILE_SYSTEM_WAS_FILESYSTEM = "FILE_SYSTEM_WAS_FILESYSTEM";
	public static final String FILTERED_USER_NAME = "FILTERED_USER_NAME";
	public static final String FILTERED_START_DATE = "FILTERED_START_DATE";
	public static final String FILTERED_END_DATE = "FILTERED_END_DATE";
	public static final String FILTERED_FILESYSTEM = "FILTERED_FILESYSTEM";
	
	/* 
	 * Given -optionally- username, start date, start time, end date, end time and file system
	 * return username as key 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 == 8){
			conf.set(FILTERED_USER_NAME, programArgs[2]);
			conf.setBoolean(USER_NAME_WAS_FILTERED, true);
			conf.set(FILTERED_START_DATE, programArgs[3] + " " + programArgs[4]);
			conf.set(FILTERED_END_DATE, programArgs[5] + " " + programArgs[6]);
			conf.setBoolean(DATE_WAS_FILTERED, true);
			conf.set(FILTERED_FILESYSTEM, programArgs[7]);
			conf.setBoolean(FILE_SYSTEM_WAS_FILESYSTEM, true);
		} else if(programArgs.length == 7){
			conf.set(FILTERED_USER_NAME, programArgs[2]);
			conf.setBoolean(USER_NAME_WAS_FILTERED, true);
			conf.set(FILTERED_START_DATE, programArgs[3] + " " + programArgs[4]);
			conf.set(FILTERED_END_DATE, programArgs[5] + " " + programArgs[6]);
			conf.setBoolean(DATE_WAS_FILTERED, true);
			conf.setBoolean(FILE_SYSTEM_WAS_FILESYSTEM, false);
		} else if(programArgs.length == 3){
			conf.set(FILTERED_USER_NAME, programArgs[2]);
			conf.setBoolean(USER_NAME_WAS_FILTERED, true);
			conf.setBoolean(DATE_WAS_FILTERED, false);
			conf.setBoolean(FILE_SYSTEM_WAS_FILESYSTEM, false);
		} else if(programArgs.length == 2){
			conf.setBoolean(USER_NAME_WAS_FILTERED, false);
			conf.setBoolean(DATE_WAS_FILTERED, false);
			conf.setBoolean(FILE_SYSTEM_WAS_FILESYSTEM, false);
		} else{
			System.err.println("Usage: SearchSpaceUsed <in> <out> <optional - username> <optional - startdate> " +
		    			"<optional - starttime> <optional - enddate> <optional - endtime> <optional - filesystem>");
			System.exit(1);
		}
		@SuppressWarnings("deprecation")
		Job job = new Job(conf, "Search Space Used");
		job.setJarByClass(SearchSpaceUsed.class);
		job.setMapperClass(SearchSpaceUsedMapper.class);
		job.setReducerClass(SearchSpaceUsedReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.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 SearchSpaceUsedMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
			/* parse log entry file */
			String[] logInInfo = value.toString().split(",");
			for(int i = 1; i < logInInfo.length; i++){
				logInInfo[i] = logInInfo[i].replace(" ", "");
			}
			Configuration conf = context.getConfiguration();
			if(conf.getBoolean(DATE_WAS_FILTERED, false)){
				String logInDate = logInInfo[DATE_TIME_IDX];
				SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				Date startDate = null;
				Date endDate = null;
				Date logInEntryDate = null;
				try{
					startDate = sdf.parse(conf.get(FILTERED_START_DATE));			//start date provided by user
					endDate = sdf.parse(conf.get(FILTERED_END_DATE));			    //end date provided by user
					logInEntryDate = sdf.parse(logInDate);							//current log entry date
				}catch(ParseException e){
					System.err.println(e);
					System.exit(1);
				}
				if(startDate.after(logInEntryDate) || endDate.before(logInEntryDate)){
					return;
				}
			}
			
			/* retrieve username */
			String userName = logInInfo[USERNAME_IDX];
			String filteredUserName = new String();

			boolean userNameFiltered = conf.getBoolean(USER_NAME_WAS_FILTERED, false);
			if(userNameFiltered){
				filteredUserName = conf.get(FILTERED_USER_NAME);
			} else{
				filteredUserName = userName;
			}
			
			/*retrieve filesystem*/
			String filteredFileSystem = new String();
			String system = logInInfo[FILESYSTEM_IDX];
			boolean fileSystemWasFiltered = conf.getBoolean(FILE_SYSTEM_WAS_FILESYSTEM, false);
			if(fileSystemWasFiltered){
				filteredFileSystem = conf.get(FILTERED_FILESYSTEM);
			} else{
				filteredFileSystem = system;
			}
			
			if(filteredUserName.equals(userName)){
				final Text user = new Text(userName);
	
				if(filteredFileSystem.equals(system)){
					/* retrieve quota used */
					long quotaUsed = Long.parseLong(logInInfo[SPACE_USED_IDX]);
					final LongWritable spaceUsed = new LongWritable(quotaUsed);
					context.write(user, spaceUsed);
				}
			}
		}
	}
	
	
	public static class SearchSpaceUsedReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
		public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException{
			long quotaUsed = 0;
			for(LongWritable value: values){
				quotaUsed += value.get();
			}
			context.write(key, new LongWritable(quotaUsed));
		}
	}
}
