# Create Accounting on Snowflake.r

# -------------------------------------------------------------------------------- # 
#                                    EXECUTION                                     # 
# -------------------------------------------------------------------------------- # 

system("touch ~/git/orch/log/DeNormalizing/starting_creating_accounting.txt")


## PULL GIT CHANGES
try({
  setGitBranchToSystem()
  .g()
}, silent=FALSE)

## Set Science
setScience(proj="DeNormalizing", subProj="Accounting", quiet=TRUE, load=FALSE, create=FALSE, subl=FALSE)
snowflake_inuse <- TRUE
setPrompt(full=TRUE)
options(sfGetCon_quiet_connex = FALSE)
options(snowflake_msg_sf=FALSE)

keep_the_old_table = FALSE

## SETUP LOG
try({
  logname <- sprintf("BI_%s_Creation", "Accounting")
  previousLogFile <- getLastSinkFileFromDisk(logname, dir=logDir)
  sinkFile <- newLogFile(name=logname)
  sinkOn(file=sinkFile)
})


## CHECK CURRENT MAX DATES
{
  setSnowflake(wh="cron_jobs_small")

  qry.max_dates.production <- "SELECT storeid, count(*) as rws, sum(gross) as gross, max(accountingperiodid) as accountingperiodid FROM production.fact_sales WHERE storeid in (1, 187, 286) GROUP BY storeid"
  DT.max_dates.production <- sfQry(qry.max_dates.production, wh="cron_jobs_small")
  addDateCols.periodid_(DT.max_dates.production, periodCols="accountingperiodid", newCol.nms="accounting_month", dropPeriodCols=TRUE)
  setkey(DT.max_dates.production)

  qry.max_dates_bi <- "SELECT storeid, count(*) as rws, sum(GROSS_REVENUE_USD) as gross, max(accounting_month) as accounting_month FROM bi.accounting WHERE storeid in (1, 187, 286) GROUP BY storeid"
  DT.max_dates.bi <- sfQry(qry.max_dates_bi, wh="cron_jobs_small")
  setkey(DT.max_dates.bi)
}

NO_UPDATE_NEEDED_FOR_BI <- nrow(DT.max_dates.production) == nrow(DT.max_dates.bi) &&
      all({(is.na(DT.max_dates.production) & is.na(DT.max_dates.bi)) | 
      (DT.max_dates.production == DT.max_dates.bi)})

