D-Net

package eu.dnetlib.usagestats.export;

import java.io.*;

import java.net.URLDecoder;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.SQLException;

import java.sql.Statement;

import java.sql.Timestamp;

import java.text.SimpleDateFormat;

import java.util.*;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.RemoteIterator;

import org.apache.log4j.Logger;

import org.json.simple.JSONArray;

import org.json.simple.JSONObject;

import org.json.simple.parser.JSONParser;

public class PiwikStatsDB {

    private String logPath;

    private String logRepoPath;

    private String logPortalPath;

    private Statement stmt = null;

    private final Logger log = Logger.getLogger(this.getClass());

    private String CounterRobotsURL;

    private ArrayList robotsList;

    public PiwikStatsDB(String logRepoPath, String logPortalPath) throws Exception {

        this.logRepoPath = logRepoPath;

        this.logPortalPath = logPortalPath;

        this.createTables();

        this.createTmpTables();

    }

    public ArrayList getRobotsList() {

        return robotsList;

    }

    public void setRobotsList(ArrayList robotsList) {

        this.robotsList = robotsList;

    }

    public String getCounterRobotsURL() {

        return CounterRobotsURL;

    }

    public void setCounterRobotsURL(String CounterRobotsURL) {

        this.CounterRobotsURL = CounterRobotsURL;

    }

    private void connectDB() throws Exception {

        try {

            ConnectDB connectDB = new ConnectDB();

        } catch (Exception e) {

            log.error("Connect to db failed: " + e);

            throw new Exception("Failed to connect to db: " + e.toString(), e);

        }

    }

    private void createTables() throws Exception {

        try {

            stmt = ConnectDB.DB_CONNECTION.createStatement();

            String sqlCreateTablePiwikLog = "CREATE TABLE IF NOT EXISTS piwiklog(source INTEGER, id_visit TEXT, country TEXT, action TEXT, url TEXT, entity_id TEXT, source_item_type TEXT, timestamp TEXT, referrer_name TEXT, agent TEXT, PRIMARY KEY(source, id_visit, action, timestamp, entity_id));";

            String sqlcreateRulePiwikLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS "

                    + " ON INSERT TO piwiklog "

                    + " WHERE (EXISTS ( SELECT piwiklog.source, piwiklog.id_visit,"

                    + "piwiklog.action, piwiklog.\"timestamp\", piwiklog.entity_id "

                    + "FROM piwiklog "

                    + "WHERE piwiklog.source = new.source AND piwiklog.id_visit = new.id_visit AND piwiklog.action = new.action AND piwiklog.entity_id = new.entity_id AND piwiklog.\"timestamp\" = new.\"timestamp\")) DO INSTEAD NOTHING;";

            stmt.executeUpdate(sqlCreateTablePiwikLog);

            stmt.executeUpdate(sqlcreateRulePiwikLog);

            String sqlCopyPublicPiwiklog = "insert into piwiklog select * from public.piwiklog;";

            stmt.executeUpdate(sqlCopyPublicPiwiklog);

            String sqlCreateTablePortalLog = "CREATE TABLE IF NOT EXISTS process_portal_log(source INTEGER, id_visit TEXT, country TEXT, action TEXT, url TEXT, entity_id TEXT, source_item_type TEXT, timestamp TEXT, referrer_name TEXT, agent TEXT, PRIMARY KEY(source, id_visit, timestamp));";

            String sqlcreateRulePortalLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS "

                    + " ON INSERT TO process_portal_log "

                    + " WHERE (EXISTS ( SELECT process_portal_log.source, process_portal_log.id_visit,"

                    + "process_portal_log.\"timestamp\" "

                    + "FROM process_portal_log "

                    + "WHERE process_portal_log.source = new.source AND process_portal_log.id_visit = new.id_visit AND process_portal_log.\"timestamp\" = new.\"timestamp\")) DO INSTEAD NOTHING;";

            stmt.executeUpdate(sqlCreateTablePortalLog);

            stmt.executeUpdate(sqlcreateRulePortalLog);

            stmt.close();

            ConnectDB.DB_CONNECTION.close();

            log.info("Usage Tables Created");

        } catch (Exception e) {

            log.error("Failed to create tables: " + e);

            throw new Exception("Failed to create tables: " + e.toString(), e);

            //System.exit(0);

        }

    }

