deobligations.R

# Set working directory to the folder containing this script
setwd(dirname(rstudioapi::getActiveDocumentContext()$path))

# load required packages
library(tidyverse)
library(httr)
library(jsonlite)
library(janitor)

# don't use scientific notation
options(scipen=999)

#####
# setting up to download transactions data for grants and contracts in our data
contracts_file <- "data/africa_contracts_list.csv"
grants_file    <- "data/africa_grants_list.csv"

out_dir <- "data/deobligations"   # where batch folders will be created

base_url <- "https://api.usaspending.gov/api/v2/download/transactions/"
batch_size <- 500   
pause_between_checks <- 5
poll_timeout <- 60 * 60  # seconds

# helper functions 

# safe first-non-null utility
first_non_null <- function(...) {
  xs <- list(...)
  for (x in xs) {
    if (!is.null(x) && !(is.atomic(x) && length(x) == 0) && !(is.character(x) && x == "")) return(x)
  }
  NULL
}

# submit POST request for a batch of award_ids
request_bulk_download <- function(award_ids_batch) {
  body <- list(
    filters = list(award_ids = unname(as.list(award_ids_batch))),
    subawards = FALSE,
    download_type = "transactions",
    file_format = "csv"
  )
  j <- toJSON(body, auto_unbox = TRUE)
  resp <- POST(base_url,
               body = j,
               content_type_json(),
               accept_json())
  parsed <- tryCatch(content(resp, as = "parsed", simplifyVector = TRUE),
                     error = function(e) list(raw = content(resp, as = "text", encoding = "UTF-8")))
  raw_text <- content(resp, as = "text", encoding = "UTF-8")
  list(http_status = status_code(resp), parsed = parsed, raw = raw_text)
}

# poll a status_url until finished/failed (returns parsed JSON on success)
poll_until_ready <- function(status_url, check_interval = 5, timeout = poll_timeout) {
  start <- Sys.time()
  repeat {
    Sys.sleep(check_interval)
    s <- GET(status_url)
    s_parsed <- tryCatch(content(s, as = "parsed", simplifyVector = TRUE),
                         error = function(e) list(raw = content(s, as = "text", encoding = "UTF-8")))
    status_val <- tolower(first_non_null(s_parsed$status, s_parsed$state, s_parsed$job_status, s_parsed$job_state))
    if (!is.null(status_val) && status_val %in% c("finished", "complete")) return(s_parsed)
    if (!is.null(status_val) && status_val %in% c("failed", "error")) stop("Download job failed: ", jsonlite::toJSON(s_parsed, auto_unbox = TRUE))
    if (as.numeric(difftime(Sys.time(), start, units = "secs")) > timeout) stop("Timeout waiting for download to be ready.")
  }
}

# download a file_url, unzip to dest_dir, return vector of files extracted
download_and_unzip <- function(file_url, dest_dir) {
  dest_zip <- file.path(dest_dir, basename(file_url))
  # download
  tryCatch({
    download.file(file_url, dest_zip, mode = "wb", quiet = TRUE)
  }, error = function(e) stop("download.file failed: ", conditionMessage(e)))
  # unzip
  unzip(dest_zip, exdir = dest_dir)
  # list CSVs (include other files if present)
  list.files(dest_dir, full.names = TRUE, recursive = FALSE)
}

# robust extraction of status_url/file_url from parsed JSON
extract_status_url <- function(parsed) {
  first_non_null(parsed$status_url, parsed$statusUrl, parsed$download_status_url, parsed$download_status$download_status_url, parsed$download_request_status_url)
}
extract_file_url <- function(parsed) {
  first_non_null(parsed$file_url, parsed$fileUrl, parsed$fileURL, parsed$download_link, parsed$download_url)
}

####
# load data on our cancelled contracts and grants

# read CSVs (expecting a column containing award ids)
africa_contracts <- read_csv(contracts_file, show_col_types = FALSE)
africa_grants    <- read_csv(grants_file, show_col_types = FALSE)

# find award-id column in a data.frame (case-insensitive list of possibles)
find_award_col <- function(df) {
  possible <- c("award_id", "awardid", "AwardID", "award__id", "piid", "generated_internal_id", "generatedinternalid")
  nm <- names(df)
  found <- nm[tolower(nm) %in% possible]
  if (length(found) > 0) return(found[1])
  stop("No award-id column found in input file. Expected one of: ", paste(possible, collapse = ", "))
}

award_col_contracts <- find_award_col(africa_contracts)
award_col_grants    <- find_award_col(africa_grants)

