# Get CPA Sample

# Project Setup

    library(tidyverse)
    source("1_code/functions.R")

# Install GitHub

Custom Library for Location LookUps

    devtools::install_github(
      repo = "https://github.com/MatthiasUckert/Rlookup",
      quiet = TRUE
    )

    devtools::install_github(
      repo = "https://github.com/MatthiasUckert/RFsel",
      quiet = TRUE
    )

# Clean Individual Raw Data

## Paths & Files

    .dir_ind_tables <- "0_data/RawData/2020_12_19_ind/tables/"
    .dir_ind_htmls <- "0_data/RawData/2020_12_19_ind/htmls/"

    .fil_ind_tables <- lfc(.dir_ind_tables, "rds")
    .fil_ind_htmls <- lfc(.dir_ind_htmls, "html")

## Read Data

### Read Tablular Data

    tab_ind_tables_raw <- cache_obj(
      .expr = {
        future::plan("multisession", workers = 10)
        tab_ind_tables <- furrr::future_map(
          .x = .fil_ind_tables,
          .f = readr::read_rds,
          .options = furrr::furrr_options(seed = TRUE),
          .progress = TRUE
        ) %>% dplyr::bind_rows()
        future::plan("default")
        tab_ind_tables
      },
      .name = "ind_tables_raw",
      .dir = "2_output/",
      .rerun = FALSE
    )

### Read HTML Data

    tab_ind_htmls_raw <- cache_obj(
      .expr = {
        future::plan("multisession", workers = 10)
        tab_ind_htmls_raw <- furrr::future_imap(
          .x = .fil_ind_htmls,
          .f = ~ get_info_map(.x, .y, "individual"),
          .options = furrr::furrr_options(seed = TRUE),
          .progress = TRUE
        ) %>% dplyr::bind_rows()
        future::plan("default")
        tab_ind_htmls_raw
      },
      .name = "ind_htmls_raw",
      .dir = "2_output/",
      .rerun = FALSE
    )

## Clean Data

### Clean Tabular Data

    tab_ind_tables_mod <- cache_obj(
      .expr = clean_tables_raw(tab_ind_tables_raw),
      .name = "ind_tables_mod",
      .dir = "2_output/",
      .rerun = FALSE
    )

### Clean HTML Data

    tab_ind_htmls_mod <- cache_obj(
      .expr = clean_htmls_raw(tab_ind_htmls_raw),
      .name = "ind_htmls_mod",
      .dir = "2_output/",
      .rerun = FALSE
    )

## Merge Tables

    tab_ind_raw <- merge_tables_and_htmls(tab_ind_tables_mod, tab_ind_htmls_mod)

## Clean-Up Environment

    rm(tab_ind_htmls_mod, tab_ind_htmls_raw, tab_ind_tables_mod, tab_ind_tables_raw)
    invisible(gc())

# Get Location Data

## Lookup Address Data

### Prepare Address Table

    tab_address0 <- prep_address_table(tab_ind_raw)

### Lookup States

    tab_state <- cache_obj(
      .expr = Rlookup::lookup_location(
        .str     = unique(na.omit(tab_address0$state)),
        .type    = "state",
        .match   = "complete",
        .min_sim = .85
      ) %>%
        dplyr::filter(state_n == 1) %>%
        dplyr::select(name_raw, state_code = state_state_code) %>%
        tidyr::unnest(state_code),
      .name = "lookup_state",
      .dir = "2_output/",
      .rerun = FALSE
    )

    tab_address1 <- tab_address0 %>%
      dtplyr::lazy_dt() %>%
      dplyr::left_join(tab_state, by = c("state" = "name_raw")) %>%
      dplyr::mutate(
        country = dplyr::if_else(!is.na(state_code), "USA", NA_character_)
      ) %>%
      tibble::as_tibble()

### Lookup Countries

    tab_country <- cache_obj(
      .expr = Rlookup::lookup_location(
        .str     = unique(na.omit(.tmp$address_adj)),
        .type    = "country",
        .match   = "partial",
        .min_sim = .95
      ) %>%
        dplyr::filter(!is.na(country_n)) %>%
        dplyr::select(name_raw, country_code = country_country_code) %>%
        tidyr::unnest(country_code) %>%
        dplyr::arrange(name_raw, factor(country_code, levels = c("USA"))) %>%
        dplyr::distinct(name_raw, .keep_all = TRUE),
      .name = "lookup_country",
      .dir = "2_output/",
      .rerun = FALSE
    )

    tab_address2 <- tab_address1 %>%
      dtplyr::lazy_dt() %>%
      dplyr::left_join(tab_country, by = c("address_adj" = "name_raw")) %>%
      dplyr::mutate(country_code = dplyr::case_when(
        !is.na(country) ~ country,
        !is.na(country_code) ~ country_code
      )) %>%
      tibble::as_tibble()