    private void createTmpTables() throws Exception {

        try {

            Statement stmt = ConnectDB.DB_CONNECTION.createStatement();

            String sqlCreateTmpTablePiwikLog = "CREATE TABLE IF NOT EXISTS piwiklogtmp(source INTEGER, id_visit TEXT, country TEXT, action TEXT, url TEXT, entity_id TEXT, source_item_type TEXT, timestamp TEXT, referrer_name TEXT, agent TEXT, PRIMARY KEY(source, id_visit, action, timestamp, entity_id));";

            String sqlcreateTmpRulePiwikLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS "

                    + " ON INSERT TO piwiklogtmp "

                    + " WHERE (EXISTS ( SELECT piwiklogtmp.source, piwiklogtmp.id_visit,"

                    + "piwiklogtmp.action, piwiklogtmp.\"timestamp\", piwiklogtmp.entity_id "

                    + "FROM piwiklogtmp "

                    + "WHERE piwiklogtmp.source = new.source AND piwiklogtmp.id_visit = new.id_visit AND piwiklogtmp.action = new.action AND piwiklogtmp.entity_id = new.entity_id AND piwiklogtmp.\"timestamp\" = new.\"timestamp\")) DO INSTEAD NOTHING;";

            stmt.executeUpdate(sqlCreateTmpTablePiwikLog);

            stmt.executeUpdate(sqlcreateTmpRulePiwikLog);

            //String sqlCopyPublicPiwiklog="insert into piwiklog select * from public.piwiklog;";

            //stmt.executeUpdate(sqlCopyPublicPiwiklog);

            String sqlCreateTmpTablePortalLog = "CREATE TABLE IF NOT EXISTS process_portal_log_tmp(source INTEGER, id_visit TEXT, country TEXT, action TEXT, url TEXT, entity_id TEXT, source_item_type TEXT, timestamp TEXT, referrer_name TEXT, agent TEXT, PRIMARY KEY(source, id_visit, timestamp));";

            String sqlcreateTmpRulePortalLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS "

                    + " ON INSERT TO process_portal_log_tmp "

                    + " WHERE (EXISTS ( SELECT process_portal_log_tmp.source, process_portal_log_tmp.id_visit,"

                    + "process_portal_log_tmp.\"timestamp\" "

                    + "FROM process_portal_log_tmp "

                    + "WHERE process_portal_log_tmp.source = new.source AND process_portal_log_tmp.id_visit = new.id_visit AND process_portal_log_tmp.\"timestamp\" = new.\"timestamp\")) DO INSTEAD NOTHING;";

            stmt.executeUpdate(sqlCreateTmpTablePortalLog);

            stmt.executeUpdate(sqlcreateTmpRulePortalLog);

            stmt.close();

            log.info("Usage Tmp Tables Created");

        } catch (Exception e) {

            log.error("Failed to create tmptables: " + e);

            throw new Exception("Failed to create tmp tables: " + e.toString(), e);

            //System.exit(0);

        }

    }

    public void processLogs() throws Exception {

        try {

            ReadCounterRobotsList counterRobots = new ReadCounterRobotsList(this.getCounterRobotsURL());

            this.robotsList = counterRobots.getRobotsPatterns();

            processRepositoryLog();

            log.info("repository process done");

            removeDoubleClicks();

            log.info("removing double clicks done");

            cleanOAI();

            log.info("cleaning oai done");

            viewsStats();

            downloadsStats();

            processPortalLog();

            log.info("portal process done");

            portalStats();

            log.info("portal stats done");

            updateProdTables();

        } catch (Exception e) {

            log.error("Failed to process logs: " + e);

            throw new Exception("Failed to process logs: " + e.toString(), e);

        }

    }

    public void usageStats() throws Exception {

        try {

            //resultStats();

            //dataSourceStats();

            //organizationsStats();

            //projectsStats();

            //repositoryViewsStats();

            //repositoryDownloadsStats();

            viewsStats();

            downloadsStats();

            log.info("stat tables and views done");

        } catch (Exception e) {

            log.error("Failed to create usage stats: " + e);

            throw new Exception("Failed to create usage stats: " + e.toString(), e);

        }

    }

    public void processRepositoryLog() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        Statement stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        //ArrayList<String> jsonFiles = listHdfsDir(logPath + "repolog");

        File dir = new File(this.logRepoPath);

        File[] jsonFiles = dir.listFiles();

        PreparedStatement prepStatem = ConnectDB.DB_CONNECTION.prepareStatement("INSERT INTO piwiklogtmp (source, id_visit, country, action, url, entity_id, source_item_type, timestamp, referrer_name, agent) VALUES (?,?,?,?,?,?,?,?,?,?)");

        int batch_size = 0;

        JSONParser parser = new JSONParser();

