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

    private final String dbUserName;

    private final String dbPassword;

    private final String logPath;

    private Connection conn = null;

    private Statement stmt = null;

        this.dbUrl = dbUrl;

        this.dbSchema = "shadow";

        this.dbUserName = dbUserName;

        this.dbPassword = dbPassword;

        this.logPath = logPath;

        connectDB();

        createTables();

    }

        try {

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

        } catch (Exception e) {

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

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

    }

            stmt.executeUpdate(sqlcreateRulePiwikLog);

            stmt.executeUpdate(sqlCopyPublicPiwiklog);

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

            stmt.close();

            conn.close();

            log.info("Usage Tables Created");

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

            //System.exit(0);

        }

    }

            log.info("repository process done");

            removeDoubleClicks();

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

            log.info("cleaning oai done");

        } catch (Exception e) {

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

        }

    }

            //dataSourceStats();

            //organizationsStats();

            //projectsStats();

            //repositoryViewsStats();

            //repositoryDownloadsStats();

            viewsStats();

            downloadsStats();

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

        } catch (Exception e) {

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

        }

    }

    // Import repository Logs to DB

        if (conn.isClosed())

        for (String jsonFile : jsonFiles) {

            JSONParser parser = new JSONParser();

                JSONObject jsonObjectRow = (JSONObject) aJsonArray;

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

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

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

                JSONArray actionDetails = (JSONArray) jsonObjectRow.get(("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 url = actionDetailsObj.get("url").toString();

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

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

                        prepStatem.setString(2, idVisit);

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

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

        stmt.close();

        conn.commit();

        stmt.close();

        conn.commit();

        conn.close();

    }

        if (conn.isClosed())

            connectDB();

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

        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.executeUpdate(sql);

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

        stmt.executeUpdate(sql);

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

        stmt.executeUpdate(sql);

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

        stmt.executeUpdate(sql);

        conn.commit();

        conn.close();

    }

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

//    }

//        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

        if (conn.isClosed())

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

        stmt.executeUpdate(sql);

        conn.commit();

        conn.close();

    }

    private void repositoryDownloadsStats() throws Exception {

        if (conn.isClosed())

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

        conn.commit();

        conn.close();

    }

    // Import OPENAIRE Logs to DB

                JSONObject jsonObjectRow = (JSONObject) aJsonArray;

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

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

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

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

                    JSONObject actionDetailsObj = (JSONObject) actionDetail;

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

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

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

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

                    String sourceItemType = "";

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

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

                    prepStatem.setInt(1, idSite);

                    prepStatem.setString(2, idVisit);

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

                        conn.commit();

                        batch_size = 0;

                    }

                }

            }

        }

        prepStatem.executeBatch();

        conn.commit();

        stmt.close();

        conn.close();

    }

        if (conn.isClosed())

        stmt.close();

        conn.commit();

        stmt.close();

        conn.commit();

        stmt.close();

        conn.commit();

        stmt.close();

        conn.commit();

        sql = "DROP TABLE process_portal_log;";

        stmt.close();

        conn.commit();

        conn.close();

    }

        if (conn.isClosed())

        conn.setAutoCommit(false);

        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/%';";

        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/%';";

        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/%';";