award_ids <- c(africa_contracts[[award_col_contracts]], africa_grants[[award_col_grants]]) %>%
  as.character() %>%
  str_trim() %>%
  discard(~ is.na(.) || . == "") %>%
  unique()

message("Total unique award_ids to process: ", length(award_ids))

####
# main download function

process_downloads_only <- function(award_ids, batch_size = batch_size) {
  batches <- split(award_ids, ceiling(seq_along(award_ids) / batch_size))
  summary_rows <- list()
  
  for (i in seq_along(batches)) {
    batch <- batches[[i]]
    message(sprintf("[%s] Submitting batch %d/%d (n=%d)", Sys.time(), i, length(batches), length(batch)))
    batch_out_dir <- file.path(out_dir, paste0("batch_", i))
    dir.create(batch_out_dir, recursive = TRUE, showWarnings = FALSE)
    
    attempt <- tryCatch({
      post_resp <- request_bulk_download(batch)
      # if HTTP error
      if (post_resp$http_status >= 400) stop("POST returned HTTP ", post_resp$http_status, " — raw: ", substr(post_resp$raw, 1, 2000))
      
      parsed_post <- post_resp$parsed
      status_url <- extract_status_url(parsed_post)
      if (is.null(status_url)) stop("No status_url found in POST response. Response parsed: ", jsonlite::toJSON(parsed_post, auto_unbox = TRUE))
      
      message("Polling status_url: ", status_url)
      sjson <- poll_until_ready(status_url, check_interval = pause_between_checks)
      
      file_url <- extract_file_url(sjson)
      if (is.null(file_url)) stop("No file_url found in finished job JSON: ", jsonlite::toJSON(sjson, auto_unbox = TRUE))
      
      message("Downloading file_url: ", file_url)
      downloaded_files <- download_and_unzip(file_url, batch_out_dir)
      
      list(success = TRUE, status = "finished", file_url = file_url, files = downloaded_files, details = NULL)
    },
    error = function(e) {
      list(success = FALSE, status = "failed", file_url = NA_character_, files = character(0), details = conditionMessage(e))
    })
    
    # record summary row for this batch
    summary_rows[[i]] <- tibble(
      batch = i,
      n_award_ids = length(batch),
      batch_dir = batch_out_dir,
      status = attempt$status,
      file_url = ifelse(is.null(attempt$file_url), NA_character_, attempt$file_url),
      n_files_downloaded = length(attempt$files),
      downloaded_files = paste(basename(attempt$files), collapse = "; "),
      error = ifelse(is.null(attempt$details), NA_character_, attempt$details),
      timestamp = as.character(Sys.time())
    )
    
    # polite pause
    Sys.sleep(1)
  }
  
  bind_rows(summary_rows)
}

# download files
download_summary <- process_downloads_only(award_ids, batch_size = batch_size)


############
# load transactions from downloads

# directory where batches were unzipped
root <- "data/deobligations"

# function to read CSVs
read_csv_safe <- function(file) {
  tryCatch({
    message("Reading: ", file)
    df <- read_csv(file, guess_max = 200000, show_col_types = FALSE)
    df <- janitor::clean_names(df)
    df$source_file <- basename(file)
    df
  }, error = function(e){
    warning("Failed to read ", file, ": ", conditionMessage(e))
    NULL
  })
}

# helper: return typeof() map for a tibble
col_types_map <- function(df) {
  if (is.null(df)) return(tibble())
  tibble(name = names(df), type = map_chr(df, typeof))
}

# function to harmonize list-of-dfs
harmonize_and_bind <- function(dfs) {
  # drop NULLs
  dfs <- compact(dfs)
  if (length(dfs) == 0) return(tibble())
  
  # all column names across dfs
  all_cols <- dfs %>% map(names) %>% flatten_chr() %>% unique()
  
  # build a table of name -> set of types present
  types_by_col <- map_dfr(dfs, col_types_map, .id = "df_id") %>%
    group_by(name) %>%
    summarize(types = list(unique(type)), n_types = length(unique(type)), .groups = "drop")
  
  # columns with mixed types
  mixed_cols <- types_by_col %>% filter(n_types > 1) %>% pull(name)
  message("Columns with mixed types (will coerce to character): ", paste(mixed_cols, collapse = ", "))
  
  # coerce mixed cols to character in each df
  dfs_fixed <- map(dfs, function(df){
    for (col in intersect(names(df), mixed_cols)) {
      df[[col]] <- as.character(df[[col]])
    }
    # also ensure all columns in all_cols exist in df (fill with NA_character_)
    missing_cols <- setdiff(all_cols, names(df))
    if (length(missing_cols) > 0) df[missing_cols] <- NA_character_
    # reorder to consistent column ordering
    df <- df[all_cols]
    df
  })
  
  # bind rows
  out <- bind_rows(dfs_fixed)
  
  # convert back some known numeric/date columns
  if ("federal_action_obligation" %in% names(out)) {
    out <- mutate(out, federal_action_obligation = as.numeric(federal_action_obligation))
  }
  if ("modification_number" %in% names(out)) {
    out <- mutate(out, modification_number = as.character(modification_number)) # often non-numeric
  }
  if ("action_date" %in% names(out)) {
    out <- mutate(out, action_date = as.character(action_date)) # keep as character; parse separately if desired
  }
  
  out
}