        for (File jsonFile : jsonFiles) {

            System.out.println(jsonFile);

            JSONArray jsonArray = (JSONArray) parser.parse(readHDFSFile(jsonFile.getAbsolutePath()));

            for (Object aJsonArray : jsonArray) {

                JSONObject jsonObjectRow = (JSONObject) aJsonArray;

                int idSite = Integer.parseInt(jsonObjectRow.get("idSite").toString());

                String idVisit = jsonObjectRow.get("idVisit").toString();

                String country = jsonObjectRow.get("country").toString();

                String referrerName = jsonObjectRow.get("referrerName").toString();

                String agent = jsonObjectRow.get("browser").toString();

                boolean botFound = false;

                Iterator it = robotsList.iterator();

                while (it.hasNext()) {

                    // Create a Pattern object

                    Pattern r = Pattern.compile(it.next().toString());

                    // Now create matcher object.

                    Matcher m = r.matcher(agent);

                    if (m.find()) {

                        //System.out.println("Found value: " + m.group(0));

                        botFound = true;

                        break;

                    }

                }

                if (botFound == false) {

                    String sourceItemType = "repItem";

                    JSONArray actionDetails = (JSONArray) jsonObjectRow.get(("actionDetails"));

                    for (Object actionDetail : actionDetails) {

                        JSONObject actionDetailsObj = (JSONObject) actionDetail;

                        if (actionDetailsObj.get("customVariables") != null) {

                            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                            simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

                            Timestamp timestamp = new Timestamp(Long.parseLong(actionDetailsObj.get("timestamp").toString()) * 1000);

                            String url = actionDetailsObj.get("url").toString();

                            String oaipmh = ((JSONObject) ((JSONObject) actionDetailsObj.get("customVariables")).get("1")).get("customVariablePageValue1").toString();

                            String action = actionDetailsObj.get("type").toString();

                            prepStatem.setInt(1, idSite);

                            prepStatem.setString(2, idVisit);

                            prepStatem.setString(3, country);

                            prepStatem.setString(4, action);

                            prepStatem.setString(5, url);

                            prepStatem.setString(6, oaipmh);

                            prepStatem.setString(7, sourceItemType);

                            prepStatem.setString(8, simpleDateFormat.format(timestamp));

                            prepStatem.setString(9, referrerName);

                            prepStatem.setString(10, agent);

                            prepStatem.addBatch();

                            batch_size++;

                            if (batch_size == 10000) {

                                prepStatem.executeBatch();

                                ConnectDB.DB_CONNECTION.commit();

                                batch_size = 0;

                            }

                        }

                    }

                }

            }

        }

        prepStatem.executeBatch();

        ConnectDB.DB_CONNECTION.commit();

        stmt.close();

        //conn.close();

    }

       public void removeDoubleClicks() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        Statement stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        //clean download double clicks

        String sql = "DELETE FROM piwiklogtmp p WHERE EXISTS (SELECT DISTINCT p1.source, p1.id_visit, p1.action, p1.entity_id, p1.timestamp FROM piwiklogtmp p1, piwiklogtmp p2 WHERE p1.source!='5' AND p1.source=p2.source AND p1.id_visit=p2.id_visit AND p1.entity_id=p2.entity_id AND p1.action=p2.action AND p1.action='download' AND p1.timestamp!=p2.timestamp AND p1.timestamp<p2.timestamp AND extract(EPOCH FROM p2.timestamp::timestamp-p1.timestamp::timestamp)<30 AND p.source=p1.source AND p.id_visit=p1.id_visit AND p.action=p1.action AND p.entity_id=p1.entity_id AND p.timestamp=p1.timestamp);";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        //clean view double clicks

        sql = "DELETE FROM piwiklogtmp p WHERE EXISTS (SELECT DISTINCT p1.source, p1.id_visit, p1.action, p1.entity_id, p1.timestamp from piwiklogtmp p1, piwiklogtmp p2 WHERE p1.source!='5' AND p1.source=p2.source AND p1.id_visit=p2.id_visit AND p1.entity_id=p2.entity_id AND p1.action=p2.action AND p1.action='action' AND p1.timestamp!=p2.timestamp AND p1.timestamp<p2.timestamp AND extract(EPOCH FROM p2.timestamp::timestamp-p1.timestamp::timestamp)<10 AND p.source=p1.source AND p.id_visit=p1.id_visit AND p.action=p1.action AND p.entity_id=p1.entity_id AND p.timestamp=p1.timestamp);";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        //conn.close();

    }

         public void viewsStats() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        Statement stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        //String sql = "CREATE OR REPLACE VIEW result_views_monthly AS SELECT entity_id AS id, COUNT(entity_id) as views, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

        String sql = "CREATE OR REPLACE VIEW result_views_monthly_tmp AS SELECT entity_id AS id, COUNT(entity_id) as views, SUM(CASE WHEN referrer_name LIKE '%openaire%' THEN 1 ELSE 0 END) AS openaire_referrer, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklogtmp where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

        stmt.executeUpdate(sql);

        // sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count, max(openaire_referrer) AS openaire INTO views_stats FROM result_views_monthly p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        sql = "CREATE TABLE IF NOT EXISTS views_stats_tmp AS SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count, max(openaire_referrer) AS openaire FROM result_views_monthly_tmp p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        stmt.executeUpdate(sql);

