nc-campaign-finance/process/R/out_receipts.R

out_receipts_payer <- function(
  path_data_prep_receipts,
  path_out_report_list
) {
  lg_info_target(lg_get_logger())

  out <- path("data-out", "receipts_payer", "receipts_payer.parquet")
  dir_create(path_dir(out))

  report_list <- out_open_dataset_db(path_out_report_list)

  receipts <-
    prep_open_dataset_db(path_data_prep_receipts) |>
    semi_join(report_list, by = "report_id")

  receipts_payer <-
    receipts |>
    receipts_replace_na() |>
    distinct(
      org_name,
      is_org,
      is_us,
      profession,
      employers_name,
      street_1,
      city,
      state,
      full_zip,
      country_name
    ) |>
    add_address_lookup(postal_code = full_zip, name = "address_lookup") |>
    mutate(payer_id = row_number(), .before = 1)

  arrow::write_parquet(collect(receipts_payer), out)

  dirname(out)
}

receipts_replace_na <- function(receipts) {
  receipts |>
    tidyr::replace_na(list(
      org_name = "",
      profession = "",
      employers_name = "",
      street_1 = "",
      city = "",
      state = "",
      full_zip = "",
      country_name = ""
    ))
}

out_receipts <- function(
  path_data_prep_receipts,
  path_out_receipts_payer,
  path_out_report_list
) {
  lg_info_target(lg_get_logger())

  out <- path("data-out", "receipts", "receipts.parquet")
  dir_create(path_dir(out))

  report_list <- out_open_dataset_db(path_out_report_list)
  receipts_payer <-
    out_open_dataset_db(path_out_receipts_payer) |>
    select(-address_lookup)

  receipts <-
    prep_open_dataset_db(path_data_prep_receipts) |>
    semi_join(report_list, by = "report_id")

  # Replace payer info with payer_id
  cols_payer_common <- intersect(colnames(receipts_payer), colnames(receipts))

  receipts <-
    receipts |>
    receipts_replace_na() |>
    left_join(receipts_payer, by = cols_payer_common) |>
    select(-any_of(cols_payer_common), -name_sort) |>
    mutate(
      is_donation = receipt_type_code %in% c("IND", "CPCM", "GEN", "PPTY", "OUTS", "NFPC"),
      .after = receipt_type_code
    ) |>
    relocate(payer_id, .after = report_id) |>
    collect() |>
    mutate(across(occur_date, lubridate::mdy))

  arrow::write_parquet(receipts, out)

  dirname(out)
}