### Lookup Cities

    tab_city <- cache_obj(
      .expr = Rlookup::lookup_location(
        .str = unique(na.omit(tab_address2$city)),
        .type = "city",
        .match = "complete",
        .min_sim = .5,
        .max_return = 20
      ) %>%
        dplyr::filter(!is.na(city_n)) %>%
        dplyr::left_join(
          y = dplyr::distinct(tab_address2, city, country_code),
          by = c("name_raw" = "city")
        ) %>%
        dplyr::arrange(country_code) %>%
        dplyr::distinct(id, .keep_all = TRUE) %>%
        tidyr::unnest(c(city_country_code, city_name_match, city_sim)) %>%
        dplyr::filter(city_country_code == country_code | is.na(country_code)) %>%
        dplyr::mutate(
          country_code = dplyr::if_else(is.na(country_code), city_country_code, country_code),
          city_name = stringr::str_to_title(city_name_match)
        ) %>%
        dplyr::distinct(id, .keep_all = TRUE) %>%
        dplyr::select(city = name_raw, city_country = country_code, city_name),
      .name = "lookup_city",
      .dir = "2_output/",
      .rerun = FALSE
    )

    tab_address3 <- tab_address2 %>%
      dplyr::left_join(tab_city, by = "city") %>%
      dplyr::mutate(country_code = dplyr::case_when(
        is.na(country_code) ~ city_country,
        TRUE ~ country_code
      )) %>%
      dplyr::mutate(
        city_type = ifelse(is.na(city_name), 0L, 1L),
        city_name = ifelse(is.na(city_name), city, city_name),
        city_name = stringr::str_to_title(city_name),
        city_name = trimws(gsub("\\s+", " ", city_name))
      ) %>%
      dplyr::select(
        address_type, address_raw, address_adj, city_type,
        city_raw = city,
        city_name, state_code, country_code
      )

## Final Address Table

    tab_address_fin <- tab_address3

## Clean-Up Environment

    rm(tab_address0, tab_address1, tab_address2, tab_address3)
    rm(tab_city, tab_country, tab_state)
    invisible(gc())

# Get ZIP Codes

## Paths

    .fil_hud <- lfc("0_data/location_data/hud/", .reg = "xlsx")
    .path_dol <- "0_data/location_data/dol/fs11_gpci_by_msa-ZIP.xlsx"

## Prepare crosswalk files

Note: for multiple cbsa/tract/cd/… we take the zipcode with the highest residential ration

    tab_hud <- get_hud_data(.fil_hud)
    tab_dol <- get_dol_data(.path_dol)

    tab_city_state <- tab_address_fin %>%
      dtplyr::lazy_dt() %>%
      dplyr::filter(!is.na(state_code)) %>%
      dplyr::distinct(city_name, state_code) %>%
      dplyr::mutate(id = paste0(city_name, "_", state_code)) %>%
      dplyr::filter(!grepl("[[:punct:]]", city_name)) %>%
      dplyr::filter(!grepl("\\d", city_name)) %>%
      dplyr::filter(!city_name == "") %>%
      dplyr::filter(!is.na(city_name)) %>%
      tibble::as_tibble()

## Get USPS Data

    fil_usps <- "2_output/scrape_usps.fst"
    dir_usps <- fs::dir_create("2_output/htmls_usps")
    tmp_usps <- cache_obj(
      .expr = {
        tab_use <- dplyr::filter(tab_city_state, !id %in% names(lfc(dir_usps)))
        lst_use <- split(tab_use, tab_use$id)
        lst_map <- split(lst_use, ceiling(seq_along(lst_use) %% 3))

        dir_map <- normalizePath(dpath_dloc())
        docker_id <- RFsel::sel_docker_start("debug", "firefox", "latest", 4446, 5902, dir_map)
        future::plan("multiprocess", workers = 3)
        furrr::future_walk(
          .x = lst_map,
          .f = ~ usps_zip_map(4446, .x, .wait = 5),
          .options = furrr::furrr_options(seed = TRUE)
        )
        future::plan("default")

        dplyr::bind_rows(purrr::map(lfc(dir_usps), extract_zip), .id = "id")
      },
      .name = "scrape_usps",
      .dir = "2_output/",
      .rerun = FALSE
    )