//        sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count INTO pageviews_stats FROM result_views_monthly p, datasource d, result_oids ro where p.source='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        sql = "CREATE TABLE IF NOT EXISTS pageviews_stats_tmp AS SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count FROM result_views_monthly_tmp p, datasource d, result_oids ro where p.source='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        stmt.executeUpdate(sql);

        sql = "DROP VIEW IF EXISTS result_views_monthly_tmp;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        ConnectDB.DB_CONNECTION.close();

    }

    public void viewsStats(String piwikid) throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        //String sql = "CREATE OR REPLACE VIEW result_views_monthly AS SELECT entity_id AS id, COUNT(entity_id) as views, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

        String sql = "CREATE OR REPLACE VIEW result_views_monthly" + piwikid + " AS SELECT entity_id AS id, COUNT(entity_id) as views, SUM(CASE WHEN referrer_name LIKE '%openaire%' THEN 1 ELSE 0 END) AS openaire_referrer, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog" + piwikid + " where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

        stmt.executeUpdate(sql);

        // sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count, max(openaire_referrer) AS openaire INTO views_stats FROM result_views_monthly p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        sql = "CREATE TABLE IF NOT EXISTS views_stats" + piwikid + " AS SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count, max(openaire_referrer) AS openaire FROM result_views_monthly" + piwikid + " p, datasource d, result_oids ro where p.source!='109' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        stmt.executeUpdate(sql);

//        sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count INTO pageviews_stats FROM result_views_monthly p, datasource d, result_oids ro where p.source='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        sql = "CREATE TABLE IF NOT EXISTS pageviews_stats" + piwikid + " AS SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(views) AS count FROM result_views_monthly" + piwikid + " p, datasource d, result_oids ro where p.source='109' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        stmt.executeUpdate(sql);

        sql = "DROP VIEW IF EXISTS result_views_monthly" + piwikid + ";";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        ConnectDB.DB_CONNECTION.close();

    }

     private void downloadsStats() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        Statement stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        //String sql = "CREATE OR REPLACE VIEW result_downloads_monthly as select entity_id AS id, COUNT(entity_id) as downloads, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklog where action='download' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

        String sql = "CREATE OR REPLACE VIEW result_downloads_monthly_tmp as select entity_id AS id, COUNT(entity_id) as downloads, SUM(CASE WHEN referrer_name LIKE '%openaire%' THEN 1 ELSE 0 END) AS openaire_referrer, extract('year' from timestamp::date) ||'/'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') AS month, source FROM piwiklogtmp where action='download' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

        stmt.executeUpdate(sql);

        //sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(downloads) AS count INTO downloads_stats FROM result_downloads_monthly p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

//        sql = "SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(downloads) AS count, max(openaire_referrer) AS openaire INTO downloads_stats FROM result_downloads_monthly p, datasource d, result_oids ro where p.source!='5' AND p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        sql = "CREATE TABLE IF NOT EXISTS downloads_stats_tmp AS SELECT 'OpenAIRE'::TEXT as source, d.id as repository_id, ro.id as result_id, month as date, max(downloads) AS count, max(openaire_referrer) AS openaire FROM result_downloads_monthly_tmp p, datasource d, result_oids ro where p.source=d.piwik_id and p.id=ro.orid group by repository_id, result_id, date ORDER BY repository_id, result_id, date;";

        stmt.executeUpdate(sql);

        sql = "DROP VIEW IF EXISTS result_downloads_monthly_tmp;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        ConnectDB.DB_CONNECTION.close();

    }

    public void finalizeStats() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        Calendar startCalendar = Calendar.getInstance();

        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));

        Calendar endCalendar = Calendar.getInstance();

        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);

        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);

//        String sql = "SELECT to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS full_date INTO full_dates FROM generate_series(0, " + diffMonth + ", 1) AS offs;";

        String sql = "CREATE TABLE IF NOT EXISTS full_dates AS SELECT to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS full_date FROM generate_series(0, " + diffMonth + ", 1) AS offs;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX full_dates_full_date ON full_dates USING btree(full_date);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX views_stats_source ON views_stats USING btree(source);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX views_stats_repository_id ON views_stats USING btree(repository_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX views_stats_result_id ON views_stats USING btree(result_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX views_stats_date ON views_stats USING btree(date);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX pageviews_stats_source ON pageviews_stats USING btree(source);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX pageviews_stats_repository_id ON pageviews_stats USING btree(repository_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX pageviews_stats_result_id ON pageviews_stats USING btree(result_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX pageviews_stats_date ON pageviews_stats USING btree(date);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX downloads_stats_source ON downloads_stats USING btree(source);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX downloads_stats_repository_id ON downloads_stats USING btree(repository_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX downloads_stats_result_id ON downloads_stats USING btree(result_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX downloads_stats_date ON downloads_stats USING btree(date);";

        stmt.executeUpdate(sql);

