# This code creates non-answers and NER indicators on question level

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

pg <- dbConnect(PostgreSQL())
dbExecute(pg, "SET work_mem TO '10GB'") # Speeds-up creation of temporary tables

# Define PG data ----

# Project schema
rs <- 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"))
regexes <- tbl(pg, sql("SELECT * FROM non_answer_old.regexes"))
processed_files <- tbl(pg,sql("SELECT * FROM non_answer_old.processed_files"))
non_answers <- tbl(pg,sql("SELECT * FROM non_answer_old.non_answers"))
ner_class_alt_7 <- tbl(pg, sql("SELECT * FROM se_features_old.ner_class_alt_7"))
ner_to_exclude <- tbl(pg, sql("SELECT * FROM se_features_old.ner_to_exclude"))
topic_brvb <- tbl(pg, sql("SELECT * FROM se_features_old.topic_brvb"))
topic_mpr <- tbl(pg, sql("SELECT * FROM se_features_old.topic_mpr"))
fog_measure <- tbl(pg, sql("SELECT * FROM se_features_old.fog_measure"))
tone_measure <- tbl(pg, sql("SELECT * FROM se_features_old.tone_measure"))
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()) %>%
    # WARNING: universe of products
    inner_join(ner_class_alt_7 %>%
                   select(file_name, last_update) %>%
                   distinct()) %>%
    select(file_name, last_update) %>%
    distinct() %>%
    compute()

# Get ner_organization indicator has_product = 1
# Get ner details {has_money, has_percent, has_date, has_time, has_location}
ner_alt <- 
    ner_class_alt_7 %>%
    # Extract individual ner org
    mutate(ner_organization = sql("(ner_tags->'ORGANIZATION')::text"), 
           ner_money = sql("(ner_tags->'MONEY')::text"), 
           ner_percent = sql("(ner_tags->'PERCENT')::text"), 
           ner_date = sql("(ner_tags->'DATE')::text"), 
           ner_time = sql("(ner_tags->'TIME')::text"), 
           ner_loc = sql("(ner_tags->'LOCATION')::text")) %>%
    # WARNING: This drops empty ner_organization
    mutate(ner_org = sql("substring(unnest(string_to_array(ner_organization, ',')),'\\w+')"),
           ner_money = sql("substring(unnest(string_to_array(ner_money, ',')),'\\w+')"),
           ner_percent = sql("substring(unnest(string_to_array(ner_percent, ',')),'\\w+')"),
           ner_date = sql("substring(unnest(string_to_array(ner_date, ',')),'\\w+')"),
           ner_time = sql("substring(unnest(string_to_array(ner_time, ',')),'\\w+')"),
           ner_loc = sql("substring(unnest(string_to_array(ner_loc, ',')),'\\w+')")) %>%
    # Exclude financial/regulator abbrevations
    left_join(ner_to_exclude %>%
                  rename(ner_org = ner_upper) %>%
                  mutate(is_exclude = T) %>%
                  select(ner_org, is_exclude)) %>%
    mutate(ner_org = if_else(!is.na(is_exclude), NA, ner_org)) %>%
    # Aggregate back to {file_name, last_update, context, section, speaker_number} level
    group_by(file_name, last_update, context, section, speaker_number) %>%
    summarize(has_money = bool_or(!is.na(ner_money)),   
              has_percent = bool_or(!is.na(ner_percent)),  
              has_date = bool_or(!is.na(ner_date)),   
              has_time = bool_or(!is.na(ner_time)),  
              has_loc = bool_or(!is.na(ner_loc)),  
              has_product = bool_or(!is.na(ner_org)), 
              ner_organization = sql("array_agg(DISTINCT ner_org)")) %>%
    ungroup() %>%
    distinct() %>%
    compute()

