Ingest_Flex_Freeze.Rmd

---
title: "Ingest_Flex_Freeze"
author: "Arush Mohan"
date: "`r Sys.Date()`"
output: html_document
params:
  report_date: 20240901
  freeze_window: 7
---


```{r setup, include=FALSE}
knitr::opts_chunk$set(echo = TRUE)
source('src/functions.R')

```



## Defining folder directories

```{r}

# Extract the date from the main folder name
main_folder_date <- as.Date(ymd(params$report_date), format = "%Y-%m-%d")

# Calculate the date for freeze files
freeze_date <- ymd(params$report_date) - days(params$freeze_window)

# Define folders
main_folder <- paste0('data/', main_folder_date)
dir_path_flex <- file.path(main_folder, "Flex")
dir_path_freeze <- file.path(main_folder, "Freeze")

```


## Listing the files in the directory

```{r}

# List all Excel files in the directories
file_list_flex <- list.files(path = dir_path_flex, pattern = "*.xlsx", full.names = TRUE)
file_list_freeze <- list.files(path = dir_path_freeze, pattern = "*.xlsx", full.names = TRUE)

```



## Combining file lists and differentiating flex/freeze

```{r}

# Combine both file lists and add a datatype column
file_list <- tibble(
  file_path = c(file_list_flex, file_list_freeze),
  datatype = c(rep("flex", length(file_list_flex)), rep("freeze", length(file_list_freeze))),
  reportdate = c(rep(main_folder_date, length(file_list_flex)), rep(freeze_date, length(file_list_freeze)))
) %>%
  mutate(provider_name = text_normalisation(str_extract(basename(file_path), "^[^_]+"))) 

```



## Defining columns and sheet parameters

``` {r}
# Read the column names
indicator_list <- read_csv("input/Provider_indicator_list.csv")
provider_list <- file_list %>%
  select(provider_name) %>%
  distinct()

# Build full list
Indicators <- expand_grid(Provider = provider_list$provider_name, Indicator = indicator_list$Indicator)

# Modifications
## read
Indicator_edits <- read_csv('input/Provider_inconsistency_adjustment.csv')
  
## remove missing
Missing_indicators <- Indicator_edits %>%
  filter(Issue == 'Missing')

Indicators <- Indicators %>%
  filter(!paste0(Provider, '_', Indicator) %in% paste0(Missing_indicators$Provider, '_', Missing_indicators$Indicator))

## Add new - NEEDS TESTING
New_indicators <- Indicator_edits %>%
  filter(Issue == 'Additional') %>%
  select(Provider, Indicator)

Indicators <- Indicators %>%
  bind_rows(New_indicators)

# Read combined sheet parameters and sheet names from CSV
sheet_parameters <- read_csv("input/Sheet_parameters.csv") %>%
  mutate(Provider  = text_normalisation(Provider))

```

## Extracting the data

```{r}
# Read and combine all files into one data frame
combined_data <- file_list %>%
  mutate(data = map(.x = file_path, ~ingest_raw_data_file(file_path = .x))) %>% # function in function.R
  #filter(map_lgl(data, ~ !is.null(.x))) %>% # I don't think you need this
  unnest(cols = c(data)) %>%
  select(-file_path)
  

```
## Pivoting data to wide format for secondary use

```{r}

# Pivoting data to wide format
wide_data <- combined_data %>%
  pivot_wider(
    names_from = Indicator,
    values_from = Activity
  )

# Adding week number based on calendar year
wide_data <- wide_data %>%
  mutate(
    WeekNumber = paste("Week", week(reportdate))
  )

wide_data$provider_name <- toupper(wide_data$provider_name) #capitalising all the provider names

wide_data <- wide_data %>%
  mutate(provider_name = ifelse(provider_name == "INHEALTH", "InHealth", provider_name))

```


## Processing for upload
```{r}
# Replace spaces, hyphens, or multiple underscores with a single underscore. Make all column names lowercase
wide_data <- clean_column_names(wide_data)


```




## Creating Intermediate Excel Files for QA
```{r}

# Write combined dataset to a new Excel file
write_csv(wide_data, paste0('output/', "wide_dataset_combined_", ymd(params$report_date), ".csv"))


```