//        sql = "SELECT coalesce(ds.source, vs.source) as source, coalesce(ds.repository_id, vs.repository_id) as repository_id, coalesce(ds.result_id, vs.result_id) as result_id, coalesce(ds.date, vs.date) as date, coalesce(ds.count, 0) as downloads, coalesce(vs.count, 0) as views, coalesce(ds.openaire, 0) as openaire_downloads, coalesce(vs.openaire, 0) as openaire_views INTO usage_stats FROM downloads_stats AS ds FULL OUTER JOIN views_stats AS vs ON ds.source=vs.source AND ds.repository_id=vs.repository_id AND ds.result_id=vs.result_id AND ds.date=vs.date;";

        sql = "CREATE TABLE IF NOT EXISTS usage_stats AS SELECT coalesce(ds.source, vs.source) as source, coalesce(ds.repository_id, vs.repository_id) as repository_id, coalesce(ds.result_id, vs.result_id) as result_id, coalesce(ds.date, vs.date) as date, coalesce(ds.count, 0) as downloads, coalesce(vs.count, 0) as views, coalesce(ds.openaire, 0) as openaire_downloads, coalesce(vs.openaire, 0) as openaire_views FROM downloads_stats AS ds FULL OUTER JOIN views_stats AS vs ON ds.source=vs.source AND ds.repository_id=vs.repository_id AND ds.result_id=vs.result_id AND ds.date=vs.date;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX usage_stats_source ON usage_stats USING btree(source);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX usage_stats_repository_id ON usage_stats USING btree(repository_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX usage_stats_result_id ON usage_stats USING btree(result_id);";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX usage_stats_date ON usage_stats USING btree(date);";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        ConnectDB.DB_CONNECTION.close();

    }

    //views stats

//    private void viewsStatsOLD() throws Exception {

//        if (conn.isClosed())

//            connectDB();

//

//        stmt = conn.createStatement();

//        conn.setAutoCommit(false);

//

//        Calendar startCalendar = Calendar.getInstance();

//        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));

//        Calendar endCalendar = Calendar.getInstance();

//        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);

//        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);

//

//        String sql = "CREATE OR REPLACE VIEW result_views_monthly as select entity_id AS id, COUNT(entity_id) as views, extract('year' from timestamp::date) ||'-'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM piwiklog where action='action' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

//        stmt.executeUpdate(sql);

//

//        sql = "SELECT d.id, d.new_date AS month, CASE when rdm.views IS NULL THEN 0 ELSE rdm.views END, d.source INTO result_views_sushi FROM (SELECT distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date, rdsm.source FROM generate_series(0, " + diffMonth + ", 1) AS offs, result_views_monthly rdsm) d LEFT JOIN (SELECT id, month, views, source FROM result_views_monthly) rdm ON d.new_date=rdm.month AND d.id=rdm.id AND d.source=rdm.source ORDER BY d.source, d.id, d.new_date;";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_views_sushi_id ON result_views_sushi USING btree (id);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_views_sushi_month ON result_views_sushi USING btree (month);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_views_sushi_source ON result_views_sushi USING btree (source);";

//        stmt.executeUpdate(sql);

//

//        sql = "SELECT roid.id, extract('year' from month::date) ||'/'|| LPAD(CAST(extract('month' from month::date) AS VARCHAR), 2, '0') as month, max(views) as views, source INTO result_views FROM result_views_sushi rvs, result_oids roid WHERE rvs.id=roid.orid GROUP BY source, roid.id, month ORDER BY source, roid.id, month;";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_views_id ON result_views USING btree (id);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_views_month ON result_views USING btree (month);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_views_source ON result_views USING btree (source);";

//        stmt.executeUpdate(sql);

//

//

//        sql = "DROP VIEW IF EXISTS result_views_monthly;";

//        stmt.executeUpdate(sql);

//

//        stmt.close();

//        conn.commit();

//        conn.close();

//    }

    //downloads stats

//    private void downloadsStatsOLD() throws Exception {

//        if (conn.isClosed())

//            connectDB();

//

//        stmt = conn.createStatement();

//        conn.setAutoCommit(false);

//

//        Calendar startCalendar = Calendar.getInstance();

//        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));

//        Calendar endCalendar = Calendar.getInstance();

//        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);

//        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);

//

//        String sql = "CREATE OR REPLACE VIEW result_downloads_monthly as select entity_id AS id, COUNT(entity_id) as downloads, extract('year' from timestamp::date) ||'-'|| LPAD(CAST(extract('month' from timestamp::date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM piwiklog where action='download' and (source_item_type='oaItem' or source_item_type='repItem') group by id, month, source order by source, id, month;";

//        stmt.executeUpdate(sql);

//

//        sql = "SELECT d.id, d.new_date AS month, CASE when rdm.downloads IS NULL THEN 0 ELSE rdm.downloads END, d.source INTO result_downloads_sushi FROM (SELECT distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date, rdsm.source FROM generate_series(0, " + diffMonth + ", 1) AS offs, result_downloads_monthly rdsm) d LEFT JOIN (SELECT id, month, downloads, source FROM result_downloads_monthly) rdm ON d.new_date=rdm.month AND d.id=rdm.id AND d.source=rdm.source ORDER BY d.source, d.id, d.new_date;";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_downloads_sushi_id ON result_downloads_sushi USING btree (id);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_downloads_sushi_month ON result_downloads_sushi USING btree (month);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_downloads_sushi_source ON result_downloads_sushi USING btree (source);";