# Combine topics (same number of rows in all topic tables with all FALSE included)
# WARNING: IGNORE _kls
topic <- 
    topic_brvb %>%
    rename(has_forward_brvb = forward_looking, 
           has_earn_brvb = earnings) %>%
    left_join(topic_mpr %>%
                  rename(has_earn_mpr = finance,
                         has_forward_mpr = forward_looking)) %>%
    # Aggregate back to {file_name, last_update, context, section, speaker_number} level
    group_by(file_name, last_update, context, section, speaker_number) %>%
    summarize_at(.vars = vars(matches("has_")),
                 .funs = funs(bool_or(.))) %>%
    ungroup() %>%
    distinct() %>%
    compute()

# Get tone {positive, negative, uncertainty}
# WARNING: {positive, negative, uncertainty} are counts and thus have to be scaled 
#   by the number of words
# WARNING: tone is based on streetevents_old data that is slightly different from original data; 
#   for that reason do NOT use last_update
tone_alt <- 
    tone_measure %>%
    select(file_name, context, section, speaker_number, positive, negative, uncertainty) %>%
    distinct() %>%
    compute()

# Get fog {fog, fog_words, fog_sents}
# WARNING: fog is index (already scaled)
# WARNING: fog is based on streetevents_old data that is slightly different from original data; 
#   for that reason do NOT use last_update
fog_alt <- 
    fog_measure %>%
    select(file_name, context, section, speaker_number, fog) %>%
    distinct() %>%
    compute()

# Get number of words and number of numbers
# 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.
# Sum_num: Number of numbers appeared in a passage, excluding year (199X/20XX).
word_counts_alt <- 
    word_counts %>%
    select(file_name, context, section, speaker_number, sum, sum_num) %>%
    distinct() %>%
    compute()

# Get non-answer data with type indicators
nonanswer <-
    non_answers %>%
    # Process only relevant files
    inner_join(se_link_alt) %>%
    # non_answers have multiple obs. per {file_name, last_update}
    # e.g., file_name == '5495803_T' & answer_nums == '{107,107,108,108,109,109}'
    distinct() %>%
    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") %>%
    filter(!is.na(category)) %>%
    group_by(file_name, last_update, speaker_number) %>%
    summarize(is_refuse = bool_or(category == 'REFUSE'),
              is_unable = bool_or(category == 'UNABLE'),
              is_aftercall = bool_or(category == 'AFTERCALL')) %>%
    ungroup() %>%
    mutate(is_nonans = is_refuse | is_unable | is_aftercall) %>%
    compute()

# Collapse question-level measures by answer_nums -----

# Get speaker for question_nums and answer_nums
utter <- 
    qa_pairs %>%
    inner_join(se_link_alt) %>%
    mutate(question_number = unnest(question_nums), 
           answer_number = unnest(answer_nums)) %>%
    distinct() %>%
    compute()

# Collapse nonanswers and products by answer_nums
ner_nums <- 
    utter %>%
    # Require ner; ner table contains entries for all processed utterances even 
    # if ner_tags are empty (ner_tags = {})
    inner_join(ner_alt %>%
                   filter(context == "qa" & section == 1) %>%
                   rename(question_number = speaker_number)) %>%
    distinct() %>%
    # Unnest ner_organization for array_agg() below
    mutate(ner_org = sql("unnest(ner_organization)")) %>%
    # Collapse products based on qa_pairs
    group_by(file_name, answer_nums, question_nums) %>%
    summarize(has_money = bool_or(has_money),   
              has_percent = bool_or(has_percent),  
              has_date = bool_or(has_date),   
              has_time = bool_or(has_time), 
              has_loc = bool_or(has_loc), 
              has_product = bool_or(has_product), 
              ner_organization = sql("array_agg(DISTINCT ner_org)")) %>%
    ungroup() %>%
    compute()