######
# apply to Contracts_Prime files
contracts_files <- list.files("data/deobligations", pattern = "^Contracts_Prime.*\\.csv$", full.names = TRUE, recursive = TRUE)
contracts_dfs <- map(contracts_files, read_csv_safe)

africa_contracts_transactions <- harmonize_and_bind(contracts_dfs)
message("Contracts combined rows: ", nrow(africa_contracts_transactions))

# process for deobligations
contracts_deobs_edit <- africa_contracts_transactions %>%
  # ensure the transaction value column is numeric
  mutate(federal_action_obligation = as.numeric(federal_action_obligation),
         action_date = as.character(action_date)) %>%
  filter(!is.na(federal_action_obligation) & federal_action_obligation < 0) %>% # this will get negative transactions, i.e. deolbligations
  select(
    award_id = award_id_piid,
    action_date,
    modification_number,
    recipient_name,  
    primary_place_of_performance_country_code,
    primary_place_of_performance_country_name,
    federal_action_obligation,
    transaction_description
  ) %>%
  arrange(award_id, desc(action_date)) %>%
  mutate(action_date = ymd(action_date)) %>%
  filter(action_date >= "2025-02-20" & action_date < "2025-09-30") %>%  # only deobligations after Trump's inauguration and before Sep 30, 2025
  group_by(
    award_id,
    recipient_name,  
    primary_place_of_performance_country_code,
    primary_place_of_performance_country_name,
  ) %>%
  summarize(federal_action_obligation = sum(federal_action_obligation, na.rm = TRUE)) # sums if there is more than one deobligation for any contract
  
# write to CSV
write_csv(contracts_deobs_edit, "data/contracts_deobs_edit.csv", na = "")


######
# apply to  Assistance_Prime files
grants_files <- list.files("data/deobligations", pattern = "^Assistance_Prime.*\\.csv$", full.names = TRUE, recursive = TRUE)
grants_dfs <- map(grants_files, read_csv_safe)

africa_grants_transactions <- harmonize_and_bind(grants_dfs)
message("Grants combined rows: ", nrow(africa_grants_transactions))

grants_deobs_edit <- africa_grants_transactions %>%
  # ensure the transaction value column is numeric 
  mutate(federal_action_obligation = as.numeric(federal_action_obligation),
         action_date = as.character(action_date)) %>%
  filter(!is.na(federal_action_obligation) & federal_action_obligation < 0) %>% # this will get negative transactions, i.e. deobligations
  select(
    award_id = award_id_fain,
    action_date,
    modification_number,
    recipient_name,  
    primary_place_of_performance_country_code,
    primary_place_of_performance_country_name,
    federal_action_obligation,
    transaction_description
  ) %>%
  arrange(award_id, desc(action_date)) %>%
  mutate(action_date = ymd(action_date)) %>%
  filter(action_date >= "2025-02-20" & action_date < "2025-09-30") %>% # only deobligations after Trump's inauguration and before Sep 30, 2025
  group_by(
    award_id,
    recipient_name,  
    primary_place_of_performance_country_code,
    primary_place_of_performance_country_name,
  ) %>%
  summarize(federal_action_obligation = sum(federal_action_obligation, na.rm = TRUE)) # sums if there is more than one deobligation for any grant

write_csv(grants_deobs_edit, "data/grants_deobs_edit.csv", na = "")

# # some summary analysis of sums involved
# 
# grants_deobs_edit %>%
#   summarize(total = sum(federal_action_obligation))
# # -66,693,173
# 
# contracts_deobs_edit %>%
#   summarize(total = sum(federal_action_obligation))
# # -32,487,987