//        stmt.executeUpdate(sql);

//

//        sql = "SELECT roid.id, extract('year' from month::date) ||'/'|| LPAD(CAST(extract('month' from month::date) AS VARCHAR), 2, '0') as month, downloads, source INTO result_downloads FROM result_downloads_sushi rvs, result_oids roid WHERE rvs.id=roid.orid ORDER BY source, roid.id, month;";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_downloads_id ON result_downloads USING btree (id);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_downloads_month ON result_downloads USING btree (month);";

//        stmt.executeUpdate(sql);

//

//        sql = "CREATE INDEX result_downloads_source ON result_downloads USING btree (source);";

//        stmt.executeUpdate(sql);

//

//

//        sql = "DROP VIEW IF EXISTS result_downloads_monthly;";

//        stmt.executeUpdate(sql);

//

//        stmt.close();

//        conn.commit();

//        conn.close();

//    }

    //Create repository Views statistics

    private void repositoryViewsStats() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

//        String sql = "SELECT entity_id AS id , COUNT(entity_id) AS number_of_views, timestamp::date AS date, source INTO repo_view_stats FROM piwiklog WHERE source!='5' AND action=\'action\' AND source_item_type=\'repItem\' GROUP BY entity_id, date, source ORDER BY entity_id, date ASC, COUNT(entity_id) DESC;";

        String sql = "CREATE TABLE IF NOT EXISTS repo_view_stats AS SELECT entity_id AS id , COUNT(entity_id) AS number_of_views, timestamp::date AS date, source FROM piwiklog WHERE source!='5' AND action=\'action\' AND source_item_type=\'repItem\' GROUP BY entity_id, date, source ORDER BY entity_id, date ASC, COUNT(entity_id) DESC;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_id ON repo_view_stats USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_date ON repo_view_stats USING btree(date)";

        stmt.executeUpdate(sql);

//        sql = "SELECT roid.id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month, source INTO repo_view_stats_monthly_clean FROM repo_view_stats rvs, result_oids roid where rvs.id=roid.orid group by roid.id, month, source;";

        sql = "CREATE TABLE IF NOT EXISTS repo_view_stats_monthly_clean AS SELECT roid.id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month, source FROM repo_view_stats rvs, result_oids roid where rvs.id=roid.orid group by roid.id, month, source;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_monthly_clean_id ON repo_view_stats_monthly_clean USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_monthly_clean_month ON repo_view_stats_monthly_clean USING btree(month)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_monthly_clean_source ON repo_view_stats_monthly_clean USING btree(source)";

        stmt.executeUpdate(sql);

        Calendar startCalendar = Calendar.getInstance();

        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));

        Calendar endCalendar = Calendar.getInstance();

        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);

        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);

        //sql="CREATE OR REPLACE view repo_view_stats_monthly AS select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth +", 1) AS offs, repo_view_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_view_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";

//        sql = "select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source INTO repo_view_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth + ", 1) AS offs, repo_view_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_view_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";

        sql = "CREATE TABLE IF NOT EXISTS repo_view_stats_monthly AS select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth + ", 1) AS offs, repo_view_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_view_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_monthly_id ON repo_view_stats_monthly USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_monthly_month ON repo_view_stats_monthly USING btree(month)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_view_stats_monthly_source ON repo_view_stats_monthly USING btree(source)";

        stmt.executeUpdate(sql);

        sql = "CREATE OR REPLACE view repo_view_stats_monthly_sushi AS SELECT id, sum(number_of_views), extract('year' from date) ||'-'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM repo_view_stats group by id, month, source;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        ConnectDB.DB_CONNECTION.close();

    }

    //Create repository downloads statistics

    private void repositoryDownloadsStats() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

//        String sql = "SELECT entity_id AS id, COUNT(entity_id) AS number_of_downloads, timestamp::date AS date, source INTO repo_download_stats FROM piwiklog WHERE source!='5' AND action=\'download\' AND source_item_type=\'repItem\' GROUP BY entity_id, date, source ORDER BY entity_id, date ASC, COUNT(entity_id) DESC;";

        String sql = "CREATE TABLE IF NOT EXISTS repo_download_stats AS SELECT entity_id AS id, COUNT(entity_id) AS number_of_downloads, timestamp::date AS date, source FROM piwiklog WHERE source!='5' AND action=\'download\' AND source_item_type=\'repItem\' GROUP BY entity_id, date, source ORDER BY entity_id, date ASC, COUNT(entity_id) DESC;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_id ON repo_download_stats USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_date ON repo_download_stats USING btree(date)";

        stmt.executeUpdate(sql);