## Finalize ZIP Code

    tab_zip_info <- zip_get_zip_code_info(tmp_usps, tab_dol, tab_hud)

    tab_ind_loc <- zip_merge_ind_with_location(
      .tab_ind_raw = tab_ind_raw,
      .tab_merge_adress = tab_address_fin %>%
        dplyr::distinct(address_raw, address_adj, city_name, state_code, country_code),
      .tab_merge_city_state = tab_city_state %>%
        dplyr::left_join(tab_zip_info, by = "id") %>%
        dplyr::select(city_name, state_code, zip)
    )

## Clean-Up Environment

    rm(tab_address_fin, tab_city_state, tab_dol, tab_hud, tmp_usps)
    invisible(gc())

# Get Gender

    tab_gender_names <- cache_obj(
      .expr = gender_prepare_name_table(tab_ind_raw) %>%
        dplyr::mutate(id = paste(name, year_min, year_max, sep = "-")) %>%
        dplyr::distinct(),
      .name = "name_table",
      .dir = "2_output/",
      .rerun = FALSE
    )

## Get Gender (First Name only)

    tab_gender_ssa_name <- cache_obj(
      .expr = gender::gender(
        names = unique(tab_gender_names$name),
        method = "ssa",
        countries = "United States"
      ),
      .name = "gender_ssa_name",
      .dir = "2_output/",
      .rerun = FALSE
    ) %>%
      dplyr::mutate(method = "ssa_name")

    tab_gender_ipums_name <- cache_obj(
      .expr = gender::gender(
        names = unique(tab_gender_names$name),
        method = "ipums",
        countries = "United States"
      ),
      .name = "gender_ipums_name",
      .dir = "2_output/",
      .rerun = FALSE
    ) %>% dplyr::mutate(method = "ipums_name")

    tab_gender_genderize_name <- cache_obj(
      .expr = gender_genderize(
        .names = unique(tab_gender_names$name),
        .path_cache = fs::dir_create("2_output/genderize_store")
      ) %>% dplyr::mutate(method = "genderize_name"),
      .name = "gender_genderize_name",
      .dir = "2_output/",
      .rerun = FALSE
    )

## Get Gender (First Name & Years)

    tab_gender_names_year <- distinct(tab_gender_names, id, name, year_min, year_max)
    lst_gender_names_year <- purrr::map(
      .x = purrr::set_names(1:nrow(tab_gender_names_year), tab_gender_names_year$id),
      .f = ~ list(
        name = tab_gender_names_year$name[.x],
        years = c(tab_gender_names_year$year_min[.x], tab_gender_names_year$year_max[.x])
      )
    )

    tab_gender_ssa_year <- cache_obj(
      .expr = purrr::map(
        .x = lst_gender_names_year,
        .f = ~ gender_year(.x, "ssa"),
        .progress = TRUE
      ) %>% dplyr::bind_rows(.id = "id") %>%
        dplyr::mutate(method = "ssa_year"),
      .name = "gender_ssa_year",
      .dir = "2_output/",
      .rerun = FALSE
    )

    tab_gender_ipums_year <- cache_obj(
      .expr = purrr::map(
        .x = lst_gender_names_year,
        .f = ~ gender_year(.x, "ipums"),
        .progress = TRUE
      ) %>% dplyr::bind_rows(.id = "id") %>%
        dplyr::mutate(method = "ipums_year"),
      .name = "gender_ipums_year",
      .dir = "2_output/",
      .rerun = FALSE
    )

## Gender by Method

    tab_gender_method <- gender_get_table_by_method(
      .tab_gender_names          = tab_gender_names,
      .tab_gender_genderize_name = tab_gender_genderize_name,
      .tab_gender_ipums_name     = tab_gender_ipums_name,
      .tab_gender_ssa_name       = tab_gender_ssa_name,
      .tab_gender_ipums_year     = tab_gender_ipums_year,
      .tab_gender_ssa_year       = tab_gender_ssa_year
    )

    tab_gender <- tab_gender_method %>%
      dtplyr::lazy_dt() %>%
      dplyr::group_by(ind_id) %>%
      dplyr::arrange(-guess_diff, .by_group = TRUE) %>%
      dplyr::slice(1) %>%
      dplyr::ungroup() %>%
      tibble::as_tibble()