# Collapse topics by answer_nums
topic_nums <- 
    utter %>%
    # Merge topic of a question
    inner_join(topic %>%
                   filter(context == "qa" & section == 1) %>%
                   rename(question_number = speaker_number)) %>%
    distinct() %>%
    # Collapse topics based on qa_pairs
    group_by(file_name, answer_nums, question_nums) %>%
    summarize_at(.vars = vars(matches("has_")), 
                 .funs = funs(bool_or(.))) %>%
    ungroup() %>%
    compute()

# Collapse tone by answer_nums
tone_nums <- 
    utter %>%
    # Merge tone of a question
    inner_join(tone_alt %>%
                   filter(context == "qa" & section == 1) %>%
                   rename(question_number = speaker_number)) %>%
    distinct() %>%
    # Collapse tone based on qa_pairs
    group_by(file_name, answer_nums, question_nums) %>%
    summarize_at(.vars = vars(positive, negative, uncertainty), 
                 .funs = funs(sum(.))) %>%
    ungroup() %>%
    mutate(has_positive = (positive > 0), 
           has_negative = (negative > 0),
           has_uncertainty = (uncertainty > 0)) %>%
    compute()

# Collapse fog by answer_nums
fog_nums <- 
    utter %>%
    # Merge fog of a question
    inner_join(fog_alt %>%
                   filter(context == "qa" & section == 1) %>%
                   rename(question_number = speaker_number)) %>%
    distinct() %>%
    # Collapse fog based on qa_pairs
    group_by(file_name, answer_nums, question_nums) %>%
    summarize(fog = mean(fog, na.rm = TRUE)) %>%
    ungroup() %>%
    compute()

# Collapse word_counts by answer_nums
word_quest_nums <- 
    utter %>%
    # Merge word_counts of a question
    inner_join(word_counts_alt %>%
                   filter(context == "qa" & section == 1) %>%
                   rename(question_number = speaker_number)) %>%
    distinct() %>%
    # Collapse word_counts based on qa_pairs
    group_by(file_name, answer_nums, question_nums) %>%
    summarize(words_quest = sum(sum), 
              nums = sum(sum_num)) %>%
    ungroup() %>%
    mutate(has_nums = (nums > 0)) %>%
    compute()

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

# Collapse non-answers
nonanswer_nums <- 
    utter %>%
    # Nonanswer data is only(!) when answer contains nonanswers---left_join(!)
    left_join(nonanswer %>%
                  rename(answer_number = speaker_number)) %>%
    distinct() %>%
    # Collapse nonanswers based on based on qa_pairs
    group_by(file_name, last_update, answer_nums, question_nums) %>%
    summarize(is_refuse = bool_or(is_refuse),
              is_unable = bool_or(is_unable),
              is_aftercall = bool_or(is_aftercall),
              is_nonans = bool_or(is_nonans)) %>%
    ungroup() %>%
    compute()

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

nonans_ner <- 
    nonanswer_nums %>%
    left_join(topic_nums) %>%
    # WARNING: product_num is incomplete when product mention is missing
    left_join(ner_nums) %>%
    left_join(tone_nums) %>%
    left_join(fog_nums) %>%
    left_join(word_ans_nums) %>%
    left_join(word_quest_nums) %>%
    # Compute frac for {positive, negative, uncertainty}
    mutate_at(.vars = vars(positive, negative, uncertainty), 
              .funs = funs(as.numeric(.)/words_quest)) %>%
    select(-words_quest, -nums) %>%
    distinct() %>%
    # Count answers
    group_by(file_name) %>%
    mutate(all_ans = n()) %>%
    ungroup() %>%
    # Fill in has_product and is_
    mutate_at(.vars = vars(matches("has_|is_")), 
              .funs = funs(coalesce(., FALSE))) %>%
    compute(name = "nonans_ner", temporary = FALSE)

db_comment <- paste0("CREATED USING azakolyukina/nonans_product/create_nonans_ner.R ON ", Sys.time())
dbGetQuery(pg, paste0("COMMENT ON TABLE nonans_ner IS '",
                      db_comment, "';"))

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

dbDisconnect(pg)
