D-Net

package eu.dnetlib.usagestats.export;

import java.io.*;

import java.net.URLDecoder;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import java.sql.Statement;

import java.sql.Timestamp;

import java.text.SimpleDateFormat;

import java.util.*;

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 final String dbUrl;

    private final String dbSchema;

    private final String dbUserName;

    private final String dbPassword;

    private final String logPath;

    private Connection conn = null;

    private Statement stmt = null;

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

    public PiwikStatsDB(String dbUrl, String dbUserName, String dbPassword, String logPath) throws Exception {

        this.dbUrl = dbUrl;

        this.dbSchema = "shadow";

        this.dbUserName = dbUserName;

        this.dbPassword = dbPassword;

        this.logPath = logPath;

        connectDB();

        createTables();

    }

    private void connectDB() throws Exception {

        try {

            Class.forName("org.postgresql.Driver");

            conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword);

            stmt = conn.createStatement();

            String sqlSetSearchPath = "SET search_path TO " + dbSchema + ";";

            stmt.executeUpdate(sqlSetSearchPath);

            //log.info("Opened database successfully");

        } 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 = conn.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();

            conn.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);

        }

    }

    protected void processLogs() throws Exception {

        try {

            processRepositoryLog();

            log.info("repository process done");

            removeDoubleClicks();

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

            cleanOAI();

            log.info("cleaning oai done");

            processPortalLog();

            log.info("portal process done");

            portalStats();

            log.info("portal stats done");

        } catch (Exception e) {

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

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

        }

    }

    protected 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);

        }

    }

    // Import repository Logs to DB

    private void processRepositoryLog() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

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

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

        int batch_size = 0;

        for (String jsonFile : jsonFiles) {

            JSONParser parser = new JSONParser();

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

            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();

                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();

                            conn.commit();

                            batch_size = 0;

                        }

                    }

                }

            }

        }

        prepStatem.executeBatch();

        conn.commit();

        stmt.close();

        conn.close();

    }

    private void removeDoubleClicks() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

        //clean download double clicks

        String sql = "DELETE FROM piwiklog p WHERE EXISTS (SELECT DISTINCT p1.source, p1.id_visit, p1.action, p1.entity_id, p1.timestamp FROM piwiklog p1, piwiklog 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();

        conn.commit();

        stmt = conn.createStatement();

        //clean view double clicks

        sql = "DELETE FROM piwiklog p WHERE EXISTS (SELECT DISTINCT p1.source, p1.id_visit, p1.action, p1.entity_id, p1.timestamp from piwiklog p1, piwiklog 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();

        conn.commit();

        conn.close();

    }

    private void viewsStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.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 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 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;";

        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;";

        stmt.executeUpdate(sql);

        sql = "DROP VIEW IF EXISTS result_views_monthly;";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    private void downloadsStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.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 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 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 '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;";

        stmt.executeUpdate(sql);

        sql = "DROP VIEW IF EXISTS result_downloads_monthly;";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    public void finalizeStats() 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 = "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;";

        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);

        stmt.close();

        conn.commit();

        conn.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 (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.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;";

        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;";

        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";

        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();

        conn.commit();

        conn.close();

    }

    //Create repository downloads statistics

    private void repositoryDownloadsStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.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;";

        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;";

        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";

        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();

        conn.commit();

        conn.close();

    }

    // Import OPENAIRE Logs to DB

    private void processPortalLog() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

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

        PreparedStatement prepStatem = conn.prepareStatement("INSERT INTO process_portal_log (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 (String jsonFile : jsonFiles) {

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

            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();

                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();

                        conn.commit();

                        batch_size = 0;

                    }

                }

            }

        }

        prepStatem.executeBatch();

        conn.commit();

        stmt.close();

        conn.close();

    }

    private void portalStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

        sql = "DROP TABLE process_portal_log;";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    private void cleanOAI() throws Exception {

        if (conn.isClosed())

            connectDB();

        conn.setAutoCommit(false);

        stmt = conn.createStatement();

        String sql = "UPDATE piwiklog 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();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog 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();

        conn.commit();

        stmt = conn.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();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog 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();

        conn.commit();

        stmt = conn.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();

        conn.commit();

        stmt = conn.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();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog 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();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog 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();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog 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();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog 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();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog 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);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

        sql = "UPDATE piwiklog SET entity_id = regexp_replace(entity_id, '^oai:rihuc.huc.min-saude.pt/','oai:rihuc.huc.min-saude.pt:') WHERE entity_id LIKE 'oai:rihuc.huc.min-saude.pt/%';";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        stmt = conn.createStatement();

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

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    //Create OpenAIRE's portal datasource statistics

    private void dataSourceStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();