## Clean-Up Environment

    rm(tab_gender_genderize_name, tab_gender_ipums_name, tab_gender_ipums_year)
    rm(tab_gender_names, tab_gender_names_year, tab_gender_ssa_name, tab_gender_ssa_year)
    rm(lst_gender_names_year)
    invisible(gc())

# Get Race

## Paths & Files

    .path_race_census <- "/Users/matthiasuckert/Dropbox/LicensingMinority/LicensingMinority_Data/Data_Raceguesses/raceguess_census.dta"
    .path_race_floter <- "/Users/matthiasuckert/Dropbox/LicensingMinority/LicensingMinority_Data/Data_Raceguesses/raceguess_flvoter.dta"

## Prepare Race Name Tables

    tab_ind_race <- race_prepare_ind_names(
      .tab_ind_raw = tab_ind_raw,
      .tab_zip_info = tab_zip_info,
      .tab_ind_loc = tab_ind_loc,
      .tab_gender = tab_gender
    )

    tmp_ind_race <- dplyr::filter(tab_ind_race, !state %in% c("GU", "VI", "CNMI"))

    tab_race_surname <- tab_ind_race %>%
      dplyr::distinct(state, surname)

    tab_county <- tmp_ind_race %>%
      dplyr::distinct(state, surname, county) %>%
      dplyr::filter(!is.na(county))

    tab_county_age <- tmp_ind_race %>%
      dplyr::distinct(state, surname, county, age) %>%
      dplyr::filter(!is.na(county), !is.na(age))

    tab_county_sex <- tmp_ind_race %>%
      dplyr::distinct(state, surname, county, sex) %>%
      dplyr::filter(!is.na(county), !is.na(sex))

    tab_county_age_sex <- tmp_ind_race %>%
      dplyr::distinct(state, surname, county, age, sex) %>%
      dplyr::filter(!is.na(county), !is.na(sex), !is.na(age))

    lst_race <- list(
      prr_surname = cache_obj(
        .expr = predictrace::predict_race(distinct(tab_surname, surname)[[1]]),
        .name = "prr_surname",
        .dir = "2_output/",
        .rerun = FALSE
      ),
      wru_surname = cache_obj(
        .expr = wru_surname(tab_surname),
        .name = "wru_surname",
        .dir = "2_output/",
        .rerun = FALSE
      ),
      wru_county = cache_obj(
        .expr = wru_county(tab_county),
        .name = "wru_county",
        .dir = "2_output/",
        .rerun = FALSE
      ),
      wru_county_age = cache_obj(
        .expr = wru_county_age(tab_county_age),
        .name = "wru_county_age",
        .dir = "2_output/",
        .rerun = FALSE
      ),
      wru_county_sex = cache_obj(
        .expr = wru_county_sex(tab_county_sex),
        .name = "wru_county_sex",
        .dir = "2_output/",
        .rerun = FALSE
      ),
      wru_county_age_sex = cache_obj(
        .expr = wru_county_age_sex(tab_county_age_sex),
        .name = "wru_county_age_sex",
        .dir = "2_output/",
        .rerun = FALSE
      )
    )

    tab_prr_surname <- lst_race$prr_surname %>%
      dplyr::select(
        surname = name,
        pred.whi = probability_white,
        pred.bla = probability_black,
        pred.his = probability_hispanic,
        pred.asi = probability_asian,
        pred.oth = probability_american_indian
      )
    lst_race <- purrr::imap(lst_race, ~ dplyr::mutate(.x, method = .y))

    tab_race_method <- race_get_table_by_method(
      .lst_race = lst_race,
      .tab_ind_race = tab_ind_race,
      .path_race_census = .path_race_census,
      .path_race_floter = .path_race_floter
    )

    tab_race <- tab_race_method %>%
      dtplyr::lazy_dt() %>%
      dplyr::group_by(ind_id) %>%
      dplyr::arrange(ind_id, -guess_diff, .by_group = TRUE) %>%
      dplyr::slice(1) %>%
      dplyr::ungroup() %>%
      tibble::as_tibble()

## Clean-Up Environment

    rm(tab_county, tab_county_age, tab_county_age_sex, tab_county_sex)
    rm(tab_race_surname, tab_prr_surname, lst_race)
    invisible(gc())

# Build Disc. Action Dictionary

## Paths

    .dir_ind_htmls <- "0_data/00_ind_data/2020_12_19_ind/htmls/"
    .fil_ind_htmls <- lft(.dir_ind_htmls, "html")
    .path_add <- "0_data/additional_dictionary_action.xlsx"
    .path_dict_edit <- "0_data/dict_edit.xlsx"

