# This code creates non-answer rates on conference call level

library(RPostgreSQL)
library(dplyr, warn.conflicts = FALSE)

pg <- dbConnect(PostgreSQL())
dbExecute(pg, "SET work_mem TO '1GB'") 

# Define PG data ----

# Project schema
dbGetQuery(pg, "SET search_path TO nonans_product, public")
se_link <- tbl(pg, "se_link")

# Outside of project schema
qa_pairs <- tbl(pg, sql("SELECT * FROM streetevents_old.qa_pairs"))
processed_files <- tbl(pg,sql("SELECT * FROM non_answer_old.processed_files"))
non_answers <- tbl(pg, sql("SELECT * FROM non_answer_old.non_answers"))
regexes <- tbl(pg, sql("SELECT * FROM non_answer_old.regexes"))
word_counts <- tbl(pg, sql("SELECT * FROM se_features_old.word_counts"))

# Get latest calls by update
se_link_alt <-
    se_link %>%
    # WARNING: universe of nonanswers
    # nonanswers are extracted only from processed files
    inner_join(processed_files %>%
                   select(file_name, last_update) %>%
                   distinct()) %>%
    select(file_name, last_update) %>%
    distinct() %>%
    compute()

# Get speaker for answer_nums
utter <- 
    qa_pairs %>%
    inner_join(se_link_alt) %>%
    # Unnest answer_nums
    mutate(speaker_number = sql("unnest(answer_nums)"), 
           context = sql("'qa'::text"), 
           answer = TRUE) %>%
    select(file_name, last_update, answer_nums, speaker_number, context, answer) %>%
    distinct() %>%
    compute()

# Get non-answer data with type indicators at speaker_number-level
# i.e., utterance-level
nonans_cat <-
    utter %>%
    left_join(non_answers) %>%
    mutate(has_non_answer = !is.na(non_answers)) %>%
    mutate(non_answer = sql("unnest(non_answers)")) %>%
    mutate(regex_id = sql("(non_answer->'regex_id')::text::integer")) %>%
    left_join(regexes, by = "regex_id") %>%
    select(file_name, last_update, answer_nums, category) %>%
    compute()
 
nonans_alt <-
    utter %>%
    left_join(non_answers) %>%
    left_join(nonans_cat, by = c("file_name", "last_update", "answer_nums")) %>%
    mutate(category = coalesce(category, 'NONE')) %>%
    # Collapse on answer-level, i.e., collapse the string of answers to the same analyst
    group_by(file_name, last_update, answer_nums) %>%
    summarize(is_refuse = bool_or(category == 'REFUSE'),
              is_unable = bool_or(category == 'UNABLE'),
              is_aftercall = bool_or(category == 'AFTERCALL')) %>%
    ungroup() %>%
    mutate(is_either_category = is_refuse | is_unable | is_aftercall) %>%
    compute()

# Aggregate non-answers by call
nonans_rate_all <- 
    nonans_alt %>%
    group_by(file_name, last_update) %>%
    summarize(all_ans = n(),
              nonans = coalesce(sum(as.integer(is_either_category)), 0),
              refuse = coalesce(sum(as.integer(is_refuse)), 0),
              unable = coalesce(sum(as.integer(is_unable)), 0),
              aftercall = coalesce(sum(as.integer(is_aftercall)), 0)) %>%
    mutate_at(.vars = vars(nonans, refuse, unable, aftercall), 
              .funs = funs(rate = ./NULLIF(all_ans, 0))) %>%
    ungroup() %>%
    compute()

# Get number of words
# WARNING: word_counts is based on streetevents_old data that is slightly different from original data; 
#   for that reason do NOT use last_update
# Sum: Number of words in a passage.
word_counts_alt <- 
    word_counts %>%
    select(file_name, context, section, speaker_number, sum) %>%
    distinct() %>%
    compute()

# Collapse word_counts by answer_nums
word_ans_nums <- 
    utter %>%
    # Merge word_counts of a question
    inner_join(word_counts_alt %>%
                   filter(context == "qa" & section == 1)) %>%
    distinct() %>%
    # Collapse word_counts based on qa_pairs
    group_by(file_name, answer_nums) %>%
    summarize(words_ans = sum(sum)) %>%
    ungroup() %>%
    compute()

# Aggregate response length by call
# WARNING: check that no duplicates by {file_name, answer_nums}
word_ans_all <- 
    word_ans_nums %>%
    group_by(file_name) %>%
    summarize(words_ans_avg = mean(words_ans, na.rm = TRUE)) %>%
    ungroup() %>%
    compute()

dbGetQuery(pg, "DROP TABLE IF EXISTS nonans_rate")

nonans_rate <-
    se_link_alt %>%
    inner_join(nonans_rate_all) %>%
    left_join(word_ans_all) %>%
    compute(name = "nonans_rate", temporary = FALSE)

db_comment <- paste0("CREATED USING azakolyukina/nonans_product/create_nonans_rate.R ON ", Sys.time())
dbGetQuery(pg, sprintf("COMMENT ON TABLE nonans_rate IS '%s'", db_comment))

dbGetQuery(pg, "ALTER TABLE nonans_rate OWNER TO nonans_product")
dbGetQuery(pg, "GRANT SELECT ON nonans_rate TO nonans_product_access")
dbGetQuery(pg, "VACUUM ANALYZE nonans_rate")

dbDisconnect(pg)