//        sql = "SELECT roid.id, sum(number_of_downloads), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month, source INTO repo_download_stats_monthly_clean FROM repo_download_stats rvs, result_oids roid WHERE rvs.id=roid.orid GROUP BY roid.id, month, source;";

        sql = "CREATE TABLE IF NOT EXISTS repo_download_stats_monthly_clean AS SELECT roid.id, sum(number_of_downloads), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month, source FROM repo_download_stats rvs, result_oids roid WHERE rvs.id=roid.orid GROUP BY roid.id, month, source;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_monthly_clean_id ON repo_download_stats_monthly_clean USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_monthly_clean_month ON repo_download_stats_monthly_clean USING btree(month)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_monthly_clean_source ON repo_download_stats_monthly_clean USING btree(source)";

        stmt.executeUpdate(sql);

        Calendar startCalendar = Calendar.getInstance();

        startCalendar.setTime(new SimpleDateFormat("yyyy-MM-dd").parse("2016-01-01"));

        Calendar endCalendar = Calendar.getInstance();

        int diffYear = endCalendar.get(Calendar.YEAR) - startCalendar.get(Calendar.YEAR);

        int diffMonth = diffYear * 12 + endCalendar.get(Calendar.MONTH) - startCalendar.get(Calendar.MONTH);

        //sql="CREATE OR REPLACE view repo_download_stats_monthly AS select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth +", 1) AS offs, repo_download_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_download_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";

        // sql = "select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source INTO repo_download_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth + ", 1) AS offs, repo_download_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_download_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";

        sql = "CREATE TABLE IF NOT EXISTS repo_download_stats_monthly AS select d.id, d.new_date AS month, case when rdm.sum is null then 0 else rdm.sum end, d.source from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date, rdsm.source from generate_series(0, " + diffMonth + ", 1) AS offs, repo_download_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum, source from repo_download_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id and d.source=rdm.source order by d.id, d.new_date";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_monthly_id ON repo_download_stats_monthly USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_monthly_month ON repo_download_stats_monthly USING btree(month)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX repo_download_stats_monthly_source ON repo_download_stats_monthly USING btree(source)";

        stmt.executeUpdate(sql);

        sql = "CREATE OR REPLACE view repo_download_stats_monthly_sushi AS SELECT id, sum(number_of_downloads), extract('year' from date) ||'-'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') ||'-01' AS month, source FROM repo_download_stats group by id, month, source;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        ConnectDB.DB_CONNECTION.close();

    }