## Extract Dictionary Entries

    tab_dict_info_raw <- cache_obj(
      .expr = {
        future::plan("multisession", workers = 10)
        tmp_ <- furrr::future_map(
          .x = .fil_ind_htmls$path,
          .f = dict_s_extract_infos_raw,
          .options = furrr::furrr_options(seed = TRUE),
          .progress = TRUE
        ) %>% purrr::transpose()
        future::plan("default")
        dplyr::bind_rows(tmp_$result)
      },
      .name = "raw_dict_entries",
      .dir = "2_output/",
      .rerun = FALSE
    )

## Adjust Dictionary Entries

    tab_dict_info_adj <- cache_obj(
      .expr = dict_adjust_infos(tab_dict_info_raw),
      .name = "adj_dict_entries",
      .dir = "2_output/",
      .rerun = FALSE
    )

    .tab_add <- dplyr::mutate(openxlsx::read.xlsx(.path_add), state = as.character(state))

    tab_dict_info_adj <- dplyr::bind_rows(tab_dict_info_adj, .tab_add)

## Merge Dictionary

    tab_status <- openxlsx::read.xlsx(.path_dict_edit, "license_status") %>%
      dplyr::filter(type_adj == "license_status") %>%
      dplyr::select(license_status = var, state, active, status, reason)

    tab_ind_dict <- tab_ind_raw %>%
      dtplyr::lazy_dt() %>%
      dplyr::select(ind_id, entry_id, state, license_status, license_basis, license_type) %>%
      dplyr::left_join(tab_status, by = c("state", "license_status")) %>%
      tibble::as_tibble()

    stats_license_status <- tab_ind_raw %>%
      dtplyr::lazy_dt() %>%
      dplyr::group_by(license_status, state) %>%
      dplyr::summarise(
        n_licenses = dplyr::n_distinct(license_num),
        n_individual = dplyr::n_distinct(ind_id),
        .groups = "drop"
      ) %>%
      tibble::as_tibble()

# Get Discc. Action

## Paths

    .path_action <- "0_data/disciplinary_action/2021-02-15_disciplinary_action.xlsx"

## Read Files

    .tab_action <- openxlsx::read.xlsx(.path_action)

## Transform Files

    tab_action <- .tab_action %>%
      dtplyr::lazy_dt() %>%
      dplyr::group_by(state, license_status, desc) %>%
      dplyr::summarise(action = mean(update, na.rm = TRUE), .groups = "drop") %>%
      tibble::as_tibble()

    unique(tab_action$action)
    any(duplicated(paste0(tab_action$state, tab_action$license_status)))

    tab_ind_disc <- tab_ind_raw %>%
      dtplyr::lazy_dt() %>%
      dplyr::select(ind_id, entry_id, state, license_status) %>%
      dplyr::mutate(
        license_status = trimws(license_status),
        license_status = dplyr::if_else(license_status == "", NA_character_, license_status),
        license_status = gsub("/", "_", license_status)
        ) %>%
      dplyr::left_join(tab_action, by = c("state", "license_status")) %>%
      tibble::as_tibble()

    nrow(tab_ind_disc) == nrow(tab_ind_raw)
    any(is.na(tab_ind_disc$action))

# Finalize Sample

    fin_zip_info <- tab_zip_info %>%
      dplyr::select(zip, cbsa_hud, msa_no_dol)
    fin_zip_ind <- tab_ind_loc %>%
      dtplyr::lazy_dt() %>%
      dplyr::select(ind_id, entry_id, same_address, address_type, address_adj, city_name, state_code, country_code, zip) %>%
      dplyr::left_join(tab_zip_info, by = "zip") %>%
      tibble::as_tibble()

    fin_cpa_ind <- tab_ind_raw %>%
      dtplyr::lazy_dt() %>%
      dplyr::left_join(dplyr::select(fin_zip_ind, -state), by = c("ind_id", "entry_id")) %>%
      dplyr::left_join(tab_gender, by = "ind_id", suffix = c("_gender", "_race")) %>%
      dplyr::left_join(tab_race, by = "ind_id", suffix = c("_gender", "_race")) %>%
      dplyr::select(-action) %>%
      dplyr::left_join(
        y = dplyr::select(tab_ind_disc, ind_id, entry_id, action),
        by = c("ind_id", "entry_id")
        ) %>%
      dplyr::select(-common_city_list, -area_code_list) %>%
      tibble::as_tibble()

    colnames(fin_cpa_ind)

    rio::export(fin_cpa_ind, "2_output/CPASample.dta")