package scripts;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;

public class AddCategoryScript {
	
	private static String tweet_table = "bb_tweets";
	private static int fetchsize = 10000;
	private static TopicClassification classifier;
	private static int batchcounter = 0;
	static ResultSet rs = null;
	static ArrayList<Tuple<Long, String>> list = null;
	private static Semaphore sem = new Semaphore(5);

	public static void main(String[] args) {
		// http://alias-i.com/lingpipe/demos/tutorial/classify/read-me.html
		System.out.println("LOAD Model Newsgroup Model ... ");
		classifier = new TopicClassification();
		try {
			classifier.trainclassifier();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.out.println("LOAD LingPipe 20 Newsgroup Model >>> DONE \n");
		
//		worker();
		
		Connection c = getConnection();
		String query = "Select tweet_id, tweet_content from "+tweet_table+" where category is null";
		try {
			c.setAutoCommit(false);
			Statement st = c.createStatement();
			st.setFetchSize(fetchsize);
			rs = st.executeQuery(query);
			list = new ArrayList<>();
			int counter = 0;
			while (rs.next()) {
				long id = Long.parseLong(rs.getString(1));
				String text = rs.getString(2);
				Tuple<Long, String> tup = new Tuple<>(id, text);
				list.add(tup);
				counter++;
				if ( counter == fetchsize) {
					counter = 0;
					addCategories(list);
					list.clear();
					if (batchcounter++ == 100) {
						System.out.println("-");
						batchcounter = 0;
					} else
						System.out.print("-");
				}
			}
		}  catch (SQLException e) {
			e.printStackTrace();
		}
	}
	
	public static class Tuple<A, B> {
		
		private A a;
		private B b;

		public Tuple(A a, B b) {
			this.a = a;
			this.b = b;
		}

		public A getA() {
			return a;
		}

		public void setA(A a) {
			this.a = a;
		}

		public B getB() {
			return b;
		}

		public void setB(B b) {
			this.b = b;
		}
		
	}
	
	
	private static void addCategories(ArrayList<Tuple<Long, String>> list) throws SQLException {
		
		int batchsize = 2000;
//		for (Tuple<Long, String> t : list) {
//			t.setB(classifier.getCategory(t.getB()));
//		}
		
		List<Tuple<Long, String>> collect = list.parallelStream().map(s -> new Tuple<Long, String>(s.getA(), 
				classifier.getCategory(s.getB()))).collect(Collectors.toList());
		
		
		Connection c = getConnection();
		c.setAutoCommit(false);
		
		Statement st = c.createStatement();
		// Update
		int counter = 0;
		for (Tuple<Long, String> t : collect) {
			String query = "Update " + tweet_table + " set category = '" + t.getB() + "' Where tweet_id = " + t.getA();
			// batch Update
			st.addBatch(query);
			counter++;
			if (counter == batchsize) {
				st.executeBatch();
				c.commit();
				counter = 0;
			}
		}
		st.close();
		c.close();
	}
	

	private static Connection getConnection() {
		Connection c = null;
		String host = "db.dbvis.de";
		String port = "5432";
		String dbname = "socialoceandb";
		String username = "socialocean";
		String pw = "blFDvUic4DL0V3ODkvbK";
		
		String connection_str = "jdbc:postgresql://"+host+":"+port+"/"+dbname+"?ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory";
		try {
			c = DriverManager.getConnection(connection_str.trim(), username.trim(), pw.trim());
		} catch (SQLException e) {
			System.err.println("Could not connect to DB \n"
					+ ""+connection_str);
			e.printStackTrace();
		}
		return c;
	}
		
}

@@ -237,47 +237,8 @@
 st;%0A
-import java.util.concurrent.Semaphore;%0A
 impo
@@ -564,58 +564,8 @@
 ull;
-%0A%09private static Semaphore sem = new Semaphore(5);
 %0A%0A%09p
@@ -2128,19 +2128,16 @@
  b;%0A%09%09%7D%0A
-%09%09%0A
 %09%7D%0A%09%0A%09%0A%09
@@ -2250,17 +2250,17 @@
 hsize = 
-2
+1
 000;%0A//%09

clean up of code -- DONE