## IF THESE ARE ALL THE SAME -- NOTHING TO DO 
if (NO_UPDATE_NEEDED_FOR_BI) {

  msg.no_update <- timeStamp("No updates neceessary for BI.ACCOUNTING at", sep=" ", human=TRUE)
  try(notifyAndEmail(subj="Accounting BI ETL -- Nothing to update", msg=msg.no_update))
  catnn(msg.no_update)
  
} else {
## OTHER WISE CONTINUE AS NORMAL  

  {
    catheader("The BI table is different from the PRODUCTION table -- proceeding with update")
    catn("\nPRODUCTION: ---------")
    print(DT.max_dates.production)
    catn("\nBI: ---------")
    print(DT.max_dates.bi)
    catn("=======================================")
  }

  ## NOTIFY THAT IT IS RUNNING
  try(notifyAndEmail(subj="Starting Accounting BI ETL", msg=timeStamp("Starting Accounting BI ETL at", sep=" ", human=TRUE)))


  ## WRAP EXECUTION IN try()
  try ({

  ## DEFAULT TO SNOWFLAKE FOR NOW
  assignIfNotExist(wh, getSnowflakeWH())
  assignIfNotExist(dbname, getSnowflakeDB())

  # if (wh %in% c("", "cron_jobs_small")) wh <- "BI_ETL_JOBS_STANDARD"
  wh <- "BI_ETL_JOBS_STANDARD"
  dbname <- "prod"
  schema <- "bi"
  tbl_dest <- "Accounting"
  tbl_tmp <- sprintf("%s_next", tbl_dest)

  setSnowflake(wh=wh, dbname=dbname, start=TRUE)
  sfWaitForWarehouse(wh=wh, N.seconds=30, max_iterations=100, verbose.waiting=TRUE)

  ## Check if tbl_tmp exists. If so, FAIL.
  if (qTableExists(dbname=dbname, schema=schema, tbl=tbl_tmp, snowflake_inuse=TRUE)) {
    ## move the logfile from zArchive
    accidentallyZarchivedLogFile <- as.path(dirname(previousLogFile), "zArchive", basename(previousLogFile))
    if (file.exists(accidentallyZarchivedLogFile) && !file.exists(previousLogFile)) {
      message("About to abort... moving previous log file back\n\t  to '", previousLogFile, "'")
      print(file.rename(accidentallyZarchivedLogFile, previousLogFile))
    }
    stop("Table ", dbschematbl(dbname=dbname, schema=schema, tbl=tbl_tmp), " already exists.\nThis might mean that the previous run ended with error or is still running (or some other issue).\nAborting.\n")
  }

  ## TODO, change the Analytics folder then update this to use tbl_dest
  folder <- src.p("BI_tables", "Accounting")
  comment <- sprintf("Table last updated %s EST", timeStamp(human=TRUE, tz="America/New_York"))

  files <- extractFilesFromFolder(folder, ext="sql", full=TRUE)
  files %<>% filegroup_adjust_for_snowflake(snowflake_inuse=snowflake_inuse)

  ## REMOVE the CREATE files, since we are using SELECT to create
  files %<>% {.[!grepl("^01_A. create_", names(.))]}
  verboseMsg(verbose, "Will execute the following files: ", pasteQand(names(files)), minw=120)

  for (file in files) {
    try(cat("\n\n ------- EXECUTING SQL FILE:  ", path.unexpand(file), "-------- \n"))
    qry_whole <- sqlFileToQry(file, clear=TRUE)
    qry_splat <- strsplit(qry_whole, ";")[[1]] %>% trim %>% removeNullsAndBlanksFromList
    ## Use for-loop instead of sapply, as easier to recover from in event of crash
    # sapply(qry_splat, sfQry, wh=wh, dbname=dbname)
    for (q_splat in qry_splat)
      ## Wrapping this in TRY;  I dont like doing so since that ultimately may allow for impropper data
      try({
        sfQry(q_splat, wh=wh, dbname=dbname)
      })
  }


  ## After queries complete, replace the existing table with the temp table we were populating
  swap_in_new_table(tbl_old=tbl_dest, tbl_new=tbl_tmp, schema=schema, zArchive.old=TRUE, wh=wh, dbname=dbname)
  sfQry(sprintf("ALTER TABLE %s SET COMMENT = '%s'", dbschematbl(dbname=dbname, schema=schema, tbl=tbl_dest), comment))

  ## Show tables and dates
  print(sfShowTables(schema=schema), nrow=200)
  print(qMaxDate(schema=schema, tbl=tbl_dest, byCols=c("store_name", "storeid"), verbose=TRUE))
  })

  ## qUpdatePerms() was for redshift
  ## source the following file for snowflake to update the permissions on the newly created tables
  catheader("Updating Permissions -- after Track Level", endl=0)
  try({source("~/git/orch/src/DeNormalizing/snowflake_permissioning_update.r")})

  # ## ----------- RELEASE LEVEL ----------------------- ##
  #   ## Create Release Level
  #   try(  create_release_level_table(tbl.trk="accounting", tbl.rel="accounting_release_level", schema=schema, dbname=dbname, wh=wh, zArchive.old=keep_the_old_table, verbose=TRUE) )
  #   ## Run Confirmation Script to confirm that release and track level have the same figures
  #   try({
  #     if (confirm_release_level_table(tbl.trk="accounting", tbl.rel="accounting_release_level", schema=schema, dbname=dbname, wh=wh, zArchive.old=keep_the_old_table, verbose=TRUE))
  #           catn("CONFIRMED:  accounting_release_level is same as accounting")
  #     else
  #           catn("ERROR:  accounting_release_level is DIFFERENT from accounting")
  #   })
  #   catheader("Updating Permissions -- after Release Level", endl=0)
  #   try({source("~/git/orch/src/DeNormalizing/snowflake_permissioning_update.r")})
  # ## ----------- RELEASE LEVEL ----------------------- ##


  catn("Turning off the BI ETL JOBS STANDARD Warehouse")
  try({  sfWarehouseOff("BI_ETL_JOBS_STANDARD", verbose=TRUE)  })

  try(notifyAndEmail(subj="Completed Accounting (Release & Track Level) BI ETL", msg=timeStamp("Completed Accounting (Release & Track Level) BI ETL at", sep=" ", human=TRUE)))
  catheader("Script Completed Succesfully;  Exiting")
} ## // end of else-statement where script will not run if nothing to update


  catn("changing warehouse to CRON JOB SMALL  For log line 163")
  setSnowflake(wh="cron_jobs_small")

sinkOff()
catn("Today is ", timeStamp(human=TRUE))