D-Net

package eu.dnetlib.usagestats.export;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.FileSystem;

import org.apache.log4j.Logger;

import java.io.*;

import java.net.URL;

import java.net.URLConnection;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.Statement;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Calendar;

public class PiwikDownloadLogs {

    private final String piwikUsername;

    private final String piwikPassword;

    private final String httpProtocol;

    private final String piwikUrl;

    private final Date startDate;

    private final String tokenAuth;

    private final String logsPath;

    private final String dbUrl;

    private final String dbUserName;

    private final String dbPassword;

    /*

       The Piwik's API method 

    */

    private final String APImethod = "?module=API&method=Live.getLastVisitsDetails";

    private final String format = "&format=json";

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

    public PiwikDownloadLogs(String username, String password, String tokenAuth, String httpProtocol, String piwikURl, String sDate, String logsPath, String dbUrl, String dbUsername, String dbPassword) throws Exception{

        this.piwikUsername = username;

        this.piwikPassword = password;

        this.httpProtocol = httpProtocol;

        this.piwikUrl = piwikURl;

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

        this.startDate = sdf.parse(sDate);

        this.tokenAuth = tokenAuth;

        this.logsPath = logsPath;

        this.dbUrl = dbUrl;

        this.dbUserName = dbUsername;

        this.dbPassword = dbPassword;

    }

    private String getPiwikLogUrl(){

        return httpProtocol + "://" + piwikUrl + "/";

    }

    private String getJson(String url,String username, String password) throws Exception {

        //String cred=username+":"+password;

        //String encoded = new sun.misc.BASE64Encoder().encode (cred.getBytes());

        try {

            URL website = new URL(url);

            URLConnection connection = website.openConnection();

            //connection.setRequestProperty ("Authorization", "Basic "+encoded);

            StringBuilder response;

            try (BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream()))) {

                response = new StringBuilder();

                String inputLine;

                while ((inputLine = in.readLine()) != null) {

                    response.append(inputLine);

                    response.append("\n");

                }

            }