// Import OPENAIRE Logs to DB

    public void processPortalLog() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        Statement stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        //ArrayList<String> jsonFiles = listHdfsDir(logPath + "portallog");

        File folder = new File(this.logPortalPath);

        File[] jsonFiles = folder.listFiles();

        PreparedStatement prepStatem = ConnectDB.DB_CONNECTION.prepareStatement("INSERT INTO process_portal_log_tmp (source, id_visit, country, action, url, entity_id, source_item_type, timestamp, referrer_name, agent) VALUES (?,?,?,?,?,?,?,?,?,?)");

        int batch_size = 0;

        JSONParser parser = new JSONParser();

        for (File jsonFile : jsonFiles) {

            JSONArray jsonArray = (JSONArray) parser.parse(readHDFSFile(jsonFile.getAbsolutePath()));

            for (Object aJsonArray : jsonArray) {

                JSONObject jsonObjectRow = (JSONObject) aJsonArray;

                int idSite = Integer.parseInt(jsonObjectRow.get("idSite").toString());

                String idVisit = jsonObjectRow.get("idVisit").toString();

                String country = jsonObjectRow.get("country").toString();

                String referrerName = jsonObjectRow.get("referrerName").toString();

                String agent = jsonObjectRow.get("browser").toString();

                boolean botFound = false;

                Iterator it = robotsList.iterator();

                while (it.hasNext()) {

                    // Create a Pattern object

                    Pattern r = Pattern.compile(it.next().toString());

                    // Now create matcher object.

                    Matcher m = r.matcher(agent);

                    if (m.find()) {

                        System.out.println("Found value: " + m.group(0));

                        botFound = true;

                        break;

                    }

                }

                if (botFound == false) {

                    JSONArray actionDetails = (JSONArray) jsonObjectRow.get(("actionDetails"));

                    for (Object actionDetail : actionDetails) {

                        JSONObject actionDetailsObj = (JSONObject) actionDetail;

                        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                        simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

                        Timestamp timestamp = new Timestamp(Long.parseLong(actionDetailsObj.get("timestamp").toString()) * 1000);

                        String action = actionDetailsObj.get("type").toString();

                        String url = actionDetailsObj.get("url").toString();

                        String entityID = processPortalURL(url);

                        String sourceItemType = "";

                        if (entityID.indexOf("|") > 0) {

                            sourceItemType = entityID.substring(0, entityID.indexOf("|"));

                            entityID = entityID.substring(entityID.indexOf("|") + 1);

                        }

                        prepStatem.setInt(1, idSite);

                        prepStatem.setString(2, idVisit);

                        prepStatem.setString(3, country);

                        prepStatem.setString(4, action);

                        prepStatem.setString(5, url);

                        prepStatem.setString(6, entityID);

                        prepStatem.setString(7, sourceItemType);

                        prepStatem.setString(8, simpleDateFormat.format(timestamp));

                        prepStatem.setString(9, referrerName);

                        prepStatem.setString(10, agent);

                        prepStatem.addBatch();

                        batch_size++;

                        if (batch_size == 10000) {

                            prepStatem.executeBatch();

                            ConnectDB.DB_CONNECTION.commit();

                            batch_size = 0;

                        }

                    }

                }

            }

        }

        prepStatem.executeBatch();

        ConnectDB.DB_CONNECTION.commit();

        stmt.close();

        ConnectDB.DB_CONNECTION.close();

    }

    public void portalStats() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        Statement stmt = ConnectDB.DB_CONNECTION.createStatement();

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        String sql = "INSERT INTO piwiklogtmp SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'oaItem\', timestamp, referrer_name, agent FROM process_portal_log_tmp, result_oids roid WHERE entity_id IS NOT null AND entity_id=roid.orid AND roid.orid IS NOT null;";

        //SELECT DISTINCT process_portal_log.source, id_visit, country, action, url, roid.orid, 'oaItem', timestamp, referrer_name, agent FROM process_portal_log, result r, result_oids roid WHERE entity_id IS NOT null AND entity_id=r.original_id AND r.id = roid.id AND roid.orid IS NOT null;

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "INSERT INTO piwiklogtmp SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'datasource\', timestamp, referrer_name, agent FROM process_portal_log_tmp, datasource_oids roid WHERE entity_id IS NOT null AND entity_id=roid.orid AND roid.orid IS NOT null;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "INSERT INTO piwiklogtmp SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'organization\', timestamp, referrer_name, agent FROM process_portal_log_tmp, organization_oids roid WHERE entity_id IS NOT null AND entity_id=roid.orid AND roid.orid IS NOT null;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "INSERT INTO piwiklogtmp SELECT DISTINCT source, id_visit, country, action, url, roid.orid, \'project\', timestamp, referrer_name, agent FROM process_portal_log_tmp, project_oids roid WHERE entity_id IS NOT null AND entity_id=roid.orid AND roid.orid IS NOT null;";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        /*

        stmt = conn.createStatement();

        sql = "DROP TABLE process_portal_log;";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

         */

        ConnectDB.DB_CONNECTION.close();

    }

    private void cleanOAI() throws Exception {

        if (ConnectDB.DB_CONNECTION.isClosed()) {

            connectDB();

        }

        ConnectDB.DB_CONNECTION.setAutoCommit(false);

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        String sql = "UPDATE piwiklogmtp SET entity_id = regexp_replace(entity_id, '^oai:repositorio.chlc.min-saude.pt/','oai:repositorio.chlc.min-saude.pt:') WHERE entity_id LIKE 'oai:repositorio.chlc.min-saude.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklogtmp SET entity_id = regexp_replace(entity_id, '^oai:repositorio.hospitaldebraga.pt/','oai:repositorio.hospitaldebraga.pt:') WHERE entity_id LIKE 'oai:repositorio.hospitaldebraga.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ipl.pt/','oai:repositorio.ipl.pt:') WHERE entity_id LIKE 'oai:repositorio.ipl.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklogtmp SET entity_id = regexp_replace(entity_id, '^oai:bibliotecadigital.ipb.pt/','oai:bibliotecadigital.ipb.pt:') WHERE entity_id LIKE 'oai:bibliotecadigital.ipb.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ismai.pt/','oai:repositorio.ismai.pt:') WHERE entity_id LIKE 'oai:repositorio.ismai.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:repositorioaberto.uab.pt/','oai:repositorioaberto.uab.pt:') WHERE entity_id LIKE 'oai:repositorioaberto.uab.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklogtmp SET entity_id = regexp_replace(entity_id, '^oai:repositorio.uac.pt/','oai:repositorio.uac.pt:') WHERE entity_id LIKE 'oai:repositorio.uac.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklogtmp SET entity_id = regexp_replace(entity_id, '^oai:repositorio.insa.pt/','oai:repositorio.insa.pt:') WHERE entity_id LIKE 'oai:repositorio.insa.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklogtmp SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ipcb.pt/','oai:repositorio.ipcb.pt:') WHERE entity_id LIKE 'oai:repositorio.ipcb.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklogtmp SET entity_id = regexp_replace(entity_id, '^oai:repositorio.ispa.pt/','oai:repositorio.ispa.pt:') WHERE entity_id LIKE 'oai:repositorio.ispa.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        ConnectDB.DB_CONNECTION.commit();

        stmt = ConnectDB.DB_CONNECTION.createStatement();

        sql = "UPDATE piwiklogtmp SET entity_id = regexp_replace(entity_id, '^oai:repositorio.chporto.pt/','oai:repositorio.chporto.pt:') WHERE entity_id LIKE 'oai:repositorio.chporto.pt/%';";

        stmt.executeUpdate(sql);