import java.sql.*;
import java.util.HashMap;
import java.util.Map;
import java.io.*;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.ho.yaml.Yaml;

import Reports.ProcessedDigSales;
import Utility.FileUtility;


public class ExportReport extends Thread
{
    private static Map<String, Object> parameters = new HashMap<String, Object>();
    private static int NUM_OF_THREADS = 50;
    private static int c_nextId = 1;
    
    int thread_id;
    
    public ExportReport(){
        super();
        // Assign an ID to the thread
        thread_id = getNextId();
     }

    synchronized static int getNextId()
    {
        return c_nextId++;
    }
    
    public static void main (String args [])
    {
    	Options options = new Options();
		options.addOption("configFilePath",true,"(Optional) Config.yml file for this cli tool (Default: /mnt/data/ETL/java/accounting/config/config.yml)");
		options.addOption("localPath",true,"(Optional) Local path where files are stored. (Default: /mnt/data/tmp/processed_dig_sales)");
		options.addOption("period_id",true,"Accounting period id.");
		
		
        try  
        {  
        	/*
        	 * Getting parameters from yml files
        	 */
        	String configFilePath = "/mnt/data/ETL/java/accounting/config/config.yml";
        	String period_id = "";
        	CommandLineParser parser = new GnuParser();
			CommandLine cmd = parser.parse( options, args);
			if(cmd.hasOption("configFilePath")){
				configFilePath = cmd.getOptionValue("configFilePath");
			}
        	Map<String,Object> conf = new HashMap<String,Object>();
        	conf = (Map)Yaml.load(new File(configFilePath));
      	    String folder = (String)conf.get("folder");
      	    parameters.put("folder", folder);
      	    parameters.put("filename", folder+conf.get("filename"));
      	    parameters.put("splittedFilesExtension", conf.get("splittedFilesExtension"));
      	    parameters.put("dsn",conf.get("dsn"));
      	    parameters.put("username", conf.get("username"));
      	    parameters.put("password", conf.get("password"));
      	    NUM_OF_THREADS = (Integer)conf.get("num_of_threads");

      	    /*
      	     * Using parameters in yml file, get a connection and init a report object
      	     * ToDo: This can be further abstracted out to a factory method or class which spit out the specific report object
      	     */
            DriverManager.registerDriver (new com.mysql.jdbc.Driver());
      	    Connection conn = DriverManager.getConnection((String)parameters.get("dsn"), (String)parameters.get("username"), (String)parameters.get("password"));
    	  	ProcessedDigSales report = new ProcessedDigSales(conn);
      	    if(cmd.hasOption("period_id")){
				period_id = cmd.getOptionValue("period_id");
				parameters.put("period_id", period_id);
			}else{
				period_id = report.getMaxAccountingPeriodId();
				parameters.put("period_id", period_id);
			}
      	    System.out.println("Exporting period: "+parameters.get("period_id"));
      	    
      	    if(args.length==0 && period_id.equals("")){
      	    	HelpFormatter formatter = new HelpFormatter();
      	    	formatter.printHelp("ant run -Dargs='paramA=xxx paramB=yyy .....'", options);
      	    	System.exit(1);
  		    }
            
      	  	/*
      	  	 * Getting min_id, max_id and records per thread information from report object
      	  	 */
            Map<String, Object> record_info = report.getRecordsPerThread(parameters, NUM_OF_THREADS);
    	    parameters.put("min_id", record_info.get("min_id"));
    	    parameters.put("recordsPerThread", record_info.get("recordsPerThread"));
            
            // Create the threads
            Thread[] threadList = new Thread[NUM_OF_THREADS];
            // spawn threads
            for (int thread_id = 0; thread_id < NUM_OF_THREADS; thread_id++)
            {
                threadList[thread_id] = new ExportReport();
                threadList[thread_id].start();
            }
            // wait for all threads to end
            for (int i = 0; i < NUM_OF_THREADS; i++)
            {
                    threadList[i].join();
            }
            //gzip all files one by one
            FileUtility.gzipFiles((String)parameters.get("folder"), (String)parameters.get("splittedFilesExtension"));
        }
        catch (Exception e)
        {
        	System.out.println(e.getMessage());
            e.printStackTrace();
        }
    }
    
    
    
    public void run(){
	      Connection conn = null;
	      ResultSet     rs   = null;
	      Statement  stmt = null;
	      
		  long min = (Long)parameters.get("min_id") + ((thread_id-1) * (Long)parameters.get("recordsPerThread"));
	      long max = ((Long)parameters.get("min_id") + ((thread_id) * (Long)parameters.get("recordsPerThread"))) + (thread_id==NUM_OF_THREADS ? 0 : -1);
	      String dsn = (String)parameters.get("dsn");
	      String username = (String)parameters.get("username");
	      String password = (String)parameters.get("password");
	      
	      System.out.println("Thread "+thread_id+" extracting data from: "+min+" to: "+max);
	      
	      try
	      {    
	            conn = DriverManager.getConnection(dsn,username, password);
	            stmt = conn.createStatement (java.sql.ResultSet.TYPE_FORWARD_ONLY,java.sql.ResultSet.CONCUR_READ_ONLY);
	            stmt.setFetchSize(Integer.MIN_VALUE);
	            
	            
	            //Static method to get report specific SQL
	            String sql = ProcessedDigSales.getSQL(parameters, min, max);
	            
	            
	            rs = stmt.executeQuery (sql);
	            ResultSetMetaData rsmd = rs.getMetaData();
	            
	            BufferedWriter fw = new BufferedWriter(new FileWriter((String)parameters.get("filename")+"_"+thread_id+(String)parameters.get("splittedFilesExtension"))); 
	            
	            // Loop through the results
	            while (rs.next()){
	            	int numberOfColumns = rsmd.getColumnCount();
	            	String row = "";
	            	for (int i = 1; i < numberOfColumns + 1; i++ ) {
	            		  row += rs.getString(rsmd.getColumnLabel(i));
	            		  if(i==numberOfColumns){
	            			  row += "\n";
	            		  }else{
	            			  row += "\t";
	            		  }
	            	}
	            	fw.write(row);
	            }
	            
	            // Close all the resources
	            fw.close();
	            rs.close();
	            stmt.close();
	            if (conn != null)
	                conn.close();
	            System.out.println("Thread " + thread_id +  " is finished. ");
	      }
	      catch (Exception e)
	      {
	          System.out.println("Thread " + parameters.get("thread_id") + " got Exception: " + e);
	          e.printStackTrace();
	          return;
	      }
    }
  
    
    
   
    
    
}