            return response.toString();

        }catch (Exception e){

            log.error("Failed to get URL: " + e);

            throw new Exception("Failed to get URL: " + e.toString(), e);

        }

    }

    public void getPiwikLogs() throws Exception{

        GetPortalLogs();

        GetRepositoriesLogs();

    }

    private void GetPortalLogs() throws Exception{

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

        Calendar start = Calendar.getInstance();

        start.setTime(startDate);

        Calendar end = Calendar.getInstance();

        end.add(Calendar.DAY_OF_MONTH, -1);

        //end.setTime(getFinalDate());

        try{

            log.info("downloading logs for site with piwik_id: 5");

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

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

            PreparedStatement st = conn.prepareStatement("SELECT max(timestamp) FROM public.piwiklog WHERE source='5';");

            ResultSet rs_date = st.executeQuery();

            while(rs_date.next()){

                start.setTime(sdf.parse(rs_date.getString(1)));

            }

            rs_date.close();

            conn.close();

            for (Date date = start.getTime(); start.before(end); start.add(Calendar.DATE, 1), date = start.getTime()) {

                String period="&period=day&date="+sdf.format(date);

                String apiUrl=getPiwikLogUrl()+APImethod+"&idSite=5"+period+format+"&expanded=5&filter_limit=1000&token_auth="+tokenAuth;

                String content = getJson(apiUrl,piwikUsername,piwikPassword);

                //for (int i=1;i<10;i++){

                int i = 1;

                while(true) {

                    String apiUrlnew=apiUrl+"&filter_offset="+i*1000;

                    String contentNew = getJson(apiUrlnew,piwikUsername,piwikUsername);

                    content += contentNew;

                    i++;

                    if(contentNew.equals("[]\n")){

                        break;

                    }

                }

                flushString(content, logsPath + "portallog/" + "5_Piwiklog"+sdf.format((date))+".json");

            }

        } catch (Exception e) {

            log.error(e);

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

        }

    }

    private void GetRepositoriesLogs() throws Exception{

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

        Calendar start = Calendar.getInstance();

        start.setTime(startDate);

        Calendar end = Calendar.getInstance();

        end.add(Calendar.DAY_OF_MONTH, -1);

        //end.setTime(getFinalDate());

        try{

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

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

            Statement statement = conn.createStatement();

            ResultSet rs = statement.executeQuery("SELECT distinct piwik_id from shadow.datasource where piwik_id is not null and piwik_id!='5' order by piwik_id;");

            while(rs.next()){

                start.setTime(startDate);

                int siteId = rs.getInt(1);

                log.info("downloading logs for site with piwik_id: " + siteId);

                PreparedStatement st = conn.prepareStatement("SELECT max(timestamp) FROM public.piwiklog WHERE source=?;");

                st.setInt(1, siteId);

                ResultSet rs_date = st.executeQuery();

                while(rs_date.next()){

                    //log.info("source: " + siteId + " - date: " + rs_date.getString(1));

                    if(rs_date.getString(1) == null || rs_date.getString(1).equals("null") || rs_date.getString(1).equals("")) {

                        start = Calendar.getInstance();

                        start.add(Calendar.MONTH, -1);

                    }

                    else {

                        start.setTime(sdf.parse(rs_date.getString(1)));

                    }

                }

                rs_date.close();

                for (Date date = start.getTime(); start.before(end); start.add(Calendar.DATE, 1), date = start.getTime()) {

                    String period="&period=day&date="+sdf.format(date);

                    String apiUrl=getPiwikLogUrl()+APImethod+"&idSite="+siteId+period+format+"&expanded=5&filter_limit=1000&token_auth="+tokenAuth;

                    String content = getJson(apiUrl,piwikUsername,piwikPassword);

                    //for (int i=1;i<10;i++){

                    int i=1;

                    while(true) {

                        String apiUrlnew=apiUrl+"&filter_offset="+i*1000;

                        String contentNew = getJson(apiUrlnew,piwikUsername,piwikPassword);

                        content += contentNew;

                        i++;

                        if(contentNew.equals("[]\n")){

                            break;

                        }

                    }

                    flushString(content, logsPath + "repolog/" + siteId + "_Piwiklog"+sdf.format((date))+".json");

                }

            }

            rs.close();

            conn.close();

        } catch (Exception e) {

            log.error(e);

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

        }

    }

    private void flushString(String data, String destination) throws Exception {

        FSDataOutputStream fin;

        try {

            FileSystem fs = FileSystem.get(new Configuration());

            fin = fs.create(new Path(destination), true);

            fin.write(data.getBytes());

            fin.close();

        } catch (Exception e) {

            log.error("Failed  to write exported data to a file : ", e);

            throw new Exception("Failed  to write exported data to a file : " + e.toString(), e);

        }

    }

}

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

        conn.setAutoCommit(false);

        String sql = "SELECT  orgid AS id, max(viewcount) AS number_of_views, date INTO datasource_stats FROM (select entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date as date FROM piwiklog, datasource_oids ooid WHERE entity_id=ooid.orid AND source_item_type='datasource' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX datasource_stats_id ON datasource_stats USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX datasource_stats_date ON datasource_stats USING btree(date)";

        stmt.executeUpdate(sql);

        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO datasource_stats_monthly_clean FROM datasource_stats GROUP BY id, month;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX datasource_stats_monthly_clean_id ON datasource_stats_monthly_clean USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX datasource_stats_monthly_clean_month ON datasource_stats_monthly_clean USING btree(month)";

        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 datasource_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, datasource_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from datasource_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id 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 INTO datasource_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth + ", 1) AS offs, datasource_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from datasource_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX datasource_stats_monthly_id ON datasource_stats_monthly USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX datasource_stats_monthly_month ON datasource_stats_monthly USING btree(month)";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    //Create OpenAIRE's portal results statistics

    private void resultStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

        String sql = "SELECT orgid AS id, max(viewcount) AS number_of_views, date INTO result_stats FROM (SELECT entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date as date FROM piwiklog, result_oids ooid WHERE entity_id=ooid.orid AND source_item_type='oaItem' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX result_stats_id ON result_stats USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX result_stats_date ON result_stats USING btree(date)";

        stmt.executeUpdate(sql);

        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO result_stats_monthly_clean FROM result_stats group by id, month;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX result_stats_monthly_clean_id ON result_stats_monthly_clean USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX result_stats_monthly_clean_month ON result_stats_monthly_clean USING btree(month)";

        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 result_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, result_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from result_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id 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 INTO result_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 from generate_series(0, " + diffMonth + ", 1) AS offs, result_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from result_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX result_stats_monthly_id ON result_stats_monthly USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX result_stats_monthly_month ON result_stats_monthly USING btree(month)";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    //Create OpenAIRE's portal organization statistics

    private void organizationsStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

        String sql = "SELECT orgid AS id, max(viewcount) AS number_of_views, date INTO organization_stats FROM (SELECT entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date AS date FROM piwiklog, organization_oids ooid WHERE entity_id=ooid.orid AND source_item_type='organization' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX organization_stats_id ON organization_stats USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX organization_stats_date ON organization_stats USING btree(date)";

        stmt.executeUpdate(sql);

        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO organization_stats_monthly_clean FROM organization_stats group by id, month;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX organization_stats_monthly_clean_id ON organization_stats_monthly_clean USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX organization_stats_monthly_clean_month ON organization_stats_monthly_clean USING btree(month)";

        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 organization_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, organization_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from organization_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id 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 INTO organization_stats_monthly from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY-MM-DD') AS new_date from generate_series(0, " + diffMonth + ", 1) AS offs, organization_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from organization_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX organization_stats_monthly_id ON organization_stats_monthly USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX organization_stats_monthly_month ON organization_stats_monthly USING btree(month)";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    //Create OpenAIRE's portal projects statistics

    private void projectsStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

        String sql = "SELECT orgid AS id, max(viewcount) AS number_of_views, date INTO project_stats FROM (SELECT entity_id, ooid.id AS orgid, count(entity_id) viewcount, timestamp::date AS date FROM piwiklog, project_oids ooid WHERE entity_id=ooid.orid AND source_item_type='project' GROUP BY entity_id, ooid.id, date) AS temp GROUP BY orgid, date ORDER BY orgid, date ASC, max(viewcount) DESC;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX project_stats_id ON project_stats USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX project_stats_date ON project_stats USING btree(date)";

        stmt.executeUpdate(sql);

        sql = "SELECT id, sum(number_of_views), extract('year' from date) ||'/'|| LPAD(CAST(extract('month' from date) AS VARCHAR), 2, '0') AS month INTO project_stats_monthly_clean FROM project_stats group by id, month;";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX project_stats_monthly_clean_id ON project_stats_monthly_clean USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX project_stats_monthly_clean_month ON project_stats_monthly_clean USING btree(month)";

        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 project_stats_monthly AS select d.id, d.new_date as month, case when rdm.sum is null then 0 else rdm.sum end from (select distinct rdsm.id, to_char(date_trunc('month', ('2016-01-01'::date + interval '1 month'*offs)), 'YYYY/MM') AS new_date from generate_series(0, " + diffMonth +", 1) AS offs, project_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from project_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id 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 INTO project_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 from generate_series(0, " + diffMonth + ", 1) AS offs, project_stats_monthly_clean rdsm) d LEFT JOIN (select id, month, sum from project_stats_monthly_clean) rdm ON d.new_date=rdm.month and d.id=rdm.id order by d.id, d.new_date";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX project_stats_monthly_id ON project_stats_monthly USING btree (id)";

        stmt.executeUpdate(sql);

        sql = "CREATE INDEX project_stats_monthly_month ON project_stats_monthly USING btree(month)";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    private String processPortalURL(String url) {

        if (url.indexOf("openaire.eu") > 0) {

            try {

                url = URLDecoder.decode(url, "UTF-8");

            } catch (Exception e) {

                log.info(url);

            }

            if (url.indexOf("datasourceId=") > 0 && url.substring(url.indexOf("datasourceId=") + 13).length() >= 46) {

                url = "datasource|" + url.substring(url.indexOf("datasourceId=") + 13, url.indexOf("datasourceId=") + 59);

            } else if (url.indexOf("datasource=") > 0 && url.substring(url.indexOf("datasource=") + 11).length() >= 46) {

                url = "datasource|" + url.substring(url.indexOf("datasource=") + 11, url.indexOf("datasource=") + 57);

            } else if (url.indexOf("datasourceFilter=") > 0 && url.substring(url.indexOf("datasourceFilter=") + 17).length() >= 46) {

                url = "datasource|" + url.substring(url.indexOf("datasourceFilter=") + 17, url.indexOf("datasourceFilter=") + 63);

            } else if (url.indexOf("articleId=") > 0 && url.substring(url.indexOf("articleId=") + 10).length() >= 46) {

                url = "result|" + url.substring(url.indexOf("articleId=") + 10, url.indexOf("articleId=") + 56);

            } else if (url.indexOf("datasetId=") > 0 && url.substring(url.indexOf("datasetId=") + 10).length() >= 46) {

                url = "result|" + url.substring(url.indexOf("datasetId=") + 10, url.indexOf("datasetId=") + 56);

            } else if (url.indexOf("projectId=") > 0 && url.substring(url.indexOf("projectId=") + 10).length() >= 46 && !url.contains("oai:dnet:corda")) {

                url = "project|" + url.substring(url.indexOf("projectId=") + 10, url.indexOf("projectId=") + 56);

            } else if (url.indexOf("organizationId=") > 0 && url.substring(url.indexOf("organizationId=") + 15).length() >= 46) {

                url = "organization|" + url.substring(url.indexOf("organizationId=") + 15, url.indexOf("organizationId=") + 61);

            } else {

                url = "";

            }

        } else {

            url = "";

        }

        return url;

    }

    private ArrayList<String> listHdfsDir(String dir) throws Exception {

        FileSystem hdfs = FileSystem.get(new Configuration());

        RemoteIterator<LocatedFileStatus> Files;

        ArrayList<String> fileNames = new ArrayList<>();

        try {

            Path exportPath = new Path(hdfs.getUri() + dir);

            Files = hdfs.listFiles(exportPath, false);

            while (Files.hasNext()) {

                String fileName = Files.next().getPath().toString();

                //log.info("Found hdfs file " + fileName);

                fileNames.add(fileName);

            }

            //hdfs.close();

        } catch (Exception e) {

            log.error("HDFS file path with exported data does not exist : " + new Path(hdfs.getUri() + logPath));

            throw new Exception("HDFS file path with exported data does not exist :   " + logPath, e);

        }

        return fileNames;

    }

    private String readHDFSFile(String filename) throws Exception {

        String result;

        try {

            FileSystem fs = FileSystem.get(new Configuration());

            //log.info("reading file : " + filename);

            BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(new Path(filename))));

            StringBuilder sb = new StringBuilder();

            String line = br.readLine();

            while (line != null) {

                if (!line.equals("[]")) {

                    sb.append(line);

                }

                //sb.append(line);

                line = br.readLine();

            }

            result = sb.toString().replace("][{\"idSite\"", ",{\"idSite\"");

            if (result.equals("")) {

                result = "[]";

            }

            //fs.close();

        } catch (Exception e) {

            log.error(e);

            throw new Exception(e);

        }

        return result;

    }

}