## Upload to sandpit sections - override for freeze, simple upload for flex
```{r}

# Connecting to the Sandpit

con <- dbConnect(odbc::odbc(), 
                 dsn = "SANDPIT",
                 database = "Data_Lab_NCL",
                 TrustedConnection = TRUE)
```





###  Checking for duplicates and uploading to Sandpit - Wide Data

```{r}

# Begin a transaction
dbBegin(con)

# Error handling
tryCatch({

  # Separate the wide data into flex and freeze datasets
  wide_flex_data <- wide_data %>% filter(datatype == "flex")
  wide_freeze_data <- wide_data %>% filter(datatype == "freeze")

  # Check for existing records in freeze data
  if (nrow(wide_freeze_data) > 0) {
    unique_freeze_entries <- wide_freeze_data %>%
      select(provider_name, reportdate) %>%
      distinct()

    # SQL IN clause strings for freeze data
    freeze_provider_names <- paste0("'", unique_freeze_entries$provider_name, "'", collapse = ", ")
    freeze_report_dates <- paste0("'", unique_freeze_entries$reportdate, "'", collapse = ", ")

    # Query to check for existing freeze data
    freeze_query <- sprintf(
      "SELECT provider_name, reportdate FROM [Data_Lab_NCL].[dbo].[diagnostics_combined_data_2] 
       WHERE datatype = 'freeze' AND provider_name IN (%s) AND reportdate IN (%s)",
      freeze_provider_names, freeze_report_dates
    )

    existing_freeze_data <- dbGetQuery(con, freeze_query)

    # If existing freeze data is found raise an error
    if (nrow(existing_freeze_data) > 0) {
      stop(sprintf(
        "Duplicate freeze data found for provider_name(s): %s on reportdate(s): %s.",
        paste(existing_freeze_data$provider_name, collapse = ", "),
        paste(existing_freeze_data$reportdate, collapse = ", ")
      ))
    }
  }

  # Check for existing records in flex data
  if (nrow(wide_flex_data) > 0) {
    unique_flex_entries <- wide_flex_data %>%
      select(provider_name, reportdate) %>%
      distinct()

    # SQL IN clause strings for flex data
    flex_provider_names <- paste0("'", unique_flex_entries$provider_name, "'", collapse = ", ")
    flex_report_dates <- paste0("'", unique_flex_entries$reportdate, "'", collapse = ", ")

    # Query to check for existing flex data
    flex_query <- sprintf(
      "SELECT provider_name, reportdate FROM [Data_Lab_NCL].[dbo].[diagnostics_combined_data_2] 
       WHERE datatype = 'flex' AND provider_name IN (%s) AND reportdate IN (%s)",
      flex_provider_names, flex_report_dates
    )

    existing_flex_data <- dbGetQuery(con, flex_query)

    # If existing flex data is found raise an error
    if (nrow(existing_flex_data) > 0) {
      stop(sprintf(
        "Duplicate flex data found for provider_name(s): %s on reportdate(s): %s.",
        paste(existing_flex_data$provider_name, collapse = ", "),
        paste(existing_flex_data$reportdate, collapse = ", ")
      ))
    }
  }

  # Delete existing flex data
  if (nrow(wide_flex_data) > 0) {
    delete_statement <- sprintf(
      "DELETE FROM [Data_Lab_NCL].[dbo].[diagnostics_combined_data_2] 
       WHERE datatype = 'flex' 
       AND provider_name IN (%s)
       AND reportdate IN (%s)",
      flex_provider_names, flex_report_dates
    )
    dbExecute(con, delete_statement)
  }

  # Insert wide-format freeze data
  if (nrow(wide_freeze_data) > 0) {
    dbAppendTable(con, Id(schema = "dbo", table = "diagnostics_combined_data_2"), wide_freeze_data)
  }

  # Insert wide-format flex data
  if (nrow(wide_flex_data) > 0) {
    dbAppendTable(con, Id(schema = "dbo", table = "diagnostics_combined_data_2"), wide_flex_data)
  }

  # Commit the transaction
  dbCommit(con)

}, error = function(e) {
  # Rollback the transaction in case of error
  dbRollback(con)
  stop("Transaction failed: ", e$message)
})


```



## Disconnecting from the Sandpit
```{r}

dbDisconnect(con)

```