package eu.dnetlib.usagestats.export;

import java.io.*;

//import java.io.BufferedReader;

//import java.io.InputStreamReader;

import java.net.URL;

import java.net.URLConnection;

import java.sql.ResultSet;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Calendar;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import java.sql.Statement;

import org.json.simple.JSONArray;

import org.json.simple.JSONObject;

import org.json.simple.parser.JSONParser;

import org.apache.log4j.Logger;

/**

 * Created by tsampikos on 28/3/2017.

 */

public class SushiStats {

     private final String dbUrl;

    private final String dbSchema;

    private final String dbUserName;

    private final String dbPassword;

    private Connection conn = null;

    private Statement stmt = null;

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

    public SushiStats(String dbUrl, String dbUsername, String dbPassword) throws Exception {

        this.dbUrl = dbUrl;

        this.dbSchema = "shadow";

        this.dbUserName = dbUsername;

        this.dbPassword = dbPassword;

        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 sqlCreateTableSushiLog = "CREATE TABLE IF NOT EXISTS sushilog(source TEXT, repository TEXT, oai TEXT, date TEXT, metric_type TEXT, count INT, PRIMARY KEY(source, repository, oai, date, metric_type));";

            stmt.executeUpdate(sqlCreateTableSushiLog);

            String sqlCopyPublicSushiLog="INSERT INTO sushilog SELECT * FROM public.sushilog;";

            stmt.executeUpdate(sqlCopyPublicSushiLog);

            String sqlcreateRuleSushiLog = "CREATE OR REPLACE RULE ignore_duplicate_inserts AS " +

                " ON INSERT TO sushilog " +

                " WHERE (EXISTS ( SELECT sushilog.source, sushilog.repository," +

                "sushilog.oai, sushilog.date " +

                "FROM sushilog " +

                "WHERE sushilog.source = new.source AND sushilog.repository = new.repository AND sushilog.oai = new.oai AND sushilog.date = new.date AND sushilog.metric_type = new.metric_type)) DO INSTEAD NOTHING;";

            stmt.executeUpdate(sqlcreateRuleSushiLog);

            stmt.close();

            conn.close();

            log.info("Sushi Tables Created");

        } catch (Exception e) {

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

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

        }

    }

    public void sushiStats() throws Exception {

        if (conn.isClosed())

            connectDB();

        stmt = conn.createStatement();

        conn.setAutoCommit(false);

        //String sql = "INSERT INTO sushi_result_downloads SELECT s.source, d.id AS repository, ro.id, s.date, s.count FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total'";

        //String sql = "SELECT s.source, d.id AS repository_id, ro.id as result_id, extract('year' from s.date::date) ||'/'|| LPAD(CAST(extract('month' from s.date::date) AS VARCHAR), 2, '0') as date, s.count INTO downloads_stats FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total'";

        //String sql = "INSERT INTO downloads_stats SELECT s.source, d.id AS repository_id, ro.id as result_id, extract('year' from s.date::date) ||'/'|| LPAD(CAST(extract('month' from s.date::date) AS VARCHAR), 2, '0') as date, s.count FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total';";

        String sql = "INSERT INTO downloads_stats SELECT s.source, d.id AS repository_id, ro.id as result_id, extract('year' from s.date::date) ||'/'|| LPAD(CAST(extract('month' from s.date::date) AS VARCHAR), 2, '0') as date, s.count, '0' FROM sushilog s, datasource_oids d, result_oids ro WHERE s.repository=d.orid AND s.oai=ro.orid AND metric_type='ft_total';";

        stmt.executeUpdate(sql);

        stmt.close();

        conn.commit();

        conn.close();

    }

    public void processIrusRRReport() throws Exception {

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("YYYY-MM");

        String reportUrl = "http://www.irus.mimas.ac.uk/api/sushilite/v1_7/GetReport/?Report=RR1&Release=4&RequestorID=OpenAIRE&BeginDate=2016-01&EndDate=" + simpleDateFormat.format(new Date()) + "&RepositoryIdentifier=&ItemDataType=&NewJiscBand=&Granularity=Monthly&Callback=";

        String text = getJson(reportUrl, "", "");

        JSONParser parser = new JSONParser();

        JSONObject jsonObject = (JSONObject) parser.parse(text);

        jsonObject = (JSONObject) jsonObject.get("ReportResponse");

        jsonObject = (JSONObject) jsonObject.get("Report");

        jsonObject = (JSONObject) jsonObject.get("Report");

        jsonObject = (JSONObject) jsonObject.get("Customer");

        JSONArray jsonArray = (JSONArray) jsonObject.get("ReportItems");

        int i = 0;

        for (Object aJsonArray : jsonArray) {

            JSONObject jsonObjectRow = (JSONObject) aJsonArray;

            JSONArray itemIdentifier = (JSONArray) jsonObjectRow.get("ItemIdentifier");

            for (Object identifier : itemIdentifier) {

                JSONObject opendoar = (JSONObject) identifier;

                if(opendoar.get("Type").toString().equals("OpenDOAR")){

                    //System.out.println(i + ": " + opendoar.get("Value").toString());

                    log.info(i + ": " + opendoar.get("Value").toString());

                    i++;

                    processIrusIRReport(opendoar.get("Value").toString());

                    break;

                }

            }

            //break;

        }

    }

    private void processIrusIRReport(String opendoar) throws Exception {

        if (conn.isClosed())

            connectDB();

        conn.setAutoCommit(false);

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("YYYY-MM");

        Calendar start = Calendar.getInstance();

        start.set(Calendar.YEAR, 2016);

        start.set(Calendar.MONTH, Calendar.JANUARY);

        //start.setTime(simpleDateFormat.parse("2016-01"));

        Calendar end = Calendar.getInstance();

        end.add(Calendar.DAY_OF_MONTH, -1);